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

How to use _patch_api_gateway_entity method in localstack

Best Python code snippet using localstack_python

apigateway_starter.py

Source:apigateway_starter.py

...79            self["requestTemplates"] = request_templates80    def apigateway_models_backend_put_rest_api(self, function_id, body, query_params):81        rest_api = self.get_rest_api(function_id)82        return import_api_from_openapi_spec(rest_api, function_id, body, query_params)83    def _patch_api_gateway_entity(self, entity: Dict) -> Optional[Tuple[int, Dict, str]]:84        not_supported_attributes = ["/id", "/region_name", "/create_date"]85        patch_operations = self._get_param("patchOperations")86        model_attributes = list(entity.keys())87        for operation in patch_operations:88            if operation["path"].strip("/") in REST_API_ATTRIBUTES:89                operation["path"] = camelcase_to_underscores(operation["path"])90            path_start = operation["path"].strip("/").split("/")[0]91            path_start_usc = camelcase_to_underscores(path_start)92            if path_start not in model_attributes and path_start_usc in model_attributes:93                operation["path"] = operation["path"].replace(path_start, path_start_usc)94            if operation["path"] in not_supported_attributes:95                msg = "Invalid patch path %s" % (operation["path"])96                return 400, {}, msg97        apply_json_patch_safe(entity, patch_operations, in_place=True)98    def apigateway_response_restapis_individual(self, request, full_url, headers):99        if request.method in ["GET", "DELETE"]:100            return apigateway_response_restapis_individual_orig(self, request, full_url, headers)101        self.setup_class(request, full_url, headers)102        function_id = self.path.replace("/restapis/", "", 1).split("/")[0]103        if self.method == "PATCH":104            rest_api = self.backend.apis.get(function_id)105            if not rest_api:106                msg = "Invalid API identifier specified %s:%s" % (107                    TEST_AWS_ACCOUNT_ID,108                    function_id,109                )110                return 404, {}, msg111            if not isinstance(rest_api.__dict__, DelSafeDict):112                rest_api.__dict__ = DelSafeDict(rest_api.__dict__)113            result = _patch_api_gateway_entity(self, rest_api.__dict__)114            if result is not None:115                return result116            # fix data types after patches have been applied117            rest_api.minimum_compression_size = int(rest_api.minimum_compression_size or -1)118            endpoint_configs = rest_api.endpoint_configuration or {}119            if isinstance(endpoint_configs.get("vpcEndpointIds"), str):120                endpoint_configs["vpcEndpointIds"] = [endpoint_configs["vpcEndpointIds"]]121            return 200, {}, json.dumps(self.backend.get_rest_api(function_id).to_dict())122        # handle import rest_api via swagger file123        if self.method == "PUT":124            body = json.loads(to_str(self.body))125            rest_api = self.backend.put_rest_api(function_id, body, self.querystring)126            return 200, {}, json.dumps(rest_api.to_dict())127        return 400, {}, ""128    def apigateway_response_resource_individual(self, request, full_url, headers):129        if request.method in ["GET", "POST", "DELETE"]:130            return apigateway_response_resource_individual_orig(self, request, full_url, headers)131        self.setup_class(request, full_url, headers)132        function_id = self.path.replace("/restapis/", "", 1).split("/")[0]133        if self.method == "PATCH":134            resource_id = self.path.split("/")[4]135            resource = self.backend.get_resource(function_id, resource_id)136            if not isinstance(resource.__dict__, DelSafeDict):137                resource.__dict__ = DelSafeDict(resource.__dict__)138            result = _patch_api_gateway_entity(self, resource.__dict__)139            if result is not None:140                return result141            return 200, {}, json.dumps(resource.to_dict())142        return 404, {}, ""143    def apigateway_response_resource_methods(self, request, *args, **kwargs):144        result = apigateway_response_resource_methods_orig(self, request, *args, **kwargs)145        if self.method == "PUT" and self._get_param("requestParameters"):146            request_parameters = self._get_param("requestParameters")147            url_path_parts = self.path.split("/")148            function_id = url_path_parts[2]149            resource_id = url_path_parts[4]150            method_type = url_path_parts[6]151            resource = self.backend.get_resource(function_id, resource_id)152            resource.resource_methods[method_type]["requestParameters"] = request_parameters...

patches.py

Source:patches.py

