catalin/moto
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

basic implementation of update rest api (#3951)

Browse source 
* basic implementation of update rest api

* basic implementation of update rest api

* basic implementation of update rest api

* review comments from bblommers

Co-authored-by: rajinder saini <rajinder.saini@c02vt5k2htd6.corp.climate.com>
This commit is contained in:
rajinder 2021-05-23 09:09:02 -07:00 • committed by GitHub
commit fbbc8fc472
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 148 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

32
moto/apigateway/models.py
View file

@ -595,6 +595,8 @@ class RestAPI(CloudFormationModel):
self.region_name = region_name
self.name = name
self.description = description
self.version = kwargs.get("version") or "V1"
self.binaryMediaTypes = kwargs.get("binaryMediaTypes") or []
self.create_date = int(time.time())
self.api_key_source = kwargs.get("api_key_source") or "HEADER"
self.policy = kwargs.get("policy") or None
@ -602,7 +604,9 @@ class RestAPI(CloudFormationModel):
"types": ["EDGE"]
}
self.tags = kwargs.get("tags") or {}
self.disableExecuteApiEndpoint = (
kwargs.get("disableExecuteApiEndpoint") or False
)
self.deployments = {}
self.authorizers = {}
self.stages = {}
@ -618,13 +622,32 @@ class RestAPI(CloudFormationModel):
"id": self.id,
"name": self.name,
"description": self.description,
"version": self.version,
"binaryMediaTypes": self.binaryMediaTypes,
"createdDate": int(time.time()),
"apiKeySource": self.api_key_source,
"endpointConfiguration": self.endpoint_configuration,
"tags": self.tags,
"policy": self.policy,
"disableExecuteApiEndpoint": self.disableExecuteApiEndpoint,
}
def apply_patch_operations(self, patch_operations):
for op in patch_operations:
path = op["path"]
value = op["value"]
if op["op"] == "replace":
if "/name" in path:
self.name = value
if "/description" in path:
self.description = value
if "/apiKeySource" in path:
self.api_key_source = value
if "/binaryMediaTypes" in path:
self.binaryMediaTypes = value
if "/disableExecuteApiEndpoint" in path:
self.disableExecuteApiEndpoint = bool(value)
def get_cfn_attribute(self, attribute_name):
from moto.cloudformation.exceptions import UnformattedGetAttTemplateException
@ -905,6 +928,13 @@ class APIGatewayBackend(BaseBackend):
raise RestAPINotFound()
return rest_api
def update_rest_api(self, function_id, patch_operations):
rest_api = self.apis.get(function_id)
if rest_api is None:
raise RestAPINotFound()
self.apis[function_id].apply_patch_operations(patch_operations)
return self.apis[function_id]
def list_apis(self):
return self.apis.values()

85
moto/apigateway/responses.py
View file

@ -39,6 +39,34 @@ class APIGatewayResponse(BaseResponse):
def backend(self):
return apigateway_backends[self.region]
def __validate_api_key_source(self, api_key_source):
if api_key_source and api_key_source not in API_KEY_SOURCES:
return self.error(
"ValidationException",
(
"1 validation error detected: "
"Value '{api_key_source}' at 'createRestApiInput.apiKeySource' failed "
"to satisfy constraint: Member must satisfy enum value set: "
"[AUTHORIZER, HEADER]"
).format(api_key_source=api_key_source),
)
def __validate_endpoint_configuration(self, endpoint_configuration):
if endpoint_configuration and "types" in endpoint_configuration:
invalid_types = list(
set(endpoint_configuration["types"]) - set(ENDPOINT_CONFIGURATION_TYPES)
)
if invalid_types:
return self.error(
"ValidationException",
(
"1 validation error detected: Value '{endpoint_type}' "
"at 'createRestApiInput.endpointConfiguration.types' failed "
"to satisfy constraint: Member must satisfy enum value set: "
"[PRIVATE, EDGE, REGIONAL]"
).format(endpoint_type=invalid_types[0]),
)
def restapis(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
@ -54,32 +82,13 @@ class APIGatewayResponse(BaseResponse):
policy = self._get_param("policy")
# Param validation
if api_key_source and api_key_source not in API_KEY_SOURCES:
return self.error(
"ValidationException",
(
"1 validation error detected: "
"Value '{api_key_source}' at 'createRestApiInput.apiKeySource' failed "
"to satisfy constraint: Member must satisfy enum value set: "
"[AUTHORIZER, HEADER]"
).format(api_key_source=api_key_source),
)
response = self.__validate_api_key_source(api_key_source)
if response is not None:
return response
if endpoint_configuration and "types" in endpoint_configuration:
invalid_types = list(
set(endpoint_configuration["types"])
- set(ENDPOINT_CONFIGURATION_TYPES)
)
if invalid_types:
return self.error(
"ValidationException",
(
"1 validation error detected: Value '{endpoint_type}' "
"at 'createRestApiInput.endpointConfiguration.types' failed "
"to satisfy constraint: Member must satisfy enum value set: "
"[PRIVATE, EDGE, REGIONAL]"
).format(endpoint_type=invalid_types[0]),
)
response = self.__validate_endpoint_configuration(endpoint_configuration)
if response is not None:
return response
rest_api = self.backend.create_rest_api(
name,
@ -91,16 +100,38 @@ class APIGatewayResponse(BaseResponse):
)
return 200, {}, json.dumps(rest_api.to_dict())
def __validte_rest_patch_operations(self, patch_operations):
for op in patch_operations:
path = op["path"]
value = op["value"]
if "apiKeySource" in path:
return self.__validate_api_key_source(value)
def restapis_individual(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
function_id = self.path.replace("/restapis/", "", 1).split("/")[0]
if self.method == "GET":
rest_api = self.backend.get_rest_api(function_id)
return 200, {}, json.dumps(rest_api.to_dict())
elif self.method == "DELETE":
rest_api = self.backend.delete_rest_api(function_id)
return 200, {}, json.dumps(rest_api.to_dict())
elif self.method == "PATCH":
patch_operations = self._get_param("patchOperations")
response = self.__validte_rest_patch_operations(patch_operations)
if response is not None:
return response
try:
rest_api = self.backend.update_rest_api(function_id, patch_operations)
except RestAPINotFound as error:
return (
error.code,
{},
'{{"message":"{0}","code":"{1}"}}'.format(
error.message, error.error_type
),
)
return 200, {}, json.dumps(rest_api.to_dict())
def resources(self, request, full_url, headers):
self.setup_class(request, full_url, headers)