Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use cfn_client method in localstack

Best Python code snippet using localstack_python

Item_test.py

Source:Item_test.py

1import boto32from troposphere import Template, constants, Parameter, Ref, dynamodb, Equals, If, Output3import custom_resources.dynamodb4from integration_tests.conftest import CloudFormationStates, wait_until_stack_stable, dict_to_param_array5template = Template()6table = template.add_parameter(Parameter(7    "Table",8    Type=constants.STRING,9    AllowedValues=["1", "2"],10    Default='1',11    Description="Key"12))13key = template.add_parameter(Parameter(14    "Key",15    Type=constants.STRING,16    Default='foo',17    Description="Key"18))19value = template.add_parameter(Parameter(20    "Value",21    Type=constants.STRING,22    Default='bar',23    Description="value"24))25overwrite = template.add_parameter(Parameter(26    "Overwrite",27    Type=constants.STRING,28    AllowedValues=['true', 'false'],29    Default='false',30    Description="overwrite",31))32custom_resources.use_custom_resources_stack_name_parameter(template)33table1 = template.add_resource(dynamodb.Table(34    "Table1",35    BillingMode="PAY_PER_REQUEST",36    AttributeDefinitions=[37        dynamodb.AttributeDefinition(38            AttributeName="key",39            AttributeType="S",40        )41    ],42    KeySchema=[43        dynamodb.KeySchema(44            AttributeName="key",45            KeyType="HASH",46        )47    ],48))49template.add_output(Output(50    "Table1Name",51    Value=Ref(table1),52))53table2 = template.add_resource(dynamodb.Table(54    "Table2",55    BillingMode="PAY_PER_REQUEST",56    AttributeDefinitions=[57        dynamodb.AttributeDefinition(58            AttributeName="key",59            AttributeType="S",60        )61    ],62    KeySchema=[63        dynamodb.KeySchema(64            AttributeName="key",65            KeyType="HASH",66        )67    ],68))69template.add_output(Output(70    "Table2Name",71    Value=Ref(table2),72))73table1_selected = template.add_condition("Table1Selected", Equals(Ref(table), '1'))74table_item = template.add_resource(custom_resources.dynamodb.Item(75    "Item",76    TableName=If(table1_selected, Ref(table1), Ref(table2)),77    ItemKey={'key': {'S': Ref(key)}},78    ItemValue={'value': {'S': Ref(value)}},79    Overwrite=Ref(overwrite),80))81def test_item(cloudformation_stack_name):82    cfn_params = {83        "CustomResourcesStack": "vrt-dpc-custom-resources-2-stag",84        "Table": "1",85        "Key": "foo",86        "Value": "bar",87        "Overwrite": "false",88    }89    outputs, stack_id = create_stack(cfn_params, cloudformation_stack_name)90    update_value(cfn_params, outputs, stack_id)91    switch_table(cfn_params, outputs, stack_id)92    already_exists(cfn_params, outputs, stack_id)93    already_exists_overwrite(cfn_params, outputs, stack_id)94    delete_stack(cfn_params, outputs, stack_id)95def create_stack(cfn_params, stack_name):96    cfn_client = boto3.client('cloudformation')97    stack_id = cfn_client.create_stack(98        StackName=stack_name,99        TemplateBody=template.to_json(),100        Parameters=dict_to_param_array(cfn_params),101    )102    stack_id = stack_id['StackId']103    assert CloudFormationStates.CREATE_COMPLETE == wait_until_stack_stable(stack_id)104    stack_info = cfn_client.describe_stacks(105        StackName=stack_id,106    )107    outputs = {108        _['OutputKey']: _['OutputValue']109        for _ in stack_info['Stacks'][0]['Outputs']110    }111    ddb_client = boto3.client('dynamodb')112    item = ddb_client.get_item(113        TableName=outputs['Table1Name'],114        Key={"key": {"S": "foo"}},115    )116    assert {117               "key": {"S": "foo"},118               "value": {"S": "bar"},119           } == item['Item']120    return outputs, stack_id121def update_value(cfn_params, outputs, stack_id):122    cfn_client = boto3.client('cloudformation')123    cfn_params['Value'] = "42"124    cfn_client.update_stack(125        StackName=stack_id,126        UsePreviousTemplate=True,127        Parameters=dict_to_param_array(cfn_params),128    )129    assert CloudFormationStates.UPDATE_COMPLETE == wait_until_stack_stable(stack_id)130    ddb_client = boto3.client('dynamodb')131    item = ddb_client.get_item(132        TableName=outputs['Table1Name'],133        Key={"key": {"S": "foo"}},134    )135    assert {136               "key": {"S": "foo"},137               "value": {"S": "42"},138           } == item['Item']139def switch_table(cfn_params, outputs, stack_id):140    cfn_client = boto3.client('cloudformation')141    cfn_params['Table'] = "2"142    cfn_client.update_stack(143        StackName=stack_id,144        UsePreviousTemplate=True,145        Parameters=dict_to_param_array(cfn_params),146    )147    assert CloudFormationStates.UPDATE_COMPLETE == wait_until_stack_stable(stack_id)148    ddb_client = boto3.client('dynamodb')149    item = ddb_client.get_item(150        TableName=outputs['Table1Name'],151        Key={"key": {"S": "foo"}},152    )153    assert 'Item' not in item154    item = ddb_client.get_item(155        TableName=outputs['Table2Name'],156        Key={"key": {"S": "foo"}},157    )158    assert {159               "key": {"S": "foo"},160               "value": {"S": "42"},161           } == item['Item']162def already_exists(cfn_params, outputs, stack_id):163    ddb_client = boto3.client('dynamodb')164    ddb_client.put_item(165        TableName=outputs['Table1Name'],166        Item={167            "key": {"S": "foo"},168            "value": {"S": "already here"}169        },170    )171    cfn_client = boto3.client('cloudformation')172    cfn_params['Table'] = "1"173    cfn_client.update_stack(174        StackName=stack_id,175        UsePreviousTemplate=True,176        Parameters=dict_to_param_array(cfn_params),177    )178    assert CloudFormationStates.UPDATE_ROLLBACK_COMPLETE == wait_until_stack_stable(stack_id)179    ddb_client = boto3.client('dynamodb')180    item = ddb_client.get_item(181        TableName=outputs['Table1Name'],182        Key={"key": {"S": "foo"}},183    )184    assert {185           "key": {"S": "foo"},186           "value": {"S": "already here"},187       } == item['Item']188    item = ddb_client.get_item(189        TableName=outputs['Table2Name'],190        Key={"key": {"S": "foo"}},191    )192    assert {193               "key": {"S": "foo"},194               "value": {"S": "42"},195           } == item['Item']196def already_exists_overwrite(cfn_params, outputs, stack_id):197    cfn_client = boto3.client('cloudformation')198    cfn_params['Table'] = "1"199    cfn_params['Overwrite'] = "true"200    cfn_client.update_stack(201        StackName=stack_id,202        UsePreviousTemplate=True,203        Parameters=dict_to_param_array(cfn_params),204    )205    assert CloudFormationStates.UPDATE_COMPLETE == wait_until_stack_stable(stack_id)206    ddb_client = boto3.client('dynamodb')207    item = ddb_client.get_item(208        TableName=outputs['Table1Name'],209        Key={"key": {"S": "foo"}},210    )211    assert {212               "key": {"S": "foo"},213               "value": {"S": "42"},214           } == item['Item']215    item = ddb_client.get_item(216        TableName=outputs['Table2Name'],217        Key={"key": {"S": "foo"}},218    )219    assert 'Item' not in item220def delete_stack(cfn_params, outputs, stack_id):221    cfn_client = boto3.client('cloudformation')222    cfn_client.delete_stack(223        StackName=stack_id,224    )...

