Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

TechDebt: Enable MyPy on APIGateway-module (#5549)

Browse Source
This commit is contained in:
Bert Blommers 2022-10-11 13:16:27 +00:00 committed by GitHub
parent d60aa64abb
commit 56ca48cfdd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 718 additions and 622 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
IMPLEMENTATION_COVERAGE.md
View File

@ -49,7 +49,7 @@
## apigateway
<details>
<summary>65% implemented</summary>
<summary>62% implemented</summary>
- [X] create_api_key
- [X] create_authorizer
@ -157,12 +157,12 @@
- [ ] update_deployment
- [ ] update_documentation_part
- [ ] update_documentation_version
- [X] update_domain_name
- [ ] update_domain_name
- [ ] update_gateway_response
- [ ] update_integration
- [ ] update_integration_response
- [X] update_method
- [X] update_method_response
- [ ] update_method
- [ ] update_method_response
- [ ] update_model
- [X] update_request_validator
- [ ] update_resource

2
Makefile
View File

@ -28,7 +28,7 @@ lint:
@echo "Running pylint..."
pylint -j 0 moto tests
@echo "Running MyPy..."
mypy --install-types --non-interactive moto/acm moto/amp moto/applicationautoscaling/
mypy --install-types --non-interactive
format:
black moto/ tests/

6
docs/docs/services/apigateway.rst
View File

@ -149,12 +149,12 @@ apigateway
- [ ] update_deployment
- [ ] update_documentation_part
- [ ] update_documentation_version
- [X] update_domain_name
- [ ] update_domain_name
- [ ] update_gateway_response
- [ ] update_integration
- [ ] update_integration_response
- [X] update_method
- [X] update_method_response
- [ ] update_method
- [ ] update_method_response
- [ ] update_model
- [X] update_request_validator
- [ ] update_resource

87
moto/apigateway/exceptions.py
View File

@ -1,4 +1,5 @@
from moto.core.exceptions import JsonRESTError
from typing import Any
class ApiGatewayException(JsonRESTError):
@ -6,12 +7,12 @@ class ApiGatewayException(JsonRESTError):
class BadRequestException(ApiGatewayException):
def __init__(self, message):
def __init__(self, message: str):
super().__init__("BadRequestException", message)
class NotFoundException(ApiGatewayException):
def __init__(self, message):
def __init__(self, message: str):
super().__init__("NotFoundException", message)
@ -22,199 +23,199 @@ class AccessDeniedException(ApiGatewayException):
class ConflictException(ApiGatewayException):
code = 409
def __init__(self, message):
def __init__(self, message: str):
super().__init__("ConflictException", message)
class AwsProxyNotAllowed(BadRequestException):
def __init__(self):
def __init__(self) -> None:
super().__init__(
"Integrations of type 'AWS_PROXY' currently only supports Lambda function and Firehose stream invocations."
)
class CrossAccountNotAllowed(AccessDeniedException):
def __init__(self):
def __init__(self) -> None:
super().__init__(
"AccessDeniedException", "Cross-account pass role is not allowed."
)
class RoleNotSpecified(BadRequestException):
def __init__(self):
def __init__(self) -> None:
super().__init__("Role ARN must be specified for AWS integrations")
class IntegrationMethodNotDefined(BadRequestException):
def __init__(self):
def __init__(self) -> None:
super().__init__("Enumeration value for HttpMethod must be non-empty")
class InvalidOpenAPIDocumentException(BadRequestException):
def __init__(self, cause):
def __init__(self, cause: Any):
super().__init__(
f"Failed to parse the uploaded OpenAPI document due to: {cause.message}"
)
class InvalidOpenApiDocVersionException(BadRequestException):
def __init__(self):
def __init__(self) -> None:
super().__init__("Only OpenAPI 3.x.x are currently supported")
class InvalidOpenApiModeException(BadRequestException):
def __init__(self):
def __init__(self) -> None:
super().__init__(
'Enumeration value of OpenAPI import mode must be "overwrite" or "merge"',
)
class InvalidResourcePathException(BadRequestException):
def __init__(self):
def __init__(self) -> None:
super().__init__(
"Resource's path part only allow a-zA-Z0-9._- and curly braces at the beginning and the end and an optional plus sign before the closing brace."
)
class InvalidHttpEndpoint(BadRequestException):
def __init__(self):
def __init__(self) -> None:
super().__init__("Invalid HTTP endpoint specified for URI")
class InvalidArn(BadRequestException):
def __init__(self):
def __init__(self) -> None:
super().__init__("Invalid ARN specified in the request")
class InvalidIntegrationArn(BadRequestException):
def __init__(self):
def __init__(self) -> None:
super().__init__("AWS ARN for integration must contain path or action")
class InvalidRequestInput(BadRequestException):
def __init__(self):
def __init__(self) -> None:
super().__init__("Invalid request input")
class NoIntegrationDefined(NotFoundException):
def __init__(self):
def __init__(self) -> None:
super().__init__("No integration defined for method")
class NoIntegrationResponseDefined(NotFoundException):
code = 404
def __init__(self):
def __init__(self) -> None:
super().__init__("Invalid Response status code specified")
class NoMethodDefined(BadRequestException):
def __init__(self):
def __init__(self) -> None:
super().__init__("The REST API doesn't contain any methods")
class AuthorizerNotFoundException(NotFoundException):
code = 404
def __init__(self):
def __init__(self) -> None:
super().__init__("Invalid Authorizer identifier specified")
class StageNotFoundException(NotFoundException):
code = 404
def __init__(self):
def __init__(self) -> None:
super().__init__("Invalid stage identifier specified")
class ApiKeyNotFoundException(NotFoundException):
code = 404
def __init__(self):
def __init__(self) -> None:
super().__init__("Invalid API Key identifier specified")
class UsagePlanNotFoundException(NotFoundException):
code = 404
def __init__(self):
def __init__(self) -> None:
super().__init__("Invalid Usage Plan ID specified")
class ApiKeyAlreadyExists(ApiGatewayException):
code = 409
def __init__(self):
def __init__(self) -> None:
super().__init__("ConflictException", "API Key already exists")
class InvalidDomainName(BadRequestException):
code = 404
def __init__(self):
def __init__(self) -> None:
super().__init__("No Domain Name specified")
class DomainNameNotFound(NotFoundException):
code = 404
def __init__(self):
def __init__(self) -> None:
super().__init__("Invalid domain name identifier specified")
class InvalidRestApiId(BadRequestException):
code = 404
def __init__(self):
def __init__(self) -> None:
super().__init__("No Rest API Id specified")
class InvalidModelName(BadRequestException):
code = 404
def __init__(self):
def __init__(self) -> None:
super().__init__("No Model Name specified")
class RestAPINotFound(NotFoundException):
code = 404
def __init__(self):
def __init__(self) -> None:
super().__init__("Invalid Rest API Id specified")
class RequestValidatorNotFound(BadRequestException):
code = 400
def __init__(self):
def __init__(self) -> None:
super().__init__("Invalid Request Validator Id specified")
class ModelNotFound(NotFoundException):
code = 404
def __init__(self):
def __init__(self) -> None:
super().__init__("Invalid Model Name specified")
class ApiKeyValueMinLength(BadRequestException):
code = 400
def __init__(self):
def __init__(self) -> None:
super().__init__("API Key value should be at least 20 characters")
class MethodNotFoundException(NotFoundException):
code = 404
def __init__(self):
def __init__(self) -> None:
super().__init__("Invalid Method identifier specified")
class InvalidBasePathException(BadRequestException):
code = 400
def __init__(self):
def __init__(self) -> None:
super().__init__(
"API Gateway V1 doesn't support the slash character (/) in base path mappings. "
"To create a multi-level base path mapping, use API Gateway V2."
@ -222,64 +223,64 @@ class InvalidBasePathException(BadRequestException):
class DeploymentNotFoundException(NotFoundException):
def __init__(self):
def __init__(self) -> None:
super().__init__("Invalid Deployment identifier specified")
class InvalidRestApiIdForBasePathMappingException(BadRequestException):
code = 400
def __init__(self):
def __init__(self) -> None:
super().__init__("Invalid REST API identifier specified")
class InvalidStageException(BadRequestException):
code = 400
def __init__(self):
def __init__(self) -> None:
super().__init__("Invalid stage identifier specified")
class BasePathConflictException(ConflictException):
def __init__(self):
def __init__(self) -> None:
super().__init__("Base path already exists for this domain name")
class BasePathNotFoundException(NotFoundException):
code = 404
def __init__(self):
def __init__(self) -> None:
super().__init__("Invalid base path mapping identifier specified")
class ResourceIdNotFoundException(NotFoundException):
code = 404
def __init__(self):
def __init__(self) -> None:
super().__init__("Invalid resource identifier specified")
class VpcLinkNotFound(NotFoundException):
code = 404
def __init__(self):
def __init__(self) -> None:
super().__init__("VPCLink not found")
class ValidationException(ApiGatewayException):
code = 400
def __init__(self, message):
def __init__(self, message: str):
super().__init__("ValidationException", message)
class StageStillActive(BadRequestException):
def __init__(self):
def __init__(self) -> None:
super().__init__(
"Active stages pointing to this deployment must be moved or deleted"
)
class GatewayResponseNotFound(NotFoundException):
def __init__(self):
def __init__(self) -> None:
super().__init__("GatewayResponse not found")

7
moto/apigateway/integration_parsers/__init__.py
View File

@ -1,7 +1,12 @@
import abc
from typing import Tuple, Union
from requests.models import PreparedRequest
from ..models import Integration
class IntegrationParser:
@abc.abstractmethod
def invoke(self, request, integration):
def invoke(
self, request: PreparedRequest, integration: Integration
) -> Tuple[int, Union[str, bytes]]:
pass

6
moto/apigateway/integration_parsers/aws_parser.py
View File

@ -1,10 +1,14 @@
import requests
from . import IntegrationParser
from ..models import Integration
from typing import Tuple, Union
class TypeAwsParser(IntegrationParser):
def invoke(self, request, integration):
def invoke(
self, request: requests.PreparedRequest, integration: Integration
) -> Tuple[int, Union[str, bytes]]:
# integration.uri = arn:aws:apigateway:{region}:{subdomain.service|service}:path|action/{service_api}
# example value = 'arn:aws:apigateway:us-west-2:dynamodb:action/PutItem'
try:

6
moto/apigateway/integration_parsers/http_parser.py
View File

@ -1,6 +1,8 @@
import requests
from typing import Tuple, Union
from . import IntegrationParser
from ..models import Integration
class TypeHttpParser(IntegrationParser):
@ -8,7 +10,9 @@ class TypeHttpParser(IntegrationParser):
Parse invocations to a APIGateway resource with integration type HTTP
"""
def invoke(self, request, integration):
def invoke(
self, request: requests.PreparedRequest, integration: Integration
) -> Tuple[int, Union[str, bytes]]:
uri = integration["uri"]
requests_func = getattr(requests, integration["httpMethod"].lower())
response = requests_func(uri)

7
moto/apigateway/integration_parsers/unknown_parser.py
View File

@ -1,4 +1,7 @@
import requests
from typing import Tuple, Union
from . import IntegrationParser
from ..models import Integration
class TypeUnknownParser(IntegrationParser):
@ -6,6 +9,8 @@ class TypeUnknownParser(IntegrationParser):
Parse invocations to a APIGateway resource with an unknown integration type
"""
def invoke(self, request, integration):
def invoke(
self, request: requests.PreparedRequest, integration: Integration
) -> Tuple[int, Union[str, bytes]]:
_type = integration["type"]
raise NotImplementedError("The {0} type has not been implemented".format(_type))

993
moto/apigateway/models.py
View File

File diff suppressed because it is too large Load Diff

168
moto/apigateway/responses.py
View File

@ -1,9 +1,10 @@
import json
from typing import Any, Dict, List, Tuple
from urllib.parse import unquote
from moto.utilities.utils import merge_multiple_dicts
from moto.core.responses import BaseResponse
from .models import apigateway_backends
from .models import apigateway_backends, APIGatewayBackend
from .utils import deserialize_body
from .exceptions import InvalidRequestInput
@ -11,21 +12,23 @@ API_KEY_SOURCES = ["AUTHORIZER", "HEADER"]
AUTHORIZER_TYPES = ["TOKEN", "REQUEST", "COGNITO_USER_POOLS"]
ENDPOINT_CONFIGURATION_TYPES = ["PRIVATE", "EDGE", "REGIONAL"]
RESPONSE_TYPE = Tuple[int, Dict[str, str], str]
class APIGatewayResponse(BaseResponse):
def __init__(self):
def __init__(self) -> None:
super().__init__(service_name="apigateway")
def error(self, type_, message, status=400):
def error(self, type_: str, message: str, status: int = 400) -> RESPONSE_TYPE:
headers = self.response_headers or {}
headers["X-Amzn-Errortype"] = type_
return (status, headers, json.dumps({"__type": type_, "message": message}))
@property
def backend(self):
def backend(self) -> APIGatewayBackend:
return apigateway_backends[self.current_account][self.region]
def __validate_api_key_source(self, api_key_source):
def __validate_api_key_source(self, api_key_source: str) -> RESPONSE_TYPE: # type: ignore[return]
if api_key_source and api_key_source not in API_KEY_SOURCES:
return self.error(
"ValidationException",
@ -37,7 +40,7 @@ class APIGatewayResponse(BaseResponse):
).format(api_key_source=api_key_source),
)
def __validate_endpoint_configuration(self, endpoint_configuration):
def __validate_endpoint_configuration(self, endpoint_configuration: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
if endpoint_configuration and "types" in endpoint_configuration:
invalid_types = list(
set(endpoint_configuration["types"]) - set(ENDPOINT_CONFIGURATION_TYPES)
@ -53,7 +56,7 @@ class APIGatewayResponse(BaseResponse):
).format(endpoint_type=invalid_types[0]),
)
def restapis(self, request, full_url, headers):
def restapis(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
if self.method == "GET":
@ -62,7 +65,7 @@ class APIGatewayResponse(BaseResponse):
elif self.method == "POST":
api_doc = deserialize_body(self.body)
if api_doc:
fail_on_warnings = self._get_bool_param("failonwarnings")
fail_on_warnings = self._get_bool_param("failonwarnings") or False
rest_api = self.backend.import_rest_api(api_doc, fail_on_warnings)
return 200, {}, json.dumps(rest_api.to_dict())
@ -97,14 +100,16 @@ class APIGatewayResponse(BaseResponse):
return 200, {}, json.dumps(rest_api.to_dict())
def __validte_rest_patch_operations(self, patch_operations):
def __validte_rest_patch_operations(self, patch_operations: List[Dict[str, str]]) -> RESPONSE_TYPE: # type: ignore[return]
for op in patch_operations:
path = op["path"]
if "apiKeySource" in path:
value = op["value"]
return self.__validate_api_key_source(value)
def restapis_individual(self, request, full_url, headers):
def restapis_individual(
self, request: Any, full_url: str, headers: Dict[str, str]
) -> RESPONSE_TYPE:
self.setup_class(request, full_url, headers)
function_id = self.path.replace("/restapis/", "", 1).split("/")[0]
@ -114,12 +119,12 @@ class APIGatewayResponse(BaseResponse):
rest_api = self.backend.delete_rest_api(function_id)
elif self.method == "PUT":
mode = self._get_param("mode", "merge")
fail_on_warnings = self._get_bool_param("failonwarnings", False)
fail_on_warnings = self._get_bool_param("failonwarnings") or False
api_doc = deserialize_body(self.body)
rest_api = self.backend.put_rest_api(
function_id, api_doc, mode, fail_on_warnings
function_id, api_doc, mode=mode, fail_on_warnings=fail_on_warnings
)
elif self.method == "PATCH":
patch_operations = self._get_param("patchOperations")
@ -130,7 +135,7 @@ class APIGatewayResponse(BaseResponse):
return 200, {}, json.dumps(rest_api.to_dict())
def resources(self, request, full_url, headers):
def resources(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
function_id = self.path.replace("/restapis/", "", 1).split("/")[0]
@ -142,7 +147,7 @@ class APIGatewayResponse(BaseResponse):
json.dumps({"item": [resource.to_dict() for resource in resources]}),
)
def gateway_response(self, request, full_url, headers):
def gateway_response(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
if request.method == "PUT":
return self.put_gateway_response()
@ -151,12 +156,12 @@ class APIGatewayResponse(BaseResponse):
elif request.method == "DELETE":
return self.delete_gateway_response()
def gateway_responses(self, request, full_url, headers):
def gateway_responses(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
if request.method == "GET":
return self.get_gateway_responses()
def resource_individual(self, request, full_url, headers):
def resource_individual(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
function_id = self.path.replace("/restapis/", "", 1).split("/")[0]
resource_id = self.path.split("/")[-1]
@ -172,7 +177,9 @@ class APIGatewayResponse(BaseResponse):
resource = self.backend.delete_resource(function_id, resource_id)
return 202, {}, json.dumps(resource.to_dict())
def resource_methods(self, request, full_url, headers):
def resource_methods(
self, request: Any, full_url: str, headers: Dict[str, str]
) -> RESPONSE_TYPE:
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
function_id = url_path_parts[2]
@ -210,15 +217,11 @@ class APIGatewayResponse(BaseResponse):
self.backend.delete_method(function_id, resource_id, method_type)
return 204, {}, ""
elif self.method == "PATCH":
patch_operations = self._get_param("patchOperations")
self.backend.update_method(
function_id, resource_id, method_type, patch_operations
)
return 200, {}, ""
def resource_method_responses(self, request, full_url, headers):
def resource_method_responses(
self, request: Any, full_url: str, headers: Dict[str, str]
) -> RESPONSE_TYPE:
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
function_id = url_path_parts[2]
@ -248,15 +251,9 @@ class APIGatewayResponse(BaseResponse):
function_id, resource_id, method_type, response_code
)
return 204, {}, json.dumps(method_response)
elif self.method == "PATCH":
patch_operations = self._get_param("patchOperations")
method_response = self.backend.update_method_response(
function_id, resource_id, method_type, response_code, patch_operations
)
return 201, {}, json.dumps(method_response)
raise Exception('Unexpected HTTP method "%s"' % self.method)
def restapis_authorizers(self, request, full_url, headers):
def restapis_authorizers(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
restapi_id = url_path_parts[2]
@ -306,7 +303,7 @@ class APIGatewayResponse(BaseResponse):
authorizers = self.backend.get_authorizers(restapi_id)
return 200, {}, json.dumps({"item": authorizers})
def request_validators(self, request, full_url, headers):
def request_validators(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
restapi_id = url_path_parts[2]
@ -326,7 +323,7 @@ class APIGatewayResponse(BaseResponse):
)
return 201, {}, json.dumps(validator)
def request_validator_individual(self, request, full_url, headers):
def request_validator_individual(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
restapi_id = url_path_parts[2]
@ -345,7 +342,7 @@ class APIGatewayResponse(BaseResponse):
)
return 200, {}, json.dumps(validator)
def authorizers(self, request, full_url, headers):
def authorizers(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
restapi_id = url_path_parts[2]
@ -364,7 +361,7 @@ class APIGatewayResponse(BaseResponse):
self.backend.delete_authorizer(restapi_id, authorizer_id)
return 202, {}, "{}"
def restapis_stages(self, request, full_url, headers):
def restapis_stages(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
function_id = url_path_parts[2]
@ -395,7 +392,7 @@ class APIGatewayResponse(BaseResponse):
stages = self.backend.get_stages(function_id)
return 200, {}, json.dumps({"item": stages})
def restapis_stages_tags(self, request, full_url, headers):
def restapis_stages_tags(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
function_id = url_path_parts[4]
@ -408,12 +405,12 @@ class APIGatewayResponse(BaseResponse):
return 200, {}, json.dumps({"item": tags})
if self.method == "DELETE":
stage = self.backend.get_stage(function_id, stage_name)
for tag in stage.get("tags").copy():
if tag in self.querystring.get("tagKeys"):
for tag in stage.get("tags", {}).copy():
if tag in (self.querystring.get("tagKeys") or {}):
stage["tags"].pop(tag, None)
return 200, {}, json.dumps({"item": ""})
def stages(self, request, full_url, headers):
def stages(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
function_id = url_path_parts[2]
@ -432,15 +429,13 @@ class APIGatewayResponse(BaseResponse):
self.backend.delete_stage(function_id, stage_name)
return 202, {}, "{}"
def integrations(self, request, full_url, headers):
def integrations(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
function_id = url_path_parts[2]
resource_id = url_path_parts[4]
method_type = url_path_parts[6]
integration_response = {}
if self.method == "GET":
integration_response = self.backend.get_integration(
function_id, resource_id, method_type
@ -484,7 +479,7 @@ class APIGatewayResponse(BaseResponse):
)
return 204, {}, json.dumps(integration_response)
def integration_responses(self, request, full_url, headers):
def integration_responses(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
function_id = url_path_parts[2]
@ -520,7 +515,7 @@ class APIGatewayResponse(BaseResponse):
)
return 204, {}, json.dumps(integration_response)
def deployments(self, request, full_url, headers):
def deployments(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
function_id = self.path.replace("/restapis/", "", 1).split("/")[0]
@ -536,7 +531,7 @@ class APIGatewayResponse(BaseResponse):
)
return 201, {}, json.dumps(deployment)
def individual_deployment(self, request, full_url, headers):
def individual_deployment(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
function_id = url_path_parts[2]
@ -549,7 +544,7 @@ class APIGatewayResponse(BaseResponse):
deployment = self.backend.delete_deployment(function_id, deployment_id)
return 202, {}, json.dumps(deployment)
def apikeys(self, request, full_url, headers):
def apikeys(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
if self.method == "POST":
@ -557,19 +552,20 @@ class APIGatewayResponse(BaseResponse):
return 201, {}, json.dumps(apikey_response)
elif self.method == "GET":
include_values = self._get_bool_param("includeValues")
include_values = self._get_bool_param("includeValues") or False
apikeys_response = self.backend.get_api_keys(include_values=include_values)
return 200, {}, json.dumps({"item": apikeys_response})
def apikey_individual(self, request, full_url, headers):
def apikey_individual(
self, request: Any, full_url: str, headers: Dict[str, str]
) -> RESPONSE_TYPE:
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
apikey = url_path_parts[2]
status_code = 200
if self.method == "GET":
include_value = self._get_bool_param("includeValue")
include_value = self._get_bool_param("includeValue") or False
apikey_response = self.backend.get_api_key(
apikey, include_value=include_value
)
@ -577,12 +573,12 @@ class APIGatewayResponse(BaseResponse):
patch_operations = self._get_param("patchOperations")
apikey_response = self.backend.update_api_key(apikey, patch_operations)
elif self.method == "DELETE":
apikey_response = self.backend.delete_api_key(apikey)
status_code = 202
self.backend.delete_api_key(apikey)
return 202, {}, "{}"
return status_code, {}, json.dumps(apikey_response)
return 200, {}, json.dumps(apikey_response)
def usage_plans(self, request, full_url, headers):
def usage_plans(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
if self.method == "POST":
usage_plan_response = self.backend.create_usage_plan(json.loads(self.body))
@ -592,7 +588,7 @@ class APIGatewayResponse(BaseResponse):
usage_plans_response = self.backend.get_usage_plans(api_key_id=api_key_id)
return 200, {}, json.dumps({"item": usage_plans_response})
def usage_plan_individual(self, request, full_url, headers):
def usage_plan_individual(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
@ -602,8 +598,8 @@ class APIGatewayResponse(BaseResponse):
usage_plan_response = self.backend.get_usage_plan(usage_plan)
return 200, {}, json.dumps(usage_plan_response)
elif self.method == "DELETE":
usage_plan_response = self.backend.delete_usage_plan(usage_plan)
return 202, {}, json.dumps(usage_plan_response)
self.backend.delete_usage_plan(usage_plan)
return 202, {}, "{}"
elif self.method == "PATCH":
patch_operations = self._get_param("patchOperations")
usage_plan_response = self.backend.update_usage_plan(
@ -611,7 +607,7 @@ class APIGatewayResponse(BaseResponse):
)
return 200, {}, json.dumps(usage_plan_response)
def usage_plan_keys(self, request, full_url, headers):
def usage_plan_keys(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
@ -626,7 +622,7 @@ class APIGatewayResponse(BaseResponse):
usage_plans_response = self.backend.get_usage_plan_keys(usage_plan_id)
return 200, {}, json.dumps({"item": usage_plans_response})
def usage_plan_key_individual(self, request, full_url, headers):
def usage_plan_key_individual(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
@ -637,12 +633,10 @@ class APIGatewayResponse(BaseResponse):
usage_plan_response = self.backend.get_usage_plan_key(usage_plan_id, key_id)
return 200, {}, json.dumps(usage_plan_response)
elif self.method == "DELETE":
usage_plan_response = self.backend.delete_usage_plan_key(
usage_plan_id, key_id
)
return 202, {}, json.dumps(usage_plan_response)
self.backend.delete_usage_plan_key(usage_plan_id, key_id)
return 202, {}, "{}"
def domain_names(self, request, full_url, headers):
def domain_names(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
if self.method == "GET":
@ -661,7 +655,6 @@ class APIGatewayResponse(BaseResponse):
regional_certificate_arn = self._get_param("regionalCertificateArn")
endpoint_configuration = self._get_param("endpointConfiguration")
security_policy = self._get_param("securityPolicy")
generate_cli_skeleton = self._get_param("generateCliSkeleton")
domain_name_resp = self.backend.create_domain_name(
domain_name,
certificate_name,
@ -674,35 +667,31 @@ class APIGatewayResponse(BaseResponse):
regional_certificate_arn,
endpoint_configuration,
security_policy,
generate_cli_skeleton,
)
return 201, {}, json.dumps(domain_name_resp)
def domain_name_induvidual(self, request, full_url, headers):
def domain_name_induvidual(
self, request: Any, full_url: str, headers: Dict[str, str]
) -> RESPONSE_TYPE:
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
domain_name = url_path_parts[2]
domain_names = {}
if self.method == "GET":
if domain_name is not None:
domain_names = self.backend.get_domain_name(domain_name)
return 200, {}, json.dumps(domain_names)
return 200, {}, json.dumps(domain_names)
return 200, {}, "{}"
elif self.method == "DELETE":
if domain_name is not None:
self.backend.delete_domain_name(domain_name)
return 202, {}, json.dumps({})
elif self.method == "PATCH":
if domain_name is not None:
patch_operations = self._get_param("patchOperations")
self.backend.update_domain_name(domain_name, patch_operations)
return 200, {}, json.dumps(domain_name)
else:
msg = 'Method "%s" for API GW domain names not implemented' % self.method
return 404, {}, json.dumps({"error": msg})
def models(self, request, full_url, headers):
def models(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
rest_api_id = self.path.replace("/restapis/", "", 1).split("/")[0]
@ -715,30 +704,29 @@ class APIGatewayResponse(BaseResponse):
description = self._get_param("description")
schema = self._get_param("schema")
content_type = self._get_param("contentType")
cli_input_json = self._get_param("cliInputJson")
generate_cli_skeleton = self._get_param("generateCliSkeleton")
model = self.backend.create_model(
rest_api_id,
name,
content_type,
description,
schema,
cli_input_json,
generate_cli_skeleton,
)
return 201, {}, json.dumps(model)
def model_induvidual(self, request, full_url, headers):
def model_induvidual(
self, request: Any, full_url: str, headers: Dict[str, str]
) -> RESPONSE_TYPE:
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
rest_api_id = url_path_parts[2]
model_name = url_path_parts[4]
model_info = {}
if self.method == "GET":
model_info = self.backend.get_model(rest_api_id, model_name)
return 200, {}, json.dumps(model_info)
return 200, {}, json.dumps(model_info)
return 200, {}, "{}"
def base_path_mappings(self, request, full_url, headers):
def base_path_mappings(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
@ -757,7 +745,7 @@ class APIGatewayResponse(BaseResponse):
)
return 201, {}, json.dumps(base_path_mapping_resp)
def base_path_mapping_individual(self, request, full_url, headers):
def base_path_mapping_individual(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
@ -780,7 +768,7 @@ class APIGatewayResponse(BaseResponse):
)
return 200, {}, json.dumps(base_path_mapping)
def vpc_link(self, request, full_url, headers):
def vpc_link(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
vpc_link_id = url_path_parts[-1]
@ -792,7 +780,7 @@ class APIGatewayResponse(BaseResponse):
vpc_link = self.backend.get_vpc_link(vpc_link_id=vpc_link_id)
return 200, {}, json.dumps(vpc_link)
def vpc_links(self, request, full_url, headers):
def vpc_links(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return]
self.setup_class(request, full_url, headers)
if self.method == "GET":
@ -808,7 +796,7 @@ class APIGatewayResponse(BaseResponse):
)
return 202, {}, json.dumps(vpc_link)
def put_gateway_response(self):
def put_gateway_response(self) -> RESPONSE_TYPE:
rest_api_id = self.path.split("/")[-3]
response_type = self.path.split("/")[-1]
params = json.loads(self.body)
@ -824,7 +812,7 @@ class APIGatewayResponse(BaseResponse):
)
return 201, {}, json.dumps(response)
def get_gateway_response(self):
def get_gateway_response(self) -> RESPONSE_TYPE:
rest_api_id = self.path.split("/")[-3]
response_type = self.path.split("/")[-1]
response = self.backend.get_gateway_response(
@ -832,12 +820,12 @@ class APIGatewayResponse(BaseResponse):
)
return 200, {}, json.dumps(response)
def get_gateway_responses(self):
def get_gateway_responses(self) -> RESPONSE_TYPE:
rest_api_id = self.path.split("/")[-2]
responses = self.backend.get_gateway_responses(rest_api_id=rest_api_id)
return 200, {}, json.dumps(dict(item=responses))
def delete_gateway_response(self):
def delete_gateway_response(self) -> RESPONSE_TYPE:
rest_api_id = self.path.split("/")[-3]
response_type = self.path.split("/")[-1]
self.backend.delete_gateway_response(

9
moto/apigateway/utils.py
View File

@ -2,15 +2,16 @@ import string
import json
import yaml
from moto.moto_api._internal import mock_random as random
from typing import Any, Dict
def create_id():
def create_id() -> str:
size = 10
chars = list(range(10)) + list(string.ascii_lowercase)
return "".join(str(random.choice(chars)) for x in range(size))
def deserialize_body(body):
def deserialize_body(body: str) -> Dict[str, Any]:
try:
api_doc = json.loads(body)
except json.JSONDecodeError:
@ -19,8 +20,8 @@ def deserialize_body(body):
if "openapi" in api_doc or "swagger" in api_doc:
return api_doc
return None
return {}
def to_path(prop):
def to_path(prop: str) -> str:
return "/" + prop

2
moto/core/common_models.py
View File

@ -29,7 +29,7 @@ class CloudFormationModel(BaseModel):
@classmethod
@abstractmethod
def has_cfn_attr(cls, attr):
def has_cfn_attr(cls, attr): # pylint: disable=unused-argument
# Used for validation
# If a template creates an Output for an attribute that does not exist, an error should be thrown
return True

2
moto/core/utils.py
View File

@ -186,7 +186,7 @@ def unix_time_millis(dt=None):
return unix_time(dt) * 1000.0
def path_url(url):
def path_url(url: str) -> str:
parsed_url = urlparse(url)
path = parsed_url.path
if not path:

4
moto/utilities/utils.py
View File

@ -2,8 +2,8 @@ import json
import hashlib
import pkgutil
from collections.abc import MutableMapping
from typing import Any, Dict
def str2bool(v):
@ -23,7 +23,7 @@ def load_resource(package, resource, as_json=True):
return json.loads(resource) if as_json else resource.decode("utf-8")
def merge_multiple_dicts(*args):
def merge_multiple_dicts(*args: Any) -> Dict[str, any]:
result = {}
for d in args:
result.update(d)

3
setup.cfg
View File

@ -18,9 +18,10 @@ disable = W,C,R,E
enable = anomalous-backslash-in-string, arguments-renamed, dangerous-default-value, deprecated-module, function-redefined, import-self, redefined-builtin, redefined-outer-name, reimported, pointless-statement, super-with-arguments, unused-argument, unused-import, unused-variable, useless-else-on-loop, wildcard-import
[mypy]
exclude = tests
files= moto/acm,moto/amp,moto/apigateway,moto/applicationautoscaling/
show_column_numbers=True
show_error_codes = True
disable_error_code=abstract
disallow_any_unimported=False
disallow_any_expr=False

30
tests/test_apigateway/test_apigateway.py
View File

@ -842,9 +842,21 @@ def test_non_existent_authorizer():
response = client.create_rest_api(name="my_api", description="this is my api")
api_id = response["id"]
client.get_authorizer.when.called_with(
restApiId=api_id, authorizerId="xxx"
).should.throw(ClientError)
with pytest.raises(ClientError) as exc:
client.get_authorizer(restApiId=api_id, authorizerId="xxx")
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
err["Message"].should.equal("Invalid Authorizer identifier specified")
with pytest.raises(ClientError) as exc:
client.update_authorizer(
restApiId=api_id,
authorizerId="xxx",
patchOperations=[{"op": "add", "path": "/type", "value": "sth"}],
)
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
err["Message"].should.equal("Invalid Authorizer identifier specified")
@mock_apigateway
@ -1878,18 +1890,6 @@ def test_get_domain_name_unknown_domainname():
ex.value.response["Error"]["Code"].should.equal("NotFoundException")
@mock_apigateway
def test_update_domain_name_unknown_domainname():
client = boto3.client("apigateway", region_name="us-east-1")
with pytest.raises(ClientError) as ex:
client.update_domain_name(domainName="www.google.fr", patchOperations=[])
ex.value.response["Error"]["Message"].should.equal(
"Invalid domain name identifier specified"
)
ex.value.response["Error"]["Code"].should.equal("NotFoundException")
@mock_apigateway
def test_delete_domain_name_unknown_domainname():
client = boto3.client("apigateway", region_name="us-east-1")