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

How to use describe_document method in localstack

Best Python code snippet using localstack_python

helmHelperLambda.py

Source:helmHelperLambda.py

...26        # Dict to return lambda outputs27        data = {}28        if eventType == 'Create' or eventType == 'Update':29            # Execute scripts to setup helm deployment30            helmingress_doc = describe_document(event['ResourceProperties']['HelmInstallIngressRunScript'])31            helminstall_doc = describe_document(event['ResourceProperties']['HelmInstallRunScript'])32            helmupgrade_doc = describe_document(event['ResourceProperties']['HelmUpgradeRunScript'])33            helmelb_doc = describe_document(event['ResourceProperties']['GetElbEndpointRunScript'])34            if helmingress_doc['Status'] == 'Active' and helminstall_doc['Status'] == 'Active' and helmupgrade_doc['Status'] == 'Active' and helmelb_doc['Status'] == 'Active':35                # Deploy nginx-ingress helm chart36                logger.info('Installing Nginx-ingress...')37                helmingress = ssm_sendcommand(ssm_instance, helmingress_doc['Name'], {})38                if ssm_commandstatus(helmingress['CommandId'], ssm_instance) is True:39                    logger.info('Nginx-ingress deployed successfully!')40                    if eventType == 'Create':41                        # Install helm chart42                        logger.info('Installing helm chart...')43                        helmdeploy = ssm_sendcommand(ssm_instance, helminstall_doc['Name'], {})44                        if ssm_commandstatus(helmdeploy['CommandId'], ssm_instance) is True:45                            logger.info('Helm chart installation completed successfully!')46                        else:47                            logger.error('Helm chart installation was unsuccessful')48                            cfnresponse.send(event, context, cfnresponse.FAILED, {})49                            return50                    if eventType == 'Update':51                        # First download all helper scripts52                        init_doc = describe_document(event['ResourceProperties']['HelmDownloadScript'])53                        logger.info('Downloading helper scripts...')54                        init = ssm_sendcommand(ssm_instance, init_doc['Name'], {})55                        if ssm_commandstatus(init['CommandId'], ssm_instance) is True:56                            logger.info('scripts directory was downloaded successfully!')57                        else:58                            logger.error('Helper scripts download was unsuccessful')59                            cfnresponse.send(event, context, cfnresponse.FAILED, {})60                            return61                        # Upgrade helm release62                        logger.info('Upgrading helm chart...')63                        helmdeploy = ssm_sendcommand(ssm_instance, helmupgrade_doc['Name'], {})64                        if ssm_commandstatus(helmdeploy['CommandId'], ssm_instance) is True:65                            logger.info('Helm chart upgrade completed successfully!')66                        else:67                            logger.error('Helm chart upgrade was unsuccessful')68                            cfnresponse.send(event, context, cfnresponse.FAILED, {})69                            return70                    # Get ELB to return as Stack Output71                    logger.info('Retrieving ELB URL...')72                    helmelb = ssm_sendcommand(ssm_instance, helmelb_doc['Name'], {})73                    if ssm_commandstatus(helmelb['CommandId'], ssm_instance) is True:74                        logger.info('Got ELB successfully!')75                        helmelb_output = ssm_commandoutput(helmelb['CommandId'], ssm_instance)76                        data['elb'] = helmelb_output['StandardOutputContent'].rstrip('\n')77                        logger.info('Signalling success to CloudFormation...')78                        cfnresponse.send(event, context, cfnresponse.SUCCESS, data, physicalResourceId)79                    else:80                        logger.error('Get Elb command was unsuccessful')81                        cfnresponse.send(event, context, cfnresponse.FAILED, {})82                else:83                    logger.error('Nginx-ingress deployment was unsuccessful')84                    cfnresponse.send(event, context, cfnresponse.FAILED, {})85            else:86                logger.error('SSM commands are not active')87                cfnresponse.send(event, context, cfnresponse.FAILED, {})88        if eventType == 'Delete':89            response = cfn_client.describe_stacks(StackName=event['ResourceProperties']['StackName'])90            status = response['Stacks'][0]['StackStatus']91            if status == 'UPDATE_ROLLBACK_COMPLETE_CLEANUP_IN_PROGRESS':92                logger.info('Skip deleting ACS helm chart')93                cfnresponse.send(event, context, cfnresponse.SUCCESS, data, physicalResourceId)94            else:95                helmdelacs_doc = describe_document(event['ResourceProperties']['HelmDeleteAcsRunScript'])96                helmdel_doc = describe_document(event['ResourceProperties']['HelmDeleteIngressRunScript'])97                if helmdelacs_doc['Status'] == 'Active' and helmdel_doc['Status'] == 'Active':98                    # Delete ACS99                    logger.info('Deleting ACS helm chart...')100                    helmdelacs = ssm_sendcommand(ssm_instance, helmdelacs_doc['Name'], {})101                    if ssm_commandstatus(helmdelacs['CommandId'], ssm_instance) is True:102                        logger.info('ACS helm chart purged successfully!')103                        # Delete the ingress104                        sgId = describe_sg(event['ResourceProperties']['VPCID'], event['ResourceProperties']['EKSName'])105                        # Only delete ingress if exists.106                        # If stack creation was unable to create ingress at the first place this code will not run107                        if sgId is not None:108                            # Delete nginx-ingress ELB as it is not fully managed by CFN109                            logger.info('Deleting Nginx-ingress helm chart...')110                            helmdel = ssm_sendcommand(ssm_instance, helmdel_doc['Name'], {})111                            if ssm_commandstatus(helmdel['CommandId'], ssm_instance) is True:112                                logger.info('Nginx-ingress helm chart purged successfully!')113                            # Revoke elb SecurityGroup rule from node sg and then delete elb SecurityGroup created by nginx-ingess114                            revoke_ingress(event['ResourceProperties']['NodeSecurityGroup'], sgId)115                            # Wait for nginx-ingress security group id to disassociate from ingress ELB network interface(s)116                            netInt = list_interfaces(event['ResourceProperties']['VPCID'], sgId)117                            for int in netInt:118                                while True:119                                    if describe_interfaces(int) != None:120                                        if describe_interfaces(int) == sgId:121                                            logger.info('NetworkInterfaceId "{int}" is still associated with ingress security group "{sgId}", waiting...'.format(int=int, sgId=sgId))122                                            time.sleep(5)123                                    else:124                                        logger.info('NetworkInterfaceId "{int}" does not exists anymore'.format(int=int))125                                        break126                            if delete_sg_status(sgId) is True:127                                logger.info('Deleted nginx-ingress SecurityGroup Id: "{sgId}" successfully'.format(sgId=sgId))128                        else:129                            logger.info('No ingress security group was found.  Exiting.')130                        logger.info('Signalling success to CloudFormation...')131                        cfnresponse.send(event, context, cfnresponse.SUCCESS, data, physicalResourceId)132                    else:133                        logger.error('Failed to delete ACS helm chart')134                        cfnresponse.send(event, context, cfnresponse.FAILED, {})135                else:136                    logger.error('SSM commands are not active')137                    cfnresponse.send(event, context, cfnresponse.FAILED, {})138    except Exception as err:139        logger.error('Helm Helper lambda Error: "{type}": "{message}"'.format(type=type(err), message=str(err)))140        cfnresponse.send(event, context, cfnresponse.FAILED, {})141def describe_document(doc_name):142    '''A function to return SSM Document details'''143    try:144        response = ssm_client.describe_document( Name=doc_name )145        return response['Document']146    except Exception as err:147        logger.error('SSM Document Describe error - "{type}": "{message}"'.format(type=type(err), message=str(err)))148        return err149def ec2_instanceId(autoscaling):150    '''A function to return SSM managed EC2 instances'''151    try:152        response = ec2_client.describe_instances(153                        Filters=[154                            {155                                'Name': 'tag:aws:autoscaling:groupName',156                                'Values': [autoscaling]157                            },158                            {...

