Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use apply_json_patch_safe method in localstack
Best Python code snippet using localstack_python
provider.py
Source:provider.py 
...167 region_details = APIGatewayRegion.get()168 authorizer = find_api_subentity_by_id(rest_api_id, authorizer_id, "authorizers")169 if authorizer is None:170 raise NotFoundException(f"Authorizer not found: {authorizer_id}")171 result = apply_json_patch_safe(authorizer, patch_operations)172 result = normalize_authorizer(result)173 auth_list = region_details.authorizers[rest_api_id]174 for i in range(len(auth_list)):175 if auth_list[i]["id"] == authorizer_id:176 auth_list[i] = result177 result = to_authorizer_response_json(rest_api_id, result)178 return Authorizer(**result)179 # accounts180 def get_account(181 self,182 context: RequestContext,183 ) -> Account:184 region_details = APIGatewayRegion.get()185 result = to_account_response_json(region_details.account)186 return Account(**result)187 def update_account(188 self, context: RequestContext, patch_operations: ListOfPatchOperation = None189 ) -> Account:190 region_details = APIGatewayRegion.get()191 apply_json_patch_safe(region_details.account, patch_operations, in_place=True)192 result = to_account_response_json(region_details.account)193 return Account(**result)194 # documentation parts195 def get_documentation_parts(196 self, context: RequestContext, request: GetDocumentationPartsRequest197 ) -> DocumentationParts:198 region_details = APIGatewayRegion.get()199 # This function returns either a list or a single entity (depending on the path)200 api_id = request["restApiId"]201 auth_list = region_details.documentation_parts.get(api_id) or []202 result = [to_documentation_part_response_json(api_id, a) for a in auth_list]203 result = {"item": result}204 return result205 def get_documentation_part(206 self, context: RequestContext, rest_api_id: String, documentation_part_id: String207 ) -> DocumentationPart:208 entity = find_api_subentity_by_id(rest_api_id, documentation_part_id, "documentation_parts")209 if entity is None:210 raise NotFoundException(f"Documentation part not found: {documentation_part_id}")211 return to_documentation_part_response_json(rest_api_id, entity)212 def create_documentation_part(213 self,214 context: RequestContext,215 rest_api_id: String,216 location: DocumentationPartLocation,217 properties: String,218 ) -> DocumentationPart:219 region_details = APIGatewayRegion.get()220 entity_id = short_uid()[:6] # length 6 for AWS parity / Terraform compatibility221 entry = {222 "id": entity_id,223 "restApiId": rest_api_id,224 "location": location,225 "properties": properties,226 }227 region_details.documentation_parts.setdefault(rest_api_id, []).append(entry)228 result = to_documentation_part_response_json(rest_api_id, entry)229 return DocumentationPart(**result)230 def update_documentation_part(231 self,232 context: RequestContext,233 rest_api_id: String,234 documentation_part_id: String,235 patch_operations: ListOfPatchOperation = None,236 ) -> DocumentationPart:237 region_details = APIGatewayRegion.get()238 entity = find_api_subentity_by_id(rest_api_id, documentation_part_id, "documentation_parts")239 if entity is None:240 raise NotFoundException(f"Documentation part not found: {documentation_part_id}")241 result = apply_json_patch_safe(entity, patch_operations)242 auth_list = region_details.documentation_parts[rest_api_id]243 for i in range(len(auth_list)):244 if auth_list[i]["id"] == documentation_part_id:245 auth_list[i] = result246 result = to_documentation_part_response_json(rest_api_id, result)247 return DocumentationPart(**result)248 def delete_documentation_part(249 self, context: RequestContext, rest_api_id: String, documentation_part_id: String250 ) -> None:251 region_details = APIGatewayRegion.get()252 auth_list = region_details.documentation_parts[rest_api_id]253 for i in range(len(auth_list)):254 if auth_list[i]["id"] == documentation_part_id:255 del auth_list[i]256 break257 # base path mappings258 def get_base_path_mappings(259 self,260 context: RequestContext,261 domain_name: String,262 position: String = None,263 limit: NullableInteger = None,264 ) -> BasePathMappings:265 region_details = APIGatewayRegion.get()266 mappings_list = region_details.base_path_mappings.get(domain_name) or []267 result = [268 to_base_mapping_response_json(domain_name, m["basePath"], m) for m in mappings_list269 ]270 return BasePathMappings(items=result)271 def get_base_path_mapping(272 self, context: RequestContext, domain_name: String, base_path: String273 ) -> BasePathMapping:274 region_details = APIGatewayRegion.get()275 mappings_list = region_details.base_path_mappings.get(domain_name) or []276 mapping = ([m for m in mappings_list if m["basePath"] == base_path] or [None])[0]277 if mapping is None:278 raise NotFoundException(f"Base path mapping not found: {domain_name} - {base_path}")279 result = to_base_mapping_response_json(domain_name, base_path, mapping)280 return BasePathMapping(**result)281 def create_base_path_mapping(282 self,283 context: RequestContext,284 domain_name: String,285 rest_api_id: String,286 base_path: String = None,287 stage: String = None,288 ) -> BasePathMapping:289 region_details = APIGatewayRegion.get()290 # Note: "(none)" is a special value in API GW:291 # https://docs.aws.amazon.com/apigateway/api-reference/link-relation/basepathmapping-by-base-path292 base_path = base_path or "(none)"293 entry = {294 "domainName": domain_name,295 "restApiId": rest_api_id,296 "basePath": base_path,297 "stage": stage,298 }299 region_details.base_path_mappings.setdefault(domain_name, []).append(entry)300 result = to_base_mapping_response_json(domain_name, base_path, entry)301 return BasePathMapping(**result)302 def update_base_path_mapping(303 self,304 context: RequestContext,305 domain_name: String,306 base_path: String,307 patch_operations: ListOfPatchOperation = None,308 ) -> BasePathMapping:309 region_details = APIGatewayRegion.get()310 mappings_list = region_details.base_path_mappings.get(domain_name) or []311 mapping = ([m for m in mappings_list if m["basePath"] == base_path] or [None])[0]312 if mapping is None:313 raise NotFoundException(314 f"Not found: mapping for domain name {domain_name}, "315 f"base path {base_path} in list {mappings_list}"316 )317 patch_operations = ensure_list(patch_operations)318 for operation in patch_operations:319 if operation["path"] == "/restapiId":320 operation["path"] = "/restApiId"321 result = apply_json_patch_safe(mapping, patch_operations)322 for i in range(len(mappings_list)):323 if mappings_list[i]["basePath"] == base_path:324 mappings_list[i] = result325 result = to_base_mapping_response_json(domain_name, base_path, result)326 return BasePathMapping(**result)327 def delete_base_path_mapping(328 self, context: RequestContext, domain_name: String, base_path: String329 ) -> None:330 region_details = APIGatewayRegion.get()331 mappings_list = region_details.base_path_mappings.get(domain_name) or []332 for i in range(len(mappings_list)):333 if mappings_list[i]["basePath"] == base_path:334 del mappings_list[i]335 return336 raise NotFoundException(f"Base path mapping {base_path} for domain {domain_name} not found")337 # client certificates338 def get_client_certificate(339 self, context: RequestContext, client_certificate_id: String340 ) -> ClientCertificate:341 region_details = APIGatewayRegion.get()342 result = region_details.client_certificates.get(client_certificate_id)343 if result is None:344 raise NotFoundException(f"Client certificate ID {client_certificate_id} not found")345 return ClientCertificate(**result)346 def get_client_certificates(347 self, context: RequestContext, position: String = None, limit: NullableInteger = None348 ) -> ClientCertificates:349 region_details = APIGatewayRegion.get()350 result = list(region_details.client_certificates.values())351 return ClientCertificates(items=result)352 def generate_client_certificate(353 self, context: RequestContext, description: String = None, tags: MapOfStringToString = None354 ) -> ClientCertificate:355 region_details = APIGatewayRegion.get()356 cert_id = short_uid()357 creation_time = now_utc()358 entry = {359 "description": description,360 "tags": tags,361 "clientCertificateId": cert_id,362 "createdDate": creation_time,363 "expirationDate": creation_time + 60 * 60 * 24 * 30, # assume 30 days validity364 "pemEncodedCertificate": "testcert-123", # TODO return proper certificate!365 }366 region_details.client_certificates[cert_id] = entry367 result = to_client_cert_response_json(entry)368 return ClientCertificate(**result)369 def update_client_certificate(370 self,371 context: RequestContext,372 client_certificate_id: String,373 patch_operations: ListOfPatchOperation = None,374 ) -> ClientCertificate:375 region_details = APIGatewayRegion.get()376 entity = region_details.client_certificates.get(client_certificate_id)377 if entity is None:378 raise NotFoundException(f'Client certificate ID "{client_certificate_id}" not found')379 result = apply_json_patch_safe(entity, patch_operations)380 result = to_client_cert_response_json(result)381 return ClientCertificate(**result)382 def delete_client_certificate(383 self, context: RequestContext, client_certificate_id: String384 ) -> None:385 region_details = APIGatewayRegion.get()386 entity = region_details.client_certificates.pop(client_certificate_id, None)387 if entity is None:388 raise NotFoundException(f'VPC link ID "{client_certificate_id}" not found for deletion')389 # VPC links390 def create_vpc_link(391 self,392 context: RequestContext,393 name: String,394 target_arns: ListOfString,395 description: String = None,396 tags: MapOfStringToString = None,397 ) -> VpcLink:398 region_details = APIGatewayRegion.get()399 link_id = short_uid()400 entry = {"id": link_id, "status": "AVAILABLE"}401 region_details.vpc_links[link_id] = entry402 result = to_vpc_link_response_json(entry)403 return VpcLink(**result)404 def get_vpc_links(405 self, context: RequestContext, position: String = None, limit: NullableInteger = None406 ) -> VpcLinks:407 region_details = APIGatewayRegion.get()408 result = region_details.vpc_links.values()409 result = [to_vpc_link_response_json(r) for r in result]410 result = {"items": result}411 return result412 def get_vpc_link(self, context: RequestContext, vpc_link_id: String) -> VpcLink:413 region_details = APIGatewayRegion.get()414 vpc_link = region_details.vpc_links.get(vpc_link_id)415 if vpc_link is None:416 raise NotFoundException(f'VPC link ID "{vpc_link_id}" not found')417 result = to_vpc_link_response_json(vpc_link)418 return VpcLink(**result)419 def update_vpc_link(420 self,421 context: RequestContext,422 vpc_link_id: String,423 patch_operations: ListOfPatchOperation = None,424 ) -> VpcLink:425 region_details = APIGatewayRegion.get()426 vpc_link = region_details.vpc_links.get(vpc_link_id)427 if vpc_link is None:428 raise NotFoundException(f'VPC link ID "{vpc_link_id}" not found')429 result = apply_json_patch_safe(vpc_link, patch_operations)430 result = to_vpc_link_response_json(result)431 return VpcLink(**result)432 def delete_vpc_link(self, context: RequestContext, vpc_link_id: String) -> None:433 region_details = APIGatewayRegion.get()434 vpc_link = region_details.vpc_links.pop(vpc_link_id, None)435 if vpc_link is None:436 raise NotFoundException(f'VPC link ID "{vpc_link_id}" not found for deletion')437 # request validators438 def get_request_validators(439 self,440 context: RequestContext,441 rest_api_id: String,442 position: String = None,443 limit: NullableInteger = None,444 ) -> RequestValidators:445 region_details = APIGatewayRegion.get()446 auth_list = region_details.validators.get(rest_api_id) or []447 result = [to_validator_response_json(rest_api_id, a) for a in auth_list]448 return RequestValidators(items=result)449 def get_request_validator(450 self, context: RequestContext, rest_api_id: String, request_validator_id: String451 ) -> RequestValidator:452 region_details = APIGatewayRegion.get()453 auth_list = region_details.validators.get(rest_api_id) or []454 validator = ([a for a in auth_list if a["id"] == request_validator_id] or [None])[0]455 if validator is None:456 raise NotFoundException(457 f"Validator {request_validator_id} for API Gateway {rest_api_id} not found"458 )459 result = to_validator_response_json(rest_api_id, validator)460 return RequestValidator(**result)461 def create_request_validator(462 self,463 context: RequestContext,464 rest_api_id: String,465 name: String = None,466 validate_request_body: Boolean = None,467 validate_request_parameters: Boolean = None,468 ) -> RequestValidator:469 region_details = APIGatewayRegion.get()470 # length 6 for AWS parity and TF compatibility471 validator_id = short_uid()[:6]472 entry = {473 "id": validator_id,474 "name": name,475 "restApiId": rest_api_id,476 "validateRequestBody": validate_request_body,477 "validateRequestPparameters": validate_request_parameters,478 }479 region_details.validators.setdefault(rest_api_id, []).append(entry)480 return RequestValidator(**entry)481 def update_request_validator(482 self,483 context: RequestContext,484 rest_api_id: String,485 request_validator_id: String,486 patch_operations: ListOfPatchOperation = None,487 ) -> RequestValidator:488 region_details = APIGatewayRegion.get()489 auth_list = region_details.validators.get(rest_api_id) or []490 validator = ([a for a in auth_list if a["id"] == request_validator_id] or [None])[0]491 if validator is None:492 raise NotFoundException(493 f"Validator {request_validator_id} for API Gateway {rest_api_id} not found"494 )495 result = apply_json_patch_safe(validator, patch_operations)496 entry_list = region_details.validators[rest_api_id]497 for i in range(len(entry_list)):498 if entry_list[i]["id"] == request_validator_id:499 entry_list[i] = result500 result = to_validator_response_json(rest_api_id, result)501 return RequestValidator(**result)502 def delete_request_validator(503 self, context: RequestContext, rest_api_id: String, request_validator_id: String504 ) -> None:505 region_details = APIGatewayRegion.get()506 auth_list = region_details.validators.get(rest_api_id, [])507 for i in range(len(auth_list)):508 if auth_list[i]["id"] == request_validator_id:509 del auth_list[i]...
test_apigateway.py
Source:test_apigateway.py
1import unittest2from localstack.services.apigateway import apigateway_listener3from localstack.services.apigateway.helpers import apply_json_patch_safe4from localstack.utils.aws import templating5from localstack.utils.common import clone6class ApiGatewayPathsTest(unittest.TestCase):7 def test_extract_query_params(self):8 path, query_params = apigateway_listener.extract_query_string_params(9 "/foo/bar?foo=foo&bar=bar&bar=baz"10 )11 self.assertEqual("/foo/bar", path)12 self.assertEqual({"foo": "foo", "bar": ["bar", "baz"]}, query_params)13 def test_extract_path_params(self):14 params = apigateway_listener.extract_path_params("/foo/bar", "/foo/{param1}")15 self.assertEqual({"param1": "bar"}, params)16 params = apigateway_listener.extract_path_params("/foo/bar1/bar2", "/foo/{param1}/{param2}")17 self.assertEqual({"param1": "bar1", "param2": "bar2"}, params)18 params = apigateway_listener.extract_path_params("/foo/bar", "/foo/bar")19 self.assertEqual({}, params)20 params = apigateway_listener.extract_path_params("/foo/bar/baz", "/foo/{proxy+}")21 self.assertEqual({"proxy+": "bar/baz"}, params)22 def test_path_matches(self):23 path, details = apigateway_listener.get_resource_for_path("/foo/bar", {"/foo/{param1}": {}})24 self.assertEqual("/foo/{param1}", path)25 path, details = apigateway_listener.get_resource_for_path(26 "/foo/bar", {"/foo/bar": {}, "/foo/{param1}": {}}27 )28 self.assertEqual("/foo/bar", path)29 path, details = apigateway_listener.get_resource_for_path(30 "/foo/bar/baz", {"/foo/bar": {}, "/foo/{proxy+}": {}}31 )32 self.assertEqual("/foo/{proxy+}", path)33 result = apigateway_listener.get_resource_for_path(34 "/foo/bar", {"/foo/bar1": {}, "/foo/bar2": {}}35 )36 self.assertEqual(None, result)37 result = apigateway_listener.get_resource_for_path(38 "/foo/bar", {"/{param1}/bar1": {}, "/foo/bar2": {}}39 )40 self.assertEqual(None, result)41 path_args = {"/{param1}/{param2}/foo/{param3}": {}, "/{param}/bar": {}}42 path, details = apigateway_listener.get_resource_for_path("/foo/bar", path_args)43 self.assertEqual("/{param}/bar", path)44 path_args = {"/{param1}/{param2}": {}, "/{param}/bar": {}}45 path, details = apigateway_listener.get_resource_for_path("/foo/bar", path_args)46 self.assertEqual("/{param}/bar", path)47 path_args = {"/{param1}/{param2}": {}, "/{param1}/bar": {}}48 path, details = apigateway_listener.get_resource_for_path("/foo/baz", path_args)49 self.assertEqual("/{param1}/{param2}", path)50 path_args = {"/{param1}/{param2}/baz": {}, "/{param1}/bar/{param2}": {}}51 path, details = apigateway_listener.get_resource_for_path("/foo/bar/baz", path_args)52 self.assertEqual("/{param1}/{param2}/baz", path)53 path_args = {"/{param1}/{param2}/baz": {}, "/{param1}/{param2}/{param2}": {}}54 path, details = apigateway_listener.get_resource_for_path("/foo/bar/baz", path_args)55 self.assertEqual("/{param1}/{param2}/baz", path)56 path_args = {"/foo123/{param1}/baz": {}}57 result = apigateway_listener.get_resource_for_path("/foo/bar/baz", path_args)58 self.assertEqual(None, result)59class TestVelocityUtil(unittest.TestCase):60 def test_render_template_values(self):61 util = templating.VelocityUtil()62 encoded = util.urlEncode("x=a+b")63 self.assertEqual("x%3Da%2Bb", encoded)64 decoded = util.urlDecode("x=a+b")65 self.assertEqual("x=a b", decoded)66 escaped = util.escapeJavaScript("it's")67 self.assertEqual(r"it\'s", escaped)68class TestJSONPatch(unittest.TestCase):69 def test_apply_json_patch(self):70 apply = apply_json_patch_safe71 # test replacing array index72 subject = {"root": [{"arr": ["1", "abc"]}]}73 result = apply(clone(subject), {"op": "replace", "path": "/root/0/arr/0", "value": 2})74 self.assertEqual({"arr": [2, "abc"]}, result["root"][0])75 # test replacing endpoint config type76 operation = {"op": "replace", "path": "/endpointConfiguration/types/0", "value": "EDGE"}77 subject = {78 "id": "b5d563g3yx",79 "endpointConfiguration": {"types": ["REGIONAL"], "vpcEndpointIds": []},80 }81 result = apply(clone(subject), operation)82 self.assertEqual(["EDGE"], result["endpointConfiguration"]["types"])83 # test replacing endpoint config type84 operation = {"op": "add", "path": "/features/-", "value": "feat2"}85 subject = {"features": ["feat1"]}86 result = apply(clone(subject), operation)...
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?
