Module taskcat.testing
None
None
View Source
from ._cfn_test import CFNTest # noqa: F401
from ._lint_test import LintTest # noqa: F401
from ._unit_test import UnitTest # noqa: F401
__all__ = ["CFNTest"]
Sub-modules
Classes
CFNTest
class CFNTest(
config: taskcat._config.Config,
printer: Union[taskcat._tui.TerminalPrinter, NoneType] = None,
test_names: str = 'ALL',
regions: str = 'ALL',
skip_upload: bool = False,
lint_disable: bool = False,
no_delete: bool = False,
keep_failed: bool = False,
dont_wait_for_delete: bool = True,
_extra_tags: list = None
)
View Source
class CFNTest(BaseTest): # pylint: disable=too-many-instance-attributes
"""
Tests Cloudformation template by making sure the stack can properly deploy
in the specified regions.
"""
def __init__(
self,
config: Config,
printer: Union[TerminalPrinter, None] = None,
test_names: str = "ALL",
regions: str = "ALL",
skip_upload: bool = False,
lint_disable: bool = False,
no_delete: bool = False,
keep_failed: bool = False,
dont_wait_for_delete: bool = True,
_extra_tags: list = None,
):
"""The constructor creates a test from the given Config object.
Args:
config (Config): A pre-configured Taskcat Config instance.
printer (Union[TerminalPrinter, None], optional): A printer object that will handle Test output. Defaults to TerminalPrinter.
test_names (str, optional): A comma separated list of tests to run. Defaults to "ALL".
regions (str, optional): A comma separated list of regions to test in. Defaults to "ALL".
skip_upload (bool, optional): Use templates in an existing cloudformation bucket. Defaults to False.
lint_disable (bool, optional): Disable linting with cfn-lint. Defaults to False.
no_delete (bool, optional): Don't delete stacks after test is complete. Defaults to False.
keep_failed (bool, optional): Don't delete failed stacks. Defaults to False.
dont_wait_for_delete (bool, optional): Exits immediately after calling stack_delete. Defaults to True.
""" # noqa: B950
super().__init__(config)
self.test_definition: Stacker
self.test_names = test_names
self.regions = regions
self.skip_upload = skip_upload
self.lint_disable = lint_disable
self.no_delete = no_delete
self.keep_failed = keep_failed
self.dont_wait_for_delete = dont_wait_for_delete
self._extra_tags = _extra_tags if _extra_tags else []
if printer is None:
self.printer = TerminalPrinter(minimalist=True)
else:
self.printer = printer
def run(self) -> None:
"""Deploys the required Test resources in AWS.
Raises:
TaskCatException: If skip_upload is set without specifying s3_bucket in config.
TaskCatException: If linting fails with errors.
"""
_trim_regions(self.regions, self.config)
_trim_tests(self.test_names, self.config)
boto3_cache = Boto3Cache()
templates = self.config.get_templates()
if self.skip_upload and not self.config.config.project.s3_bucket:
raise TaskCatException(
"cannot skip_buckets without specifying s3_bucket in config"
)
buckets = self.config.get_buckets(boto3_cache)
if not self.skip_upload:
# 1. lint
if not self.lint_disable:
lint = TaskCatLint(self.config, templates)
errors = lint.lints[1]
lint.output_results()
if errors or not lint.passed:
raise TaskCatException("Lint failed with errors")
# 2. build lambdas
if self.config.config.project.package_lambda:
LambdaBuild(self.config, self.config.project_root)
# 3. s3 sync
stage_in_s3(
buckets, self.config.config.project.name, self.config.project_root, []
)
regions = self.config.get_regions(boto3_cache)
parameters = self.config.get_rendered_parameters(buckets, regions, templates)
tests = self.config.get_tests(templates, regions, buckets, parameters)
# pre-hooks
execute_hooks("prehooks", self.config, tests, parameters)
self.test_definition = Stacker(
self.config.config.project.name,
tests,
shorten_stack_name=self.config.config.project.shorten_stack_name,
tags=self._extra_tags,
)
self.test_definition.create_stacks()
# post-hooks
# TODO: pass in outputs, once there is a standard interface for a test_definition
execute_hooks("posthooks", self.config, tests, parameters)
self.printer.report_test_progress(stacker=self.test_definition)
self.passed = True
self.result = self.test_definition.stacks
def clean_up(self) -> None: # noqa: C901
"""Deletes the Test related resources in AWS.
Raises:
TaskCatException: If one or more stacks failed to create.
"""
if not hasattr(self, "test_definition"):
LOG.warning("No stacks were created... skipping cleanup.")
return
status = self.test_definition.status()
# Delete Stacks
if self.no_delete:
LOG.info("Skipping delete due to cli argument")
elif self.keep_failed:
if len(status["COMPLETE"]) > 0:
LOG.info("deleting successful stacks")
self.test_definition.delete_stacks({"status": "CREATE_COMPLETE"})
else:
self.test_definition.delete_stacks()
if not self.dont_wait_for_delete:
self.printer.report_test_progress(stacker=self.test_definition)
# TODO: summarise stack statusses (did they complete/delete ok) and print any
# error events
# Delete Templates and Buckets
buckets = self.config.get_buckets()
if not self.no_delete or (
self.keep_failed is True and len(status["FAILED"]) == 0
):
deleted: ListType[str] = []
for test in buckets.values():
for bucket in test.values():
if (bucket.name not in deleted) and not bucket.regional_buckets:
bucket.delete(delete_objects=True)
deleted.append(bucket.name)
# 9. raise if something failed
# - grabbing the status again to ensure everything deleted OK.
status = self.test_definition.status()
if len(status["FAILED"]) > 0:
raise TaskCatException(
f'One or more stacks failed to create: {status["FAILED"]}'
)
def report(
self,
output_directory: str = "./taskcat_outputs",
):
"""Generates a report of the status of Cloudformation stacks.
Args:
output_directory (str, optional): The directory to save the report in. Defaults to "./taskcat_outputs".
""" # noqa: B950
report_path = Path(output_directory).resolve()
report_path.mkdir(exist_ok=True)
cfn_logs = _CfnLogTools()
cfn_logs.createcfnlogs(self.test_definition, report_path)
ReportBuilder(
self.test_definition, report_path / "index.html"
).generate_report()
Ancestors (in MRO)
- taskcat.testing.base_test.BaseTest
- taskcat.testing._abstract_test.Test
- abc.ABC
Static methods
from_dict
def from_dict(
input_config: dict,
project_root: str = './',
regions: str = 'ALL',
enable_sig_v2: bool = False
) -> ~T
Creates a Test from a Taskcat configuration in dictionary form.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| input_config | dict | A Taskcat configuration in the form of a dict. | None |
| project_root | str | The path to the directory with your template and config file. Defaults to "./". | "./" |
| regions | str | A comma separated list of regions to test in. Defaults to "ALL". | "ALL" |
| enable_sig_v2 | bool | Enable legacy sigv2 requests for auto-created buckets. Defaults to False. | False |
Returns:
| Type | Description |
|---|---|
| T | Returns a Test instance. |
View Source
@classmethod
def from_dict(
cls: Type[T],
input_config: dict,
project_root: str = "./",
regions: str = "ALL",
enable_sig_v2: bool = False,
) -> T:
"""Creates a Test from a Taskcat configuration in dictionary form.
Args:
input_config (dict): A Taskcat configuration in the form of a dict.
project_root (str, optional): The path to the directory with your template and config file. Defaults to "./".
regions (str, optional): A comma separated list of regions to test in. Defaults to "ALL".
enable_sig_v2 (bool, optional): Enable legacy sigv2 requests for auto-created buckets. Defaults to False.
Returns:
T: Returns a Test instance.
""" # noqa: B950
project_root_path: Path = Path(project_root).expanduser().resolve()
# pylint: disable=too-many-arguments
args = _build_args(enable_sig_v2, regions, GLOBAL_ARGS.profile)
sources = [
{"source": "Manual", "config": input_config},
{"source": "CliArgument", "config": args},
]
config = Config(
uid=uuid.uuid4(), project_root=project_root_path, sources=sources
)
return cls(config)
from_file
def from_file(
project_root: str = './',
input_file: str = './.taskcat.yml',
regions: str = 'ALL',
enable_sig_v2: bool = False
) -> ~T
Creates a Test from a Taskcat config file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| project_root | str | The path to the directory with your template and config file. Defaults to "./". | "./" |
| input_file | str | The name of the Taskcat confile file. Defaults to "./.taskcat.yml". | "./.taskcat.yml" |
| regions | str | A comma separated list of regions to test in. Defaults to "ALL". | "ALL" |
| enable_sig_v2 | bool | Enable legacy sigv2 requests for auto-created buckets. Defaults to False. | False |
Returns:
| Type | Description |
|---|---|
| T | Returns a Test instance. |
View Source
@classmethod
def from_file(
cls: Type[T],
project_root: str = "./",
input_file: str = "./.taskcat.yml",
regions: str = "ALL",
enable_sig_v2: bool = False,
) -> T:
"""Creates a Test from a Taskcat config file.
Args:
project_root (str, optional): The path to the directory with your template and config file. Defaults to "./".
input_file (str, optional): The name of the Taskcat confile file. Defaults to "./.taskcat.yml".
regions (str, optional): A comma separated list of regions to test in. Defaults to "ALL".
enable_sig_v2 (bool, optional): Enable legacy sigv2 requests for auto-created buckets. Defaults to False.
Returns:
T: Returns a Test instance.
""" # noqa: B950
project_root_path: Path = Path(project_root).expanduser().resolve()
input_file_path: Path = project_root_path / input_file
# pylint: disable=too-many-arguments
args = _build_args(enable_sig_v2, regions, GLOBAL_ARGS.profile)
config = Config.create(
project_root=project_root_path,
project_config_path=input_file_path,
args=args
# TODO: detect if input file is taskcat config or CloudFormation template
)
return cls(config)
Instance variables
config
passed
result
Methods
clean_up
def clean_up(
self
) -> None
Deletes the Test related resources in AWS.
Raises:
| Type | Description |
|---|---|
| TaskCatException | If one or more stacks failed to create. |
View Source
def clean_up(self) -> None: # noqa: C901
"""Deletes the Test related resources in AWS.
Raises:
TaskCatException: If one or more stacks failed to create.
"""
if not hasattr(self, "test_definition"):
LOG.warning("No stacks were created... skipping cleanup.")
return
status = self.test_definition.status()
# Delete Stacks
if self.no_delete:
LOG.info("Skipping delete due to cli argument")
elif self.keep_failed:
if len(status["COMPLETE"]) > 0:
LOG.info("deleting successful stacks")
self.test_definition.delete_stacks({"status": "CREATE_COMPLETE"})
else:
self.test_definition.delete_stacks()
if not self.dont_wait_for_delete:
self.printer.report_test_progress(stacker=self.test_definition)
# TODO: summarise stack statusses (did they complete/delete ok) and print any
# error events
# Delete Templates and Buckets
buckets = self.config.get_buckets()
if not self.no_delete or (
self.keep_failed is True and len(status["FAILED"]) == 0
):
deleted: ListType[str] = []
for test in buckets.values():
for bucket in test.values():
if (bucket.name not in deleted) and not bucket.regional_buckets:
bucket.delete(delete_objects=True)
deleted.append(bucket.name)
# 9. raise if something failed
# - grabbing the status again to ensure everything deleted OK.
status = self.test_definition.status()
if len(status["FAILED"]) > 0:
raise TaskCatException(
f'One or more stacks failed to create: {status["FAILED"]}'
)
report
def report(
self,
output_directory: str = './taskcat_outputs'
)
Generates a report of the status of Cloudformation stacks.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| output_directory | str | The directory to save the report in. Defaults to "./taskcat_outputs". | "./taskcat_outputs" |
View Source
def report(
self,
output_directory: str = "./taskcat_outputs",
):
"""Generates a report of the status of Cloudformation stacks.
Args:
output_directory (str, optional): The directory to save the report in. Defaults to "./taskcat_outputs".
""" # noqa: B950
report_path = Path(output_directory).resolve()
report_path.mkdir(exist_ok=True)
cfn_logs = _CfnLogTools()
cfn_logs.createcfnlogs(self.test_definition, report_path)
ReportBuilder(
self.test_definition, report_path / "index.html"
).generate_report()
run
def run(
self
) -> None
Deploys the required Test resources in AWS.
Raises:
| Type | Description |
|---|---|
| TaskCatException | If skip_upload is set without specifying s3_bucket in config. |
| TaskCatException | If linting fails with errors. |
View Source
def run(self) -> None:
"""Deploys the required Test resources in AWS.
Raises:
TaskCatException: If skip_upload is set without specifying s3_bucket in config.
TaskCatException: If linting fails with errors.
"""
_trim_regions(self.regions, self.config)
_trim_tests(self.test_names, self.config)
boto3_cache = Boto3Cache()
templates = self.config.get_templates()
if self.skip_upload and not self.config.config.project.s3_bucket:
raise TaskCatException(
"cannot skip_buckets without specifying s3_bucket in config"
)
buckets = self.config.get_buckets(boto3_cache)
if not self.skip_upload:
# 1. lint
if not self.lint_disable:
lint = TaskCatLint(self.config, templates)
errors = lint.lints[1]
lint.output_results()
if errors or not lint.passed:
raise TaskCatException("Lint failed with errors")
# 2. build lambdas
if self.config.config.project.package_lambda:
LambdaBuild(self.config, self.config.project_root)
# 3. s3 sync
stage_in_s3(
buckets, self.config.config.project.name, self.config.project_root, []
)
regions = self.config.get_regions(boto3_cache)
parameters = self.config.get_rendered_parameters(buckets, regions, templates)
tests = self.config.get_tests(templates, regions, buckets, parameters)
# pre-hooks
execute_hooks("prehooks", self.config, tests, parameters)
self.test_definition = Stacker(
self.config.config.project.name,
tests,
shorten_stack_name=self.config.config.project.shorten_stack_name,
tags=self._extra_tags,
)
self.test_definition.create_stacks()
# post-hooks
# TODO: pass in outputs, once there is a standard interface for a test_definition
execute_hooks("posthooks", self.config, tests, parameters)
self.printer.report_test_progress(stacker=self.test_definition)
self.passed = True
self.result = self.test_definition.stacks