Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use generate_arn method in localstack

Best Python code snippet using localstack_python

model.py

Source:model.py

...79            fromlist=['url_bases', 'url_paths'],80        )81        return urls_module82    @staticmethod83    def generate_arn(region_name, info=''):84        base_arn = BASE_SAGEMAKER_ARN.format(region=region_name, account=ACCOUNT_ID)85        return base_arn + info86    def describe_endpoint(self, endpoint_name):87        if endpoint_name not in self.endpoints:88            raise ValueError('Endpoint {} does not exist'.format(endpoint_name))89        endpoint = self.endpoints[endpoint_name]90        config = self.endpoint_configs[endpoint['EndpointConfigName']]91        return {92            'Endpoint': endpoint['EndpointName'],93            'EndpointConfigName': endpoint['EndpointConfigName'],94            'EndpointArn': endpoint['arn'],95            'EndpointStatus': endpoint['latest_operation'].status(),96            'CreationTime': endpoint['created_time'],97            'LastModifiedTime': endpoint['latest_operation'].created_time,98            'ProductionVariants': config['ProductionVariants'],99        }100    def delete_endpoint(self, endpoint_name):101        if endpoint_name not in self.endpoints:102            raise ClientError(103                {104                    "Error": {105                        "Code": "ValidationException",106                        "Message": "Could not find endpoint ...",107                    }108                },109                "DeleteEndpoint",110            )111        del self.endpoints[endpoint_name]112    def delete_endpoint_config(self, config_name):113        if config_name not in self.endpoint_configs:114            raise ClientError(115                {116                    "Error": {117                        "Code": "ValidationException",118                        "Message": "Could not find endpoint configuration ...",119                    }120                },121                "DeleteEndpointConfiguration",122            )123        del self.endpoint_configs[config_name]124    def delete_model(self, model_name):125        if model_name not in self.models:126            raise ClientError(127                {128                    "Error": {129                        "Code": "ValidationException",130                        "Message": "Could not find model ...",131                    }132                },133                "DeleteModel",134            )135        del self.models[model_name]136    def create_model(137        self, model_name, tags, primary_container, execution_role_arn, region_name138    ):139        if model_name in self.models:140            raise ValueError('Model {} already exists'.format(model_name))141        info = {142            "Containers": [primary_container],143            'PrimaryContainer': primary_container,144            'ExecutionRoleArn': execution_role_arn,145            'Tags': tags,146        }147        self.models[model_name] = info148        return {149            'resource': info,150            'arn': self.generate_arn(region_name, ':model/{}'.format(model_name)),151        }152    def create_endpoint_config(153        self, endpoint_config_name, production_variants, region_name154    ):155        if endpoint_config_name in self.endpoint_configs:156            raise ValueError(157                'Endpoint configuration {} already exists'.format(endpoint_config_name)158            )159        for production_variant in production_variants:160            if "ModelName" not in production_variant:161                raise ValueError('ModelName is required for ProductionVariants')162            elif production_variant['ModelName'] not in self.models:163                raise ValueError(164                    'Model {} does not exist'.format(production_variant['ModelName'])165                )166            model = self.models[production_variant['ModelName']]167            production_variant['DeployedImages'] = [168                {'SpecifiedImage': model['PrimaryContainer']['Image']}169            ]170        info = {171            'EndpointConfigName': endpoint_config_name,172            'ProductionVariants': production_variants,173        }174        self.endpoint_configs[endpoint_config_name] = info175        return {176            'resource': info,177            'arn': self.generate_arn(178                region_name, ':endpoint-config/{}'.format(endpoint_config_name)179            ),180        }181    def create_endpoint(self, endpoint_name, endpoint_config_name, region_name):182        if endpoint_name in self.endpoints:183            raise ValueError('Endpoint {} already exists'.format(endpoint_name))184        if endpoint_config_name not in self.endpoint_configs:185            raise ValueError(186                'Endpoint configuration {} does not exist'.format(endpoint_config_name)187            )188        created_time = time.time()189        info = {190            'EndpointName': endpoint_name,191            'EndpointConfigName': endpoint_config_name,192            'arn': self.generate_arn(region_name, ':endpoint/{}'.format(endpoint_name)),193            'created_time': created_time,194            'latest_operation': EndpointOperation.create_successful(created_time),195        }196        self.endpoints[endpoint_name] = info197        return {'resource': info, 'arn': info['arn']}198    def update_endpoint(self, endpoint_name, config_name):199        if endpoint_name not in self.endpoints:200            raise ValueError('Endpoint {} does not exist'.format(endpoint_name))201        if config_name not in self.endpoint_configs:202            raise ValueError(203                'Endpoint configuration {} does not exist'.format(config_name)204            )205        endpoint = self.endpoints[endpoint_name]206        endpoint['EndpointConfigName'] = config_name...

vpc_manager.py

Source:vpc_manager.py

...25        match_vpc = self.get_vpc(vpc_id, vpcs)26        match_subnet = self.get_subnet(subnet_id, subnets)27        if match_vpc is not None:28            vpc_data.update({29                'vpc_arn': self.generate_arn('vpc', match_vpc.get('OwnerId'), match_vpc.get('VpcId'), region_name),30                'vpc_id': match_vpc.get('VpcId'),31                'cidr': match_vpc.get('CidrBlock'),32                'vpc_name': self.generate_name(match_vpc),33            })34        if match_subnet is not None:35            subnet_data.update({36                'subnet_name': self.generate_name(match_subnet),37                'subnet_arn': match_subnet.get('SubnetArn'),38                'subnet_id': match_subnet.get('SubnetId'),39                'cidr':  match_subnet.get('CidrBlock'),40            })41        return VPC(vpc_data, strict=False), Subnet(subnet_data, strict=False)42    @staticmethod43    def get_vpc(vpc_id, vpcs):44        for vpc in vpcs:45            if vpc_id == vpc['VpcId']:46                return vpc47        return None48    @staticmethod49    def get_subnet(subnet_id, subnets):50        for subnet in subnets:51            if subnet_id == subnet['SubnetId']:52                return subnet53        return None54    @staticmethod55    def generate_arn(resource_type, owner_id, resource_id, region_name):56        return f'arn:aws:ec2:{region_name}:{owner_id}:{resource_type}/{resource_id}'57    @staticmethod58    def generate_name(resource):59        for resource_tag in resource.get('Tags', []):60            if resource_tag['Key'] == "Name":61                return resource_tag["Value"]...

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.