test_document_model_view_set.py

Source:test_document_model_view_set.py

1from django.test import TestCase2from rest_framework.test import APIClient3from backend.models import UserModel, RoleModel, TenantModel, AwsEnvironmentModel, Document4from datetime import datetime5from unittest import mock678@mock.patch('backend.views.document_model_view_set.ControlResourceUseCase')9class DocumentViewSetTestCase(TestCase):1011    api_path = '/api/tenants/{}/aws-environments/{}/regions/ap-northeast-1/documents/{}'1213    @staticmethod14    def _create_aws_env_model(name, aws_account_id, tenant):15        now = datetime.now()16        aws = AwsEnvironmentModel.objects.create(17            name=name,18            aws_account_id=aws_account_id,19            aws_role="test_role",20            aws_external_id="test_external_id",21            tenant=tenant,22            created_at=now,23            updated_at=now24        )25        aws.save()26        return aws2728    @staticmethod29    def _create_role_model(id, role_name):30        now = datetime.now()31        return RoleModel.objects.create(32            id=id,33            role_name=role_name,34            created_at=now,35            updated_at=now36        )3738    @staticmethod39    def _create_tenant_model(tenant_name):40        now = datetime.now()41        return TenantModel.objects.create(42            tenant_name=tenant_name,43            created_at=now,44            updated_at=now45        )4647    @staticmethod48    def _create_user_model(email, name, password, tenant, role):49        now = datetime.now()50        user_model = UserModel(51            email=email,52            name=name,53            password=password,54            tenant=tenant,55            role=role,56            created_at=now,57            updated_at=now,58        )59        user_model.save()60        return user_model6162    @classmethod63    def setUpClass(cls):64        super(DocumentViewSetTestCase, cls).setUpClass()65        # Company1ã«æå±ããMASTERã¦ã¼ã¶ã¼ã®ä½æ66        role_model = cls._create_role_model(2, "test_role")67        tenant_model1 = cls._create_tenant_model("test_tenant_users_in_tenant_1")68        # Company1ã«æå±ããAWSç°å¢ã®ä½æ69        aws1 = cls._create_aws_env_model("test_name1", "test_aws1", tenant_model1)7071        user1 = cls._create_user_model(72            email="test_email",73            name="test_name",74            password="test_password",75            tenant=tenant_model1,76            role=role_model,77        )78        user1.aws_environments.add(aws1)79        # Company1ã«æå±ããUSERã¦ã¼ã¶ã¼ã®ä½æ80        role_model_user = cls._create_role_model(3, "test_role")81        user2 = cls._create_user_model(82            email="test_email_USER",83            name="test_name",84            password="test_password",85            tenant=tenant_model1,86            role=role_model_user,87        )88        user2.aws_environments.add(aws1)8990        # Company2ã«æå±ããã¦ã¼ã¶ã¼ã®ä½æ91        tenant_model2 = cls._create_tenant_model("test_tenant_users_in_tenant_2")9293        cls._create_user_model(94            email="test_email2",95            name="test_name2",96            password="test_password2",97            tenant=tenant_model2,98            role=role_model,99        )100101        # Company2ã«æå±ããAWSç°å¢ã®ä½æ102        cls._create_aws_env_model("test_name2", "test_aws2", tenant_model2)103104    # ãã°ã¤ã³ãã¦ããªãç¶æã§APIãä½¿ç¨ã§ããªããã¨ãç¢ºèªãã105    def test_not_login(self, use_case):106        client = APIClient()107        # æ¤è¨¼å¯¾è±¡ã®å®è¡108        response = client.get(self.api_path.format(1, 1, ""), format='json')109        self.assertEqual(response.status_code, 401)110111    # æ£å¸¸ç³»112    def test_list_documents(self, use_case: mock.Mock):113        client = APIClient()114        user_model = UserModel.objects.get(email="test_email")115        client.force_authenticate(user=user_model)116117        # Company1ã®IDãåå¾118        tenant_id = TenantModel.objects.get(tenant_name="test_tenant_users_in_tenant_1").id119        # AWSç°å¢ã®IDãåå¾120        aws_id = AwsEnvironmentModel.objects.get(aws_account_id="test_aws1").id121122        fetch_documents = use_case.return_value.fetch_documents123        fetch_documents.return_value = []124125        # æ¤è¨¼å¯¾è±¡ã®å®è¡126        response = client.get(127            path=self.api_path.format(tenant_id, aws_id, ""),128            format='json')129130        fetch_documents.assert_called_once()131        self.assertEqual(response.status_code, 200)132133    # AWSç°å¢ãåå¨ããªãå ´å134    def test_list_documents_no_aws_env(self, use_case: mock.Mock):135        client = APIClient()136        user_model = UserModel.objects.get(email="test_email")137        client.force_authenticate(user=user_model)138139        fetch_documents = use_case.return_value.fetch_documents140        fetch_documents.return_value = []141142        # æ¤è¨¼å¯¾è±¡ã®å®è¡143        response = client.get(144            path=self.api_path.format(-1, -1, ""),145            format='json')146147        fetch_documents.assert_not_called()148        self.assertEqual(response.status_code, 404)149150    # æ£å¸¸ç³»151    def test_get_document(self, use_case: mock.Mock):152        client = APIClient()153        user_model = UserModel.objects.get(email="test_email")154        client.force_authenticate(user=user_model)155156        # Company1ã®IDãåå¾157        tenant_id = TenantModel.objects.get(tenant_name="test_tenant_users_in_tenant_1").id158        # AWSç°å¢ã®IDãåå¾159        aws_id = AwsEnvironmentModel.objects.get(aws_account_id="test_aws1").id160161        describe_document = use_case.return_value.describe_document162        describe_document.return_value = Document(name="test", parameters=[])163164        # æ¤è¨¼å¯¾è±¡ã®å®è¡165        response = client.get(166            path=self.api_path.format(tenant_id, aws_id, "test/"),167            format='json')168169        describe_document.assert_called_once()170        self.assertEqual(response.status_code, 200)171172    # AWSã®ç°å¢ãåå¨ããªãå ´å173    def test_get_document_no_aws_env(self, use_case: mock.Mock):174        client = APIClient()175        user_model = UserModel.objects.get(email="test_email")176        client.force_authenticate(user=user_model)177178        describe_document = use_case.return_value.describe_document179        describe_document.return_value = Document(name="test", parameters=[])180181        # æ¤è¨¼å¯¾è±¡ã®å®è¡182        response = client.get(183            path=self.api_path.format(-1, -1, "test/"),184            format='json')185186        describe_document.assert_not_called()
...