...42        self, function_id: str, body: Dict, query_params: Dict43    ):44        rest_api = self.get_rest_api(function_id)45        return import_api_from_openapi_spec(rest_api, body, query_params)46    def _patch_api_gateway_entity(self, entity: Dict) -> Optional[Tuple[int, Dict, str]]:47        not_supported_attributes = ["/id", "/region_name", "/create_date"]48        patch_operations = self._get_param("patchOperations")49        model_attributes = list(entity.keys())50        for operation in patch_operations:51            if operation["path"].strip("/") in REST_API_ATTRIBUTES:52                operation["path"] = camelcase_to_underscores(operation["path"])53            path_start = operation["path"].strip("/").split("/")[0]54            path_start_usc = camelcase_to_underscores(path_start)55            if path_start not in model_attributes and path_start_usc in model_attributes:56                operation["path"] = operation["path"].replace(path_start, path_start_usc)57            if operation["path"] in not_supported_attributes:58                msg = f'Invalid patch path {operation["path"]}'59                return 400, {}, msg60        apply_json_patch_safe(entity, patch_operations, in_place=True)61        # apply some type fixes - TODO refactor/generalize62        if "disable_execute_api_endpoint" in entity:63            entity["disableExecuteApiEndpoint"] = bool(entity.pop("disable_execute_api_endpoint"))64        if "binary_media_types" in entity:65            entity["binaryMediaTypes"] = ensure_list(entity.pop("binary_media_types"))66    def apigateway_response_restapis_individual(self, request, full_url, headers):67        if request.method in ["GET", "DELETE"]:68            return apigateway_response_restapis_individual_orig(self, request, full_url, headers)69        self.setup_class(request, full_url, headers)70        function_id = self.path.replace("/restapis/", "", 1).split("/")[0]71        if self.method == "PATCH":72            rest_api = self.backend.apis.get(function_id)73            if not rest_api:74                msg = "Invalid API identifier specified %s:%s" % (75                    get_aws_account_id(),76                    function_id,77                )78                raise NotFoundException(msg)79            if not isinstance(rest_api.__dict__, DelSafeDict):80                rest_api.__dict__ = DelSafeDict(rest_api.__dict__)81            result = _patch_api_gateway_entity(self, rest_api.__dict__)82            if result is not None:83                return result84            # fix data types after patches have been applied85            rest_api.minimum_compression_size = int(rest_api.minimum_compression_size or -1)86            endpoint_configs = rest_api.endpoint_configuration or {}87            if isinstance(endpoint_configs.get("vpcEndpointIds"), str):88                endpoint_configs["vpcEndpointIds"] = [endpoint_configs["vpcEndpointIds"]]89            return 200, {}, json.dumps(self.backend.get_rest_api(function_id).to_dict())90        # handle import rest_api via swagger file91        if self.method == "PUT":92            body = parse_json_or_yaml(to_str(self.body))93            rest_api = self.backend.put_rest_api(function_id, body, self.querystring)94            return 200, {}, json.dumps(rest_api.to_dict())95        return 400, {}, ""96    def apigateway_response_resource_individual(self, request, full_url, headers):97        if request.method in ["GET", "POST", "DELETE"]:98            return apigateway_response_resource_individual_orig(self, request, full_url, headers)99        self.setup_class(request, full_url, headers)100        function_id = self.path.replace("/restapis/", "", 1).split("/")[0]101        if self.method == "PATCH":102            resource_id = self.path.split("/")[4]103            resource = self.backend.get_resource(function_id, resource_id)104            if not isinstance(resource.__dict__, DelSafeDict):105                resource.__dict__ = DelSafeDict(resource.__dict__)106            result = _patch_api_gateway_entity(self, resource.__dict__)107            if result is not None:108                return result109            return 200, {}, json.dumps(resource.to_dict())110        return 404, {}, ""111    def apigateway_response_resource_methods(self, request, *args, **kwargs):112        result = apigateway_response_resource_methods_orig(self, request, *args, **kwargs)113        if self.method == "PUT" and self._get_param("requestParameters"):114            request_parameters = self._get_param("requestParameters")115            url_path_parts = self.path.split("/")116            function_id = url_path_parts[2]117            resource_id = url_path_parts[4]118            method_type = url_path_parts[6]119            resource = self.backend.get_resource(function_id, resource_id)120            resource.resource_methods[method_type]["requestParameters"] = request_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.