Best Python code snippet using localstack_python
cloudformation_info.py
Source:cloudformation_info.py
1#!/usr/bin/python2# Copyright: Ansible Project3# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)4from __future__ import absolute_import, division, print_function5__metaclass__ = type6ANSIBLE_METADATA = {'metadata_version': '1.1',7 'status': ['preview'],8 'supported_by': 'community'}9DOCUMENTATION = '''10---11module: cloudformation_info12short_description: Obtain information about an AWS CloudFormation stack13description:14 - Gets information about an AWS CloudFormation stack.15 - This module was called C(cloudformation_facts) before Ansible 2.9, returning C(ansible_facts).16 Note that the M(cloudformation_info) module no longer returns C(ansible_facts)!17requirements:18 - boto3 >= 1.0.019 - python >= 2.620version_added: "2.2"21author:22 - Justin Menga (@jmenga)23 - Kevin Coming (@waffie1)24options:25 stack_name:26 description:27 - The name or id of the CloudFormation stack. Gathers information on all stacks by default.28 type: str29 all_facts:30 description:31 - Get all stack information for the stack.32 type: bool33 default: false34 stack_events:35 description:36 - Get stack events for the stack.37 type: bool38 default: false39 stack_template:40 description:41 - Get stack template body for the stack.42 type: bool43 default: false44 stack_resources:45 description:46 - Get stack resources for the stack.47 type: bool48 default: false49 stack_policy:50 description:51 - Get stack policy for the stack.52 type: bool53 default: false54 stack_change_sets:55 description:56 - Get stack change sets for the stack57 type: bool58 default: false59 version_added: '2.10'60extends_documentation_fragment:61 - aws62 - ec263'''64EXAMPLES = '''65# Note: These examples do not set authentication details, see the AWS Guide for details.66# Get summary information about a stack67- cloudformation_info:68 stack_name: my-cloudformation-stack69 register: output70- debug:71 msg: "{{ output['cloudformation']['my-cloudformation-stack'] }}"72# When the module is called as cloudformation_facts, return values are published73# in ansible_facts['cloudformation'][<stack_name>] and can be used as follows.74# Note that this is deprecated and will stop working in Ansible 2.13.75- cloudformation_facts:76 stack_name: my-cloudformation-stack77- debug:78 msg: "{{ ansible_facts['cloudformation']['my-cloudformation-stack'] }}"79# Get stack outputs, when you have the stack name available as a fact80- set_fact:81 stack_name: my-awesome-stack82- cloudformation_info:83 stack_name: "{{ stack_name }}"84 register: my_stack85- debug:86 msg: "{{ my_stack.cloudformation[stack_name].stack_outputs }}"87# Get all stack information about a stack88- cloudformation_info:89 stack_name: my-cloudformation-stack90 all_facts: true91# Get stack resource and stack policy information about a stack92- cloudformation_info:93 stack_name: my-cloudformation-stack94 stack_resources: true95 stack_policy: true96# Fail if the stack doesn't exist97- name: try to get facts about a stack but fail if it doesn't exist98 cloudformation_info:99 stack_name: nonexistent-stack100 all_facts: yes101 failed_when: cloudformation['nonexistent-stack'] is undefined102'''103RETURN = '''104stack_description:105 description: Summary facts about the stack106 returned: if the stack exists107 type: dict108stack_outputs:109 description: Dictionary of stack outputs keyed by the value of each output 'OutputKey' parameter and corresponding value of each110 output 'OutputValue' parameter111 returned: if the stack exists112 type: dict113 sample:114 ApplicationDatabaseName: dazvlpr01xj55a.ap-southeast-2.rds.amazonaws.com115stack_parameters:116 description: Dictionary of stack parameters keyed by the value of each parameter 'ParameterKey' parameter and corresponding value of117 each parameter 'ParameterValue' parameter118 returned: if the stack exists119 type: dict120 sample:121 DatabaseEngine: mysql122 DatabasePassword: "***"123stack_events:124 description: All stack events for the stack125 returned: only if all_facts or stack_events is true and the stack exists126 type: list127stack_policy:128 description: Describes the stack policy for the stack129 returned: only if all_facts or stack_policy is true and the stack exists130 type: dict131stack_template:132 description: Describes the stack template for the stack133 returned: only if all_facts or stack_template is true and the stack exists134 type: dict135stack_resource_list:136 description: Describes stack resources for the stack137 returned: only if all_facts or stack_resourses is true and the stack exists138 type: list139stack_resources:140 description: Dictionary of stack resources keyed by the value of each resource 'LogicalResourceId' parameter and corresponding value of each141 resource 'PhysicalResourceId' parameter142 returned: only if all_facts or stack_resourses is true and the stack exists143 type: dict144 sample:145 AutoScalingGroup: "dev-someapp-AutoscalingGroup-1SKEXXBCAN0S7"146 AutoScalingSecurityGroup: "sg-abcd1234"147 ApplicationDatabase: "dazvlpr01xj55a"148stack_change_sets:149 description: A list of stack change sets. Each item in the list represents the details of a specific changeset150 returned: only if all_facts or stack_change_sets is true and the stack exists151 type: list152'''153import json154import traceback155from functools import partial156from ansible.module_utils._text import to_native157from ansible.module_utils.aws.core import AnsibleAWSModule158from ansible.module_utils.ec2 import (camel_dict_to_snake_dict, AWSRetry, boto3_tag_list_to_ansible_dict)159try:160 import botocore161except ImportError:162 pass # handled by AnsibleAWSModule163class CloudFormationServiceManager:164 """Handles CloudFormation Services"""165 def __init__(self, module):166 self.module = module167 self.client = module.client('cloudformation')168 @AWSRetry.exponential_backoff(retries=5, delay=5)169 def describe_stacks_with_backoff(self, **kwargs):170 paginator = self.client.get_paginator('describe_stacks')171 return paginator.paginate(**kwargs).build_full_result()['Stacks']172 def describe_stacks(self, stack_name=None):173 try:174 kwargs = {'StackName': stack_name} if stack_name else {}175 response = self.describe_stacks_with_backoff(**kwargs)176 if response is not None:177 return response178 self.module.fail_json(msg="Error describing stack(s) - an empty response was returned")179 except (botocore.exceptions.BotoCoreError, botocore.exceptions.ClientError) as e:180 if 'does not exist' in e.response['Error']['Message']:181 # missing stack, don't bail.182 return {}183 self.module.fail_json_aws(e, msg="Error describing stack " + stack_name)184 @AWSRetry.exponential_backoff(retries=5, delay=5)185 def list_stack_resources_with_backoff(self, stack_name):186 paginator = self.client.get_paginator('list_stack_resources')187 return paginator.paginate(StackName=stack_name).build_full_result()['StackResourceSummaries']188 def list_stack_resources(self, stack_name):189 try:190 return self.list_stack_resources_with_backoff(stack_name)191 except (botocore.exceptions.BotoCoreError, botocore.exceptions.ClientError) as e:192 self.module.fail_json_aws(e, msg="Error listing stack resources for stack " + stack_name)193 @AWSRetry.exponential_backoff(retries=5, delay=5)194 def describe_stack_events_with_backoff(self, stack_name):195 paginator = self.client.get_paginator('describe_stack_events')196 return paginator.paginate(StackName=stack_name).build_full_result()['StackEvents']197 def describe_stack_events(self, stack_name):198 try:199 return self.describe_stack_events_with_backoff(stack_name)200 except (botocore.exceptions.BotoCoreError, botocore.exceptions.ClientError) as e:201 self.module.fail_json_aws(e, msg="Error listing stack events for stack " + stack_name)202 @AWSRetry.exponential_backoff(retries=5, delay=5)203 def list_stack_change_sets_with_backoff(self, stack_name):204 paginator = self.client.get_paginator('list_change_sets')205 return paginator.paginate(StackName=stack_name).build_full_result()['Summaries']206 @AWSRetry.exponential_backoff(retries=5, delay=5)207 def describe_stack_change_set_with_backoff(self, **kwargs):208 paginator = self.client.get_paginator('describe_change_set')209 return paginator.paginate(**kwargs).build_full_result()210 def describe_stack_change_sets(self, stack_name):211 changes = []212 try:213 change_sets = self.list_stack_change_sets_with_backoff(stack_name)214 for item in change_sets:215 changes.append(self.describe_stack_change_set_with_backoff(216 StackName=stack_name,217 ChangeSetName=item['ChangeSetName']))218 return changes219 except (botocore.exceptions.BotoCoreError, botocore.exceptions.ClientError) as e:220 self.module.fail_json_aws(e, msg="Error describing stack change sets for stack " + stack_name)221 @AWSRetry.exponential_backoff(retries=5, delay=5)222 def get_stack_policy_with_backoff(self, stack_name):223 return self.client.get_stack_policy(StackName=stack_name)224 def get_stack_policy(self, stack_name):225 try:226 response = self.get_stack_policy_with_backoff(stack_name)227 stack_policy = response.get('StackPolicyBody')228 if stack_policy:229 return json.loads(stack_policy)230 return dict()231 except (botocore.exceptions.BotoCoreError, botocore.exceptions.ClientError) as e:232 self.module.fail_json_aws(e, msg="Error getting stack policy for stack " + stack_name)233 @AWSRetry.exponential_backoff(retries=5, delay=5)234 def get_template_with_backoff(self, stack_name):235 return self.client.get_template(StackName=stack_name)236 def get_template(self, stack_name):237 try:238 response = self.get_template_with_backoff(stack_name)239 return response.get('TemplateBody')240 except (botocore.exceptions.BotoCoreError, botocore.exceptions.ClientError) as e:241 self.module.fail_json_aws(e, msg="Error getting stack template for stack " + stack_name)242def to_dict(items, key, value):243 ''' Transforms a list of items to a Key/Value dictionary '''244 if items:245 return dict(zip([i.get(key) for i in items], [i.get(value) for i in items]))246 else:247 return dict()248def main():249 argument_spec = dict(250 stack_name=dict(),251 all_facts=dict(required=False, default=False, type='bool'),252 stack_policy=dict(required=False, default=False, type='bool'),253 stack_events=dict(required=False, default=False, type='bool'),254 stack_resources=dict(required=False, default=False, type='bool'),255 stack_template=dict(required=False, default=False, type='bool'),256 stack_change_sets=dict(required=False, default=False, type='bool'),257 )258 module = AnsibleAWSModule(argument_spec=argument_spec, supports_check_mode=True)259 is_old_facts = module._name == 'cloudformation_facts'260 if is_old_facts:261 module.deprecate("The 'cloudformation_facts' module has been renamed to 'cloudformation_info', "262 "and the renamed one no longer returns ansible_facts", version='2.13')263 service_mgr = CloudFormationServiceManager(module)264 if is_old_facts:265 result = {'ansible_facts': {'cloudformation': {}}}266 else:267 result = {'cloudformation': {}}268 for stack_description in service_mgr.describe_stacks(module.params.get('stack_name')):269 facts = {'stack_description': stack_description}270 stack_name = stack_description.get('StackName')271 # Create stack output and stack parameter dictionaries272 if facts['stack_description']:273 facts['stack_outputs'] = to_dict(facts['stack_description'].get('Outputs'), 'OutputKey', 'OutputValue')274 facts['stack_parameters'] = to_dict(facts['stack_description'].get('Parameters'),275 'ParameterKey', 'ParameterValue')276 facts['stack_tags'] = boto3_tag_list_to_ansible_dict(facts['stack_description'].get('Tags'))277 # Create optional stack outputs278 all_facts = module.params.get('all_facts')279 if all_facts or module.params.get('stack_resources'):280 facts['stack_resource_list'] = service_mgr.list_stack_resources(stack_name)281 facts['stack_resources'] = to_dict(facts.get('stack_resource_list'),282 'LogicalResourceId', 'PhysicalResourceId')283 if all_facts or module.params.get('stack_template'):284 facts['stack_template'] = service_mgr.get_template(stack_name)285 if all_facts or module.params.get('stack_policy'):286 facts['stack_policy'] = service_mgr.get_stack_policy(stack_name)287 if all_facts or module.params.get('stack_events'):288 facts['stack_events'] = service_mgr.describe_stack_events(stack_name)289 if all_facts or module.params.get('stack_change_sets'):290 facts['stack_change_sets'] = service_mgr.describe_stack_change_sets(stack_name)291 if is_old_facts:292 result['ansible_facts']['cloudformation'][stack_name] = facts293 else:294 result['cloudformation'][stack_name] = camel_dict_to_snake_dict(facts, ignore_list=('stack_outputs',295 'stack_parameters',296 'stack_policy',297 'stack_resources',298 'stack_tags',299 'stack_template'))300 module.exit_json(changed=False, **result)301if __name__ == '__main__':...
__init__.py
Source:__init__.py
1import logging2import boto33import botocore.exceptions4def deploy_stack(stack_name, template_body, resource_types=None, capabilities=None, parameters=None, tags=None,5 timeout_mins=5):6 logger = logging.getLogger(__name__)7 client = boto3.client('cloudformation')8 create_stack_args = get_create_stack_args(stack_name, template_body, resource_types,9 capabilities, parameters, tags, timeout_mins)10 update_stack_args = get_update_stack_args(stack_name, template_body, resource_types,11 capabilities, parameters, tags)12 logger.info('Deploying stack: {}'.format(stack_name))13 try:14 stacks = client.describe_stacks(StackName=stack_name)15 stack = stacks['Stacks'][0]16 except botocore.exceptions.ClientError:17 stack = None18 if stack is not None and stack['StackStatus'] in ['CREATE_FAILED']:19 replace_stack(client, stack_name, **create_stack_args)20 elif stack is not None:21 try:22 update_stack(client, stack_name, **update_stack_args)23 except botocore.exceptions.ClientError as ex:24 error_code = ex.response['Error']['Code']25 if error_code == 'ValidationError':26 logger.warning('Validation error: {} (stack={})'.format(ex, stack_name))27 else:28 raise ex29 else:30 create_stack(client, stack_name, **create_stack_args)31 return client.describe_stacks(StackName=stack_name)['Stacks'][0]32def deploy_stack_by_replacement(stack_name, template_body, resource_types=None, capabilities=None, parameters=None,33 tags=None, timeout_mins=5):34 create_stack_args = get_create_stack_args(35 stack_name, template_body, resource_types, capabilities, parameters, tags, timeout_mins36 )37 logger = logging.getLogger(__name__)38 client = boto3.client('cloudformation')39 logger.info('Deploying stack: {}'.format(stack_name))40 try:41 stacks = client.describe_stacks(StackName=stack_name)42 stack = stacks['Stacks'][0]43 except botocore.exceptions.ClientError:44 stack = None45 if stack is not None:46 replace_stack(client, stack_name, **create_stack_args)47 else:48 create_stack(client, stack_name, **create_stack_args)49 return client.describe_stacks(StackName=stack_name)['Stacks'][0]50def create_stack(client, stack_name, **kwargs):51 logger = logging.getLogger(__name__)52 logger.info('Creating stack: {}'.format(stack_name))53 client.create_stack(**kwargs)54 logger.info('Waiting for stack create to complete: {}'.format(stack_name))55 waiter = client.get_waiter('stack_create_complete')56 waiter.wait(StackName=stack_name)57 logger.info('Stack created: {}'.format(stack_name))58def update_stack(client, stack_name, **kwargs):59 logger = logging.getLogger(__name__)60 logger.info('Updating stack: {}'.format(stack_name))61 client.update_stack(**kwargs)62 logger.info('Waiting for stack update to complete: {}'.format(stack_name))63 waiter = client.get_waiter('stack_update_complete')64 waiter.wait(StackName=stack_name)65 logger.info('Stack updated: {}'.format(stack_name))66def replace_stack(client, stack_name, **kwargs):67 logger = logging.getLogger(__name__)68 logger.info('Replacing stack: {}'.format(stack_name))69 client.delete_stack(StackName=stack_name)70 logger.info('Waiting for stack delete to complete: {}'.format(stack_name))71 waiter = client.get_waiter('stack_delete_complete')72 waiter.wait(StackName=stack_name)73 logger.info('Stack deleted, recreating: {}'.format(stack_name))74 client.create_stack(**kwargs)75 logger.info('Waiting for stack create to complete: {}'.format(stack_name))76 waiter = client.get_waiter('stack_create_complete')77 waiter.wait(StackName=stack_name)78 logger.info('Stack replaced: {}'.format(stack_name))79def get_create_stack_args(stack_name, template_body, resource_types=None, capabilities=None, parameters=None,80 tags=None, timeout_mins=5):81 return apply_resource_types_to_stack_args(82 apply_capabilities_to_stack_args(83 apply_tags_to_stack_args(84 apply_parameters_to_stack_args(dict(85 StackName=stack_name,86 TemplateBody=template_body,87 DisableRollback=True,88 TimeoutInMinutes=timeout_mins89 ), parameters), tags), capabilities), resource_types)90def get_update_stack_args(stack_name, template_body, resource_types=None, capabilities=None, parameters=None,91 tags=None):92 return apply_resource_types_to_stack_args(93 apply_capabilities_to_stack_args(94 apply_tags_to_stack_args(95 apply_parameters_to_stack_args(dict(96 StackName=stack_name,97 TemplateBody=template_body98 ), parameters), tags), capabilities), resource_types)99def apply_resource_types_to_stack_args(args, resource_types=None):100 if resource_types is not None:101 args['ResourceTypes'] = resource_types102 return args103def apply_capabilities_to_stack_args(args, capabilities=None):104 if capabilities is not None:105 args['Capabilities'] = capabilities106 return args107def apply_parameters_to_stack_args(args, parameters=None):108 if parameters is not None:109 param_list = [{'ParameterKey': key, 'ParameterValue': parameters[key]} for key in parameters]110 args['Parameters'] = param_list111 return args112def apply_tags_to_stack_args(args, tags=None):113 if tags is not None:114 tag_list = [{'Key': key, 'Value': tags[key]} for key in tags]115 args['Tags'] = tag_list...
cloudformation_factory.py
Source:cloudformation_factory.py
1import logging2import boto33import os4import sys5import botocore.exceptions6import json7def setup_logging(log_stream=sys.stdout, log_level=logging.INFO):8 log = logging.getLogger(__name__)9 out_hdlr = logging.StreamHandler(log_stream)10 out_hdlr.setFormatter(logging.Formatter('%(asctime)s %(message)s'))11 out_hdlr.setLevel(logging.INFO)12 log.addHandler(out_hdlr)13 log.setLevel(log_level)14 return log15class CloudformationFactory:16 def __init__(self, config={}, logger=None):17 self.region = 'us-east-2'18 self.cfn = boto3.client('cloudformation', region_name=self.region)19 assert logger is not None20 self.logger = logger21 def upsert_stack(self, stack_name, template, params):22 if self.check_update(stack_name, template, params):23 self.create_stack(stack_name, template, params)24 else:25 self.update_stack(stack_name, template, params)26 def update_stack(self, stack_name, template, parameters=[], wait=True):27 # assert stack_name not in self.stacks.keys()28 # TODO: remove the hard-coded capabilities param29 try:30 self.cfn.update_stack(StackName=stack_name, TemplateBody=template, Parameters=parameters,Capabilities=['CAPABILITY_NAMED_IAM'])31 # self.logger.info(f'Beginning stack update {stack_name}')32 if wait:33 waiter = self.cfn.get_waiter('stack_update_complete')34 waiter.wait(35 StackName=stack_name,36 WaiterConfig={37 'Delay': 10,38 'MaxAttempts': 18039 }40 )41 self.logger.info(f'stack update complete for stack: {stack_name}')42 except Exception as e:43 print(e)44 def create_stack(self, stack_name, template, parameters=[], wait=True):45 # assert stack_name not in self.stacks.keys()46 # TODO: remove the hard-coded capabilities param47 self.cfn.create_stack(StackName=stack_name, TemplateBody=template, Parameters=parameters,Capabilities=['CAPABILITY_NAMED_IAM'])48 self.logger.info(f'Beginning stack create {stack_name}')49 if wait:50 waiter = self.cfn.get_waiter('stack_create_complete')51 waiter.wait(52 StackName=stack_name,53 WaiterConfig={54 'Delay': 20,55 'MaxAttempts': 18056 }57 )58 self.logger.info(f'stack create complete for stack: {stack_name}')59 def check_update(self, stack_name, template, parameters):60 if self._stack_exists(stack_name):61 print('Updating {}'.format(stack_name))62 return False63 else:64 print('Creating {}'.format(stack_name))65 return True66 def _stack_exists(self, stack_name):67 stacks = self.cfn.list_stacks()['StackSummaries']68 for stack in stacks:69 if stack['StackStatus'] == 'DELETE_COMPLETE':70 continue71 if stack_name == stack['StackName']:72 return True73 return False74if __name__ == '__main__':75 logger = setup_logging()76 cfn_handle = CloudformationFactory(logger=logger)77 elb = False78 if elb:79 params = []80 stack_name = 'demo-loadbalancer'81 with open('load_balancer.yml', 'r') as fp:82 template = fp.read()83 cfn_handle.upsert_stack(stack_name, template, params)84 else:85 stack_name = 'webapp-demo'86 with open('webapp.yml', 'r') as fp:87 template = fp.read()88 outputs = boto3.client('cloudformation', region_name='us-east-2').describe_stacks(StackName='demo-loadbalancer')['Stacks'][0]['Outputs']89 target_group = [output['OutputValue'] for output in outputs if output['OutputKey'] == 'AppTargetGroupArn'][0]90 print(target_group)91 params = [92 {93 'ParameterKey': 'TargetGroupArn',94 'ParameterValue': target_group,95 }96 ]...
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!