clusterInitHelperLambda.py

Source:clusterInitHelperLambda.py

...25        # Dict to return lambda outputs26        data = {}27        if eventType == 'Create':28            # First download all helper scripts29            init_doc = describe_document(event['ResourceProperties']['HelmDownloadScript'])30            logger.info('Downloading helper scripts...')31            init = ssm_sendcommand(ssm_instance, init_doc['Name'], {})32            if ssm_commandstatus(init['CommandId'], ssm_instance) is True:33                logger.info('scripts directory was downloaded successfully!')34                # Execute scripts to setup helm deployment35                helminit_doc = describe_document(event['ResourceProperties']['HelmInitRunScript'])36                helmfluentd_doc = describe_document(event['ResourceProperties']['HelmInstallFluentdRunScript'])37                if helminit_doc['Status'] == 'Active' and helmfluentd_doc['Status'] == 'Active':38                    # Deploy Tiller with Helm init39                    logger.info('Initialising helm...')40                    helminit = ssm_sendcommand(ssm_instance, helminit_doc['Name'], {})41                    if ssm_commandstatus(helminit['CommandId'], ssm_instance) is True:42                        logger.info('Tiller was deployed successfully!')43                        # Deploy Fluentd helm chart44                        logger.info('Installing fluentd...')45                        helmdeployfluentd = ssm_sendcommand(ssm_instance, helmfluentd_doc['Name'], {})46                        if ssm_commandstatus(helmdeployfluentd['CommandId'], ssm_instance) is True:47                            logger.info('Fluentd installation completed successfully!')48                            logger.info('Signalling success to CloudFormation...')49                            cfnresponse.send(event, context, cfnresponse.SUCCESS, data, physicalResourceId)50                        else:51                            logger.error('Fluentd installation was unsuccessful')52                            cfnresponse.send(event, context, cfnresponse.FAILED, {})53                    else:54                        logger.error('Tiller deployment was unsuccessful')55                        cfnresponse.send(event, context, cfnresponse.FAILED, {})56                else:57                    logger.error('SSM commands are not active')58                    cfnresponse.send(event, context, cfnresponse.FAILED, {})59            else:60                logger.error('Helper scripts download was unsuccessful')61                cfnresponse.send(event, context, cfnresponse.FAILED, {})62        else:63            logger.info('Nothing to do for update or delete, signalling success to CloudFormation...')64            cfnresponse.send(event, context, cfnresponse.SUCCESS, data, physicalResourceId)65    except Exception as err:66        logger.error('Cluster Init Helper Lambda Error: "{type}": "{message}"'.format(type=type(err), message=str(err)))67        cfnresponse.send(event, context, cfnresponse.FAILED, {})68def describe_document(doc_name):69    '''A function to return SSM Document details'''70    try:71        response = ssm_client.describe_document( Name=doc_name )72        return response['Document']73    except Exception as err:74        logger.error('SSM Document Describe error - "{type}": "{message}"'.format(type=type(err), message=str(err)))75        return err76def ec2_instanceId(autoscaling):77    '''A function to return SSM managed EC2 instances'''78    try:79        response = ec2_client.describe_instances(80                        Filters=[81                            {82                                'Name': 'tag:aws:autoscaling:groupName',83                                'Values': [autoscaling]84                            },85                            {...

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.