diff --git a/moto/apigateway/models.py b/moto/apigateway/models.py index 85f3a49f8..75541664c 100644 --- a/moto/apigateway/models.py +++ b/moto/apigateway/models.py @@ -659,23 +659,45 @@ class UsagePlanKey(BaseModel, dict): class RestAPI(CloudFormationModel): + + PROP_ID = "id" + PROP_NAME = "name" + PROP_DESCRIPTON = "description" + PROP_VERSION = "version" + PROP_BINARY_MEDIA_TYPES = "binaryMediaTypes" + PROP_CREATED_DATE = "createdDate" + PROP_API_KEY_SOURCE = "apiKeySource" + PROP_ENDPOINT_CONFIGURATION = "endpointConfiguration" + PROP_TAGS = "tags" + PROP_POLICY = "policy" + PROP_DISABLE_EXECUTE_API_ENDPOINT = "disableExecuteApiEndpoint" + PROP_MINIMUM_COMPRESSION_SIZE = "minimumCompressionSize" + + # operations + OPERATION_ADD = "add" + OPERATION_REPLACE = "replace" + OPERATION_REMOVE = "remove" + OPERATION_PATH = "path" + OPERATION_VALUE = "value" + OPERATION_OP = "op" + def __init__(self, id, region_name, name, description, **kwargs): super(RestAPI, self).__init__() self.id = id 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.version = kwargs.get(RestAPI.PROP_VERSION) or "V1" + self.binaryMediaTypes = kwargs.get(RestAPI.PROP_BINARY_MEDIA_TYPES) 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 + self.policy = kwargs.get(RestAPI.PROP_POLICY) or None self.endpoint_configuration = kwargs.get("endpoint_configuration") or { "types": ["EDGE"] } - self.tags = kwargs.get("tags") or {} + self.tags = kwargs.get(RestAPI.PROP_TAGS) or {} self.disableExecuteApiEndpoint = ( - kwargs.get("disableExecuteApiEndpoint") or False + kwargs.get(RestAPI.PROP_DISABLE_EXECUTE_API_ENDPOINT) or False ) self.minimum_compression_size = kwargs.get("minimum_compression_size") self.deployments = {} @@ -690,35 +712,49 @@ class RestAPI(CloudFormationModel): def to_dict(self): return { - "id": self.id, - "name": self.name, - "description": self.description, - "version": self.version, - "binaryMediaTypes": self.binaryMediaTypes, - "createdDate": self.create_date, - "apiKeySource": self.api_key_source, - "endpointConfiguration": self.endpoint_configuration, - "tags": self.tags, - "policy": self.policy, - "disableExecuteApiEndpoint": self.disableExecuteApiEndpoint, - "minimumCompressionSize": self.minimum_compression_size, + self.PROP_ID: self.id, + self.PROP_NAME: self.name, + self.PROP_DESCRIPTON: self.description, + self.PROP_VERSION: self.version, + self.PROP_BINARY_MEDIA_TYPES: self.binaryMediaTypes, + self.PROP_CREATED_DATE: self.create_date, + self.PROP_API_KEY_SOURCE: self.api_key_source, + self.PROP_ENDPOINT_CONFIGURATION: self.endpoint_configuration, + self.PROP_TAGS: self.tags, + self.PROP_POLICY: self.policy, + self.PROP_DISABLE_EXECUTE_API_ENDPOINT: self.disableExecuteApiEndpoint, + self.PROP_MINIMUM_COMPRESSION_SIZE: self.minimum_compression_size, } def apply_patch_operations(self, patch_operations): + def to_path(prop): + return "/" + prop + for op in patch_operations: - path = op["path"] - value = op["value"] - if op["op"] == "replace": - if "/name" in path: + path = op[self.OPERATION_PATH] + value = "" + if self.OPERATION_VALUE in op: + value = op[self.OPERATION_VALUE] + operaton = op[self.OPERATION_OP] + if operaton == self.OPERATION_REPLACE: + if to_path(self.PROP_NAME) in path: self.name = value - if "/description" in path: + if to_path(self.PROP_DESCRIPTON) in path: self.description = value - if "/apiKeySource" in path: + if to_path(self.PROP_API_KEY_SOURCE) in path: self.api_key_source = value - if "/binaryMediaTypes" in path: - self.binaryMediaTypes = value - if "/disableExecuteApiEndpoint" in path: + if to_path(self.PROP_BINARY_MEDIA_TYPES) in path: + self.binaryMediaTypes = [value] + if to_path(self.PROP_DISABLE_EXECUTE_API_ENDPOINT) in path: self.disableExecuteApiEndpoint = bool(value) + elif operaton == self.OPERATION_ADD: + if to_path(self.PROP_BINARY_MEDIA_TYPES) in path: + self.binaryMediaTypes.append(value) + elif operaton == self.OPERATION_REMOVE: + if to_path(self.PROP_BINARY_MEDIA_TYPES) in path: + self.binaryMediaTypes.remove(value) + if to_path(self.PROP_DESCRIPTON) in path: + self.description = "" def get_cfn_attribute(self, attribute_name): from moto.cloudformation.exceptions import UnformattedGetAttTemplateException diff --git a/moto/apigateway/responses.py b/moto/apigateway/responses.py index 30fed63cd..e91a10c4f 100644 --- a/moto/apigateway/responses.py +++ b/moto/apigateway/responses.py @@ -111,8 +111,8 @@ class APIGatewayResponse(BaseResponse): def __validte_rest_patch_operations(self, patch_operations): for op in patch_operations: path = op["path"] - value = op["value"] if "apiKeySource" in path: + value = op["value"] return self.__validate_api_key_source(value) def restapis_individual(self, request, full_url, headers): diff --git a/moto/logs/responses.py b/moto/logs/responses.py index 57219eb2e..e1549d18a 100644 --- a/moto/logs/responses.py +++ b/moto/logs/responses.py @@ -110,15 +110,15 @@ class LogsResponse(BaseResponse): if metric_name and not metric_namespace: raise InvalidParameterException( - constraint=f"If you include the metricName parameter in your request, " - f"you must also include the metricNamespace parameter.", + constraint=f'{"If you include the metricName parameter in your request, "}' + f'{"you must also include the metricNamespace parameter."}', parameter="metricNamespace", value=metric_namespace, ) if metric_namespace and not metric_name: raise InvalidParameterException( - constraint=f"If you include the metricNamespace parameter in your request, " - f"you must also include the metricName parameter.", + constraint=f'{"If you include the metricNamespace parameter in your request, "}' + f'{"you must also include the metricName parameter."}', parameter="metricName", value=metric_name, ) diff --git a/moto/sagemaker/responses.py b/moto/sagemaker/responses.py index ebeee0279..c92527c6b 100644 --- a/moto/sagemaker/responses.py +++ b/moto/sagemaker/responses.py @@ -302,7 +302,7 @@ class SageMakerResponse(BaseResponse): errors = [] if max_results and max_results not in max_results_range: errors.append( - "Value '%s' at 'maxResults' failed to satisfy constraint: Member must have value less than or equal to %s".format( + "Value '{0}' at 'maxResults' failed to satisfy constraint: Member must have value less than or equal to {1}".format( max_results, max_results_range[-1] ) ) diff --git a/moto/transcribe/models.py b/moto/transcribe/models.py index 3cff073c8..b5abc9431 100644 --- a/moto/transcribe/models.py +++ b/moto/transcribe/models.py @@ -181,7 +181,9 @@ class FakeTranscriptionJob(BaseObject): ) if self.output_key is not None else transcript_file_uri - + "{1}.json".format(self.output_key, self.transcription_job_name) + + "{transcription_job_name}.json".format( + transcription_job_name=self.transcription_job_name + ) ) self.output_location_type = "CUSTOMER_BUCKET" else: diff --git a/tests/test_apigateway/test_apigateway.py b/tests/test_apigateway/test_apigateway.py index c4af04cb7..3d7c77a2b 100644 --- a/tests/test_apigateway/test_apigateway.py +++ b/tests/test_apigateway/test_apigateway.py @@ -43,7 +43,7 @@ def test_create_and_get_rest_api(): @mock_apigateway -def test_upate_rest_api(): +def test_update_rest_api(): client = boto3.client("apigateway", region_name="us-west-2") response = client.create_rest_api(name="my_api", description="this is my api") api_id = response["id"] @@ -87,7 +87,7 @@ def test_upate_rest_api(): @mock_apigateway -def test_upate_rest_api_invalid_api_id(): +def test_update_rest_api_invalid_api_id(): client = boto3.client("apigateway", region_name="us-west-2") patchOperations = [ {"op": "replace", "path": "/apiKeySource", "value": "AUTHORIZER"} @@ -97,6 +97,27 @@ def test_upate_rest_api_invalid_api_id(): ex.value.response["Error"]["Code"].should.equal("NotFoundException") +@mock_apigateway +def test_update_rest_api_operation_add_remove(): + client = boto3.client("apigateway", region_name="us-west-2") + response = client.create_rest_api(name="my_api", description="this is my api") + api_id = response["id"] + patchOperations = [ + {"op": "add", "path": "/binaryMediaTypes", "value": "image/png"}, + {"op": "add", "path": "/binaryMediaTypes", "value": "image/jpeg"}, + ] + response = client.update_rest_api(restApiId=api_id, patchOperations=patchOperations) + response["binaryMediaTypes"].should.equal(["image/png", "image/jpeg"]) + response["description"].should.equal("this is my api") + patchOperations = [ + {"op": "remove", "path": "/binaryMediaTypes", "value": "image/png"}, + {"op": "remove", "path": "/description"}, + ] + response = client.update_rest_api(restApiId=api_id, patchOperations=patchOperations) + response["binaryMediaTypes"].should.equal(["image/jpeg"]) + response["description"].should.equal("") + + @mock_apigateway def test_list_and_delete_apis(): client = boto3.client("apigateway", region_name="us-west-2")