KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use change_set_name method in localstack

Best Python code snippet using localstack_python

__init__.py

Source:__init__.py

1import logging2import re 3import yaml4from boto3.session import Session5from botocore.exceptions import ClientError, WaiterError6from datetime import datetime, timedelta7from sys import stdout8class GoUtils:9    def __init__(self, args):10        self.args = args11        self.run_time = datetime.now().strftime('%Y%m%d%H%M%S')12        self.acceptable_stack_states = ['CREATE_COMPLETE', 'UPDATE_COMPLETE', 'UPDATE_ROLLBACK_COMPLETE']13        self.init_logging()14        self.init_aws()15        with open('config.yml', 'r') as frameworkYAML:16            self.config = yaml.load(frameworkYAML)17    def init_aws(self):18        if self.args.profile:19            self.aws = Session(profile_name=self.args.profile, region_name=self.args.region)20        else:21            self.aws = Session(region_name=self.args.region)22        self.appstream = self.aws.client('appstream')23        self.cfn = self.aws.resource('cloudformation')24        self.s3 = self.aws.resource('s3')25    def init_logging(self):26        log_levels = { 'DEBUG': logging.DEBUG, 'INFO': logging.INFO, 'WARNING': logging.WARNING, 'ERROR': logging.ERROR, 'CRITICAL': logging.CRITICAL }27        self.logger = logging.getLogger()28        self.logger.setLevel(log_levels[self.args.log_level])29        sh = logging.StreamHandler(stdout)30        sh.setLevel(log_levels[self.args.log_level])31        formatter = logging.Formatter('[%(levelname)s] %(asctime)s %(message)s')32        sh.setFormatter(formatter)33        self.logger.addHandler(sh)34        logging.getLogger('boto').propagate = False35        logging.getLogger('boto3').propagate = False36        logging.getLogger('botocore').propagate = False37    def summarize_change_set(self, stack_name, change_set_name):38        change_set = self.cfn.meta.client.describe_change_set(ChangeSetName=change_set_name, StackName=stack_name)39        change_template = (40            '\tResource: {resource} ({type})\n'41            '\tAction: {action}\n'42            '{details}\n\n\a'43        )44        details_template = (45            '\tMust Replace: {replace}\n'46            '\tTarget: {target}\n'47            '\tSource: {source}\n'48        )49        change_summary = '\n\nChanges:\n\n'50        for change in change_set['Changes']:51            resource = change['ResourceChange']52            details = ''53            if resource['Action'] == 'Modify':54                for detail in resource['Details']:55                    details = details_template.format(56                        replace=resource['Replacement'],57                        target=detail['Target']['Name'],58                        source=detail['ChangeSource'],59                    ) 60            change_summary += change_template.format(61                resource=resource['LogicalResourceId'],62                type=resource['ResourceType'],63                action=resource['Action'],64                details=details,65            )66        return change_summary67    def deploy_template(self, stack_name, template, tags):68        from troposphere import Tags    69        try:70            stack = self.cfn.create_stack(71                StackName=stack_name,72                TemplateBody=template.to_json(),73                Tags=Tags(tags).tags,74                Capabilities=[75                    'CAPABILITY_IAM',76                    'CAPABILITY_NAMED_IAM',77                ],78            )79            stack_exists = self.cfn.meta.client.get_waiter('stack_exists')80            stack_exists.wait(StackName=stack_name)81            try:82                self.logger.info('Creating {stack} Stack...'.format(stack=stack_name))83                stack_complete = self.cfn.meta.client.get_waiter('stack_create_complete')84                stack_complete.wait(StackName=stack_name)                85            except:86                self.logger.error('Stack Error or Timeout')87            else:88                self.logger.info('Stack Deployed!\a')89        except ClientError as e:90            if e.response['Error']['Code'] == 'AlreadyExistsException':91                stack = list(self.cfn.stacks.filter(StackName=stack_name))[0]92                if self.args.update and stack.stack_status in self.acceptable_stack_states:93                    change_set_name = '{stack}-{runtime}'.format(stack=stack_name, runtime=self.run_time)94                    self.logger.info('Creating Change Set for Existing Stack: ' + change_set_name)95                    self.cfn.meta.client.create_change_set(96                        StackName=stack_name,97                        ChangeSetName=change_set_name,98                        TemplateBody=template.to_json(),99                        Tags=Tags(tags).tags,100                        Capabilities=[101                            'CAPABILITY_IAM',102                            'CAPABILITY_NAMED_IAM',103                        ],104                    )105                    try:106                        change_set_created = self.cfn.meta.client.get_waiter('change_set_create_complete')107                        change_set_created.wait(ChangeSetName=change_set_name, StackName=stack_name)108                    except:109                        change_set = self.cfn.meta.client.describe_change_set(ChangeSetName=change_set_name, StackName=stack_name)110                        self.logger.warning('Change Set Creation Failed: {reason}'.format(reason=change_set['StatusReason']))111                        self.cfn.meta.client.delete_change_set(ChangeSetName=change_set_name, StackName=stack_name)112                    else:113                        self.logger.warning(self.summarize_change_set(stack_name, change_set_name))114                        yn = ''115                        while yn.lower() not in ['y', 'n']:116                            yn = raw_input('Do you want to execute this change set? [y/n]: ')117                        if yn == 'y':118                            self.logger.info('Executing {stack} Change Set: {change}'.format(stack=stack_name, change=change_set_name))119                            self.cfn.meta.client.execute_change_set(ChangeSetName=change_set_name, StackName=stack_name)120                            stack_updated = self.cfn.meta.client.get_waiter('stack_update_complete')121                            stack_updated.wait(StackName=stack_name)122                            self.logger.info('Change Set Deployed!\a')123                        else:124                            self.cfn.meta.client.delete_change_set(ChangeSetName=change_set_name, StackName=stack_name)125                else:126                    self.logger.info('Using Existing Stack: ' + stack_name)127            else:128                raise e129        return stack130    def get_framework_template(self):131        from troposphere import (132            GetAtt,133            Ref,134            Sub,135            Tags,136            Template,137        )138        from troposphere.apigateway import (139            BasePathMapping,140            Deployment,141            DomainName,142            Integration,143            IntegrationResponse,144            Method,145            MethodResponse,146            Resource,147            RestApi,148            Stage,149        )150        from troposphere.ec2 import (151            VPCEndpoint,152        )153        from troposphere.s3 import (154            Bucket,155            BucketPolicy,156            VersioningConfiguration,157        )158        t = Template()159        ###############160        # API Gateway #161        ###############162        api = t.add_resource(RestApi(163            'ApiGateway',164            Name=self.args.stack + 'Api',165            Description='API for portal and redirects for the Cornell AppStream Service',            166        ))167        ####################168        # Redirect Methods #169        ####################170        stack_url = "'https://shibidp.cit.cornell.edu/idp/profile/SAML2/Unsolicited/SSO?providerId=urn:amazon:webservices&target=https://appstream2.{region}.aws.amazon.com/saml?accountId={account}%26stack={stack}'"171        stack_link = '<li><a href="./{redirect_nal}">{redirect}</a></li>'172        methods = []173        stack_links = ''174        for redirect in sorted(self.config['Redirects'].keys()):175            redirect_info = self.config['Redirects'][redirect]176            redirect_nal = re.sub('\W+', '', redirect)177            redirect_url = stack_url.format(account=redirect_info['account'], region=redirect_info['region'], stack=redirect_info['stack'])178            methods.append('ApiGatewayRedirect' + redirect_nal)179            stack_links += stack_link.format(redirect=redirect, redirect_nal=redirect_nal)180            resource = t.add_resource(Resource(181                'ApiGatewayResource' + redirect_nal,182                ParentId=GetAtt(api, 'RootResourceId'),183                PathPart=redirect_nal,184                RestApiId=Ref(api),185            ))186            method = t.add_resource(Method(187                'ApiGatewayRedirect' + redirect_nal,188                AuthorizationType='None',189                HttpMethod='ANY',190                Integration=Integration(191                    Type='MOCK',192                    IntegrationResponses=[193                        IntegrationResponse(194                            ResponseParameters={195                                'method.response.header.Location': redirect_url,196                            },197                            ResponseTemplates={198                                'application/json': '{"redirect": 302}'199                            },200                            StatusCode='302',201                        ),202                    ],203                    RequestTemplates={204                        'application/json': '{"statusCode":200}'205                    },206                ),207                MethodResponses=[208                    MethodResponse(209                        ResponseParameters={210                            'method.response.header.Location': True,211                        },212                        StatusCode='302',213                    ),214                ],215                ResourceId=Ref(resource),216                RestApiId=Ref(api),217            ))            218        ###########################219        # API Gateway Root Method #220        ###########################221        with open('./include/root_integration_template.html', 'r') as rootTemplateHTML:222            rootTemplate = rootTemplateHTML.read()223        root_method = t.add_resource(Method(224            'ApiGatewayRootMethod',225            AuthorizationType='None',226            HttpMethod='ANY',227            Integration=Integration(228                Type='MOCK',229                IntegrationResponses=[230                    IntegrationResponse(231                        ResponseParameters={232                            'method.response.header.Content-Type': "'text/html'",233                        },234                        ResponseTemplates={235                            'text/html': rootTemplate.format(stack_links=stack_links),236                        },237                        StatusCode='200',238                    ),239                ],240                RequestTemplates={241                    'application/json': '{"statusCode":200}'242                },243            ),244            MethodResponses=[245                MethodResponse(246                    ResponseParameters={247                        'method.response.header.Content-Type': True,248                    },249                    StatusCode='200',250                ),251            ],252            ResourceId=GetAtt(api, 'RootResourceId'),253            RestApiId=Ref(api),254        ))255        #####################256        # API Gateway Stage #257        #####################258        api_deployment = t.add_resource(Deployment(259            'ApiGatewayDeployment' + self.run_time,260            Description='Deployment for API portal and redirects for the Cornell AppStream Service',261            RestApiId=Ref(api),262            DependsOn=methods + ['ApiGatewayRootMethod'],263        ))264        api_stage = t.add_resource(Stage(265            'ApiGatewayStage',266            DeploymentId=Ref(api_deployment),267            Description='Stage for API portal and redirects for the Cornell AppStream Service',268            RestApiId=Ref(api),269            StageName='apps',270        ))271        ######################272        # API Gateway Domain #273        ######################274        api_domain = t.add_resource(DomainName(275            'ApiGatewayDomain',276            CertificateArn=self.config['ACM_ARN'],277            DomainName=self.config['DomainName'],278        ))279        api_domain_mapping = t.add_resource(BasePathMapping(280            'ApiGatewayDomainMapping',281            DomainName=Ref(api_domain),282            RestApiId=Ref(api),283            Stage=Ref(api_stage),284        ))285        ###################286        # VPC S3 Endpoint #287        ###################288        s3_endpoint = t.add_resource(VPCEndpoint(289            'S3VPCEndpoint',290            ServiceName=Sub('com.amazonaws.${AWS::Region}.s3'),291            VpcId=self.config['VPC'],292            RouteTableIds=self.config['RouteTables'],293        ))294        ####################295        # S3 Bucket Policy #296        ####################297        sub_args = { 'bucket_name': self.config['Bucket'], 'vpc_id': self.config['VPC'] }298        with open('./include/bucket_policy.json', 'r') as bucketPolicyJSON:299            bucket_policy_document = bucketPolicyJSON.read()300        bucket_policy = t.add_resource(BucketPolicy(301            'FrameworkBucketPolicy',302            Bucket=self.config['Bucket'],303            PolicyDocument=Sub(bucket_policy_document, **sub_args),304        ))305        with open('./cloudformation/framework.json', 'w') as frameworkTemplate:306            frameworkTemplate.write(t.to_json())...

preview.py

Source:preview.py

1import logging2import botocore.exceptions3import docman.manage.exceptions4logger = logging.getLogger(5    __name__,6)7def create_change_set(change_set_name, session, stack_name, template_path, templates_bucket):8    f = open(9        template_path,10        'rb',11    )12    key = f'cft/{ change_set_name }'13    cloudformation = session.client(14        'cloudformation',15    )16    s3 = session.client(17        's3',18    )19    template_argument = dict(20    )21    try:22        s3.put_object(23            Body=f,24            Bucket=templates_bucket,25            Key=key,26        )27    except botocore.exceptions.ClientError as original_exception:28        error_code = original_exception.response['Error']['Code']29        logger.error(30            'CloudFormation stack template: cannot upload to S3 (%s/%s): %s',31            templates_bucket,32            key,33            error_code,34        )35        template_body = f.read(36        )37        template_argument['TemplateBody'] = template_body38    else:39        logger.debug(40            'CloudFormation stack template: uploaded to S3 (%s/%s)',41            templates_bucket,42            key,43        )44        template_url = f'https://{ templates_bucket }.s3.amazonaws.com/{ key }'45        template_argument['TemplateURL'] = template_url46        try:47            try:48                response = cloudformation.validate_template(49                    **template_argument,50                )51            except botocore.exceptions.ClientError as original_exception:52                error_code = original_exception.response['Error']['Code']53                logger.critical(54                    'CloudFormation stack template: validation failed: %s',55                    error_code,56                )57                e = docman.manage.exceptions.TemplatedFailedValidation(58                    original_exception=original_exception,59                )60                raise e61            else:62                logger.debug(63                    'CloudFormation stack template: validation passed',64                )65                capabilities = response['Capabilities']66                required_capabilities = ', '.join(67                    capabilities,68                )69                logger.debug(70                    'CloudFormation stack template: required capabilities: %s',71                    required_capabilities,72                )73                iterator = range(74                    2,75                )76                for attempt in iterator:77                    logger.debug(78                        'CloudFormation stack "%s": change set "%s": creation attempt #%i',79                        stack_name,80                        change_set_name,81                        attempt,82                    )83                    try:84                        cloudformation.create_change_set(85                            Capabilities=capabilities,86                            ChangeSetName=change_set_name,87                            ChangeSetType='UPDATE',88                            StackName=stack_name,89                            **template_argument,90                        )91                    except botocore.exceptions.ClientError as original_exception:92                        error_code = original_exception.response['Error']['Code']93                        logger.debug(94                            'CloudFormation stack "%s": change set "%s": cannot create: %s',95                            stack_name,96                            change_set_name,97                            error_code,98                        )99                        if error_code == 'AlreadyExistsException':100                            logger.warning(101                                'CloudFormation stack "%s": change set "%s": already exists',102                                stack_name,103                                change_set_name,104                            )105                            cloudformation.delete_change_set(106                                ChangeSetName=change_set_name,107                                StackName=stack_name,108                            )109                            logger.debug(110                                'CloudFormation stack "%s": change set "%s": deleted',111                                stack_name,112                                change_set_name,113                            )114                            continue115                        else:116                            e = docman.manage.exceptions.CannotCreateChangeSet(117                                original_exception=original_exception,118                                stack_name=stack_name,119                            )120                            raise e121                    else:122                        logger.debug(123                            'CloudFormation stack "%s": change set "%s": creation request succeeded',124                            stack_name,125                            change_set_name,126                        )127                        break128        except Exception as e:129            raise e130        finally:131            try:132                s3.delete_object(133                    Bucket=templates_bucket,134                    Key=key,135                )136            except botocore.exceptions.ClientError as original_exception:137                error_code = original_exception.response['Error']['Code']138                logger.error(139                    'CloudFormation stack template: cannot delete from S3 (%s/%s): %s',140                    templates_bucket,141                    key,142                    error_code,143                )144            except:145                logger.error(146                    'CloudFormation stack template: cannot delete from S3 (%s/%s): unknown error',147                    templates_bucket,148                    key,149                )150            else:151                logger.debug(152                    'CloudFormation stack template: deleted from S3 (%s/%s)',153                    templates_bucket,154                    key,...

infra_pipeline.py

Source:infra_pipeline.py

1import yaml2from aws_cdk import (3    core, 4    aws_iam as iam,5    aws_secretsmanager as secrets,6    aws_codebuild as codebuild,7    aws_codepipeline as codepipeline,8    aws_codepipeline_actions as actions,9    aws_chatbot as chatbot10)11class InfraPipelineStack(core.Stack):12    def __init__(self, scope: core.Construct, id: str, stacks: list, props: dict, **kwargs) -> None:13        super().__init__(scope=scope, id=id, **kwargs)14        # github info for codepipeline15        github_repo = props.get("GITHUB_INFRA_REPO")16        github_org = props.get("GITHUB_ORG")17        github_branch = props.get("GITHUB_INFRA_BRANCH", "main")18        codestar_connection_arn = props.get("CODESTAR_CONN_ARN")19        codestar_notify_arn = props.get("CODESTAR_NOTIFY_ARN")20        # create the output artifact space for the pipeline21        source_output = codepipeline.Artifact()22        synth_output = codepipeline.Artifact()23        # setup source to be the backstage app source24        source_action = actions.CodeStarConnectionsSourceAction(25            action_name="Github-Source",26            connection_arn=codestar_connection_arn,27            repo=github_repo,28            owner=github_org,29            branch=github_branch,30            output=source_output31        )32        synth_project = codebuild.PipelineProject(33            self, 34            "CodebuildProject", 35            project_name=id,36            build_spec=codebuild.BuildSpec.from_source_filename('./infra-buildspec.yml'),37            environment=codebuild.BuildEnvironment(build_image=codebuild.LinuxBuildImage.STANDARD_5_0)38        )39        40        # lookups needed by backstage stack have to have permissions added in order to synth41        codebuild_policy=iam.PolicyStatement(42            resources=["*"],43            actions=['*'],44        )45        synth_project.add_to_role_policy(codebuild_policy)46        synth_action = actions.CodeBuildAction(47            action_name="Synth",48            project=synth_project,49            input=source_output,50            outputs=[synth_output],51        )52       53        pipeline = codepipeline.Pipeline(self, "infra-pipeline", pipeline_name=id)54        pipeline.add_stage(55            stage_name="Source",56            actions=[source_action]57        )58        pipeline.add_stage(59            stage_name="Synth",60            actions=[synth_action]61        )62        63        for stack in stacks:64            change_set_name = f"{stack}-changeset"65            pipeline.add_stage(66                stage_name=f"Deploy-{stack}",67                actions=[68                    actions.CloudFormationCreateReplaceChangeSetAction(69                        action_name=f"Create-{stack}-ChangeSet",70                        stack_name=stack,71                        template_path=synth_output.at_path(f"{stack}.template.json"),72                        admin_permissions=True,73                        change_set_name=change_set_name,74                        run_order=175                    ),76                    actions.CloudFormationExecuteChangeSetAction(77                        action_name=f"Exec-{stack}-ChangeSet",78                        stack_name=stack,79                        change_set_name=change_set_name,80                        run_order=281                    )82                ]83            )84        if codestar_notify_arn is not None:85            # pull in existing slackbot channel integration86            slack_channel = chatbot.SlackChannelConfiguration.from_slack_channel_configuration_arn(self, 87                "SlackChatbot",88                slack_channel_configuration_arn=codestar_notify_arn,89            )...

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.