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

Implement APIGateway create_base_path_mapping (#4475)

Browse Source
This commit is contained in:
cm-iwata 2021-10-27 19:48:32 +09:00 committed by GitHub
parent 98ca9b82e1
commit 2218806f3d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 275 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
IMPLEMENTATION_COVERAGE.md
View File

@ -22,11 +22,11 @@
## apigateway
<details>
<summary>48% implemented</summary>
<summary>49% implemented</summary>
- [X] create_api_key
- [X] create_authorizer
- [ ] create_base_path_mapping
- [X] create_base_path_mapping
- [X] create_deployment
- [ ] create_documentation_part
- [ ] create_documentation_version
@ -2716,29 +2716,24 @@
## lambda
<details>
<summary>41% implemented</summary>
<summary>48% implemented</summary>
- [ ] add_layer_version_permission
- [X] add_permission
- [ ] create_alias
- [ ] create_code_signing_config
- [X] create_event_source_mapping
- [X] create_function
- [ ] delete_alias
- [ ] delete_code_signing_config
- [X] delete_event_source_mapping
- [X] delete_function
- [ ] delete_function_code_signing_config
- [X] delete_function_concurrency
- [ ] delete_function_event_invoke_config
- [ ] delete_layer_version
- [ ] delete_provisioned_concurrency_config
- [ ] get_account_settings
- [ ] get_alias
- [ ] get_code_signing_config
- [X] get_event_source_mapping
- [X] get_function
- [ ] get_function_code_signing_config
- [X] get_function_concurrency
- [ ] get_function_configuration
- [ ] get_function_event_invoke_config
@ -2750,11 +2745,9 @@
- [X] invoke
- [ ] invoke_async
- [ ] list_aliases
- [ ] list_code_signing_configs
- [X] list_event_source_mappings
- [ ] list_function_event_invoke_configs
- [X] list_functions
- [ ] list_functions_by_code_signing_config
- [ ] list_layer_versions
- [X] list_layers
- [ ] list_provisioned_concurrency_configs
@ -2762,7 +2755,6 @@
- [X] list_versions_by_function
- [X] publish_layer_version
- [ ] publish_version
- [ ] put_function_code_signing_config
- [X] put_function_concurrency
- [ ] put_function_event_invoke_config
- [ ] put_provisioned_concurrency_config
@ -2771,7 +2763,6 @@
- [X] tag_resource
- [X] untag_resource
- [ ] update_alias
- [ ] update_code_signing_config
- [X] update_event_source_mapping
- [X] update_function_code
- [X] update_function_configuration

40
moto/apigateway/exceptions.py
View File

@ -13,6 +13,10 @@ class AccessDeniedException(JsonRESTError):
pass
class ConflictException(JsonRESTError):
code = 409
class AwsProxyNotAllowed(BadRequestException):
def __init__(self):
super(AwsProxyNotAllowed, self).__init__(
@ -225,3 +229,39 @@ class MethodNotFoundException(NotFoundException):
super(MethodNotFoundException, self).__init__(
"NotFoundException", "Invalid Method identifier specified"
)
class InvalidBasePathException(BadRequestException):
code = 400
def __init__(self):
super(InvalidBasePathException, self).__init__(
"BadRequestException",
"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.",
)
class InvalidRestApiIdForBasePathMappingException(BadRequestException):
code = 400
def __init__(self):
super(InvalidRestApiIdForBasePathMappingException, self).__init__(
"BadRequestException", "Invalid REST API identifier specified"
)
class InvalidStageException(BadRequestException):
code = 400
def __init__(self):
super(InvalidStageException, self).__init__(
"BadRequestException", "Invalid stage identifier specified"
)
class BasePathConflictException(ConflictException):
def __init__(self):
super(BasePathConflictException, self).__init__(
"ConflictException", "Base path already exists for this domain name"
)

53
moto/apigateway/models.py
View File

@ -43,6 +43,10 @@ from .exceptions import (
RequestValidatorNotFound,
ModelNotFound,
ApiKeyValueMinLength,
InvalidBasePathException,
InvalidRestApiIdForBasePathMappingException,
InvalidStageException,
BasePathConflictException,
)
from ..core.models import responses_mock
from moto.apigateway.exceptions import MethodNotFoundException
@ -1081,6 +1085,20 @@ class Model(BaseModel, dict):
self["generateCliSkeleton"] = kwargs.get("generate_cli_skeleton")
class BasePathMapping(BaseModel, dict):
def __init__(self, domain_name, rest_api_id, **kwargs):
super(BasePathMapping, self).__init__()
self["domain_name"] = domain_name
self["restApiId"] = rest_api_id
if kwargs.get("basePath"):
self["basePath"] = kwargs.get("basePath")
else:
self["basePath"] = "(none)"
if kwargs.get("stage"):
self["stage"] = kwargs.get("stage")
class APIGatewayBackend(BaseBackend):
def __init__(self, region_name):
super(APIGatewayBackend, self).__init__()
@ -1091,6 +1109,7 @@ class APIGatewayBackend(BaseBackend):
self.domain_names = {}
self.models = {}
self.region_name = region_name
self.base_path_mappings = {}
def reset(self):
region_name = self.region_name
@ -1705,6 +1724,40 @@ class APIGatewayBackend(BaseBackend):
restApi = self.get_rest_api(restapi_id)
return restApi.update_request_validator(validator_id, patch_operations)
def create_base_path_mapping(
self, domain_name, rest_api_id, base_path=None, stage=None
):
if domain_name not in self.domain_names:
raise DomainNameNotFound()
if base_path and "/" in base_path:
raise InvalidBasePathException()
if rest_api_id not in self.apis:
raise InvalidRestApiIdForBasePathMappingException()
if stage and self.apis[rest_api_id].stages.get(stage) is None:
raise InvalidStageException()
new_base_path_mapping = BasePathMapping(
domain_name=domain_name,
rest_api_id=rest_api_id,
basePath=base_path,
stage=stage,
)
new_base_path = new_base_path_mapping.get("basePath")
if self.base_path_mappings.get(domain_name) is None:
self.base_path_mappings[domain_name] = {}
else:
if (
self.base_path_mappings[domain_name].get(new_base_path)
and new_base_path != "(none)"
):
raise BasePathConflictException()
self.base_path_mappings[domain_name][new_base_path] = new_base_path_mapping
return new_base_path_mapping
apigateway_backends = {}
for region_name in Session().get_available_regions("apigateway"):

25
moto/apigateway/responses.py
View File

@ -22,6 +22,7 @@ from .exceptions import (
NoIntegrationDefined,
NoIntegrationResponseDefined,
NotFoundException,
ConflictException,
)
API_KEY_SOURCES = ["AUTHORIZER", "HEADER"]
@ -849,3 +850,27 @@ class APIGatewayResponse(BaseResponse):
error.message, error.error_type
),
)
def base_path_mappings(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
domain_name = url_path_parts[2]
try:
if self.method == "GET":
# TODO implements
pass
elif self.method == "POST":
base_path = self._get_param("basePath")
rest_api_id = self._get_param("restApiId")
stage = self._get_param("stage")
base_path_mapping_resp = self.backend.create_base_path_mapping(
domain_name, rest_api_id, base_path, stage,
)
return 201, {}, json.dumps(base_path_mapping_resp)
except BadRequestException as e:
return self.error("BadRequestException", e.message)
except ConflictException as e:
return self.error("ConflictException", e.message, 409)

1
moto/apigateway/urls.py
View File

@ -27,6 +27,7 @@ url_paths = {
"{0}/restapis/(?P<function_id>[^/]+)/models$": response.models,
"{0}/restapis/(?P<function_id>[^/]+)/models/(?P<model_name>[^/]+)/?$": response.model_induvidual,
"{0}/domainnames/(?P<domain_name>[^/]+)/?$": response.domain_name_induvidual,
"{0}/domainnames/(?P<domain_name>[^/]+)/basepathmappings$": response.base_path_mappings,
"{0}/usageplans/(?P<usage_plan_id>[^/]+)/?$": response.usage_plan_individual,
"{0}/usageplans/(?P<usage_plan_id>[^/]+)/keys$": response.usage_plan_keys,
"{0}/usageplans/(?P<usage_plan_id>[^/]+)/keys/(?P<api_key_id>[^/]+)/?$": response.usage_plan_key_individual,

153
tests/test_apigateway/test_apigateway.py
View File

@ -2349,3 +2349,156 @@ def test_delete_domain_name_unknown_domainname():
"Invalid domain name identifier specified"
)
ex.value.response["Error"]["Code"].should.equal("NotFoundException")
@mock_apigateway
def test_create_base_path_mapping():
client = boto3.client("apigateway", region_name="us-west-2")
domain_name = "testDomain"
client.create_domain_name(
domainName=domain_name,
certificateName="test.certificate",
certificatePrivateKey="testPrivateKey",
)
response = client.create_rest_api(name="my_api", description="this is my api")
api_id = response["id"]
stage_name = "dev"
create_method_integration(client, api_id)
client.create_deployment(
restApiId=api_id, stageName=stage_name, description="1.0.1"
)
response = client.create_base_path_mapping(domainName=domain_name, restApiId=api_id)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(201)
response["basePath"].should.equal("(none)")
response["restApiId"].should.equal(api_id)
response.should_not.have.key("stage")
response = client.create_base_path_mapping(
domainName=domain_name, restApiId=api_id, stage=stage_name
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(201)
response["basePath"].should.equal("(none)")
response["restApiId"].should.equal(api_id)
response["stage"].should.equal(stage_name)
response = client.create_base_path_mapping(
domainName=domain_name, restApiId=api_id, stage=stage_name, basePath="v1"
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(201)
response["basePath"].should.equal("v1")
response["restApiId"].should.equal(api_id)
response["stage"].should.equal(stage_name)
@mock_apigateway
def test_create_base_path_mapping_with_unknown_api():
client = boto3.client("apigateway", region_name="us-west-2")
domain_name = "testDomain"
client.create_domain_name(
domainName=domain_name,
certificateName="test.certificate",
certificatePrivateKey="testPrivateKey",
)
with pytest.raises(ClientError) as ex:
client.create_base_path_mapping(
domainName=domain_name, restApiId="none-exists-api"
)
ex.value.response["Error"]["Message"].should.equal(
"Invalid REST API identifier specified"
)
ex.value.response["Error"]["Code"].should.equal("BadRequestException")
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
@mock_apigateway
def test_create_base_path_mapping_with_invalid_base_path():
client = boto3.client("apigateway", region_name="us-west-2")
domain_name = "testDomain"
client.create_domain_name(
domainName=domain_name,
certificateName="test.certificate",
certificatePrivateKey="testPrivateKey",
)
response = client.create_rest_api(name="my_api", description="this is my api")
api_id = response["id"]
stage_name = "dev"
create_method_integration(client, api_id)
client.create_deployment(
restApiId=api_id, stageName=stage_name, description="1.0.1"
)
with pytest.raises(ClientError) as ex:
client.create_base_path_mapping(
domainName=domain_name, restApiId=api_id, basePath="/v1"
)
ex.value.response["Error"]["Message"].should.equal(
"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."
)
ex.value.response["Error"]["Code"].should.equal("BadRequestException")
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
@mock_apigateway
def test_create_base_path_mapping_with_unknown_stage():
client = boto3.client("apigateway", region_name="us-west-2")
domain_name = "testDomain"
client.create_domain_name(
domainName=domain_name,
certificateName="test.certificate",
certificatePrivateKey="testPrivateKey",
)
response = client.create_rest_api(name="my_api", description="this is my api")
api_id = response["id"]
stage_name = "dev"
create_method_integration(client, api_id)
client.create_deployment(
restApiId=api_id, stageName=stage_name, description="1.0.1"
)
with pytest.raises(ClientError) as ex:
client.create_base_path_mapping(
domainName=domain_name, restApiId=api_id, stage="unknown-stage"
)
ex.value.response["Error"]["Message"].should.equal(
"Invalid stage identifier specified"
)
ex.value.response["Error"]["Code"].should.equal("BadRequestException")
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
@mock_apigateway
def test_create_base_path_mapping_with_duplicate_base_path():
client = boto3.client("apigateway", region_name="us-west-2")
domain_name = "testDomain"
client.create_domain_name(
domainName=domain_name,
certificateName="test.certificate",
certificatePrivateKey="testPrivateKey",
)
response = client.create_rest_api(name="my_api", description="this is my api")
api_id = response["id"]
base_path = "v1"
client.create_base_path_mapping(
domainName=domain_name, restApiId=api_id, basePath=base_path
)
with pytest.raises(ClientError) as ex:
client.create_base_path_mapping(
domainName=domain_name, restApiId=api_id, basePath=base_path
)
ex.value.response["Error"]["Message"].should.equal(
"Base path already exists for this domain name"
)
ex.value.response["Error"]["Code"].should.equal("ConflictException")
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(409)