cfn_stacks_factory.py

Source:cfn_stacks_factory.py

1# Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.2#3# Licensed under the Apache License, Version 2.0 (the "License"). You may not use this file except in compliance4# with the License. A copy of the License is located at5#6# http://aws.amazon.com/apache2.0/7#8# or in the "LICENSE.txt" file accompanying this file. This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES9# OR CONDITIONS OF ANY KIND, express or implied. See the License for the specific language governing permissions and10# limitations under the License.11import logging12import boto313from botocore.exceptions import ClientError14from retrying import retry15from utils import retrieve_cfn_outputs16class CfnStack:17    """Identify a CloudFormation stack."""18    def __init__(self, name, region, template):19        self.name = name20        self.region = region21        self.template = template22        self.cfn_stack_id = None23        self.__cfn_outputs = None24    @property25    def cfn_outputs(self):26        """27        Return the CloudFormation stack outputs for the stack.28        Outputs are retrieved only once and then cached.29        """30        if not self.__cfn_outputs:31            self.__cfn_outputs = retrieve_cfn_outputs(self.name, self.region)32        return self.__cfn_outputs33class CfnStacksFactory:34    """Manage creation and deletion of CloudFormation stacks."""35    def __init__(self):36        self.__created_stacks = {}37    def create_stack(self, stack):38        """39        Create a cfn stack with a given template.40        :param stack: stack to create.41        """42        name = stack.name43        region = stack.region44        id = self.__get_stack_internal_id(name, region)45        if id in self.__created_stacks:46            raise ValueError("Stack {0} already exists in region {1}".format(name, region))47        logging.info("Creating stack {0} in region {1}".format(name, region))48        self.__created_stacks[id] = stack49        try:50            cfn_client = boto3.client("cloudformation", region_name=region)51            result = cfn_client.create_stack(StackName=name, TemplateBody=stack.template)52            stack.cfn_stack_id = result["StackId"]53            final_status = self.__wait_for_stack_creation(stack.cfn_stack_id, cfn_client)54            self.__assert_stack_status(final_status, "CREATE_COMPLETE")55        except Exception as e:56            logging.error("Creation of stack {0} in region {1} failed with exception: {2}".format(name, region, e))57            raise58        logging.info("Stack {0} created successfully in region {1}".format(name, region))59    @retry(60        stop_max_attempt_number=10,61        wait_fixed=5000,62        retry_on_exception=lambda exception: isinstance(exception, ClientError),63    )64    def delete_stack(self, name, region):65        """Destroy a created cfn stack."""66        id = self.__get_stack_internal_id(name, region)67        if id in self.__created_stacks:68            logging.info("Destroying stack {0} in region {1}".format(name, region))69            try:70                stack = self.__created_stacks[id]71                cfn_client = boto3.client("cloudformation", region_name=stack.region)72                cfn_client.delete_stack(StackName=stack.name)73                final_status = self.__wait_for_stack_deletion(stack.cfn_stack_id, cfn_client)74                self.__assert_stack_status(final_status, "DELETE_COMPLETE")75            except Exception as e:76                logging.error("Deletion of stack {0} in region {1} failed with exception: {2}".format(name, region, e))77                raise78            del self.__created_stacks[id]79            logging.info("Stack {0} deleted successfully in region {1}".format(name, region))80        else:81            logging.warning("Couldn't find stack with name {0} in region. Skipping deletion.".format(name, region))82    def delete_all_stacks(self):83        """Destroy all created stacks."""84        logging.debug("Destroying all cfn stacks")85        for _, value in dict(self.__created_stacks).items():86            try:87                self.delete_stack(value.name, value.region)88            except Exception as e:89                logging.error(90                    "Failed when destroying stack {0} in region {1} with exception {2}.".format(91                        value.name, value.region, e92                    )93                )94    @retry(95        retry_on_result=lambda result: result == "CREATE_IN_PROGRESS",96        wait_fixed=5000,97        retry_on_exception=lambda e: False,98    )99    def __wait_for_stack_creation(self, name, cfn_client):100        return self.__get_stack_status(name, cfn_client)101    @retry(102        retry_on_result=lambda result: result == "DELETE_IN_PROGRESS",103        wait_fixed=5000,104        retry_on_exception=lambda e: False,105    )106    def __wait_for_stack_deletion(self, name, cfn_client):107        return self.__get_stack_status(name, cfn_client)108    @staticmethod109    def __get_stack_status(name, cfn_client):110        return cfn_client.describe_stacks(StackName=name).get("Stacks")[0].get("StackStatus")111    @staticmethod112    def __assert_stack_status(status, expected_status):113        if status != expected_status:114            raise Exception("Stack status {0} differs from expected one {1}".format(status, expected_status))115    @staticmethod116    def __get_stack_internal_id(name, region):...

