Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_lint(self, m_boto):
cwd = os.getcwd()
base_path = "/tmp/lint_test/"
mkdir(base_path)
try:
for test_case in test_cases:
config_path = Path(build_test_case(base_path, test_case)).resolve()
project_root = config_path.parent.parent
config = Config.create(
project_config_path=config_path, project_root=project_root
)
templates = config.get_templates(project_root=project_root)
lint = Lint(config=config, templates=templates)
self.assertEqual(
test_case["expected_lints"], flatten_rule(lint.lints[0])
)
finally:
shutil.rmtree(base_path)
os.chdir(cwd)
pass
def create_ephemeral_template_object(self, template_type="generic"):
test_proj = (
Path(__file__).parent / f"./data/update_ami/{template_type}"
).resolve()
c = Config.create(
project_config_path=test_proj / ".taskcat.yml", project_root=test_proj
)
templates = c.get_templates(project_root=test_proj)
return templates
def test_config(self):
base_path = "./" if os.getcwd().endswith("/tests") else "./tests/"
base_path = Path(base_path + "data/config_inheritance").resolve()
config = Config.create(
args={"project": {"build_submodules": False}},
global_config_path=base_path / ".taskcat_global.yml",
project_config_path=base_path / "./.taskcat.yml",
overrides_path=base_path / "./.taskcat_overrides.yml",
env_vars={"TASKCAT_PROJECT_PACKAGE_LAMBDA": "False"},
)
expected = {
"general": {
"parameters": {
"GlobalVar": "set_in_global",
"OverridenVar": "set_in_global",
},
"s3_bucket": "set-in-global",
},
"project": {
base_path = Path(root_path + "data/regional_client_and_bucket").resolve()
template = base_path / "templates/debug-yaml.template"
# valid config
template_dict = Config._dict_from_template(template)
self.assertEqual(True, isinstance(template_dict, dict))
# invalid path
with self.assertRaises(TaskCatException):
Config._dict_from_template(base_path / "invalid-path")
# cannot create template object
with mock.patch("taskcat._config.Template") as mock_template:
exc = ValueError("fail")
mock_template.side_effect = exc
with self.assertRaises(ValueError) as e:
Config._dict_from_template(template)
self.assertEqual(exc, e.exception)
mock_log.warning.assert_called_once()
base_path = Path(root_path + "data/standalone_template").resolve()
# metadata in taskcat, but no taskcat key
template = base_path / "test.template_no_tc_meta.yaml"
template_dict = Config._dict_from_template(template)
self.assertEqual(True, isinstance(template_dict, dict))
self.assertEqual(
True, template_dict["project"]["template"].endswith("no_tc_meta.yaml")
)
self.assertEqual({}, template_dict["tests"]["default"]["parameters"])
# empty dict taskcat metadata
template = base_path / "test.template_tc_empty_meta.yaml"
def test_nested_submodules(self):
tmp = Path(mkdtemp())
test_proj = (
Path(__file__).parent / "./data/lambda_build_with_submodules"
).resolve()
copytree(test_proj, tmp / "test")
c = Config.create(
project_config_path=tmp / "test" / ".taskcat.yml",
project_root=(tmp / "test").resolve(),
args={
"project": {
"lambda_zip_path": "lambda_functions/packages",
"lambda_source_path": "lambda_functions/source",
}
},
)
LambdaBuild(c, project_root=(tmp / "test").resolve())
path = tmp / "test"
zip_suffix = Path("lambda_functions") / "packages" / "TestFunc" / "lambda.zip"
self.assertEqual((path / "lambda_functions" / "packages").is_dir(), True)
self.assertEqual((path / zip_suffix).is_file(), True)
path = path / "submodules" / "SomeSub"
self.assertEqual((path / "lambda_functions" / "packages").is_dir(), True)
package_type = "github"
package = f"aws-quickstart/quickstart-{package}"
if package_type == "github":
if package.startswith("https://") or package.startswith("git@"):
url = package
org, repo = (
package.replace(".git", "").replace(":", "/").split("/")[-2:]
)
else:
org, repo = package.split("/")
url = f"https://github.com/{org}/{repo}.git"
path = Deploy.PKG_CACHE_PATH / org / repo
LOG.info(f"fetching git repo {url}")
self._git_clone(url, path)
self._recurse_submodules(path, url)
config = Config.create(
args={"project": {"regions": [region]}},
project_config_path=(path / ".taskcat.yml"),
project_root=path,
)
# only use one region
for test_name in config.config.tests:
config.config.tests[test_name].regions = config.config.project.regions
# if there's no test called default, take the 1st in the list
if "default" not in config.config.tests:
config.config.tests["default"] = config.config.tests[
list(config.config.tests.keys())[0]
]
# until install offers a way to run different "plans" we only need one test
for test_name in list(config.config.tests.keys()):
if test_name != "default":
del config.config.tests[test_name]
self,
project_root: str = "./",
source_folder: str = "lambda_functions/source",
zip_folder: str = "lambda_functions/packages",
config_file: str = ".taskcat.yml",
):
"""
:param project_root: base path for project
:param source_folder: folder containing the lambda source files, relative to the
project_root
:param zip_folder: folder to output zip files, relative to the project root
:param config_file: path to taskcat project config file
"""
project_root_path: Path = Path(project_root).expanduser().resolve()
project_config: Path = project_root_path / config_file
config = Config.create(
project_config_path=project_config,
project_root=project_root_path,
args={
"project": {
"lambda_zip_path": zip_folder,
"lambda_source_path": source_folder,
}
},
)
LambdaBuild(config, project_root_path)
if project_root == "./":
_project_root = Path(os.getcwd())
else:
_project_root = Path(project_root)
_c = Config.create(project_config_path=Path(_project_root / ".taskcat.yml"))
_boto3cache = Boto3Cache()
# Stripping out any test-specific regions/auth.
config_dict = _c.config.to_dict()
for _, test_config in config_dict["tests"].items():
if test_config.get("auth", None):
del test_config["auth"]
if test_config.get("regions", None):
del test_config["regions"]
new_config = Config.create(
project_config_path=Path(_project_root / ".taskcat.yml"), args=config_dict
)
# Fetching the region objects.
regions = new_config.get_regions(boto3_cache=_boto3cache)
region_key = list(regions.keys())[0]
unprocessed_templates = new_config.get_templates(
project_root=Path(_project_root)
).values()
finalized_templates = neglect_submodule_templates(
project_root=Path(_project_root), template_list=unprocessed_templates
)
amiupdater = AMIUpdater(
template_list=finalized_templates,
def __init__(
self,
input_file: str = ".taskcat.yml",
project_root: str = "./",
strict: bool = False,
):
"""
:param input_file: path to project config or CloudFormation template
:param project_root: base path for project
:param strict: fail on lint warnings as well as errors
"""
project_root_path: Path = Path(project_root).expanduser().resolve()
input_file_path: Path = project_root_path / input_file
config = Config.create(
project_root=project_root_path, project_config_path=input_file_path
)
templates = config.get_templates(project_root_path)
lint = TaskCatLint(config, templates, strict)
errors = lint.lints[1]
lint.output_results()
if errors or not lint.passed:
raise TaskCatException("Lint failed with errors")
def __init__(self, project_root: str = "./"):
"""
:param project_root: base path for project
"""
if project_root == "./":
_project_root = Path(os.getcwd())
else:
_project_root = Path(project_root)
_c = Config.create(project_config_path=Path(_project_root / ".taskcat.yml"))
_boto3cache = Boto3Cache()
# Stripping out any test-specific regions/auth.
config_dict = _c.config.to_dict()
for _, test_config in config_dict["tests"].items():
if test_config.get("auth", None):
del test_config["auth"]
if test_config.get("regions", None):
del test_config["regions"]
new_config = Config.create(
project_config_path=Path(_project_root / ".taskcat.yml"), args=config_dict
)
# Fetching the region objects.
regions = new_config.get_regions(boto3_cache=_boto3cache)
region_key = list(regions.keys())[0]