KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