cfn.py

Source:cfn.py

...12  tags_list = []13  for k, v in tags_dict.items():14    tags_list.append({tag_name_key_name: k, tag_value_key_name: v})15  return tags_list16def _get_cfn_client(profile_name: str = None, region: str = None):17  if profile_name:18    session = boto3.Session(profile_name=profile_name, region_name=region)19    return session.client('cloudformation')20  return boto3.client('cloudformation', region_name=region)21def _stack_exists(cfn_client, stack_name):22  stacks = cfn_client.list_stacks()['StackSummaries']23  for stack in stacks:24    if stack['StackStatus'] == 'DELETE_COMPLETE':25      continue26    if stack_name == stack['StackName']:27      return True28  return False29def generate(template: Template, template_path: str, template_format: TemplateFormat):30  print(f'Generating template: {template_path}')31  pathlib.Path(os.path.dirname(os.path.abspath(template_path))).mkdir(parents=True, exist_ok=True)32  if template_format == TemplateFormat.YAML:33    with open(template_path, 'w') as f:34      f.write(template.to_yaml())35  else:36    with open(template_path, 'w') as f:37      f.write(template.to_json())38def _create_stack(cfn_client, stack_params):39  stack_name = stack_params['StackName']40  print(f'Creating stack: {stack_name}')41  cfn_client.create_stack(**stack_params)42  waiter = cfn_client.get_waiter('stack_create_complete')43  print('...waiting for stack to be created...')44  waiter.wait(StackName=stack_name)45  print(f'Stack created: {stack_name}')46def _update_stack(cfn_client, stack_params):47  stack_name = stack_params['StackName']48  print(f'Updating stack: {stack_name}')49  try:50    cfn_client.update_stack(**stack_params)51    waiter = cfn_client.get_waiter('stack_update_complete')52    print('...waiting for stack to be updated...')53    waiter.wait(StackName=stack_name)54  except botocore.exceptions.ClientError as ex:55    error_message = ex.response['Error']['Message']56    if error_message == 'No updates are to be performed.':57      print(f'No changes for stack {stack_name}')58    else:59      raise60  else:61    print(f'Stack updated: {stack_name}')62def _delete_stack(cfn_client, stack_name):63  print(f'Deleting stack: {stack_name}')64  cfn_client.delete_stack(StackName=stack_name)65  waiter = cfn_client.get_waiter('stack_delete_complete')66  print('...waiting for stack to be deleted...')67  waiter.wait(StackName=stack_name)68  print(f'Stack deleted: {stack_name}')69def deploy(stack_name: str,70           template_path: str,71           template_parameters: Dict[str, str] = None,72           tags: Dict[str, str] = None,73           profile: str = None,74           region: str = None75          ):76  stack_params: Dict = {77    'Capabilities': ['CAPABILITY_IAM', 'CAPABILITY_NAMED_IAM'],78    # 'ClientRequestToken': uuid.uuid4(),79    'StackName': stack_name80  }81  with open(template_path, 'r') as template:82    stack_params['TemplateBody'] = template.read()83  stack_params['Parameters'] = []84  if template_parameters:85    for k, v in template_parameters.items():86      stack_params['Parameters'].append({'ParameterKey': k, 'ParameterValue': v})87  if tags:88    stack_params['Tags'] = _dict_to_boto3_tag_list(tags)89  cfn_client = _get_cfn_client(profile, region)90  stack_exists = _stack_exists(cfn_client, stack_name)91  if not stack_exists:92    _create_stack(cfn_client, stack_params)93  else:94    _update_stack(cfn_client, stack_params)95def undeploy(stack_name: str, profile: str = None, region: str = None):96  cfn_client = _get_cfn_client(profile, region)97  stack_exists = _stack_exists(cfn_client, stack_name)98  if stack_exists:99    _delete_stack(cfn_client, stack_name)100  else:...

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.