How to use stack_name method in localstack

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        ]...

Automation Testing Tutorials

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.