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

How to use send_automation_signal method in localstack

Best Python code snippet using localstack_python

tests.py

Source:tests.py

...93            # Since this automation requires approval to continue, the correct status at this point should be 'Waiting'94            assert ssm_doc.automation_execution_status(ssm_client, execution, False) == 'Waiting', \95                'Automation not waiting for approval'96            LOGGER.info('Approving continuation of execution')97            ssm_client.send_automation_signal(98                AutomationExecutionId=execution,99                SignalType='Approve'100            )101            # Collect asg instance lifecycle change.102            asg_status_changes = []103            asg_status_ignores = ["EnteringStandby", "Pending"]104            # Status callback to collect any necessary data105            def status_callback(_):106                collect_asg_status_change(asg_name, working_instance, asg_status_ignores, asg_status_changes)107            # Wait for SSM to finish while collecting value change (callback).108            result = ssm_doc.automation_execution_status(ssm_client, execution, status_callback=status_callback)109            # Verify instance status change.110            LOGGER.info("ASG status change sequence: " + str(asg_status_changes))111            expected_status_change_sequence = [112                "InService",113                "Standby"114            ]115            is_status_change_expected = asg_status_changes == expected_status_change_sequence116            assert is_status_change_expected, 'ASG instant lifecycle did not match expected.'117            LOGGER.info('Verifying automation executions have concluded successfully')118            assert result == 'Success', 'Document did not complete'119        finally:120            try:121                LOGGER.info('Taking instance out of standby (required for CF stack teardown)')122                as_client.exit_standby(123                    InstanceIds=[working_instance],124                    AutoScalingGroupName=asg_name)125            finally:126                test_cf_stack.delete_stack()127                ssm_doc.destroy()128    def test_exit_standby_document(self):129        available_subnets = vpcUtil.find_default_subnets()130        # ensure the account being used has a default VPC.131        assert len(available_subnets) > 0, "No default subnet available for testing"132        ssm_doc = ssm_testing.SSMTester(133            ssm_client=ssm_client,134            doc_filename=os.path.join(135                DOC_DIR,136                'Documents/aws-ASGExitStandbyWithApproval.json'),137            doc_name=EXIT_STANDBY_SSM_DOC_NAME,138            doc_type='Automation'139        )140        test_cf_stack = ssm_testing.CFNTester(141            cfn_client=cfn_client,142            template_filename=os.path.abspath(os.path.join(143                DOC_DIR,144                'Tests/CloudFormationTemplates/ASG.yml')),145            stack_name=EXIT_STANDBY_CFN_STACK_NAME146        )147        automation_role = ssm_doc.get_automation_role(sts_client, iam_client, SERVICE_ROLE_NAME)148        LOGGER.info('Creating AutoScaling Group for testing')149        stack_param = {150            'AMI': AMI_ID,151            'Subnets': available_subnets[0].id,152            'InstanceType': INSTANCE_TYPE153        }154        test_cf_stack.create_stack([155            {'ParameterKey': key, 'ParameterValue': value} for key, value in stack_param.iteritems()])156        asg_name = test_cf_stack.stack_outputs["ASGName"]157        LOGGER.info("Waiting for an instance to become ready...")158        working_instance = asg_wait_for_running_instance(159            asg_name=asg_name,160            number_of_instance=1,161            max_wait_sec=300)[0]162        LOGGER.info("Setting instance to enter standby mode")163        as_client.enter_standby(164            InstanceIds=[working_instance],165            AutoScalingGroupName=asg_name,166            ShouldDecrementDesiredCapacity=True)167        # poll the instance until it reaches the standby state168        asg_wait_for_instance_in_state(working_instance, 'Standby')169        try:170            LOGGER.info("Creating automation document")171            assert ssm_doc.create_document() == 'Active', 'Document not created successfully'172            LOGGER.info(173                "Executing SSM automation document to remove instance from standby mode on {}".format(working_instance))174            user_arn = sts_client.get_caller_identity()['Arn']175            sns_topic_arn = test_cf_stack.stack_outputs['SNSTopicArn']176            LOGGER.info("User ARN for approval: " + user_arn)177            LOGGER.info("SNS Topic ARN for approval: " + sns_topic_arn)178            execution = ssm_doc.execute_automation(179                params={'InstanceId': [working_instance],180                        'LambdaRoleArn': [automation_role],181                        'AutomationAssumeRole': [automation_role],182                        'Approvers': [user_arn],183                        'SNSTopicArn': [sns_topic_arn]})184            # since this automation requires approval to continue, the correct status at this point should be 'Waiting'185            assert ssm_doc.automation_execution_status(ssm_client, execution, False) == 'Waiting', \186                'Automation not waiting for approval'187            LOGGER.info('Approving continuation of execution')188            ssm_client.send_automation_signal(189                AutomationExecutionId=execution,190                SignalType='Approve'191            )192            # Collect asg instance lifecycle change.193            asg_status_changes = []194            asg_status_ignores = ["Pending"]195            # Status callback to collect any necessary data196            def status_callback(_):197                collect_asg_status_change(asg_name, working_instance, asg_status_ignores, asg_status_changes)198            # Wait for SSM to finish while collecting value change (callback).199            result = ssm_doc.automation_execution_status(ssm_client, execution, status_callback=status_callback)200            # Verify instance status change.201            LOGGER.info("ASG status change sequence: " + str(asg_status_changes))202            expected_status_change_sequence = [...

test_approval_document.py

Source:test_approval_document.py

...90                        'RoleName': [PROFILE_NAME],91                        'GroupName': [GROUP_NAME]})92                self.assertEqual(ssm_doc.automation_execution_status(ssm_client, execution, False), 'Waiting')93                LOGGER.info('Approving continuation of execution')94                ssm_client.send_automation_signal(95                    AutomationExecutionId=execution,96                    SignalType='Approve'97                )98                result = ssm_doc.automation_execution_status(ssm_client, execution)99                self.assertEqual(result, "Success")100                instances = ec2_client.describe_instances(101                    Filters=[102                        {'Name': 'tag:aws:cloudformation:stack-name', 'Values': [CFN_STACK_NAME]},103                        {'Name': 'instance-state-name', 'Values': ["running"]}104                    ])105                running_instances = []106                for reservation in instances["Reservations"]:107                    for instance in reservation["Instances"]:108                        print instance...

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.