Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def stackdelete(self, testdata_list):
"""
This function deletes the CloudFormation stacks of the given tests.
:param testdata_list: List of TestData objects
"""
for test in testdata_list:
for stack in test.get_test_stacks():
stackdata = CommonTools(stack['StackId']).parse_stack_info()
region = stackdata['region']
stack_name = stackdata['stack_name']
cfn = self._boto_client.get('cloudformation', region=region)
cfn.delete_stack(StackName=stack_name)
def write_logs(self, stack_id, logpath):
"""
This function writes the event logs of the given stack and all the child stacks to a given file.
:param stack_id: Stack Id
:param logpath: Log file path
:return:
"""
stackinfo = CommonTools(stack_id).parse_stack_info()
stackname = str(stackinfo["stack_name"])
region = str(stackinfo["region"])
# Get stack resources
cfnlogs = self.get_cfnlogs(stackname, region)
if len(cfnlogs) != 0:
if cfnlogs[0]["ResourceStatus"] != "CREATE_COMPLETE":
if "ResourceStatusReason" in cfnlogs[0]:
reason = cfnlogs[0]["ResourceStatusReason"]
else:
reason = "Unknown"
else:
reason = "Stack launch was successful"
msg = "StackName: %s \n" % stackname
)
)
# if resource is a stack and has a physical resource id
# (NOTE: physical id will be missing if stack creation is failed)
if (
resource.get("ResourceType") == "AWS::CloudFormation::Stack"
and "PhysicalResourceId" in resource
):
if include_stacks:
d = {
"logicalId": resource.get("LogicalResourceId"),
"physicalId": resource.get("PhysicalResourceId"),
"resourceType": resource.get("ResourceType"),
}
l_resources.append(d)
stackdata = CommonTools(
str(resource.get("PhysicalResourceId"))
).parse_stack_info()
region = stackdata["region"]
self.get_resources_helper(
resource.get("PhysicalResourceId"),
region,
l_resources,
include_stacks,
)
# else if resource is not a stack and has a physical resource id
# (NOTE: physical id will be missing if stack creation is failed)
elif (
resource.get("ResourceType") != "AWS::CloudFormation::Stack"
and "PhysicalResourceId" in resource
):
d = {
def collect_resources(self, testdata_list, logpath):
"""
This function collects the AWS resources information created by the
CloudFormation stack for generating the report.
:param testdata_list: List of TestData object
:param logpath: Log file path
"""
resource = {}
print(PrintMsg.INFO + "(Collecting Resources)")
for test in testdata_list:
for stack in test.get_test_stacks():
stackinfo = CommonTools(stack['StackId']).parse_stack_info()
# Get stack resources
resource[stackinfo['region']] = (
CfnResourceTools(self._boto_client).get_resources(
str(stackinfo['stack_name']),
str(stackinfo['region'])
)
)
extension = '.txt'
test_logpath = '{}/{}-{}-{}{}'.format(
logpath,
stackinfo['stack_name'],
stackinfo['region'],
'resources',
extension)
# Write resource logs
def stackcheck(self, stack_id):
"""
Given the stack id, this function returns the status of the stack as
a list with stack name, region, and status as list items, in the respective
order.
:param stack_id: CloudFormation stack id
:return: List containing the stack name, region and stack status in the
respective order.
"""
stackdata = CommonTools(stack_id).parse_stack_info()
region = stackdata['region']
stack_name = stackdata['stack_name']
test_info = []
cfn = self._boto_client.get('cloudformation', region=region)
# noinspection PyBroadException
try:
test_query = (cfn.describe_stacks(StackName=stack_name))
for result in test_query['Stacks']:
test_info.append(stack_name)
test_info.append(region)
test_info.append(result.get('StackStatus'))
if result.get(
'StackStatus') == 'CREATE_IN_PROGRESS' or result.get('StackStatus') == 'DELETE_IN_PROGRESS':
test_info.append(1)
else: