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

How to use execute_change_set method in localstack

Best Python code snippet using localstack_python

main.py

Source:main.py

...105            raise Exception(error)106        return "change_set_create_complete"107    def delete_change_set(self):108        client.delete_change_set(**self.parameters)109    def execute_change_set(self):110        try:111            client.execute_change_set(**self.parameters)112        except client.exceptions.ChangeSetNotFoundException as error:113            logger.error(114                "Change set %s was not found", self.parameters["ChangeSetName"]115            )116            raise Exception(error)117        return "stack_update_complete"118    def wait(self, waiter_name):119        logger.info("waiting for operation %s to complete", waiter_name)120        waiter = client.get_waiter(waiter_name)121        try:122            # TODO: Fix this, it should use the paremeters provided by the build_cloudformation_parameters123            parameters = {"StackName": self.stack_name}124            if "change_set" in waiter_name:125                parameters["ChangeSetName"] = f"{self.stack_name}-{self.commit}"126            waiter.wait(**parameters)127            logger.info(f"operation {operation} finished!")128        except botocore.exceptions.WaiterError as error:129            logger.error(130                "An error happened waiting for the operation %s, %s", operation, error131            )132            raise Exception(error)133    def generate_parameters(self):134        logger.info("generating parameters")135        for file in os.listdir(self.stack_parameters_path):136            if self.environment in file:137                parameters_file = os.path.join(self.stack_parameters_path, file)138                if "enc" in file:139                    subprocess.run(140                        [141                            "sops",142                            "-d",143                            parameters_file,144                            ">",145                            self.stack_tmp_parameters_file,146                        ],147                        stdout=subprocess.DEVNULL,148                    )149                else:150                    logger.debug("copying file")151                    shutil.copyfile(parameters_file, self.stack_tmp_parameters_file)152                break153    def clean(self):154        logger.info("Cleaning up the parameters")155        subprocess.run(["rm", self.stack_tmp_parameters_file])156# TODO: Support decrypt of the parameters file157# def generate_parameters(parameters_path):158def usage(info, err):159    help = """160main script helps with the creation of CloudFormation stacks.161Usage:162python main stack-name <command>163Commands to work with Stacks:164create              - Creates a new stack based on the environment and the stack name165update              - Updates an already created stack166delete              - Deletes an already created stack167Commands to work with Change Sets:168create_change_set   - Creates a new change set using the stack name and the commit's SHA as change set name169execute_change_set  - Executes the current change set170delete_change_set   - Delete the current change set171The stack-name is the name of the stack's folder inside stacks/172E.g.173python main network create174    """175    print(info)176    print(help)177    sys.exit(err)178def args_validator():179    if len(sys.argv) != 3:180        usage("Invalid number of arguments", errno.EINVAL)181    if sys.argv[2] not in OPERATIONS:182        usage(f"{sys.argv[2]} is not permitted", errno.EPERM)183    stack_path = os.path.join("stacks", sys.argv[1])184    if not os.path.exists(stack_path):185        usage(f"{stack_path} does not exist", errno.EPERM)186    try:187        os.environ["PROJECT"]188        os.environt["ENVIRONMENT"]189    except:190        logger.warning(191            "PROJECT or ENVIRONMENT are not set, falling back to default values"192        )193# Run validation before creating194args_validator()195operation = sys.argv[2]196stack = Stack(sys.argv[1], sys.argv[2])197stack.validate_cloudformation_template()198try:199    if operation == "create":200        waiter_name = stack.create_stack()201    elif operation == "delete":202        waiter_name = stack.delete_stack()203    elif operation == "update":204        waiter_name = stack.update_stack()205    elif operation == "create_change_set":206        waiter_name = stack.create_change_set()207    elif operation == "delete_change_set":208        waiter_name = stack.delete_change_set()209        sys.exit(0)210    elif operation == "execute_change_set":211        waiter_name = stack.execute_change_set()212    else:213        usage(1)214    stack.wait(waiter_name)215except Exception as error:216    logger.error(error)217finally:...

deploy.py

Source:deploy.py

...25                        ChangeSetName=self.change_set_name,26                        ChangeSetType='CREATE'27                        )28            self.wait()29            self.execute_change_set()30        except ClientError as e:31            if e.response['Error']['Code'] == 'ValidationError':32                print "Removing any previously created change sets with the same name"33                self.delete_change_set()                34                print "Create UPDATE change set {}".format(self.change_set_name)35                change_set = self.client.create_change_set(36                        StackName=self.stackname,37                        TemplateBody=self.cfn_string,38                        Parameters=[39                            {40                                'ParameterKey' : 'dbTableName',41                                'ParameterValue' : self.table_name42                            }43                        ],44                        ChangeSetName=self.change_set_name,45                        ChangeSetType='UPDATE'46                        )47                self.wait()48                print type(self.get_change_status())49                if self.get_change_status() is not 'True':50                    raise Exception("Change Set Cannot be excuted as the table would be replaced")51                else:52                    self.execute_change_set()53            else:54                print "Unexpected error: {}".format(e)55    def wait(self):56        print "Waiting for change set create complete"57        waiter = self.client.get_waiter('change_set_create_complete')58        try:59            waiter.wait(60                ChangeSetName=self.change_set_name,61                StackName=self.stackname,62                WaiterConfig={63                    'Delay': 30,64                    'MaxAttempts' : 12065                }66            )67            print "Change Set Complete"68        except WaiterError:69            raise Exception("Your Change Set Does not Contain Any Changes")70            71    def delete_change_set(self):72        self.client.delete_change_set(73            ChangeSetName=self.change_set_name,74            StackName=self.stackname75        )76            77    def get_change_status(self):78        describe_response = self.client.describe_change_set(79            ChangeSetName=self.change_set_name,80            StackName=self.stackname81        )82        # Boolean value for the modify action if replacement is true than the resource is deleted and recreated83        #print describe_response["Changes"][0]["ResourceChange"]["Replacement"]84        return describe_response["Changes"][0]["ResourceChange"]["Replacement"]85    def execute_change_set(self):86        print "Executing change set {} on stack {}".format(self.change_set_name, self.stackname)87        self.client.execute_change_set(88            ChangeSetName=self.change_set_name,89            StackName=self.stackname90            )91def main():92    with open('dynamodb.yml') as f:93        template_string = f.read()94    cf1 = cloudformation( cfn_string=template_string, 95                         stackname='jp-test-1', 96                         table_name="myTableName3", 97                         change_set_name="change-set-1")98    cf1.create_change_set()99if __name__ == "__main__":...

execute.py

Source:execute.py

...34 yes,35 change_set=change_set_name,36 command_path=path37 )...

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.