Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use get_deploy_templates method in localstack

Best Python code snippet using localstack_python

apigateway.py

Source:apigateway.py

...18        client = aws_stack.connect_to_service("apigateway")19        result = client.get_gateway_response(restApiId=api_id, responseType=props["ResponseType"])20        return result if "responseType" in result else None21    @staticmethod22    def get_deploy_templates():23        return {24            "create": {25                "function": "put_gateway_response",26                "parameters": {27                    "restApiId": "RestApiId",28                    "responseType": "ResponseType",29                    "statusCode": "StatusCode",30                    "responseParameters": "ResponseParameters",31                    "responseTemplates": "ResponseTemplates",32                },33            }34        }35class GatewayRequestValidator(GenericBaseModel):36    @staticmethod37    def cloudformation_type():38        return "AWS::ApiGateway::RequestValidator"39    def get_physical_resource_id(self, attribute=None, **kwargs):40        return self.props.get("id")41    def fetch_state(self, stack_name, resources):42        client = aws_stack.connect_to_service("apigateway")43        props = self.props44        api_id = self.resolve_refs_recursively(stack_name, props["RestApiId"], resources)45        name = self.resolve_refs_recursively(stack_name, props["Name"], resources)46        result = client.get_request_validators(restApiId=api_id).get("items", [])47        result = [r for r in result if r.get("name") == name]48        return result[0] if result else None49    @staticmethod50    def get_deploy_templates():51        return {52            "create": {53                "function": "create_request_validator",54                "parameters": {55                    "name": "Name",56                    "restApiId": "RestApiId",57                    "validateRequestBody": "ValidateRequestBody",58                    "validateRequestParameters": "ValidateRequestParameters",59                },60            },61            "delete": {62                "function": "delete_request_validator",63                "parameters": {"restApiId": "RestApiId", "requestValidatorId": "id"},64            },65        }66class GatewayRestAPI(GenericBaseModel):67    @staticmethod68    def cloudformation_type():69        return "AWS::ApiGateway::RestApi"70    def get_physical_resource_id(self, attribute=None, **kwargs):71        return self.props.get("id")72    def fetch_state(self, stack_name, resources):73        apis = aws_stack.connect_to_service("apigateway").get_rest_apis()["items"]74        api_name = self.props.get("Name") or self.resource_id75        api_name = self.resolve_refs_recursively(stack_name, api_name, resources)76        result = list(filter(lambda api: api["name"] == api_name, apis))77        return result[0] if result else None78    @staticmethod79    def get_deploy_templates():80        def _api_id(params, resources, resource_id, **kwargs):81            resource = GatewayRestAPI(resources[resource_id])82            return resource.physical_resource_id or resource.get_physical_resource_id()83        return {84            "create": {85                "function": "create_rest_api",86                "parameters": {"name": "Name", "description": "Description"},87            },88            "delete": {89                "function": "delete_rest_api",90                "parameters": {91                    "restApiId": _api_id,92                },93            },94        }95class GatewayDeployment(GenericBaseModel):96    @staticmethod97    def cloudformation_type():98        return "AWS::ApiGateway::Deployment"99    def fetch_state(self, stack_name, resources):100        api_id = self.props.get("RestApiId")101        api_id = self.resolve_refs_recursively(stack_name, api_id, resources)102        if not api_id:103            return None104        client = aws_stack.connect_to_service("apigateway")105        result = client.get_deployments(restApiId=api_id)["items"]106        # TODO possibly filter results by stage name or other criteria107        return result[0] if result else None108    def get_physical_resource_id(self, attribute=None, **kwargs):109        return self.props.get("id")110    @staticmethod111    def get_deploy_templates():112        return {113            "create": {114                "function": "create_deployment",115                "parameters": {116                    "restApiId": "RestApiId",117                    "stageName": "StageName",118                    "stageDescription": "StageDescription",119                    "description": "Description",120                },121            }122        }123class GatewayResource(GenericBaseModel):124    @staticmethod125    def cloudformation_type():126        return "AWS::ApiGateway::Resource"127    def fetch_state(self, stack_name, resources):128        props = self.props129        api_id = props.get("RestApiId") or self.resource_id130        api_id = self.resolve_refs_recursively(stack_name, api_id, resources)131        parent_id = self.resolve_refs_recursively(stack_name, props.get("ParentId"), resources)132        if not api_id or not parent_id:133            return None134        api_resources = aws_stack.connect_to_service("apigateway").get_resources(restApiId=api_id)[135            "items"136        ]137        target_resource = list(138            filter(139                lambda res: res.get("parentId") == parent_id140                and res["pathPart"] == props["PathPart"],141                api_resources,142            )143        )144        if not target_resource:145            return None146        path = aws_stack.get_apigateway_path_for_resource(147            api_id, target_resource[0]["id"], resources=api_resources148        )149        result = list(filter(lambda res: res["path"] == path, api_resources))150        return result[0] if result else None151    @staticmethod152    def get_deploy_templates():153        def get_apigw_resource_params(params, **kwargs):154            result = {155                "restApiId": params.get("RestApiId"),156                "pathPart": params.get("PathPart"),157                "parentId": params.get("ParentId"),158            }159            if not result.get("parentId"):160                # get root resource id161                apigw = aws_stack.connect_to_service("apigateway")162                resources = apigw.get_resources(restApiId=result["restApiId"])["items"]163                root_resource = ([r for r in resources if r["path"] == "/"] or [None])[0]164                if not root_resource:165                    raise Exception(166                        "Unable to find root resource for REST API %s" % result["restApiId"]167                    )168                result["parentId"] = root_resource["id"]169            return result170        return {171            "create": {172                "function": "create_resource",173                "parameters": get_apigw_resource_params,174            }175        }176class GatewayMethod(GenericBaseModel):177    @staticmethod178    def cloudformation_type():179        return "AWS::ApiGateway::Method"180    def fetch_state(self, stack_name, resources):181        props = self.props182        api_id = self.resolve_refs_recursively(stack_name, props["RestApiId"], resources)183        res_id = self.resolve_refs_recursively(stack_name, props["ResourceId"], resources)184        if not api_id or not res_id:185            return None186        res_obj = aws_stack.connect_to_service("apigateway").get_resource(187            restApiId=api_id, resourceId=res_id188        )189        match = [190            v191            for (k, v) in res_obj.get("resourceMethods", {}).items()192            if props["HttpMethod"] in (v.get("httpMethod"), k)193        ]194        int_props = props.get("Integration") or {}195        if int_props.get("Type") == "AWS_PROXY":196            match = [197                m198                for m in match199                if m.get("methodIntegration", {}).get("type") == "AWS_PROXY"200                and m.get("methodIntegration", {}).get("httpMethod")201                == int_props.get("IntegrationHttpMethod")202            ]203        return match[0] if match else None204    def update_resource(self, new_resource, stack_name, resources):205        props = new_resource["Properties"]206        client = aws_stack.connect_to_service("apigateway")207        integration = props.get("Integration")208        kwargs = {209            "restApiId": props["RestApiId"],210            "resourceId": props["ResourceId"],211            "httpMethod": props["HttpMethod"],212            "requestParameters": props.get("RequestParameters") or {},213        }214        if integration:215            kwargs["type"] = integration["Type"]216            if integration.get("IntegrationHttpMethod"):217                kwargs["integrationHttpMethod"] = integration.get("IntegrationHttpMethod")218            if integration.get("Uri"):219                kwargs["uri"] = integration.get("Uri")220            kwargs["requestParameters"] = integration.get("RequestParameters") or {}221            return client.put_integration(**kwargs)222        kwargs["authorizationType"] = props.get("AuthorizationType")223        return client.put_method(**kwargs)224    def get_physical_resource_id(self, attribute=None, **kwargs):225        props = self.props226        result = "%s-%s-%s" % (227            props.get("RestApiId"),228            props.get("ResourceId"),229            props.get("HttpMethod"),230        )231        return result232    @classmethod233    def get_deploy_templates(cls):234        """235        https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-apigateway-method.html236        https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-apitgateway-method-methodresponse.html237        https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-apitgateway-method-integration.html238        https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-apitgateway-method-integration-integrationresponse.html239        """240        def _subresources(resource_id, resources, resource_type, func, stack_name):241            apigateway = aws_stack.connect_to_service("apigateway")242            resource = cls(resources[resource_id])243            props = resource.props244            integration = props.get("Integration")245            if integration:246                api_id = resource.resolve_refs_recursively(247                    stack_name, props["RestApiId"], resources248                )249                res_id = resource.resolve_refs_recursively(250                    stack_name, props["ResourceId"], resources251                )252                kwargs = {}253                if integration.get("Uri"):254                    uri = resource.resolve_refs_recursively(255                        stack_name, integration.get("Uri"), resources256                    )257                    # Moto has a validate method on Uri for integration_type "HTTP" | "HTTP_PROXY" that does not accept258                    # Uri value without path, we need to add path ("/") if not exists259                    if integration.get("Type") in ["HTTP", "HTTP_PROXY"]:260                        rs = urlparse(uri)261                        if not rs.path:262                            uri = "{}/".format(uri)263                    kwargs["uri"] = uri264                if integration.get("IntegrationHttpMethod"):265                    kwargs["integrationHttpMethod"] = integration["IntegrationHttpMethod"]266                if integration.get("RequestTemplates"):267                    kwargs["requestTemplates"] = integration["RequestTemplates"]268                if integration.get("Credentials"):269                    kwargs["credentials"] = integration["Credentials"]270                if integration.get("RequestParameters"):271                    kwargs["requestParameters"] = integration["RequestParameters"]272                apigateway.put_integration(273                    restApiId=api_id,274                    resourceId=res_id,275                    httpMethod=props["HttpMethod"],276                    type=integration["Type"],277                    **kwargs,278                )279            responses = props.get("MethodResponses") or []280            for response in responses:281                api_id = resource.resolve_refs_recursively(282                    stack_name, props["RestApiId"], resources283                )284                res_id = resource.resolve_refs_recursively(285                    stack_name, props["ResourceId"], resources286                )287                apigateway.put_method_response(288                    restApiId=api_id,289                    resourceId=res_id,290                    httpMethod=props["HttpMethod"],291                    statusCode=str(response["StatusCode"]),292                    responseParameters=response.get("ResponseParameters", {}),293                )294        return {295            "create": [296                {297                    "function": "put_method",298                    "parameters": {299                        "restApiId": "RestApiId",300                        "resourceId": "ResourceId",301                        "httpMethod": "HttpMethod",302                        "apiKeyRequired": "ApiKeyRequired",303                        "authorizationType": "AuthorizationType",304                        "authorizerId": "AuthorizerId",305                        "requestParameters": "RequestParameters",306                    },307                },308                {309                    "function": _subresources  # dynamic mapping for additional sdk calls for this CFn resource310                },311            ]312        }313class GatewayStage(GenericBaseModel):314    @staticmethod315    def cloudformation_type():316        return "AWS::ApiGateway::Stage"317    def fetch_state(self, stack_name, resources):318        api_id = self.props.get("RestApiId") or self.resource_id319        api_id = self.resolve_refs_recursively(stack_name, api_id, resources)320        if not api_id:321            return None322        result = aws_stack.connect_to_service("apigateway").get_stage(323            restApiId=api_id, stageName=self.props["StageName"]324        )325        return result326    def get_physical_resource_id(self, attribute=None, **kwargs):327        return self.props.get("id")328    @staticmethod329    def get_deploy_templates():330        def get_params(params, **kwargs):331            result = keys_to_lower(params)332            param_names = [333                "restApiId",334                "stageName",335                "deploymentId",336                "description",337                "cacheClusterEnabled",338                "cacheClusterSize",339                "variables",340                "documentationVersion",341                "canarySettings",342                "tracingEnabled",343                "tags",344            ]345            result = select_attributes(result, param_names)346            result["tags"] = {t["key"]: t["value"] for t in result.get("tags", [])}347            return result348        return {"create": {"function": "create_stage", "parameters": get_params}}349class GatewayUsagePlan(GenericBaseModel):350    @staticmethod351    def cloudformation_type():352        return "AWS::ApiGateway::UsagePlan"353    def fetch_state(self, stack_name, resources):354        plan_name = self.props.get("UsagePlanName")355        plan_name = self.resolve_refs_recursively(stack_name, plan_name, resources)356        result = aws_stack.connect_to_service("apigateway").get_usage_plans().get("items", [])357        result = [r for r in result if r["name"] == plan_name]358        return (result or [None])[0]359    @staticmethod360    def get_deploy_templates():361        return {362            "create": {363                "function": "create_usage_plan",364                "parameters": {365                    "name": "UsagePlanName",366                    "description": "Description",367                    "apiStages": lambda_keys_to_lower("ApiStages"),368                    "quota": lambda_keys_to_lower("Quota"),369                    "throttle": lambda_keys_to_lower("Throttle"),370                    "tags": params_list_to_dict("Tags"),371                },372            }373        }374    def get_physical_resource_id(self, attribute=None, **kwargs):375        return self.props.get("id")376class GatewayApiKey(GenericBaseModel):377    @staticmethod378    def cloudformation_type():379        return "AWS::ApiGateway::ApiKey"380    def fetch_state(self, stack_name, resources):381        props = self.props382        key_name = self.resolve_refs_recursively(stack_name, props.get("Name"), resources)383        cust_id = props.get("CustomerId")384        result = aws_stack.connect_to_service("apigateway").get_api_keys().get("items", [])385        result = [386            r387            for r in result388            if r.get("name") == key_name and cust_id in (None, r.get("customerId"))389        ]390        return (result or [None])[0]391    @staticmethod392    def get_deploy_templates():393        return {394            "create": {395                "function": "create_api_key",396                "parameters": {397                    "description": "Description",398                    "customerId": "CustomerId",399                    "name": "Name",400                    "value": "Value",401                    "enabled": "Enabled",402                    "stageKeys": lambda_keys_to_lower("StageKeys"),403                    "tags": params_list_to_dict("Tags"),404                },405                "types": {"enabled": bool},406            }407        }408    def get_physical_resource_id(self, attribute=None, **kwargs):409        return self.props.get("id")410class GatewayUsagePlanKey(GenericBaseModel):411    @staticmethod412    def cloudformation_type():413        return "AWS::ApiGateway::UsagePlanKey"414    def fetch_state(self, stack_name, resources):415        client = aws_stack.connect_to_service("apigateway")416        key_id = self.resolve_refs_recursively(stack_name, self.props.get("KeyId"), resources)417        key_type = self.resolve_refs_recursively(stack_name, self.props.get("KeyType"), resources)418        plan_id = self.resolve_refs_recursively(419            stack_name, self.props.get("UsagePlanId"), resources420        )421        result = client.get_usage_plan_keys(usagePlanId=plan_id).get("items", [])422        result = [r for r in result if r["id"] == key_id and key_type in [None, r.get("type")]]423        return (result or [None])[0]424    @staticmethod425    def get_deploy_templates():426        return {427            "create": {428                "function": "create_usage_plan_key",429                "parameters": lambda_keys_to_lower(),430            }431        }432    def get_physical_resource_id(self, attribute=None, **kwargs):433        return self.props.get("id")434# TODO: add tests for this resource type435class GatewayDomain(GenericBaseModel):436    @staticmethod437    def cloudformation_type():438        return "AWS::ApiGateway::DomainName"439    def fetch_state(self, stack_name, resources):440        return aws_stack.connect_to_service("apigateway").get_domain_name(441            domainName=self.props["DomainName"]442        )443    @staticmethod444    def get_deploy_templates():445        return {446            "create": {447                "function": "create_domain_name",448                "parameters": lambda_keys_to_lower(),449            }450        }451    def get_physical_resource_id(self, attribute=None, **kwargs):452        return self.props.get("domainName")453# TODO: add tests for this resource type454class GatewayBasePathMapping(GenericBaseModel):455    @staticmethod456    def cloudformation_type():457        return "AWS::ApiGateway::BasePathMapping"458    def fetch_state(self, stack_name, resources):459        resources = (460            aws_stack.connect_to_service("apigateway")461            .get_base_path_mappings(domainName=self.props.get("DomainName"))462            .get("items", [])463        )464        comparable = (465            [self.props.get("BasePath")] if self.props.get("BasePath") else [None, "", "(none)"]466        )467        return next(iter([res for res in resources if res.get("basePath") in comparable]))468    @classmethod469    def get_deploy_templates(cls):470        def _create_base_path_mapping(resource_id, resources, *args, **kwargs):471            resource = cls(resources[resource_id])472            props = resource.props473            kwargs = {474                "domainName": props.get("DomainName"),475                "restApiId": props.get("RestApiId"),476                **({"basePath": props.get("BasePath")} if props.get("BasePath") else {}),477                **({"stage": props.get("Stage")} if props.get("Stage") else {}),478            }479            aws_stack.connect_to_service("apigateway").create_base_path_mapping(**kwargs)480        return {"create": {"function": _create_base_path_mapping}}481    def get_physical_resource_id(self, attribute=None, **kwargs):482        return self.props.get("id")483class GatewayModel(GenericBaseModel):484    @staticmethod485    def cloudformation_type():486        return "AWS::ApiGateway::Model"487    def fetch_state(self, stack_name, resources):488        client = aws_stack.connect_to_service("apigateway")489        api_id = self.resolve_refs_recursively(stack_name, self.props["RestApiId"], resources)490        items = client.get_models(restApiId=api_id)["items"]491        if not items:492            return None493        model_name = self.resolve_refs_recursively(stack_name, self.props["Name"], resources)494        models = [item for item in items if item["name"] == model_name]495        if models:496            return models[0]497        return None498    @staticmethod499    def get_deploy_templates():500        return {501            "create": {502                "function": "create_model",503                "parameters": {504                    "name": "Name",505                    "restApiId": "RestApiId",506                },507                "defaults": {"contentType": "application/json"},508            }509        }510class GatewayAccount(GenericBaseModel):511    @staticmethod512    def cloudformation_type():513        return "AWS::ApiGateway::Account"514    @staticmethod515    def get_deploy_templates():...

ec2.py

Source:ec2.py

...17        return (route_tables or [None])[0]18    def get_physical_resource_id(self, attribute=None, **kwargs):19        return self.physical_resource_id or self.props.get("RouteTableId")20    @staticmethod21    def get_deploy_templates():22        return {23            "create": {24                "function": "create_route_table",25                "parameters": {26                    "VpcId": "VpcId",27                    "TagSpecifications": lambda params, **kwargs: [28                        {"ResourceType": "route-table", "Tags": params.get("Tags")}29                    ],30                },31            },32            "delete": {33                "function": "delete_route_table",34                "parameters": {"RouteTableId": "RouteTableId"},35            },36        }37class EC2Route(GenericBaseModel):38    @staticmethod39    def cloudformation_type():40        return "AWS::EC2::Route"41    def fetch_state(self, stack_name, resources):42        client = aws_stack.connect_to_service("ec2")43        props = self.props44        dst_cidr = self.resolve_refs_recursively(45            stack_name, props.get("DestinationCidrBlock"), resources46        )47        dst_cidr6 = self.resolve_refs_recursively(48            stack_name, props.get("DestinationIpv6CidrBlock"), resources49        )50        table_id = self.resolve_refs_recursively(stack_name, props.get("RouteTableId"), resources)51        route_tables = client.describe_route_tables()["RouteTables"]52        route_table = ([t for t in route_tables if t["RouteTableId"] == table_id] or [None])[0]53        if route_table:54            routes = route_table.get("Routes", [])55            route = [56                r57                for r in routes58                if r.get("DestinationCidrBlock") == (dst_cidr or "_not_set_")59                or r.get("DestinationIpv6CidrBlock") == (dst_cidr6 or "_not_set_")60            ]61            return (route or [None])[0]62    def get_physical_resource_id(self, attribute=None, **kwargs):63        props = self.props64        return generate_route_id(65            props.get("RouteTableId"),66            props.get("DestinationCidrBlock"),67            props.get("DestinationIpv6CidrBlock"),68        )69    @staticmethod70    def get_deploy_templates():71        return {72            "create": {73                "function": "create_route",74                "parameters": {75                    "DestinationCidrBlock": "DestinationCidrBlock",76                    "DestinationIpv6CidrBlock": "DestinationIpv6CidrBlock",77                    "RouteTableId": "RouteTableId",78                },79            },80            "delete": {81                "function": "delete_route",82                "parameters": {83                    "DestinationCidrBlock": "DestinationCidrBlock",84                    "DestinationIpv6CidrBlock": "DestinationIpv6CidrBlock",85                    "RouteTableId": "RouteTableId",86                },87            },88        }89class EC2InternetGateway(GenericBaseModel):90    @staticmethod91    def cloudformation_type():92        return "AWS::EC2::InternetGateway"93    def fetch_state(self, stack_name, resources):94        client = aws_stack.connect_to_service("ec2")95        gateways = client.describe_internet_gateways()["InternetGateways"]96        tags = self.props.get("Tags")97        gateway = [g for g in gateways if (g.get("Tags") or []) == (tags or [])]98        return (gateway or [None])[0]99    def get_physical_resource_id(self, attribute=None, **kwargs):100        return self.props.get("InternetGatewayId")101    @staticmethod102    def get_deploy_templates():103        def _create_params(params, **kwargs):104            return {105                "TagSpecifications": [106                    {"ResourceType": "internet-gateway", "Tags": params.get("Tags", [])}107                ]108            }109        return {110            "create": {111                "function": "create_internet_gateway",112                "parameters": _create_params,113            }114        }115class EC2SubnetRouteTableAssociation(GenericBaseModel):116    @staticmethod117    def cloudformation_type():118        return "AWS::EC2::SubnetRouteTableAssociation"119    def fetch_state(self, stack_name, resources):120        client = aws_stack.connect_to_service("ec2")121        props = self.props122        table_id = self.resolve_refs_recursively(stack_name, props.get("RouteTableId"), resources)123        gw_id = self.resolve_refs_recursively(stack_name, props.get("GatewayId"), resources)124        route_tables = client.describe_route_tables()["RouteTables"]125        route_table = ([t for t in route_tables if t["RouteTableId"] == table_id] or [None])[0]126        if route_table:127            associations = route_table.get("Associations", [])128            association = [a for a in associations if a.get("GatewayId") == gw_id]129            return (association or [None])[0]130    def get_physical_resource_id(self, attribute=None, **kwargs):131        return self.props.get("RouteTableAssociationId")132    @staticmethod133    def get_deploy_templates():134        return {135            "create": {136                "function": "associate_route_table",137                "parameters": {138                    "GatewayId": "GatewayId",139                    "RouteTableId": "RouteTableId",140                    "SubnetId": "SubnetId",141                },142            }143        }144class EC2VPCGatewayAttachment(GenericBaseModel):145    @staticmethod146    def cloudformation_type():147        return "AWS::EC2::VPCGatewayAttachment"148    def fetch_state(self, stack_name, resources):149        client = aws_stack.connect_to_service("ec2")150        props = self.props151        igw_id = self.resolve_refs_recursively(152            stack_name, props.get("InternetGatewayId"), resources153        )154        vpngw_id = self.resolve_refs_recursively(stack_name, props.get("VpnGatewayId"), resources)155        gateways = []156        if igw_id:157            gateways = client.describe_internet_gateways()["InternetGateways"]158            gateways = [g for g in gateways if g["InternetGatewayId"] == igw_id]159        elif vpngw_id:160            gateways = client.describe_vpn_gateways()["VpnGateways"]161            gateways = [g for g in gateways if g["VpnGatewayId"] == vpngw_id]162        gateway = (gateways or [{}])[0]163        attachments = gateway.get("Attachments") or gateway.get("VpcAttachments") or []164        result = [a for a in attachments if a.get("State") in ("attached", "available")]165        if result:166            return gateway167    def get_physical_resource_id(self, attribute=None, **kwargs):168        props = self.props169        gw_id = props.get("VpnGatewayId") or props.get("InternetGatewayId")170        attachment = (props.get("Attachments") or props.get("VpcAttachments") or [{}])[0]171        if attachment:172            result = "%s-%s" % (gw_id, attachment.get("VpcId"))173            return result174    @classmethod175    def get_deploy_templates(cls):176        def _attach_gateway(resource_id, resources, *args, **kwargs):177            client = aws_stack.connect_to_service("ec2")178            resource = cls(resources[resource_id])179            props = resource.props180            igw_id = props.get("InternetGatewayId")181            vpngw_id = props.get("VpnGatewayId")182            vpc_id = props.get("VpcId")183            if igw_id:184                client.attach_internet_gateway(VpcId=vpc_id, InternetGatewayId=igw_id)185            elif vpngw_id:186                client.attach_vpn_gateway(VpcId=vpc_id, VpnGatewayId=vpngw_id)187        return {"create": {"function": _attach_gateway}}188class SecurityGroup(GenericBaseModel):189    @staticmethod190    def cloudformation_type():191        return "AWS::EC2::SecurityGroup"192    def fetch_state(self, stack_name, resources):193        props = self.props194        group_id = props.get("GroupId")195        group_name = props.get("GroupName")196        client = aws_stack.connect_to_service("ec2")197        if group_id:198            resp = client.describe_security_groups(GroupIds=[group_id])199        else:200            resp = client.describe_security_groups(GroupNames=[group_name])201        return (resp["SecurityGroups"] or [None])[0]202    def get_physical_resource_id(self, attribute=None, **kwargs):203        if self.physical_resource_id:204            return self.physical_resource_id205        if attribute in REF_ID_ATTRS:206            props = self.props207            return props.get("GroupId") or props.get("GroupName")208    @staticmethod209    def add_defaults(resource, stack_name: str):210        role_name = resource.get("Properties", {}).get("GroupName")211        if not role_name:212            resource["Properties"]["GroupName"] = generate_default_name(213                stack_name, resource["LogicalResourceId"]214            )215    @staticmethod216    def get_deploy_templates():217        return {218            "create": {219                "function": "create_security_group",220                "parameters": {221                    "GroupName": "GroupName",222                    "VpcId": "VpcId",223                    "Description": "GroupDescription",224                },225            },226            "delete": {227                "function": "delete_security_group",228                "parameters": {"GroupId": "PhysicalResourceId"},229            },230        }231class EC2Subnet(GenericBaseModel):232    @staticmethod233    def cloudformation_type():234        return "AWS::EC2::Subnet"235    def fetch_state(self, stack_name, resources):236        client = aws_stack.connect_to_service("ec2")237        props = self.props238        filters = [239            {"Name": "cidr-block", "Values": [props["CidrBlock"]]},240            {"Name": "vpc-id", "Values": [props["VpcId"]]},241        ]242        subnets = client.describe_subnets(Filters=filters)["Subnets"]243        return (subnets or [None])[0]244    def get_physical_resource_id(self, attribute=None, **kwargs):245        return self.props.get("SubnetId")246    @staticmethod247    def get_deploy_templates():248        return {249            "create": {250                "function": "create_subnet",251                "parameters": {252                    "VpcId": "VpcId",253                    "CidrBlock": "CidrBlock",254                    "OutpostArn": "OutpostArn",255                    "Ipv6CidrBlock": "Ipv6CidrBlock",256                    "AvailabilityZone": "AvailabilityZone"257                    # TODO: add TagSpecifications258                },259            },260            "delete": {261                "function": "delete_subnet",262                "parameters": {"SubnetId": "PhysicalResourceId"},263            },264        }265class EC2VPC(GenericBaseModel):266    @staticmethod267    def cloudformation_type():268        return "AWS::EC2::VPC"269    def fetch_state(self, stack_name, resources):270        client = aws_stack.connect_to_service("ec2")271        resp = client.describe_vpcs(Filters=[{"Name": "cidr", "Values": [self.props["CidrBlock"]]}])272        return (resp["Vpcs"] or [None])[0]273    @classmethod274    def get_deploy_templates(cls):275        def _pre_delete(resource_id, resources, resource_type, func, stack_name):276            res = cls(resources[resource_id])277            vpc_id = res.state.get("VpcId")278            if vpc_id:279                ec2_client = aws_stack.connect_to_service("ec2")280                resp = ec2_client.describe_route_tables(281                    Filters=[282                        {"Name": "vpc-id", "Values": [vpc_id]},283                        {"Name": "association.main", "Values": ["false"]},284                    ]285                )286                for rt in resp["RouteTables"]:287                    ec2_client.delete_route_table(RouteTableId=rt["RouteTableId"])288        return {289            "create": {290                "function": "create_vpc",291                "parameters": {292                    "CidrBlock": "CidrBlock",293                    "InstanceTenancy": "InstanceTenancy"294                    # TODO: add TagSpecifications295                },296            },297            "delete": [298                {"function": _pre_delete},299                {300                    "function": "delete_vpc",301                    "parameters": {"VpcId": "PhysicalResourceId"},302                },303            ],304        }305    def get_physical_resource_id(self, attribute=None, **kwargs):306        return self.physical_resource_id or self.props.get("VpcId")307class EC2NatGateway(GenericBaseModel):308    @staticmethod309    def cloudformation_type():310        return "AWS::EC2::NatGateway"311    def fetch_state(self, stack_name, resources):312        client = aws_stack.connect_to_service("ec2")313        props = self.props314        subnet_id = self.resolve_refs_recursively(stack_name, props.get("SubnetId"), resources)315        assoc_id = self.resolve_refs_recursively(stack_name, props.get("AllocationId"), resources)316        result = client.describe_nat_gateways(317            Filters=[{"Name": "subnet-id", "Values": [subnet_id]}]318        )319        result = result["NatGateways"]320        result = [321            gw322            for gw in result323            if assoc_id in [ga["AllocationId"] for ga in gw["NatGatewayAddresses"]]324        ]325        return (result or [None])[0]326    @staticmethod327    def get_deploy_templates():328        return {329            "create": {330                "function": "create_nat_gateway",331                "parameters": {332                    "SubnetId": "SubnetId",333                    "AllocationId": "AllocationId"334                    # TODO: add TagSpecifications335                },336            },337            "delete": {338                "function": "delete_nat_gateway",339                "parameters": {"NatGatewayId": "PhysicalResourceId"},340            },341        }342    def get_physical_resource_id(self, attribute=None, **kwargs):343        return self.physical_resource_id or self.props.get("NatGatewayId")344class EC2Instance(GenericBaseModel):345    @staticmethod346    def cloudformation_type():347        return "AWS::EC2::Instance"348    def fetch_state(self, stack_name, resources):349        instance_id = self.get_physical_resource_id()350        if not instance_id:351            return352        return self._get_state()353    def update_resource(self, new_resource, stack_name, resources):354        instance_id = self.get_physical_resource_id()355        props = new_resource["Properties"]356        groups = props.get("SecurityGroups", props.get("SecurityGroupIds"))357        client = aws_stack.connect_to_service("ec2")358        client.modify_instance_attribute(359            Groups=groups,360            InstanceId=instance_id,361            InstanceType={"Value": props["InstanceType"]},362        )363        return self._get_state(client)364    def _get_state(self, client=None):365        instance_id = self.get_physical_resource_id()366        client = client or aws_stack.connect_to_service("ec2")367        resp = client.describe_instances(InstanceIds=[instance_id])368        reservation = (resp.get("Reservations") or [{}])[0]369        result = (reservation.get("Instances") or [None])[0]370        return result371    def get_physical_resource_id(self, attribute=None, **kwargs):372        return self.physical_resource_id or self.props.get("InstanceId")373    def get_cfn_attribute(self, attribute_name):374        if attribute_name in REF_ID_ATTRS:375            return self.props.get("InstanceId")376        if attribute_name == "PublicIp":377            return self.props.get("PublicIpAddress") or "127.0.0.1"378        if attribute_name == "PublicDnsName":379            return self.props.get("PublicDnsName")380        if attribute_name == "AvailabilityZone":381            return (382                self.props.get("Placement", {}).get("AvailabilityZone")383                or f"{aws_stack.get_region()}a"384            )385        return super(EC2Instance, self).get_cfn_attribute(attribute_name)386    @staticmethod387    def get_deploy_templates():388        return {389            "create": {390                "function": "create_instances",391                "parameters": {392                    "InstanceType": "InstanceType",393                    "SecurityGroups": "SecurityGroups",394                    "KeyName": "KeyName",395                    "ImageId": "ImageId",396                },397                "defaults": {"MinCount": 1, "MaxCount": 1},398            },399            "delete": {400                "function": "terminate_instances",401                "parameters": {...

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.