Feature: APIGatewayV2 (#4840)

This commit is contained in:
Bert Blommers 2022-02-08 20:12:51 -01:00 committed by GitHub
parent 947357a718
commit 094d00a37a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
33 changed files with 4642 additions and 37 deletions

View File

@ -22,7 +22,7 @@
## apigateway
<details>
<summary>55% implemented</summary>
<summary>59% implemented</summary>
- [X] create_api_key
- [X] create_authorizer
@ -38,7 +38,7 @@
- [X] create_stage
- [X] create_usage_plan
- [X] create_usage_plan_key
- [ ] create_vpc_link
- [X] create_vpc_link
- [X] delete_api_key
- [X] delete_authorizer
- [X] delete_base_path_mapping
@ -59,7 +59,7 @@
- [X] delete_stage
- [X] delete_usage_plan
- [X] delete_usage_plan_key
- [ ] delete_vpc_link
- [X] delete_vpc_link
- [ ] flush_stage_authorizers_cache
- [ ] flush_stage_cache
- [ ] generate_client_certificate
@ -107,8 +107,8 @@
- [X] get_usage_plan_key
- [X] get_usage_plan_keys
- [X] get_usage_plans
- [ ] get_vpc_link
- [ ] get_vpc_links
- [X] get_vpc_link
- [X] get_vpc_links
- [ ] import_api_keys
- [ ] import_documentation_parts
- [ ] import_rest_api
@ -146,6 +146,84 @@
- [ ] update_vpc_link
</details>
## apigatewayv2
<details>
<summary>58% implemented</summary>
- [X] create_api
- [ ] create_api_mapping
- [X] create_authorizer
- [ ] create_deployment
- [ ] create_domain_name
- [X] create_integration
- [X] create_integration_response
- [X] create_model
- [X] create_route
- [X] create_route_response
- [ ] create_stage
- [X] create_vpc_link
- [ ] delete_access_log_settings
- [X] delete_api
- [ ] delete_api_mapping
- [X] delete_authorizer
- [X] delete_cors_configuration
- [ ] delete_deployment
- [ ] delete_domain_name
- [X] delete_integration
- [X] delete_integration_response
- [X] delete_model
- [X] delete_route
- [X] delete_route_request_parameter
- [X] delete_route_response
- [ ] delete_route_settings
- [ ] delete_stage
- [X] delete_vpc_link
- [ ] export_api
- [X] get_api
- [ ] get_api_mapping
- [ ] get_api_mappings
- [X] get_apis
- [X] get_authorizer
- [ ] get_authorizers
- [ ] get_deployment
- [ ] get_deployments
- [ ] get_domain_name
- [ ] get_domain_names
- [X] get_integration
- [X] get_integration_response
- [X] get_integration_responses
- [X] get_integrations
- [X] get_model
- [ ] get_model_template
- [ ] get_models
- [X] get_route
- [X] get_route_response
- [ ] get_route_responses
- [X] get_routes
- [ ] get_stage
- [ ] get_stages
- [X] get_tags
- [X] get_vpc_link
- [X] get_vpc_links
- [ ] import_api
- [X] reimport_api
- [ ] reset_authorizers_cache
- [X] tag_resource
- [X] untag_resource
- [X] update_api
- [ ] update_api_mapping
- [X] update_authorizer
- [ ] update_deployment
- [ ] update_domain_name
- [X] update_integration
- [X] update_integration_response
- [X] update_model
- [X] update_route
- [ ] update_route_response
- [ ] update_stage
- [X] update_vpc_link
</details>
## application-autoscaling
<details>
<summary>60% implemented</summary>
@ -3289,7 +3367,7 @@
## lambda
<details>
<summary>41% implemented</summary>
<summary>43% implemented</summary>
- [ ] add_layer_version_permission
- [X] add_permission
@ -3308,7 +3386,7 @@
- [ ] delete_provisioned_concurrency_config
- [ ] get_account_settings
- [ ] get_alias
- [ ] get_code_signing_config
- [X] get_code_signing_config
- [X] get_event_source_mapping
- [X] get_function
- [ ] get_function_code_signing_config
@ -5275,7 +5353,6 @@
- amplifybackend
- amplifyuibuilder
- apigatewaymanagementapi
- apigatewayv2
- appconfig
- appconfigdata
- appflow

View File

@ -41,7 +41,7 @@ apigateway
- [X] create_stage
- [X] create_usage_plan
- [X] create_usage_plan_key
- [ ] create_vpc_link
- [X] create_vpc_link
- [X] delete_api_key
- [X] delete_authorizer
- [X] delete_base_path_mapping
@ -62,7 +62,7 @@ apigateway
- [X] delete_stage
- [X] delete_usage_plan
- [X] delete_usage_plan_key
- [ ] delete_vpc_link
- [X] delete_vpc_link
- [ ] flush_stage_authorizers_cache
- [ ] flush_stage_cache
- [ ] generate_client_certificate
@ -110,8 +110,12 @@ apigateway
- [X] get_usage_plan_key
- [X] get_usage_plan_keys
- [X] get_usage_plans
- [ ] get_vpc_link
- [ ] get_vpc_links
- [X] get_vpc_link
- [X] get_vpc_links
Pagination has not yet been implemented
- [ ] import_api_keys
- [ ] import_documentation_parts
- [ ] import_rest_api

View File

@ -0,0 +1,131 @@
.. _implementedservice_apigatewayv2:
.. |start-h3| raw:: html
<h3>
.. |end-h3| raw:: html
</h3>
============
apigatewayv2
============
.. autoclass:: moto.apigatewayv2.models.ApiGatewayV2Backend
|start-h3| Example usage |end-h3|
.. sourcecode:: python
@mock_apigatewayv2
def test_apigatewayv2_behaviour:
boto3.client("apigatewayv2")
...
|start-h3| Implemented features for this service |end-h3|
- [X] create_api
The following parameters are not yet implemented:
CredentialsArn, RouteKey, Tags, Target
- [ ] create_api_mapping
- [X] create_authorizer
- [ ] create_deployment
- [ ] create_domain_name
- [X] create_integration
- [X] create_integration_response
- [X] create_model
- [X] create_route
- [X] create_route_response
The following parameters are not yet implemented: ResponseModels, ResponseParameters
- [ ] create_stage
- [X] create_vpc_link
- [ ] delete_access_log_settings
- [X] delete_api
- [ ] delete_api_mapping
- [X] delete_authorizer
- [X] delete_cors_configuration
- [ ] delete_deployment
- [ ] delete_domain_name
- [X] delete_integration
- [X] delete_integration_response
- [X] delete_model
- [X] delete_route
- [X] delete_route_request_parameter
- [X] delete_route_response
- [ ] delete_route_settings
- [ ] delete_stage
- [X] delete_vpc_link
- [ ] export_api
- [X] get_api
- [ ] get_api_mapping
- [ ] get_api_mappings
- [X] get_apis
Pagination is not yet implemented
- [X] get_authorizer
- [ ] get_authorizers
- [ ] get_deployment
- [ ] get_deployments
- [ ] get_domain_name
- [ ] get_domain_names
- [X] get_integration
- [X] get_integration_response
- [X] get_integration_responses
- [X] get_integrations
Pagination is not yet implemented
- [X] get_model
- [ ] get_model_template
- [ ] get_models
- [X] get_route
- [X] get_route_response
- [ ] get_route_responses
- [X] get_routes
Pagination is not yet implemented
- [ ] get_stage
- [ ] get_stages
- [X] get_tags
- [X] get_vpc_link
- [X] get_vpc_links
- [ ] import_api
- [X] reimport_api
Only YAML is supported at the moment. Full OpenAPI-support is not guaranteed. Only limited validation is implemented
- [ ] reset_authorizers_cache
- [X] tag_resource
- [X] untag_resource
- [X] update_api
The following parameters have not yet been implemented: CredentialsArn, RouteKey, Target
- [ ] update_api_mapping
- [X] update_authorizer
- [ ] update_deployment
- [ ] update_domain_name
- [X] update_integration
- [X] update_integration_response
- [X] update_model
- [X] update_route
- [ ] update_route_response
- [ ] update_stage
- [X] update_vpc_link

View File

@ -44,7 +44,7 @@ lambda
- [ ] delete_provisioned_concurrency_config
- [ ] get_account_settings
- [ ] get_alias
- [ ] get_code_signing_config
- [X] get_code_signing_config
- [X] get_event_source_mapping
- [X] get_function
- [ ] get_function_code_signing_config

View File

@ -26,7 +26,8 @@ def lazy_load(
mock_acm = lazy_load(".acm", "mock_acm")
mock_apigateway = lazy_load(".apigateway", "mock_apigateway")
mock_appsync = lazy_load(".appsync", "mock_appsync", boto3_name="appsync")
mock_apigatewayv2 = lazy_load(".apigatewayv2", "mock_apigatewayv2")
mock_appsync = lazy_load(".appsync", "mock_appsync")
mock_athena = lazy_load(".athena", "mock_athena")
mock_applicationautoscaling = lazy_load(
".applicationautoscaling", "mock_applicationautoscaling"
@ -63,6 +64,11 @@ mock_ec2 = lazy_load(".ec2", "mock_ec2")
mock_ec2instanceconnect = lazy_load(".ec2instanceconnect", "mock_ec2instanceconnect")
mock_ecr = lazy_load(".ecr", "mock_ecr")
mock_ecs = lazy_load(".ecs", "mock_ecs")
mock_efs = lazy_load(".efs", "mock_efs")
mock_eks = lazy_load(".eks", "mock_eks")
mock_elasticache = lazy_load(
".elasticache", "mock_elasticache", boto3_name="elasticache"
)
mock_elastictranscoder = lazy_load(".elastictranscoder", "mock_elastictranscoder")
mock_elb = lazy_load(".elb", "mock_elb")
mock_elbv2 = lazy_load(".elbv2", "mock_elbv2")
@ -81,9 +87,22 @@ mock_iam = lazy_load(".iam", "mock_iam")
mock_iot = lazy_load(".iot", "mock_iot")
mock_iotdata = lazy_load(".iotdata", "mock_iotdata", boto3_name="iot-data")
mock_kinesis = lazy_load(".kinesis", "mock_kinesis")
mock_kinesisvideo = lazy_load(".kinesisvideo", "mock_kinesisvideo")
mock_kinesisvideoarchivedmedia = lazy_load(
".kinesisvideoarchivedmedia",
"mock_kinesisvideoarchivedmedia",
boto3_name="kinesis-video-archived-media",
)
mock_kms = lazy_load(".kms", "mock_kms")
mock_logs = lazy_load(".logs", "mock_logs")
mock_managedblockchain = lazy_load(".managedblockchain", "mock_managedblockchain")
mock_mediaconnect = lazy_load(".mediaconnect", "mock_mediaconnect")
mock_medialive = lazy_load(".medialive", "mock_medialive")
mock_mediapackage = lazy_load(".mediapackage", "mock_mediapackage")
mock_mediastore = lazy_load(".mediastore", "mock_mediastore")
mock_mediastoredata = lazy_load(
".mediastoredata", "mock_mediastoredata", boto3_name="mediastore-data"
)
mock_mq = lazy_load(".mq", "mock_mq", boto3_name="mq")
mock_opsworks = lazy_load(".opsworks", "mock_opsworks")
mock_organizations = lazy_load(".organizations", "mock_organizations")
@ -105,6 +124,7 @@ mock_route53resolver = lazy_load(
mock_s3 = lazy_load(".s3", "mock_s3")
mock_s3control = lazy_load(".s3control", "mock_s3control")
mock_sagemaker = lazy_load(".sagemaker", "mock_sagemaker")
mock_sdb = lazy_load(".sdb", "mock_sdb")
mock_secretsmanager = lazy_load(".secretsmanager", "mock_secretsmanager")
mock_ses = lazy_load(".ses", "mock_ses")
mock_sns = lazy_load(".sns", "mock_sns")
@ -115,6 +135,7 @@ mock_stepfunctions = lazy_load(
".stepfunctions", "mock_stepfunctions", backend="stepfunction_backends"
)
mock_sts = lazy_load(".sts", "mock_sts")
mock_support = lazy_load(".support", "mock_support")
mock_swf = lazy_load(".swf", "mock_swf")
mock_timestreamwrite = lazy_load(
".timestreamwrite", "mock_timestreamwrite", boto3_name="timestream-write"
@ -123,27 +144,7 @@ mock_transcribe = lazy_load(".transcribe", "mock_transcribe")
XRaySegment = lazy_load(".xray", "XRaySegment")
mock_xray = lazy_load(".xray", "mock_xray")
mock_xray_client = lazy_load(".xray", "mock_xray_client")
mock_kinesisvideo = lazy_load(".kinesisvideo", "mock_kinesisvideo")
mock_kinesisvideoarchivedmedia = lazy_load(
".kinesisvideoarchivedmedia",
"mock_kinesisvideoarchivedmedia",
boto3_name="kinesis-video-archived-media",
)
mock_medialive = lazy_load(".medialive", "mock_medialive")
mock_support = lazy_load(".support", "mock_support")
mock_mediaconnect = lazy_load(".mediaconnect", "mock_mediaconnect")
mock_mediapackage = lazy_load(".mediapackage", "mock_mediapackage")
mock_mediastore = lazy_load(".mediastore", "mock_mediastore")
mock_eks = lazy_load(".eks", "mock_eks")
mock_mediastoredata = lazy_load(
".mediastoredata", "mock_mediastoredata", boto3_name="mediastore-data"
)
mock_efs = lazy_load(".efs", "mock_efs")
mock_wafv2 = lazy_load(".wafv2", "mock_wafv2")
mock_sdb = lazy_load(".sdb", "mock_sdb")
mock_elasticache = lazy_load(
".elasticache", "mock_elasticache", boto3_name="elasticache"
)
class MockAll(ContextDecorator):

View File

@ -238,3 +238,10 @@ class BasePathNotFoundException(NotFoundException):
super().__init__(
"NotFoundException", "Invalid base path mapping identifier specified"
)
class VpcLinkNotFound(NotFoundException):
code = 404
def __init__(self):
super().__init__("NotFoundException", "VPCLink not found")

View File

@ -49,6 +49,7 @@ from .exceptions import (
InvalidStageException,
BasePathConflictException,
BasePathNotFoundException,
VpcLinkNotFound,
)
from ..core.models import responses_mock
from moto.apigateway.exceptions import MethodNotFoundException
@ -712,6 +713,17 @@ class UsagePlanKey(BaseModel, dict):
self["value"] = value
class VpcLink(BaseModel, dict):
def __init__(self, name, description, target_arns, tags):
super().__init__()
self["id"] = create_id()
self["name"] = name
self["description"] = description
self["targetArns"] = target_arns
self["tags"] = tags
self["status"] = "AVAILABLE"
class RestAPI(CloudFormationModel):
PROP_ID = "id"
@ -1161,6 +1173,7 @@ class APIGatewayBackend(BaseBackend):
self.models = {}
self.region_name = region_name
self.base_path_mappings = {}
self.vpc_links = {}
def reset(self):
region_name = self.region_name
@ -1881,5 +1894,26 @@ class APIGatewayBackend(BaseBackend):
return base_path_mapping
def create_vpc_link(self, name, description, target_arns, tags):
vpc_link = VpcLink(
name, description=description, target_arns=target_arns, tags=tags
)
self.vpc_links[vpc_link["id"]] = vpc_link
return vpc_link
def delete_vpc_link(self, vpc_link_id):
self.vpc_links.pop(vpc_link_id, None)
def get_vpc_link(self, vpc_link_id):
if vpc_link_id not in self.vpc_links:
raise VpcLinkNotFound
return self.vpc_links[vpc_link_id]
def get_vpc_links(self):
"""
Pagination has not yet been implemented
"""
return list(self.vpc_links.values())
apigateway_backends = BackendDict(APIGatewayBackend, "apigateway")

View File

@ -909,3 +909,34 @@ class APIGatewayResponse(BaseResponse):
return self.error("BadRequestException", e.message)
except InvalidStageException as e:
return self.error("BadRequestException", e.message)
def vpc_link(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
vpc_link_id = url_path_parts[-1]
try:
if self.method == "DELETE":
self.backend.delete_vpc_link(vpc_link_id=vpc_link_id)
return 200, {}, "{}"
if self.method == "GET":
vpc_link = self.backend.get_vpc_link(vpc_link_id=vpc_link_id)
return 200, {}, json.dumps(vpc_link)
except NotFoundException as e:
return self.error("NotFoundException", e.message, 404)
def vpc_links(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "GET":
vpc_links = self.backend.get_vpc_links()
return 200, {}, json.dumps({"item": vpc_links})
if self.method == "POST":
name = self._get_param("name")
description = self._get_param("description")
target_arns = self._get_param("targetArns")
tags = self._get_param("tags")
vpc_link = self.backend.create_vpc_link(
name=name, description=description, target_arns=target_arns, tags=tags
)
return 200, {}, json.dumps(vpc_link)

View File

@ -1,4 +1,5 @@
from .responses import APIGatewayResponse
from ..apigatewayv2.urls import url_paths as url_paths_v2
response = APIGatewayResponse()
@ -34,4 +35,9 @@ url_paths = {
"{0}/usageplans/(?P<usage_plan_id>[^/]+)/keys/(?P<api_key_id>[^/]+)/?$": response.usage_plan_key_individual,
"{0}/restapis/(?P<function_id>[^/]+)/requestvalidators$": response.request_validators,
"{0}/restapis/(?P<api_id>[^/]+)/requestvalidators/(?P<validator_id>[^/]+)/?$": response.request_validator_individual,
"{0}/vpclinks$": response.vpc_links,
"{0}/vpclinks/(?P<vpclink_id>[^/]+)": response.vpc_link,
}
# Also manages the APIGatewayV2
url_paths.update(url_paths_v2)

View File

@ -0,0 +1,5 @@
"""apigatewayv2 module initialization; sets value for base decorator."""
from .models import apigatewayv2_backends
from ..core.models import base_decorator
mock_apigatewayv2 = base_decorator(apigatewayv2_backends)

View File

@ -0,0 +1,95 @@
from moto.core.exceptions import JsonRESTError
class APIGatewayV2Error(JsonRESTError):
pass
class ApiNotFound(APIGatewayV2Error):
code = 404
def __init__(self, api_id):
super().__init__(
"NotFoundException", f"Invalid API identifier specified {api_id}"
)
class AuthorizerNotFound(APIGatewayV2Error):
code = 404
def __init__(self, authorizer_id):
super().__init__(
"NotFoundException",
f"Invalid Authorizer identifier specified {authorizer_id}",
)
class ModelNotFound(APIGatewayV2Error):
code = 404
def __init__(self, model_id):
super().__init__(
"NotFoundException", f"Invalid Model identifier specified {model_id}"
)
class RouteResponseNotFound(APIGatewayV2Error):
code = 404
def __init__(self, rr_id):
super().__init__(
"NotFoundException", f"Invalid RouteResponse identifier specified {rr_id}"
)
class BadRequestException(APIGatewayV2Error):
code = 400
def __init__(self, message):
super().__init__("BadRequestException", message)
class IntegrationNotFound(APIGatewayV2Error):
code = 404
def __init__(self, integration_id):
super().__init__(
"NotFoundException",
f"Invalid Integration identifier specified {integration_id}",
)
class IntegrationResponseNotFound(APIGatewayV2Error):
code = 404
def __init__(self, int_res_id):
super().__init__(
"NotFoundException",
f"Invalid IntegrationResponse identifier specified {int_res_id}",
)
class RouteNotFound(APIGatewayV2Error):
code = 404
def __init__(self, route_id):
super().__init__(
"NotFoundException", f"Invalid Route identifier specified {route_id}"
)
class VpcLinkNotFound(APIGatewayV2Error):
code = 404
def __init__(self, vpc_link_id):
super().__init__(
"NotFoundException", f"Invalid VpcLink identifier specified {vpc_link_id}"
)
class UnknownProtocol(APIGatewayV2Error):
def __init__(self):
super().__init__(
"BadRequestException",
"Invalid protocol specified. Must be one of [HTTP, WEBSOCKET]",
)

1484
moto/apigatewayv2/models.py Normal file

File diff suppressed because it is too large Load Diff

View File

@ -0,0 +1,780 @@
"""Handles incoming apigatewayv2 requests, invokes methods, returns responses."""
import json
from functools import wraps
from moto.core.responses import BaseResponse
from urllib.parse import unquote
from .exceptions import APIGatewayV2Error, UnknownProtocol
from .models import apigatewayv2_backends
def error_handler(f):
@wraps(f)
def _wrapper(*args, **kwargs):
try:
return f(*args, **kwargs)
except APIGatewayV2Error as e:
return e.code, e.get_headers(), e.get_body()
return _wrapper
class ApiGatewayV2Response(BaseResponse):
"""Handler for ApiGatewayV2 requests and responses."""
@property
def apigatewayv2_backend(self):
"""Return backend instance specific for this region."""
return apigatewayv2_backends[self.region]
@error_handler
def apis(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "POST":
return self.create_api()
if self.method == "GET":
return self.get_apis()
@error_handler
def api(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "GET":
return self.get_api()
if self.method == "PATCH":
return self.update_api()
if self.method == "PUT":
return self.reimport_api()
if self.method == "DELETE":
return self.delete_api()
@error_handler
def authorizer(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "DELETE":
return self.delete_authorizer()
if self.method == "GET":
return self.get_authorizer()
if self.method == "PATCH":
return self.update_authorizer()
def authorizers(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "POST":
return self.create_authorizer()
@error_handler
def cors(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "DELETE":
return self.delete_cors_configuration()
def route_request_parameter(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "DELETE":
return self.delete_route_request_parameter()
@error_handler
def model(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "DELETE":
return self.delete_model()
if self.method == "GET":
return self.get_model()
if self.method == "PATCH":
return self.update_model()
def models(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "POST":
return self.create_model()
@error_handler
def integration(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "DELETE":
return self.delete_integration()
if self.method == "GET":
return self.get_integration()
if self.method == "PATCH":
return self.update_integration()
@error_handler
def integrations(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "GET":
return self.get_integrations()
if self.method == "POST":
return self.create_integration()
@error_handler
def integration_response(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "DELETE":
return self.delete_integration_response()
if self.method == "GET":
return self.get_integration_response()
if self.method == "PATCH":
return self.update_integration_response()
def integration_responses(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "GET":
return self.get_integration_responses()
if self.method == "POST":
return self.create_integration_response()
@error_handler
def route(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "DELETE":
return self.delete_route()
if self.method == "GET":
return self.get_route()
if self.method == "PATCH":
return self.update_route()
@error_handler
def routes(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "GET":
return self.get_routes()
if self.method == "POST":
return self.create_route()
@error_handler
def route_response(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "DELETE":
return self.delete_route_response()
if self.method == "GET":
return self.get_route_response()
def route_responses(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "POST":
return self.create_route_response()
@error_handler
def tags(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "POST":
return self.tag_resource()
if self.method == "GET":
return self.get_tags()
if self.method == "DELETE":
return self.untag_resource()
@error_handler
def vpc_link(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if request.method == "DELETE":
return self.delete_vpc_link()
if request.method == "GET":
return self.get_vpc_link()
if request.method == "PATCH":
return self.update_vpc_link()
def vpc_links(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if request.method == "GET":
return self.get_vpc_links()
if request.method == "POST":
return self.create_vpc_link()
def create_api(self):
params = json.loads(self.body)
api_key_selection_expression = params.get("apiKeySelectionExpression")
cors_configuration = params.get("corsConfiguration")
credentials_arn = params.get("credentialsArn")
description = params.get("description")
disable_schema_validation = params.get("disableSchemaValidation")
disable_execute_api_endpoint = params.get("disableExecuteApiEndpoint")
name = params.get("name")
protocol_type = params.get("protocolType")
route_key = params.get("routeKey")
route_selection_expression = params.get("routeSelectionExpression")
tags = params.get("tags")
target = params.get("target")
version = params.get("version")
if protocol_type not in ["HTTP", "WEBSOCKET"]:
raise UnknownProtocol
api = self.apigatewayv2_backend.create_api(
api_key_selection_expression=api_key_selection_expression,
cors_configuration=cors_configuration,
credentials_arn=credentials_arn,
description=description,
disable_schema_validation=disable_schema_validation,
disable_execute_api_endpoint=disable_execute_api_endpoint,
name=name,
protocol_type=protocol_type,
route_key=route_key,
route_selection_expression=route_selection_expression,
tags=tags,
target=target,
version=version,
)
return 200, {}, json.dumps(api.to_json())
def delete_api(self):
api_id = self.path.split("/")[-1]
self.apigatewayv2_backend.delete_api(api_id=api_id)
return 200, "", "{}"
def get_api(self):
api_id = self.path.split("/")[-1]
api = self.apigatewayv2_backend.get_api(api_id=api_id)
return 200, {}, json.dumps(api.to_json())
def get_apis(self):
apis = self.apigatewayv2_backend.get_apis()
return 200, {}, json.dumps({"items": [a.to_json() for a in apis]})
def update_api(self):
api_id = self.path.split("/")[-1]
params = json.loads(self.body)
api_key_selection_expression = params.get("apiKeySelectionExpression")
cors_configuration = params.get("corsConfiguration")
description = params.get("description")
disable_schema_validation = params.get("disableSchemaValidation")
disable_execute_api_endpoint = params.get("disableExecuteApiEndpoint")
name = params.get("name")
route_selection_expression = params.get("routeSelectionExpression")
version = params.get("version")
api = self.apigatewayv2_backend.update_api(
api_id=api_id,
api_key_selection_expression=api_key_selection_expression,
cors_configuration=cors_configuration,
description=description,
disable_schema_validation=disable_schema_validation,
disable_execute_api_endpoint=disable_execute_api_endpoint,
name=name,
route_selection_expression=route_selection_expression,
version=version,
)
return 200, {}, json.dumps(api.to_json())
def reimport_api(self):
api_id = self.path.split("/")[-1]
params = json.loads(self.body)
body = params.get("body")
fail_on_warnings = (
str(self._get_param("failOnWarnings", "false")).lower() == "true"
)
api = self.apigatewayv2_backend.reimport_api(api_id, body, fail_on_warnings)
return 201, {}, json.dumps(api.to_json())
def create_authorizer(self):
api_id = self.path.split("/")[-2]
params = json.loads(self.body)
auth_creds_arn = params.get("authorizerCredentialsArn")
auth_payload_format_version = params.get("authorizerPayloadFormatVersion")
auth_result_ttl = params.get("authorizerResultTtlInSeconds")
authorizer_type = params.get("authorizerType")
authorizer_uri = params.get("authorizerUri")
enable_simple_response = params.get("enableSimpleResponses")
identity_source = params.get("identitySource")
identity_validation_expr = params.get("identityValidationExpression")
jwt_config = params.get("jwtConfiguration")
name = params.get("name")
authorizer = self.apigatewayv2_backend.create_authorizer(
api_id,
auth_creds_arn=auth_creds_arn,
auth_payload_format_version=auth_payload_format_version,
auth_result_ttl=auth_result_ttl,
authorizer_type=authorizer_type,
authorizer_uri=authorizer_uri,
enable_simple_response=enable_simple_response,
identity_source=identity_source,
identity_validation_expr=identity_validation_expr,
jwt_config=jwt_config,
name=name,
)
return 200, {}, json.dumps(authorizer.to_json())
def delete_authorizer(self):
api_id = self.path.split("/")[-3]
authorizer_id = self.path.split("/")[-1]
self.apigatewayv2_backend.delete_authorizer(api_id, authorizer_id)
return 200, {}, "{}"
def get_authorizer(self):
api_id = self.path.split("/")[-3]
authorizer_id = self.path.split("/")[-1]
authorizer = self.apigatewayv2_backend.get_authorizer(api_id, authorizer_id)
return 200, {}, json.dumps(authorizer.to_json())
def update_authorizer(self):
api_id = self.path.split("/")[-3]
authorizer_id = self.path.split("/")[-1]
params = json.loads(self.body)
auth_creds_arn = params.get("authorizerCredentialsArn")
auth_payload_format_version = params.get("authorizerPayloadFormatVersion")
auth_result_ttl = params.get("authorizerResultTtlInSeconds")
authorizer_type = params.get("authorizerType")
authorizer_uri = params.get("authorizerUri")
enable_simple_response = params.get("enableSimpleResponses")
identity_source = params.get("identitySource")
identity_validation_expr = params.get("identityValidationExpression")
jwt_config = params.get("jwtConfiguration")
name = params.get("name")
authorizer = self.apigatewayv2_backend.update_authorizer(
api_id,
authorizer_id=authorizer_id,
auth_creds_arn=auth_creds_arn,
auth_payload_format_version=auth_payload_format_version,
auth_result_ttl=auth_result_ttl,
authorizer_type=authorizer_type,
authorizer_uri=authorizer_uri,
enable_simple_response=enable_simple_response,
identity_source=identity_source,
identity_validation_expr=identity_validation_expr,
jwt_config=jwt_config,
name=name,
)
return 200, {}, json.dumps(authorizer.to_json())
def delete_cors_configuration(self):
api_id = self.path.split("/")[-2]
self.apigatewayv2_backend.delete_cors_configuration(api_id)
return 200, {}, "{}"
def create_model(self):
api_id = self.path.split("/")[-2]
params = json.loads(self.body)
content_type = params.get("contentType")
description = params.get("description")
name = params.get("name")
schema = params.get("schema")
model = self.apigatewayv2_backend.create_model(
api_id, content_type, description, name, schema
)
return 200, {}, json.dumps(model.to_json())
def delete_model(self):
api_id = self.path.split("/")[-3]
model_id = self.path.split("/")[-1]
self.apigatewayv2_backend.delete_model(api_id, model_id)
return 200, {}, "{}"
def get_model(self):
api_id = self.path.split("/")[-3]
model_id = self.path.split("/")[-1]
model = self.apigatewayv2_backend.get_model(api_id, model_id)
return 200, {}, json.dumps(model.to_json())
def update_model(self):
api_id = self.path.split("/")[-3]
model_id = self.path.split("/")[-1]
params = json.loads(self.body)
content_type = params.get("contentType")
description = params.get("description")
name = params.get("name")
schema = params.get("schema")
model = self.apigatewayv2_backend.update_model(
api_id,
model_id,
content_type=content_type,
description=description,
name=name,
schema=schema,
)
return 200, {}, json.dumps(model.to_json())
def get_tags(self):
resource_arn = unquote(self.path.split("/tags/")[1])
tags = self.apigatewayv2_backend.get_tags(resource_arn)
return 200, {}, json.dumps({"tags": tags})
def tag_resource(self):
resource_arn = unquote(self.path.split("/tags/")[1])
tags = json.loads(self.body).get("tags", {})
self.apigatewayv2_backend.tag_resource(resource_arn, tags)
return 201, {}, "{}"
def untag_resource(self):
resource_arn = unquote(self.path.split("/tags/")[1])
tag_keys = self.querystring.get("tagKeys")
self.apigatewayv2_backend.untag_resource(resource_arn, tag_keys)
return 200, {}, "{}"
def create_route(self):
api_id = self.path.split("/")[-2]
params = json.loads(self.body)
api_key_required = params.get("apiKeyRequired", False)
authorization_scopes = params.get("authorizationScopes")
authorization_type = params.get("authorizationType", "NONE")
authorizer_id = params.get("authorizerId")
model_selection_expression = params.get("modelSelectionExpression")
operation_name = params.get("operationName")
request_models = params.get("requestModels")
request_parameters = params.get("requestParameters")
route_key = params.get("routeKey")
route_response_selection_expression = params.get(
"routeResponseSelectionExpression"
)
target = params.get("target")
route = self.apigatewayv2_backend.create_route(
api_id=api_id,
api_key_required=api_key_required,
authorization_scopes=authorization_scopes,
authorization_type=authorization_type,
authorizer_id=authorizer_id,
model_selection_expression=model_selection_expression,
operation_name=operation_name,
request_models=request_models,
request_parameters=request_parameters,
route_key=route_key,
route_response_selection_expression=route_response_selection_expression,
target=target,
)
return 201, {}, json.dumps(route.to_json())
def delete_route(self):
api_id = self.path.split("/")[-3]
route_id = self.path.split("/")[-1]
self.apigatewayv2_backend.delete_route(api_id=api_id, route_id=route_id)
return 200, {}, "{}"
def delete_route_request_parameter(self):
api_id = self.path.split("/")[-5]
route_id = self.path.split("/")[-3]
request_param = self.path.split("/")[-1]
self.apigatewayv2_backend.delete_route_request_parameter(
api_id, route_id, request_param
)
return 200, {}, "{}"
def get_route(self):
api_id = self.path.split("/")[-3]
route_id = self.path.split("/")[-1]
api = self.apigatewayv2_backend.get_route(api_id=api_id, route_id=route_id)
return 200, {}, json.dumps(api.to_json())
def get_routes(self):
api_id = self.path.split("/")[-2]
apis = self.apigatewayv2_backend.get_routes(api_id=api_id)
return 200, {}, json.dumps({"items": [api.to_json() for api in apis]})
def update_route(self):
api_id = self.path.split("/")[-3]
route_id = self.path.split("/")[-1]
params = json.loads(self.body)
api_key_required = params.get("apiKeyRequired")
authorization_scopes = params.get("authorizationScopes")
authorization_type = params.get("authorizationType")
authorizer_id = params.get("authorizerId")
model_selection_expression = params.get("modelSelectionExpression")
operation_name = params.get("operationName")
request_models = params.get("requestModels")
request_parameters = params.get("requestParameters")
route_key = params.get("routeKey")
route_response_selection_expression = params.get(
"routeResponseSelectionExpression"
)
target = params.get("target")
api = self.apigatewayv2_backend.update_route(
api_id=api_id,
api_key_required=api_key_required,
authorization_scopes=authorization_scopes,
authorization_type=authorization_type,
authorizer_id=authorizer_id,
model_selection_expression=model_selection_expression,
operation_name=operation_name,
request_models=request_models,
request_parameters=request_parameters,
route_id=route_id,
route_key=route_key,
route_response_selection_expression=route_response_selection_expression,
target=target,
)
return 200, {}, json.dumps(api.to_json())
def create_route_response(self):
api_id = self.path.split("/")[-4]
route_id = self.path.split("/")[-2]
params = json.loads(self.body)
response_models = params.get("responseModels")
route_response_key = params.get("routeResponseKey")
model_selection_expression = params.get("modelSelectionExpression")
route_response = self.apigatewayv2_backend.create_route_response(
api_id,
route_id,
route_response_key,
model_selection_expression=model_selection_expression,
response_models=response_models,
)
return 200, {}, json.dumps(route_response.to_json())
def delete_route_response(self):
api_id = self.path.split("/")[-5]
route_id = self.path.split("/")[-3]
route_response_id = self.path.split("/")[-1]
self.apigatewayv2_backend.delete_route_response(
api_id, route_id, route_response_id
)
return 200, {}, "{}"
def get_route_response(self):
api_id = self.path.split("/")[-5]
route_id = self.path.split("/")[-3]
route_response_id = self.path.split("/")[-1]
route_response = self.apigatewayv2_backend.get_route_response(
api_id, route_id, route_response_id
)
return 200, {}, json.dumps(route_response.to_json())
def create_integration(self):
api_id = self.path.split("/")[-2]
params = json.loads(self.body)
connection_id = params.get("connectionId")
connection_type = params.get("connectionType")
content_handling_strategy = params.get("contentHandlingStrategy")
credentials_arn = params.get("credentialsArn")
description = params.get("description")
integration_method = params.get("integrationMethod")
integration_subtype = params.get("integrationSubtype")
integration_type = params.get("integrationType")
integration_uri = params.get("integrationUri")
passthrough_behavior = params.get("passthroughBehavior")
payload_format_version = params.get("payloadFormatVersion")
request_parameters = params.get("requestParameters")
request_templates = params.get("requestTemplates")
response_parameters = params.get("responseParameters")
template_selection_expression = params.get("templateSelectionExpression")
timeout_in_millis = params.get("timeoutInMillis")
tls_config = params.get("tlsConfig")
integration = self.apigatewayv2_backend.create_integration(
api_id=api_id,
connection_id=connection_id,
connection_type=connection_type,
content_handling_strategy=content_handling_strategy,
credentials_arn=credentials_arn,
description=description,
integration_method=integration_method,
integration_subtype=integration_subtype,
integration_type=integration_type,
integration_uri=integration_uri,
passthrough_behavior=passthrough_behavior,
payload_format_version=payload_format_version,
request_parameters=request_parameters,
request_templates=request_templates,
response_parameters=response_parameters,
template_selection_expression=template_selection_expression,
timeout_in_millis=timeout_in_millis,
tls_config=tls_config,
)
return 200, {}, json.dumps(integration.to_json())
def get_integration(self):
api_id = self.path.split("/")[-3]
integration_id = self.path.split("/")[-1]
integration = self.apigatewayv2_backend.get_integration(
api_id=api_id, integration_id=integration_id
)
return 200, {}, json.dumps(integration.to_json())
def get_integrations(self):
api_id = self.path.split("/")[-2]
integrations = self.apigatewayv2_backend.get_integrations(api_id=api_id)
return 200, {}, json.dumps({"items": [i.to_json() for i in integrations]})
def delete_integration(self):
api_id = self.path.split("/")[-3]
integration_id = self.path.split("/")[-1]
self.apigatewayv2_backend.delete_integration(
api_id=api_id, integration_id=integration_id,
)
return 200, {}, "{}"
def update_integration(self):
api_id = self.path.split("/")[-3]
integration_id = self.path.split("/")[-1]
params = json.loads(self.body)
connection_id = params.get("connectionId")
connection_type = params.get("connectionType")
content_handling_strategy = params.get("contentHandlingStrategy")
credentials_arn = params.get("credentialsArn")
description = params.get("description")
integration_method = params.get("integrationMethod")
integration_subtype = params.get("integrationSubtype")
integration_type = params.get("integrationType")
integration_uri = params.get("integrationUri")
passthrough_behavior = params.get("passthroughBehavior")
payload_format_version = params.get("payloadFormatVersion")
request_parameters = params.get("requestParameters")
request_templates = params.get("requestTemplates")
response_parameters = params.get("responseParameters")
template_selection_expression = params.get("templateSelectionExpression")
timeout_in_millis = params.get("timeoutInMillis")
tls_config = params.get("tlsConfig")
integration = self.apigatewayv2_backend.update_integration(
api_id=api_id,
connection_id=connection_id,
connection_type=connection_type,
content_handling_strategy=content_handling_strategy,
credentials_arn=credentials_arn,
description=description,
integration_id=integration_id,
integration_method=integration_method,
integration_subtype=integration_subtype,
integration_type=integration_type,
integration_uri=integration_uri,
passthrough_behavior=passthrough_behavior,
payload_format_version=payload_format_version,
request_parameters=request_parameters,
request_templates=request_templates,
response_parameters=response_parameters,
template_selection_expression=template_selection_expression,
timeout_in_millis=timeout_in_millis,
tls_config=tls_config,
)
return 200, {}, json.dumps(integration.to_json())
def create_integration_response(self):
api_id = self.path.split("/")[-4]
int_id = self.path.split("/")[-2]
params = json.loads(self.body)
content_handling_strategy = params.get("contentHandlingStrategy")
integration_response_key = params.get("integrationResponseKey")
response_parameters = params.get("responseParameters")
response_templates = params.get("responseTemplates")
template_selection_expression = params.get("templateSelectionExpression")
integration_response = self.apigatewayv2_backend.create_integration_response(
api_id=api_id,
integration_id=int_id,
content_handling_strategy=content_handling_strategy,
integration_response_key=integration_response_key,
response_parameters=response_parameters,
response_templates=response_templates,
template_selection_expression=template_selection_expression,
)
return 200, {}, json.dumps(integration_response.to_json())
def delete_integration_response(self):
api_id = self.path.split("/")[-5]
int_id = self.path.split("/")[-3]
int_res_id = self.path.split("/")[-1]
self.apigatewayv2_backend.delete_integration_response(
api_id=api_id, integration_id=int_id, integration_response_id=int_res_id
)
return 200, {}, "{}"
def get_integration_response(self):
api_id = self.path.split("/")[-5]
int_id = self.path.split("/")[-3]
int_res_id = self.path.split("/")[-1]
int_response = self.apigatewayv2_backend.get_integration_response(
api_id=api_id, integration_id=int_id, integration_response_id=int_res_id
)
return 200, {}, json.dumps(int_response.to_json())
def get_integration_responses(self):
api_id = self.path.split("/")[-4]
int_id = self.path.split("/")[-2]
int_response = self.apigatewayv2_backend.get_integration_responses(
api_id=api_id, integration_id=int_id
)
return 200, {}, json.dumps({"items": [res.to_json() for res in int_response]})
def update_integration_response(self):
api_id = self.path.split("/")[-5]
int_id = self.path.split("/")[-3]
int_res_id = self.path.split("/")[-1]
params = json.loads(self.body)
content_handling_strategy = params.get("contentHandlingStrategy")
integration_response_key = params.get("integrationResponseKey")
response_parameters = params.get("responseParameters")
response_templates = params.get("responseTemplates")
template_selection_expression = params.get("templateSelectionExpression")
integration_response = self.apigatewayv2_backend.update_integration_response(
api_id=api_id,
integration_id=int_id,
integration_response_id=int_res_id,
content_handling_strategy=content_handling_strategy,
integration_response_key=integration_response_key,
response_parameters=response_parameters,
response_templates=response_templates,
template_selection_expression=template_selection_expression,
)
return 200, {}, json.dumps(integration_response.to_json())
def create_vpc_link(self):
params = json.loads(self.body)
name = params.get("name")
sg_ids = params.get("securityGroupIds")
subnet_ids = params.get("subnetIds")
tags = params.get("tags")
vpc_link = self.apigatewayv2_backend.create_vpc_link(
name, sg_ids, subnet_ids, tags
)
return 200, {}, json.dumps(vpc_link.to_json())
def delete_vpc_link(self):
vpc_link_id = self.path.split("/")[-1]
self.apigatewayv2_backend.delete_vpc_link(vpc_link_id)
return 200, {}, "{}"
def get_vpc_link(self):
vpc_link_id = self.path.split("/")[-1]
vpc_link = self.apigatewayv2_backend.get_vpc_link(vpc_link_id)
return 200, {}, json.dumps(vpc_link.to_json())
def get_vpc_links(self):
vpc_links = self.apigatewayv2_backend.get_vpc_links()
return 200, {}, json.dumps({"items": [l.to_json() for l in vpc_links]})
def update_vpc_link(self):
vpc_link_id = self.path.split("/")[-1]
params = json.loads(self.body)
name = params.get("name")
vpc_link = self.apigatewayv2_backend.update_vpc_link(vpc_link_id, name=name)
return 200, {}, json.dumps(vpc_link.to_json())

32
moto/apigatewayv2/urls.py Normal file
View File

@ -0,0 +1,32 @@
"""apigatewayv2 base URL and path."""
from .responses import ApiGatewayV2Response
url_bases = [
r"https?://apigateway\.(.+)\.amazonaws\.com",
]
response_v2 = ApiGatewayV2Response()
url_paths = {
"{0}/v2/apis$": response_v2.apis,
"{0}/v2/apis/(?P<api_id>[^/]+)$": response_v2.api,
"{0}/v2/apis/(?P<api_id>[^/]+)/authorizers$": response_v2.authorizers,
"{0}/v2/apis/(?P<api_id>[^/]+)/authorizers/(?P<authorizer_id>[^/]+)$": response_v2.authorizer,
"{0}/v2/apis/(?P<api_id>[^/]+)/cors$": response_v2.cors,
"{0}/v2/apis/(?P<api_id>[^/]+)/integrations$": response_v2.integrations,
"{0}/v2/apis/(?P<api_id>[^/]+)/integrations/(?P<integration_id>[^/]+)$": response_v2.integration,
"{0}/v2/apis/(?P<api_id>[^/]+)/integrations/(?P<integration_id>[^/]+)/integrationresponses$": response_v2.integration_responses,
"{0}/v2/apis/(?P<api_id>[^/]+)/integrations/(?P<integration_id>[^/]+)/integrationresponses/(?P<integration_response_id>[^/]+)$": response_v2.integration_response,
"{0}/v2/apis/(?P<api_id>[^/]+)/models$": response_v2.models,
"{0}/v2/apis/(?P<api_id>[^/]+)/models/(?P<model_id>[^/]+)$": response_v2.model,
"{0}/v2/apis/(?P<api_id>[^/]+)/routes$": response_v2.routes,
"{0}/v2/apis/(?P<api_id>[^/]+)/routes/(?P<route_id>[^/]+)$": response_v2.route,
"{0}/v2/apis/(?P<api_id>[^/]+)/routes/(?P<route_id>[^/]+)/routeresponses$": response_v2.route_responses,
"{0}/v2/apis/(?P<api_id>[^/]+)/routes/(?P<route_id>[^/]+)/routeresponses/(?P<route_response_id>[^/]+)$": response_v2.route_response,
"{0}/v2/apis/(?P<api_id>[^/]+)/routes/(?P<route_id>[^/]+)/requestparameters/(?P<request_parameter>[^/]+)$": response_v2.route_request_parameter,
"{0}/v2/tags/(?P<resource_arn>.+)$": response_v2.tags,
"{0}/v2/vpclinks$": response_v2.vpc_links,
"{0}/v2/vpclinks/(?P<vpc_link_id>[^/]+)$": response_v2.vpc_link,
}

View File

@ -334,9 +334,13 @@ class LambdaFunction(CloudFormationModel, DockerModel):
# optional
self.description = spec.get("Description", "")
self.memory_size = spec.get("MemorySize", 128)
self.package_type = spec.get("PackageType", None)
self.publish = spec.get("Publish", False) # this is ignored currently
self.timeout = spec.get("Timeout", 3)
self.layers = self._get_layers_data(spec.get("Layers", []))
self.signing_profile_version_arn = spec.get("SigningProfileVersionArn")
self.signing_job_arn = spec.get("SigningJobArn")
self.code_signing_config_arn = spec.get("CodeSigningConfigArn")
self.logs_group_name = "/aws/lambda/{}".format(self.function_name)
@ -413,6 +417,12 @@ class LambdaFunction(CloudFormationModel, DockerModel):
)
return [{"Arn": lv.arn, "CodeSize": lv.code_size} for lv in layer_versions]
def get_code_signing_config(self):
return {
"CodeSigningConfigArn": self.code_signing_config_arn,
"FunctionName": self.function_name,
}
def get_configuration(self):
config = {
"CodeSha256": self.code_sha_256,
@ -426,10 +436,13 @@ class LambdaFunction(CloudFormationModel, DockerModel):
"Role": self.role,
"Runtime": self.run_time,
"State": self.state,
"PackageType": self.package_type,
"Timeout": self.timeout,
"Version": str(self.version),
"VpcConfig": self.vpc_config,
"Layers": self.layers,
"SigningProfileVersionArn": self.signing_profile_version_arn,
"SigningJobArn": self.signing_job_arn,
}
if self.environment_vars:
config["Environment"] = {"Variables": self.environment_vars}
@ -1454,6 +1467,10 @@ The following environment variables are available for fine-grained control over
fn = self.get_function(function_name)
fn.policy.del_statement(sid, revision)
def get_code_signing_config(self, function_name):
fn = self.get_function(function_name)
return fn.get_code_signing_config()
def get_policy(self, function_name):
fn = self.get_function(function_name)
return fn.policy.get_policy()

View File

@ -154,6 +154,11 @@ class LambdaResponse(BaseResponse):
else:
raise ValueError("Cannot handle request")
def code_signing_config(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if request.method == "GET":
return self._get_code_signing_config()
def function_concurrency(self, request, full_url, headers):
http_method = request.method
self.setup_class(request, full_url, headers)
@ -420,6 +425,11 @@ class LambdaResponse(BaseResponse):
else:
return 404, {}, "{}"
def _get_code_signing_config(self):
function_name = unquote(self.path.rsplit("/", 2)[-2])
resp = self.lambda_backend.get_code_signing_config(function_name)
return 200, {}, json.dumps(resp)
def _get_function_concurrency(self, request):
path_function_name = unquote(self.path.rsplit("/", 2)[-2])
function_name = self.lambda_backend.get_function(path_function_name)

View File

@ -18,6 +18,7 @@ url_paths = {
r"{0}/(?P<api_version>[^/]+)/functions/(?P<function_name>[\w_:%-]+)/policy/?$": response.policy,
r"{0}/(?P<api_version>[^/]+)/functions/(?P<function_name>[\w_:%-]+)/configuration/?$": response.configuration,
r"{0}/(?P<api_version>[^/]+)/functions/(?P<function_name>[\w_:%-]+)/code/?$": response.code,
r"{0}/(?P<api_version>[^/]+)/functions/(?P<function_name>[\w_:%-]+)/code-signing-config$": response.code_signing_config,
r"{0}/(?P<api_version>[^/]+)/functions/(?P<function_name>[\w_:%-]+)/concurrency/?$": response.function_concurrency,
r"{0}/(?P<api_version>[^/]+)/layers/?$": response.list_layers,
r"{0}/(?P<api_version>[^/]+)/layers/(?P<layer_name>[\w_-]+)/versions/?$": response.layers_versions,

View File

@ -15,13 +15,16 @@ output_file = "moto/backend_index.py"
script_dir = os.path.dirname(os.path.abspath(__file__))
output_path = os.path.join(script_dir, "..", output_file)
IGNORE_BACKENDS = ["moto_api", "instance_metadata"]
# Ignore the MotoAPI and InstanceMetadata backend, as they do not represent AWS services
# Ignore the APIGatewayV2, as it's URL's are managed by APIGateway
# Ignore S3bucket_path, as the functionality is covered in the S3 service
IGNORE_BACKENDS = ["moto_api", "instance_metadata", "apigatewayv2", "s3bucket_path"]
def iter_backend_url_patterns():
for backend, (module_name, _) in backends.BACKENDS.items():
if backend in IGNORE_BACKENDS:
break
continue
# otherwise we need to import the module
url_module_name = f"moto.{module_name}.urls"
module = importlib.import_module(url_module_name)

View File

@ -81,6 +81,7 @@ for service_name in [
extras_per_service.update(
{
"apigateway": [_dep_python_jose, _dep_python_jose_ecdsa_pin],
"apigatewayv2": [_dep_PyYAML],
"appsync": [_dep_graphql],
"awslambda": [_dep_docker],
"batch": [_dep_docker],

View File

@ -1,7 +1,13 @@
TestAccAWSAccessKey
TestAccAWSAvailabilityZones
TestAccAWSAPIGatewayV2Authorizer
TestAccAWSAPIGatewayV2IntegrationResponse
TestAccAWSAPIGatewayV2Model
TestAccAWSAPIGatewayV2Route
TestAccAWSAPIGatewayV2RouteResponse
TestAccAWSAPIGatewayV2VpcLink
TestAccAWSAppsyncApiKey
TestAccAWSAppsyncGraphqlApi
TestAccAWSAvailabilityZones
TestAccAWSBillingServiceAccount
TestAccAWSCallerIdentity
TestAccAWSCloudTrailServiceAccount

View File

@ -0,0 +1,110 @@
import boto3
import pytest
from botocore.exceptions import ClientError
from moto import mock_apigateway
@mock_apigateway
def test_get_vpc_links_empty():
client = boto3.client("apigateway", region_name="eu-west-1")
resp = client.get_vpc_links()
resp.should.have.key("items").equals([])
@mock_apigateway
def test_create_vpc_link():
client = boto3.client("apigateway", region_name="eu-west-1")
resp = client.create_vpc_link(
name="vpcl",
description="my first vpc link",
targetArns=["elb:target:arn"],
tags={"key1": "value1"},
)
resp.should.have.key("id")
resp.should.have.key("name").equals("vpcl")
resp.should.have.key("description").equals("my first vpc link")
resp.should.have.key("targetArns").equals(["elb:target:arn"])
resp.should.have.key("status").equals("AVAILABLE")
resp.should.have.key("tags").equals({"key1": "value1"})
@mock_apigateway
def test_get_vpc_link():
client = boto3.client("apigateway", region_name="eu-west-1")
vpc_link_id = client.create_vpc_link(
name="vpcl",
description="my first vpc link",
targetArns=["elb:target:arn"],
tags={"key1": "value1"},
)["id"]
resp = client.get_vpc_link(vpcLinkId=vpc_link_id)
resp.should.have.key("id")
resp.should.have.key("name").equals("vpcl")
resp.should.have.key("description").equals("my first vpc link")
resp.should.have.key("targetArns").equals(["elb:target:arn"])
resp.should.have.key("status").equals("AVAILABLE")
resp.should.have.key("tags").equals({"key1": "value1"})
@mock_apigateway
def test_get_vpc_link_unknown():
client = boto3.client("apigateway", region_name="ap-southeast-1")
with pytest.raises(ClientError) as exc:
client.get_vpc_link(vpcLinkId="unknown")
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
err["Message"].should.equal("VPCLink not found")
@mock_apigateway
def test_get_vpc_links():
client = boto3.client("apigateway", region_name="eu-west-1")
vpc_link_id = client.create_vpc_link(
name="vpcl",
description="my first vpc link",
targetArns=["elb:target:arn"],
tags={"key1": "value1"},
)["id"]
links = client.get_vpc_links()["items"]
links.should.have.length_of(1)
links[0]["id"].should.equal(vpc_link_id)
client.create_vpc_link(
name="vpcl2",
description="my first vpc link",
targetArns=["elb:target:arn"],
tags={"key2": "value2"},
)
links = client.get_vpc_links()["items"]
links.should.have.length_of(2)
@mock_apigateway
def test_delete_vpc_link():
client = boto3.client("apigateway", region_name="eu-north-1")
vpc_link_id = client.create_vpc_link(
name="vpcl",
description="my first vpc link",
targetArns=["elb:target:arn"],
tags={"key1": "value1"},
)["id"]
links = client.get_vpc_links()["items"]
links.should.have.length_of(1)
client.delete_vpc_link(vpcLinkId=vpc_link_id)
links = client.get_vpc_links()["items"]
links.should.have.length_of(0)

View File

View File

@ -0,0 +1,325 @@
"""Unit tests for apigatewayv2-supported APIs."""
import boto3
import pytest
import sure # noqa # pylint: disable=unused-import
from botocore.exceptions import ClientError
from moto import mock_apigatewayv2
# See our Development Tips on writing tests for hints on how to write good tests:
# http://docs.getmoto.org/en/latest/docs/contributing/development_tips/tests.html
@mock_apigatewayv2
def test_create_api_with_unknown_protocol_type():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
with pytest.raises(ClientError) as exc:
client.create_api(Name="test-api", ProtocolType="?")
err = exc.value.response["Error"]
err["Code"].should.equal("BadRequestException")
err["Message"].should.equal(
"Invalid protocol specified. Must be one of [HTTP, WEBSOCKET]"
)
@mock_apigatewayv2
def test_create_api_minimal():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
resp = client.create_api(Name="test-api", ProtocolType="HTTP")
resp.should.have.key("ApiId")
resp.should.have.key("ApiEndpoint").equals(
f"https://{resp['ApiId']}.execute-api.eu-west-1.amazonaws.com"
)
resp.should.have.key("ApiKeySelectionExpression").equals(
"$request.header.x-api-key"
)
resp.should.have.key("CreatedDate")
resp.should.have.key("DisableExecuteApiEndpoint").equals(False)
resp.should.have.key("Name").equals("test-api")
resp.should.have.key("ProtocolType").equals("HTTP")
resp.should.have.key("RouteSelectionExpression").equals(
"$request.method $request.path"
)
@mock_apigatewayv2
def test_create_api():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
resp = client.create_api(
ApiKeySelectionExpression="s3l3ction",
CorsConfiguration={
"AllowCredentials": True,
"AllowHeaders": ["x-header1"],
"AllowMethods": ["GET", "PUT"],
"AllowOrigins": ["google.com"],
"ExposeHeaders": ["x-header1"],
"MaxAge": 2,
},
Description="my first api",
DisableSchemaValidation=True,
DisableExecuteApiEndpoint=True,
Name="test-api",
ProtocolType="HTTP",
RouteSelectionExpression="route_s3l3ction",
Version="1.0",
)
resp.should.have.key("ApiId")
resp.should.have.key("ApiEndpoint").equals(
f"https://{resp['ApiId']}.execute-api.eu-west-1.amazonaws.com"
)
resp.should.have.key("ApiKeySelectionExpression").equals("s3l3ction")
resp.should.have.key("CreatedDate")
resp.should.have.key("CorsConfiguration").equals(
{
"AllowCredentials": True,
"AllowHeaders": ["x-header1"],
"AllowMethods": ["GET", "PUT"],
"AllowOrigins": ["google.com"],
"ExposeHeaders": ["x-header1"],
"MaxAge": 2,
}
)
resp.should.have.key("Description").equals("my first api")
resp.should.have.key("DisableExecuteApiEndpoint").equals(True)
resp.should.have.key("DisableSchemaValidation").equals(True)
resp.should.have.key("Name").equals("test-api")
resp.should.have.key("ProtocolType").equals("HTTP")
resp.should.have.key("RouteSelectionExpression").equals("route_s3l3ction")
resp.should.have.key("Version").equals("1.0")
@mock_apigatewayv2
def test_delete_api():
client = boto3.client("apigatewayv2", region_name="us-east-2")
api_id = client.create_api(Name="t", ProtocolType="HTTP")["ApiId"]
client.delete_api(ApiId=api_id)
with pytest.raises(ClientError) as exc:
client.get_api(ApiId=api_id)
exc.value.response["Error"]["Code"].should.equal("NotFoundException")
@mock_apigatewayv2
def test_delete_cors_configuration():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(
ApiKeySelectionExpression="s3l3ction",
CorsConfiguration={
"AllowCredentials": True,
"AllowHeaders": ["x-header1"],
"AllowMethods": ["GET", "PUT"],
"AllowOrigins": ["google.com"],
"ExposeHeaders": ["x-header1"],
"MaxAge": 2,
},
Description="my first api",
DisableSchemaValidation=True,
DisableExecuteApiEndpoint=True,
Name="test-api",
ProtocolType="HTTP",
RouteSelectionExpression="route_s3l3ction",
Version="1.0",
)["ApiId"]
client.delete_cors_configuration(ApiId=api_id)
resp = client.get_api(ApiId=api_id)
resp.shouldnt.have.key("CorsConfiguration")
resp.should.have.key("Description").equals("my first api")
resp.should.have.key("Name").equals("test-api")
@mock_apigatewayv2
def test_get_api_unknown():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
with pytest.raises(ClientError) as exc:
client.get_api(ApiId="unknown")
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
err["Message"].should.equal("Invalid API identifier specified unknown")
@mock_apigatewayv2
def test_get_api():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
api_id = client.create_api(Name="test-get-api", ProtocolType="WEBSOCKET")["ApiId"]
resp = client.get_api(ApiId=api_id)
resp.should.have.key("ApiId").equals(api_id)
resp.should.have.key("ApiEndpoint").equals(
f"https://{resp['ApiId']}.execute-api.ap-southeast-1.amazonaws.com"
)
resp.should.have.key("ApiKeySelectionExpression").equals(
"$request.header.x-api-key"
)
resp.should.have.key("CreatedDate")
resp.should.have.key("DisableExecuteApiEndpoint").equals(False)
resp.should.have.key("Name").equals("test-get-api")
resp.should.have.key("ProtocolType").equals("WEBSOCKET")
resp.should.have.key("RouteSelectionExpression").equals(
"$request.method $request.path"
)
@mock_apigatewayv2
def test_get_apis():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
client.get_apis().should.have.key("Items").length_of(0)
api_id_1 = client.create_api(Name="api1", ProtocolType="HTTP")["ApiId"]
api_id_2 = client.create_api(Name="api2", ProtocolType="WEBSOCKET")["ApiId"]
client.get_apis().should.have.key("Items").length_of(2)
api_ids = [i["ApiId"] for i in client.get_apis()["Items"]]
api_ids.should.contain(api_id_1)
api_ids.should.contain(api_id_2)
@mock_apigatewayv2
def test_update_api_minimal():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(
ApiKeySelectionExpression="s3l3ction",
CorsConfiguration={
"AllowCredentials": True,
"AllowHeaders": ["x-header1"],
"AllowMethods": ["GET", "PUT"],
"AllowOrigins": ["google.com"],
"ExposeHeaders": ["x-header1"],
"MaxAge": 2,
},
Description="my first api",
DisableSchemaValidation=True,
DisableExecuteApiEndpoint=True,
Name="test-api",
ProtocolType="HTTP",
RouteSelectionExpression="route_s3l3ction",
Version="1.0",
)["ApiId"]
resp = client.update_api(
ApiId=api_id,
CorsConfiguration={
"AllowCredentials": False,
"AllowHeaders": ["x-header2"],
"AllowMethods": ["GET", "PUT"],
"AllowOrigins": ["google.com"],
"ExposeHeaders": ["x-header2"],
"MaxAge": 2,
},
)
resp.should.have.key("ApiId")
resp.should.have.key("ApiEndpoint").equals(
f"https://{resp['ApiId']}.execute-api.eu-west-1.amazonaws.com"
)
resp.should.have.key("ApiKeySelectionExpression").equals("s3l3ction")
resp.should.have.key("CreatedDate")
resp.should.have.key("CorsConfiguration").equals(
{
"AllowCredentials": False,
"AllowHeaders": ["x-header2"],
"AllowMethods": ["GET", "PUT"],
"AllowOrigins": ["google.com"],
"ExposeHeaders": ["x-header2"],
"MaxAge": 2,
}
)
resp.should.have.key("Description").equals("my first api")
resp.should.have.key("DisableExecuteApiEndpoint").equals(True)
resp.should.have.key("DisableSchemaValidation").equals(True)
resp.should.have.key("Name").equals("test-api")
resp.should.have.key("ProtocolType").equals("HTTP")
resp.should.have.key("RouteSelectionExpression").equals("route_s3l3ction")
resp.should.have.key("Version").equals("1.0")
@mock_apigatewayv2
def test_update_api_empty_fields():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(
ApiKeySelectionExpression="s3l3ction",
CorsConfiguration={
"AllowCredentials": True,
"AllowHeaders": ["x-header1"],
"AllowMethods": ["GET", "PUT"],
"AllowOrigins": ["google.com"],
"ExposeHeaders": ["x-header1"],
"MaxAge": 2,
},
Description="my first api",
DisableSchemaValidation=True,
DisableExecuteApiEndpoint=True,
Name="test-api",
ProtocolType="HTTP",
RouteSelectionExpression="route_s3l3ction",
Version="1.0",
)["ApiId"]
resp = client.update_api(ApiId=api_id, Description="", Name="updated", Version="")
resp.should.have.key("ApiId")
resp.should.have.key("ApiEndpoint").equals(
f"https://{resp['ApiId']}.execute-api.eu-west-1.amazonaws.com"
)
resp.should.have.key("ApiKeySelectionExpression").equals("s3l3ction")
resp.should.have.key("Description").equals("")
resp.should.have.key("DisableExecuteApiEndpoint").equals(True)
resp.should.have.key("DisableSchemaValidation").equals(True)
resp.should.have.key("Name").equals("updated")
resp.should.have.key("ProtocolType").equals("HTTP")
resp.should.have.key("RouteSelectionExpression").equals("route_s3l3ction")
resp.should.have.key("Version").equals("")
@mock_apigatewayv2
def test_update_api():
client = boto3.client("apigatewayv2", region_name="us-east-2")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
resp = client.update_api(
ApiId=api_id,
ApiKeySelectionExpression="api_key_s3l3ction",
CorsConfiguration={
"AllowCredentials": True,
"AllowHeaders": ["X-Amz-Target"],
"AllowMethods": ["GET"],
},
CredentialsArn="credentials:arn",
Description="updated API",
DisableSchemaValidation=True,
DisableExecuteApiEndpoint=True,
Name="new name",
RouteKey="route key",
RouteSelectionExpression="route_s3l3ction",
Target="updated target",
Version="1.1",
)
resp.should.have.key("ApiId")
resp.should.have.key("ApiEndpoint").equals(
f"https://{resp['ApiId']}.execute-api.us-east-2.amazonaws.com"
)
resp.should.have.key("ApiKeySelectionExpression").equals("api_key_s3l3ction")
resp.should.have.key("CorsConfiguration").equals(
{
"AllowCredentials": True,
"AllowHeaders": ["X-Amz-Target"],
"AllowMethods": ["GET"],
}
)
resp.should.have.key("CreatedDate")
resp.should.have.key("Description").equals("updated API")
resp.should.have.key("DisableSchemaValidation").equals(True)
resp.should.have.key("DisableExecuteApiEndpoint").equals(True)
resp.should.have.key("Name").equals("new name")
resp.should.have.key("ProtocolType").equals("HTTP")
resp.should.have.key("RouteSelectionExpression").equals("route_s3l3ction")
resp.should.have.key("Version").equals("1.1")

View File

@ -0,0 +1,176 @@
import boto3
import pytest
from botocore.exceptions import ClientError
from moto import mock_apigatewayv2
@mock_apigatewayv2
def test_create_authorizer_minimum():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
resp = client.create_authorizer(
ApiId=api_id, AuthorizerType="REQUEST", IdentitySource=[], Name="auth1"
)
resp.should.have.key("AuthorizerId")
resp.should.have.key("AuthorizerType").equals("REQUEST")
resp.should.have.key("Name").equals("auth1")
@mock_apigatewayv2
def test_create_authorizer():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
resp = client.create_authorizer(
ApiId=api_id,
AuthorizerCredentialsArn="auth:creds:arn",
AuthorizerPayloadFormatVersion="2.0",
AuthorizerResultTtlInSeconds=3,
AuthorizerType="REQUEST",
AuthorizerUri="auth_uri",
EnableSimpleResponses=True,
IdentitySource=["$request.header.Authorization"],
IdentityValidationExpression="ive",
JwtConfiguration={"Audience": ["a1"], "Issuer": "moto.com"},
Name="auth1",
)
resp.should.have.key("AuthorizerId")
resp.should.have.key("AuthorizerCredentialsArn").equals("auth:creds:arn")
resp.should.have.key("AuthorizerPayloadFormatVersion").equals("2.0")
resp.should.have.key("AuthorizerResultTtlInSeconds").equals(3)
resp.should.have.key("AuthorizerType").equals("REQUEST")
resp.should.have.key("AuthorizerUri").equals("auth_uri")
resp.should.have.key("EnableSimpleResponses").equals(True)
resp.should.have.key("IdentitySource").equals(["$request.header.Authorization"])
resp.should.have.key("IdentityValidationExpression").equals("ive")
resp.should.have.key("JwtConfiguration").equals(
{"Audience": ["a1"], "Issuer": "moto.com"}
)
resp.should.have.key("Name").equals("auth1")
@mock_apigatewayv2
def test_get_authorizer():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
authorizer_id = client.create_authorizer(
ApiId=api_id, AuthorizerType="REQUEST", IdentitySource=[], Name="auth1"
)["AuthorizerId"]
resp = client.get_authorizer(ApiId=api_id, AuthorizerId=authorizer_id)
resp.should.have.key("AuthorizerId")
resp.should.have.key("AuthorizerType").equals("REQUEST")
resp.should.have.key("Name").equals("auth1")
@mock_apigatewayv2
def test_delete_authorizer():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
authorizer_id = client.create_authorizer(
ApiId=api_id, AuthorizerType="REQUEST", IdentitySource=[], Name="auth1"
)["AuthorizerId"]
client.delete_authorizer(ApiId=api_id, AuthorizerId=authorizer_id)
with pytest.raises(ClientError) as exc:
client.get_authorizer(ApiId=api_id, AuthorizerId="unknown")
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
@mock_apigatewayv2
def test_get_authorizer_unknown():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
with pytest.raises(ClientError) as exc:
client.get_authorizer(ApiId=api_id, AuthorizerId="unknown")
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
@mock_apigatewayv2
def test_update_authorizer_single():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
auth_id = client.create_authorizer(
ApiId=api_id,
AuthorizerCredentialsArn="auth:creds:arn",
AuthorizerPayloadFormatVersion="2.0",
AuthorizerResultTtlInSeconds=3,
AuthorizerType="REQUEST",
AuthorizerUri="auth_uri",
EnableSimpleResponses=True,
IdentitySource=["$request.header.Authorization"],
IdentityValidationExpression="ive",
JwtConfiguration={"Audience": ["a1"], "Issuer": "moto.com"},
Name="auth1",
)["AuthorizerId"]
resp = client.update_authorizer(ApiId=api_id, AuthorizerId=auth_id, Name="auth2")
resp.should.have.key("AuthorizerId")
resp.should.have.key("AuthorizerCredentialsArn").equals("auth:creds:arn")
resp.should.have.key("AuthorizerPayloadFormatVersion").equals("2.0")
resp.should.have.key("AuthorizerResultTtlInSeconds").equals(3)
resp.should.have.key("AuthorizerType").equals("REQUEST")
resp.should.have.key("AuthorizerUri").equals("auth_uri")
resp.should.have.key("EnableSimpleResponses").equals(True)
resp.should.have.key("IdentitySource").equals(["$request.header.Authorization"])
resp.should.have.key("IdentityValidationExpression").equals("ive")
resp.should.have.key("JwtConfiguration").equals(
{"Audience": ["a1"], "Issuer": "moto.com"}
)
resp.should.have.key("Name").equals("auth2")
@mock_apigatewayv2
def test_update_authorizer_all_attributes():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
auth_id = client.create_authorizer(
ApiId=api_id, AuthorizerType="REQUEST", IdentitySource=[], Name="auth1"
)["AuthorizerId"]
auth_id = client.update_authorizer(
ApiId=api_id,
AuthorizerId=auth_id,
AuthorizerCredentialsArn="",
AuthorizerPayloadFormatVersion="3.0",
AuthorizerResultTtlInSeconds=5,
AuthorizerType="REQUEST",
AuthorizerUri="auth_uri",
EnableSimpleResponses=False,
IdentitySource=["$request.header.Authentication"],
IdentityValidationExpression="ive2",
JwtConfiguration={"Audience": ["a2"], "Issuer": "moto.com"},
Name="auth1",
)["AuthorizerId"]
resp = client.update_authorizer(ApiId=api_id, AuthorizerId=auth_id, Name="auth2")
resp.should.have.key("AuthorizerId")
resp.should.have.key("AuthorizerCredentialsArn").equals("")
resp.should.have.key("AuthorizerPayloadFormatVersion").equals("3.0")
resp.should.have.key("AuthorizerResultTtlInSeconds").equals(5)
resp.should.have.key("AuthorizerType").equals("REQUEST")
resp.should.have.key("AuthorizerUri").equals("auth_uri")
resp.should.have.key("EnableSimpleResponses").equals(False)
resp.should.have.key("IdentitySource").equals(["$request.header.Authentication"])
resp.should.have.key("IdentityValidationExpression").equals("ive2")
resp.should.have.key("JwtConfiguration").equals(
{"Audience": ["a2"], "Issuer": "moto.com"}
)
resp.should.have.key("Name").equals("auth2")

View File

@ -0,0 +1,174 @@
import boto3
import pytest
from botocore.exceptions import ClientError
from moto import mock_apigatewayv2
@mock_apigatewayv2
def test_get_integration_responses_empty():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
int_id = client.create_integration(ApiId=api_id, IntegrationType="HTTP")[
"IntegrationId"
]
resp = client.get_integration_responses(ApiId=api_id, IntegrationId=int_id)
resp.should.have.key("Items").equals([])
@mock_apigatewayv2
def test_create_integration_response():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
int_id = client.create_integration(ApiId=api_id, IntegrationType="HTTP")[
"IntegrationId"
]
int_res = client.create_integration_response(
ApiId=api_id,
ContentHandlingStrategy="CONVERT_TO_BINARY",
IntegrationId=int_id,
IntegrationResponseKey="int_res_key",
ResponseParameters={"x-header": "header-value"},
ResponseTemplates={"t": "template"},
TemplateSelectionExpression="tse",
)
int_res.should.have.key("ContentHandlingStrategy").equals("CONVERT_TO_BINARY")
int_res.should.have.key("IntegrationResponseId")
int_res.should.have.key("IntegrationResponseKey").equals("int_res_key")
int_res.should.have.key("ResponseParameters").equals({"x-header": "header-value"},)
int_res.should.have.key("ResponseTemplates").equals({"t": "template"})
int_res.should.have.key("TemplateSelectionExpression").equals("tse")
@mock_apigatewayv2
def test_get_integration_response():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
int_id = client.create_integration(ApiId=api_id, IntegrationType="HTTP")[
"IntegrationId"
]
int_res_id = client.create_integration_response(
ApiId=api_id,
ContentHandlingStrategy="CONVERT_TO_BINARY",
IntegrationId=int_id,
IntegrationResponseKey="int_res_key",
ResponseParameters={"x-header": "header-value"},
ResponseTemplates={"t": "template"},
TemplateSelectionExpression="tse",
)["IntegrationResponseId"]
int_res = client.get_integration_response(
ApiId=api_id, IntegrationId=int_id, IntegrationResponseId=int_res_id
)
int_res.should.have.key("ContentHandlingStrategy").equals("CONVERT_TO_BINARY")
int_res.should.have.key("IntegrationResponseId")
int_res.should.have.key("IntegrationResponseKey").equals("int_res_key")
int_res.should.have.key("ResponseParameters").equals({"x-header": "header-value"},)
int_res.should.have.key("ResponseTemplates").equals({"t": "template"})
int_res.should.have.key("TemplateSelectionExpression").equals("tse")
@mock_apigatewayv2
def test_get_integration_response_unknown():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
int_id = client.create_integration(ApiId=api_id, IntegrationType="HTTP")[
"IntegrationId"
]
with pytest.raises(ClientError) as exc:
client.get_integration_response(
ApiId=api_id, IntegrationId=int_id, IntegrationResponseId="unknown"
)
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
err["Message"].should.equal(
"Invalid IntegrationResponse identifier specified unknown"
)
@mock_apigatewayv2
def test_delete_integration_response():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
int_id = client.create_integration(ApiId=api_id, IntegrationType="HTTP")[
"IntegrationId"
]
int_res_id = client.create_integration_response(
ApiId=api_id, IntegrationId=int_id, IntegrationResponseKey="int_res_key",
)["IntegrationResponseId"]
client.delete_integration_response(
ApiId=api_id, IntegrationId=int_id, IntegrationResponseId=int_res_id
)
with pytest.raises(ClientError) as exc:
client.get_integration_response(
ApiId=api_id, IntegrationId=int_id, IntegrationResponseId=int_res_id
)
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
@mock_apigatewayv2
def test_update_integration_response_single_attr():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
int_id = client.create_integration(ApiId=api_id, IntegrationType="HTTP")[
"IntegrationId"
]
int_res_id = client.create_integration_response(
ApiId=api_id, IntegrationId=int_id, IntegrationResponseKey="int_res_key",
)["IntegrationResponseId"]
res = client.update_integration_response(
ApiId=api_id,
IntegrationId=int_id,
IntegrationResponseId=int_res_id,
ContentHandlingStrategy="CONVERT_TO_BINARY",
)
res.should.have.key("ContentHandlingStrategy").equals("CONVERT_TO_BINARY")
res.should.have.key("IntegrationResponseId")
res.should.have.key("IntegrationResponseKey").equals("int_res_key")
@mock_apigatewayv2
def test_update_integration_response_multiple_attrs():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
int_id = client.create_integration(ApiId=api_id, IntegrationType="HTTP")[
"IntegrationId"
]
int_res_id = client.create_integration_response(
ApiId=api_id,
IntegrationId=int_id,
IntegrationResponseKey="int_res_key",
ContentHandlingStrategy="CONVERT_TO_BINARY",
)["IntegrationResponseId"]
res = client.update_integration_response(
ApiId=api_id,
IntegrationId=int_id,
IntegrationResponseId=int_res_id,
IntegrationResponseKey="int_res_key2",
ResponseParameters={"x-header": "header-value"},
ResponseTemplates={"t": "template"},
TemplateSelectionExpression="tse",
)
res.should.have.key("ContentHandlingStrategy").equals("CONVERT_TO_BINARY")
res.should.have.key("IntegrationResponseId")
res.should.have.key("IntegrationResponseKey").equals("int_res_key2")
res.should.have.key("ResponseParameters").equals({"x-header": "header-value"},)
res.should.have.key("ResponseTemplates").equals({"t": "template"})
res.should.have.key("TemplateSelectionExpression").equals("tse")

View File

@ -0,0 +1,317 @@
import boto3
import pytest
from botocore.exceptions import ClientError
from moto import mock_apigatewayv2
@mock_apigatewayv2
def test_get_integrations_empty():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
resp = client.get_integrations(ApiId=api_id)
resp.should.have.key("Items").equals([])
@mock_apigatewayv2
def test_create_integration_minimum():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
resp = client.create_integration(ApiId=api_id, IntegrationType="HTTP")
resp.should.have.key("IntegrationId")
resp.should.have.key("IntegrationType").equals("HTTP")
@mock_apigatewayv2
def test_create_integration_for_internet_mock():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
resp = client.create_integration(
ApiId=api_id, ConnectionType="INTERNET", IntegrationType="MOCK"
)
resp.should.have.key("IntegrationId")
resp.should.have.key("IntegrationType").equals("MOCK")
resp.should.have.key("IntegrationResponseSelectionExpression").equals(
"${integration.response.statuscode}"
)
@mock_apigatewayv2
def test_create_integration_full():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
resp = client.create_integration(
ApiId=api_id,
ConnectionId="conn_id",
ConnectionType="INTERNET",
ContentHandlingStrategy="CONVERT_TO_BINARY",
CredentialsArn="cred:arn",
Description="my full integration",
IntegrationMethod="PUT",
IntegrationType="HTTP",
IntegrationSubtype="n/a",
PassthroughBehavior="WHEN_NO_MATCH",
PayloadFormatVersion="1.0",
RequestParameters={"r": "p"},
RequestTemplates={"r": "t"},
ResponseParameters={"res": {"par": "am"}},
TemplateSelectionExpression="tse",
TimeoutInMillis=123,
TlsConfig={"ServerNameToVerify": "server"},
)
resp.should.have.key("IntegrationId")
resp.should.have.key("ConnectionId").equals("conn_id")
resp.should.have.key("ConnectionType").equals("INTERNET")
resp.should.have.key("ContentHandlingStrategy").equals("CONVERT_TO_BINARY")
resp.should.have.key("CredentialsArn").equals("cred:arn")
resp.should.have.key("Description").equals("my full integration")
resp.should.have.key("IntegrationMethod").equals("PUT")
resp.should.have.key("IntegrationType").equals("HTTP")
resp.should.have.key("IntegrationSubtype").equals("n/a")
resp.should.have.key("PassthroughBehavior").equals("WHEN_NO_MATCH")
resp.should.have.key("PayloadFormatVersion").equals("1.0")
resp.should.have.key("RequestParameters").equals({"r": "p"})
resp.should.have.key("RequestTemplates").equals({"r": "t"})
resp.should.have.key("ResponseParameters").equals({"res": {"par": "am"}})
resp.should.have.key("TemplateSelectionExpression").equals("tse")
resp.should.have.key("TimeoutInMillis").equals(123)
resp.should.have.key("TlsConfig").equals({"ServerNameToVerify": "server"})
@mock_apigatewayv2
def test_get_integration():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
integration_id = client.create_integration(ApiId=api_id, IntegrationType="HTTP")[
"IntegrationId"
]
resp = client.get_integration(ApiId=api_id, IntegrationId=integration_id)
resp.should.have.key("IntegrationId")
resp.should.have.key("IntegrationType").equals("HTTP")
@mock_apigatewayv2
def test_get_integration_unknown():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
with pytest.raises(ClientError) as exc:
client.get_integration(ApiId=api_id, IntegrationId="unknown")
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
err["Message"].should.equal("Invalid Integration identifier specified unknown")
@mock_apigatewayv2
def test_get_integrations():
client = boto3.client("apigatewayv2", region_name="us-east-2")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
integration_id = client.create_integration(ApiId=api_id, IntegrationType="HTTP")[
"IntegrationId"
]
resp = client.get_integrations(ApiId=api_id)
resp.should.have.key("Items").length_of(1)
resp["Items"][0].should.have.key("IntegrationId").equals(integration_id)
@mock_apigatewayv2
def test_delete_integration():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
integration_id = client.create_integration(ApiId=api_id, IntegrationType="HTTP")[
"IntegrationId"
]
client.delete_integration(ApiId=api_id, IntegrationId=integration_id)
with pytest.raises(ClientError) as exc:
client.get_integration(ApiId=api_id, IntegrationId=integration_id)
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
@mock_apigatewayv2
def test_update_integration_single_attr():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
int_id = client.create_integration(
ApiId=api_id,
ConnectionId="conn_id",
ConnectionType="INTERNET",
ContentHandlingStrategy="CONVERT_TO_BINARY",
CredentialsArn="cred:arn",
Description="my full integration",
IntegrationMethod="PUT",
IntegrationType="HTTP",
IntegrationSubtype="n/a",
PassthroughBehavior="WHEN_NO_MATCH",
PayloadFormatVersion="1.0",
RequestParameters={"r": "p"},
RequestTemplates={"r": "t"},
ResponseParameters={"res": {"par": "am"}},
TemplateSelectionExpression="tse",
TimeoutInMillis=123,
TlsConfig={"ServerNameToVerify": "server"},
)["IntegrationId"]
resp = client.update_integration(
ApiId=api_id, IntegrationId=int_id, Description="updated int"
)
resp.should.have.key("IntegrationId")
resp.should.have.key("ConnectionId").equals("conn_id")
resp.should.have.key("ConnectionType").equals("INTERNET")
resp.should.have.key("ContentHandlingStrategy").equals("CONVERT_TO_BINARY")
resp.should.have.key("CredentialsArn").equals("cred:arn")
resp.should.have.key("Description").equals("updated int")
resp.should.have.key("IntegrationMethod").equals("PUT")
resp.should.have.key("IntegrationType").equals("HTTP")
resp.should.have.key("IntegrationSubtype").equals("n/a")
resp.should.have.key("PassthroughBehavior").equals("WHEN_NO_MATCH")
resp.should.have.key("PayloadFormatVersion").equals("1.0")
resp.should.have.key("RequestParameters").equals({"r": "p"})
resp.should.have.key("RequestTemplates").equals({"r": "t"})
resp.should.have.key("ResponseParameters").equals({"res": {"par": "am"}})
resp.should.have.key("TemplateSelectionExpression").equals("tse")
resp.should.have.key("TimeoutInMillis").equals(123)
resp.should.have.key("TlsConfig").equals({"ServerNameToVerify": "server"})
@mock_apigatewayv2
def test_update_integration_all_attrs():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
int_id = client.create_integration(
ApiId=api_id,
ConnectionId="conn_id",
ConnectionType="INTERNET",
ContentHandlingStrategy="CONVERT_TO_BINARY",
CredentialsArn="cred:arn",
Description="my full integration",
IntegrationMethod="PUT",
IntegrationType="HTTP",
IntegrationSubtype="n/a",
PassthroughBehavior="WHEN_NO_MATCH",
PayloadFormatVersion="1.0",
RequestParameters={"r": "p"},
RequestTemplates={"r": "t"},
ResponseParameters={"res": {"par": "am"}},
TemplateSelectionExpression="tse",
TimeoutInMillis=123,
TlsConfig={"ServerNameToVerify": "server"},
)["IntegrationId"]
resp = client.update_integration(
ApiId=api_id,
IntegrationId=int_id,
ConnectionType="VPC_LINK",
ContentHandlingStrategy="CONVERT_TO_TEXT",
CredentialsArn="",
IntegrationMethod="PATCH",
IntegrationType="AWS",
PassthroughBehavior="NEVER",
)
resp.should.have.key("IntegrationId")
resp.should.have.key("ConnectionId").equals("conn_id")
resp.should.have.key("ConnectionType").equals("VPC_LINK")
resp.should.have.key("ContentHandlingStrategy").equals("CONVERT_TO_TEXT")
resp.should.have.key("CredentialsArn").equals("")
resp.should.have.key("Description").equals("my full integration")
resp.should.have.key("IntegrationMethod").equals("PATCH")
resp.should.have.key("IntegrationType").equals("AWS")
resp.should.have.key("IntegrationSubtype").equals("n/a")
resp.should.have.key("PassthroughBehavior").equals("NEVER")
resp.should.have.key("PayloadFormatVersion").equals("1.0")
resp.should.have.key("RequestParameters").equals({"r": "p"})
resp.should.have.key("RequestTemplates").equals({"r": "t"})
resp.should.have.key("ResponseParameters").equals({"res": {"par": "am"}})
resp.should.have.key("TemplateSelectionExpression").equals("tse")
resp.should.have.key("TimeoutInMillis").equals(123)
resp.should.have.key("TlsConfig").equals({"ServerNameToVerify": "server"})
@mock_apigatewayv2
def test_update_integration_request_parameters():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
resp = client.create_integration(
ApiId=api_id,
ConnectionType="INTERNET",
IntegrationMethod="ANY",
IntegrationType="HTTP_PROXY",
IntegrationUri="http://www.example.com",
PayloadFormatVersion="1.0",
RequestParameters={
"append:header.header1": "$context.requestId",
"remove:querystring.qs1": "''",
},
ResponseParameters={
"404": {"append:header.error": "$stageVariables.environmentId"},
"500": {
"append:header.header1": "$context.requestId",
"overwrite:statuscode": "403",
},
},
)
int_id = resp["IntegrationId"]
# Having an empty value between quotes ("''") is valid
resp["RequestParameters"].should.equal(
{"append:header.header1": "$context.requestId", "remove:querystring.qs1": "''"}
)
resp = client.update_integration(
ApiId=api_id,
IntegrationId=int_id,
IntegrationType="HTTP_PROXY",
RequestParameters={
"append:header.header1": "$context.accountId",
"overwrite:header.header2": "$stageVariables.environmentId",
"remove:querystring.qs1": "",
},
ResponseParameters={
"404": {},
"500": {
"append:header.header1": "$context.requestId",
"overwrite:statuscode": "403",
},
},
)
resp.should.have.key("IntegrationId")
resp.should.have.key("IntegrationMethod").equals("ANY")
resp.should.have.key("IntegrationType").equals("HTTP_PROXY")
resp.should.have.key("PayloadFormatVersion").equals("1.0")
# Having no value ("") is not valid, so that param should not be persisted
resp.should.have.key("RequestParameters").equals(
{
"append:header.header1": "$context.accountId",
"overwrite:header.header2": "$stageVariables.environmentId",
}
)
resp.should.have.key("ResponseParameters").equals(
{
"404": {},
"500": {
"append:header.header1": "$context.requestId",
"overwrite:statuscode": "403",
},
}
)

View File

@ -0,0 +1,115 @@
import boto3
import pytest
from botocore.exceptions import ClientError
from moto import mock_apigatewayv2
@mock_apigatewayv2
def test_create_model():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
resp = client.create_model(
ApiId=api_id, ContentType="app/xml", Description="desc", Name="nm", Schema="cs"
)
resp.should.have.key("ContentType").equals("app/xml")
resp.should.have.key("Description").equals("desc")
resp.should.have.key("ModelId")
resp.should.have.key("Name").equals("nm")
resp.should.have.key("Schema").equals("cs")
@mock_apigatewayv2
def test_get_model():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
model_id = client.create_model(
ApiId=api_id, ContentType="app/xml", Description="desc", Name="nm", Schema="cs"
)["ModelId"]
resp = client.get_model(ApiId=api_id, ModelId=model_id)
resp.should.have.key("ContentType").equals("app/xml")
resp.should.have.key("Description").equals("desc")
resp.should.have.key("ModelId")
resp.should.have.key("Name").equals("nm")
resp.should.have.key("Schema").equals("cs")
@mock_apigatewayv2
def test_delete_model():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
model_id = client.create_model(
ApiId=api_id, ContentType="app/xml", Description="desc", Name="nm", Schema="cs"
)["ModelId"]
client.delete_model(ApiId=api_id, ModelId=model_id)
with pytest.raises(ClientError) as exc:
client.get_model(ApiId=api_id, ModelId=model_id)
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
@mock_apigatewayv2
def test_get_model_unknown():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
with pytest.raises(ClientError) as exc:
client.get_model(ApiId=api_id, ModelId="unknown")
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
@mock_apigatewayv2
def test_update_model_single_attr():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
model_id = client.create_model(
ApiId=api_id, ContentType="app/xml", Description="desc", Name="nm", Schema="cs"
)["ModelId"]
model_id = client.get_model(ApiId=api_id, ModelId=model_id)["ModelId"]
resp = client.update_model(ApiId=api_id, ModelId=model_id, Schema="cs2")
resp.should.have.key("ContentType").equals("app/xml")
resp.should.have.key("Description").equals("desc")
resp.should.have.key("ModelId")
resp.should.have.key("Name").equals("nm")
resp.should.have.key("Schema").equals("cs2")
@mock_apigatewayv2
def test_update_model_all_attrs():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
model_id = client.create_model(
ApiId=api_id, ContentType="app/xml", Description="desc", Name="nm", Schema="cs"
)["ModelId"]
model_id = client.get_model(ApiId=api_id, ModelId=model_id)["ModelId"]
resp = client.update_model(
ApiId=api_id,
ModelId=model_id,
ContentType="app/html",
Description="html2.x",
Name="html-schema",
Schema="cs2",
)
resp.should.have.key("ContentType").equals("app/html")
resp.should.have.key("Description").equals("html2.x")
resp.should.have.key("Name").equals("html-schema")
resp.should.have.key("Schema").equals("cs2")

View File

@ -0,0 +1,94 @@
import boto3
import pytest
import sure # noqa # pylint: disable=unused-import
from botocore.exceptions import ClientError
from moto import mock_apigatewayv2
@mock_apigatewayv2
def test_reimport_api_standard_fields():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
resp = client.reimport_api(
ApiId=api_id,
Body="---\nopenapi: 3.0.1\ninfo:\n title: tf-acc-test-2983214806752144669_DIFFERENT\n version: 2.0\nx-amazon-apigateway-cors:\n allow_methods:\n - delete\n allow_origins:\n - https://www.google.de\npaths:\n \"/test\":\n get:\n x-amazon-apigateway-integration:\n type: HTTP_PROXY\n httpMethod: GET\n payloadFormatVersion: '1.0'\n uri: https://www.google.de\n",
)
resp.should.have.key("CorsConfiguration").equals({})
resp.should.have.key("Name").equals("tf-acc-test-2983214806752144669_DIFFERENT")
resp.should.have.key("Version").equals("2.0")
@mock_apigatewayv2
def test_reimport_api_failonwarnings():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-import-api", ProtocolType="HTTP")["ApiId"]
with pytest.raises(ClientError) as exc:
client.reimport_api(
ApiId=api_id,
Body='{\n "openapi": "3.0.1",\n "info": {\n "title": "Title test",\n "version": "2.0",\n "description": "Description test"\n },\n "paths": {\n "/update": {\n "get": {\n "x-amazon-apigateway-integration": {\n "type": "HTTP_PROXY",\n "httpMethod": "GET",\n "payloadFormatVersion": "1.0",\n "uri": "https://www.google.de"\n },\n "responses": {\n "200": {\n "description": "Response description",\n "content": {\n "application/json": {\n "schema": {\n "$ref": "#/components/schemas/ModelThatDoesNotExist"\n }\n }\n }\n }\n }\n }\n }\n }\n}\n',
FailOnWarnings=True,
)
err = exc.value.response["Error"]
err["Code"].should.equal("BadRequestException")
err["Message"].should.equal(
"Warnings found during import:\n\tParse issue: attribute paths.'/update'(get).responses.200.content.schema.#/components/schemas/ModelThatDoesNotExist is missing"
)
# Title should not have been updated
resp = client.get_api(ApiId=api_id)
resp.should.have.key("Name").equals("test-import-api")
@mock_apigatewayv2
def test_reimport_api_do_not_failonwarnings():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-import-api", ProtocolType="HTTP")["ApiId"]
client.reimport_api(
ApiId=api_id,
Body='{\n "openapi": "3.0.1",\n "info": {\n "title": "Title test",\n "version": "2.0",\n "description": "Description test"\n },\n "paths": {\n "/update": {\n "get": {\n "x-amazon-apigateway-integration": {\n "type": "HTTP_PROXY",\n "httpMethod": "GET",\n "payloadFormatVersion": "1.0",\n "uri": "https://www.google.de"\n },\n "responses": {\n "200": {\n "description": "Response description",\n "content": {\n "application/json": {\n "schema": {\n "$ref": "#/components/schemas/ModelThatDoesNotExist"\n }\n }\n }\n }\n }\n }\n }\n }\n}\n',
FailOnWarnings=False,
)
# Title should have been updated
resp = client.get_api(ApiId=api_id)
resp.should.have.key("Name").equals("Title test")
@mock_apigatewayv2
def test_reimport_api_routes_and_integrations():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
resp = client.create_api(Name="test-import-api", ProtocolType="HTTP")
api_id = resp["ApiId"]
client.reimport_api(
ApiId=api_id,
Body='{\n "openapi": "3.0.1",\n "info": {\n "title": "tf-acc-test-121780722915762033_DIFFERENT",\n "version": "1.0"\n },\n "paths": {\n "/test": {\n "get": {\n "x-amazon-apigateway-integration": {\n "type": "HTTP_PROXY",\n "httpMethod": "GET",\n "payloadFormatVersion": "1.0",\n "uri": "https://www.google.de"\n }\n }\n }\n }\n}\n',
)
resp = client.get_integrations(ApiId=api_id)
resp.should.have.key("Items").length_of(1)
integration = resp["Items"][0]
integration.should.have.key("ConnectionType").equals("INTERNET")
integration.should.have.key("IntegrationId")
integration.should.have.key("IntegrationMethod").equals("GET")
integration.should.have.key("IntegrationType").equals("HTTP_PROXY")
integration.should.have.key("IntegrationUri").equals("https://www.google.de")
integration.should.have.key("PayloadFormatVersion").equals("1.0")
integration.should.have.key("TimeoutInMillis").equals(30000)
resp = client.get_routes(ApiId=api_id)
resp.should.have.key("Items").length_of(1)
route = resp["Items"][0]
route.should.have.key("ApiKeyRequired").equals(False)
route.should.have.key("AuthorizationScopes").equals([])
route.should.have.key("RequestParameters").equals({})
route.should.have.key("RouteId")
route.should.have.key("RouteKey").equals("GET /test")
route.should.have.key("Target").should.equal(
f"integrations/{integration['IntegrationId']}"
)

View File

@ -0,0 +1,298 @@
import boto3
import pytest
from botocore.exceptions import ClientError
from moto import mock_apigatewayv2
@mock_apigatewayv2
def test_get_routes_empty():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
resp = client.get_routes(ApiId=api_id)
resp.should.have.key("Items").equals([])
@mock_apigatewayv2
def test_create_route_minimal():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
resp = client.create_route(ApiId=api_id, RouteKey="GET /")
resp.should.have.key("ApiKeyRequired").equals(False)
resp.should.have.key("AuthorizationType").equals("NONE")
resp.should.have.key("RouteId")
resp.should.have.key("RouteKey").equals("GET /")
resp.shouldnt.have.key("AuthorizationScopes")
resp.shouldnt.have.key("AuthorizerId")
resp.shouldnt.have.key("ModelSelectionExpression")
resp.shouldnt.have.key("OperationName")
resp.shouldnt.have.key("RequestModels")
resp.shouldnt.have.key("RouteResponseSelectionExpression")
resp.shouldnt.have.key("Target")
@mock_apigatewayv2
def test_create_route_full():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
api_id = client.create_api(
Name="test-api",
ProtocolType="WEBSOCKET",
RouteSelectionExpression="${request.method}",
)["ApiId"]
resp = client.create_route(
ApiId=api_id,
ApiKeyRequired=True,
AuthorizerId="auth_id",
AuthorizationScopes=["scope1", "scope2"],
AuthorizationType="CUSTOM",
ModelSelectionExpression="mse",
OperationName="OP",
RequestModels={"req": "uest"},
RequestParameters={"action": {"Required": True}},
RouteKey="GET /",
RouteResponseSelectionExpression="$default",
Target="t",
)
resp.should.have.key("ApiKeyRequired").equals(True)
resp.should.have.key("AuthorizationType").equals("CUSTOM")
resp.should.have.key("AuthorizationScopes").equals(["scope1", "scope2"])
resp.should.have.key("AuthorizerId").equals("auth_id")
resp.should.have.key("RouteId")
resp.should.have.key("RouteKey").equals("GET /")
resp.should.have.key("ModelSelectionExpression").equals("mse")
resp.should.have.key("OperationName").equals("OP")
resp.should.have.key("RequestModels").equals({"req": "uest"})
resp.should.have.key("RequestParameters").equals({"action": {"Required": True}})
resp.should.have.key("RouteResponseSelectionExpression").equals("$default")
resp.should.have.key("Target").equals("t")
@mock_apigatewayv2
def test_delete_route():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
route_id = client.create_route(ApiId=api_id, RouteKey="GET /")["RouteId"]
client.delete_route(ApiId=api_id, RouteId=route_id)
resp = client.get_routes(ApiId=api_id)
resp.should.have.key("Items").length_of(0)
@mock_apigatewayv2
def test_get_route():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
route_id = client.create_route(ApiId=api_id, RouteKey="GET /")["RouteId"]
resp = client.get_route(ApiId=api_id, RouteId=route_id)
resp.should.have.key("ApiKeyRequired").equals(False)
resp.should.have.key("AuthorizationType").equals("NONE")
resp.should.have.key("RouteId")
resp.should.have.key("RouteKey").equals("GET /")
resp.shouldnt.have.key("AuthorizationScopes")
resp.shouldnt.have.key("AuthorizerId")
resp.shouldnt.have.key("ModelSelectionExpression")
resp.shouldnt.have.key("OperationName")
resp.shouldnt.have.key("RouteResponseSelectionExpression")
resp.shouldnt.have.key("Target")
@mock_apigatewayv2
def test_get_route_unknown():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
with pytest.raises(ClientError) as exc:
client.get_route(ApiId=api_id, RouteId="unknown")
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
err["Message"].should.equal("Invalid Route identifier specified unknown")
@mock_apigatewayv2
def test_get_routes():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
client.create_route(ApiId=api_id, RouteKey="GET /")
resp = client.get_routes(ApiId=api_id)
resp.should.have.key("Items").length_of(1)
@mock_apigatewayv2
def test_update_route_single_attribute():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
route_id = client.create_route(ApiId=api_id, RouteKey="GET /")["RouteId"]
resp = client.update_route(ApiId=api_id, RouteId=route_id, RouteKey="POST /")
resp.should.have.key("ApiKeyRequired").equals(False)
resp.should.have.key("AuthorizationType").equals("NONE")
resp.should.have.key("RouteId").equals(route_id)
resp.should.have.key("RouteKey").equals("POST /")
@mock_apigatewayv2
def test_update_route_all_attributes():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
route_id = client.create_route(ApiId=api_id, ApiKeyRequired=True, RouteKey="GET /")[
"RouteId"
]
resp = client.update_route(
ApiId=api_id,
RouteId=route_id,
ApiKeyRequired=False,
AuthorizationScopes=["scope"],
AuthorizerId="auth_id",
AuthorizationType="JWT",
ModelSelectionExpression="mse",
OperationName="OP",
RequestModels={"req": "uest"},
RequestParameters={"action": {"Required": True}},
RouteResponseSelectionExpression="$default",
Target="t",
)
resp.should.have.key("ApiKeyRequired").equals(False)
resp.should.have.key("AuthorizationType").equals("JWT")
resp.should.have.key("AuthorizationScopes").equals(["scope"])
resp.should.have.key("AuthorizerId").equals("auth_id")
resp.should.have.key("RouteId")
resp.should.have.key("RouteKey").equals("GET /")
resp.should.have.key("ModelSelectionExpression").equals("mse")
resp.should.have.key("OperationName").equals("OP")
resp.should.have.key("RequestModels").equals({"req": "uest"})
resp.should.have.key("RequestParameters").equals({"action": {"Required": True}})
resp.should.have.key("RouteResponseSelectionExpression").equals("$default")
resp.should.have.key("Target").equals("t")
@mock_apigatewayv2
def test_delete_route_request_parameter():
client = boto3.client("apigatewayv2", region_name="us-east-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
route_id = client.create_route(
ApiId=api_id,
RequestParameters={
"action": {"Required": True},
"route.request.header.authorization": {"Required": False},
"zparam": {"Required": False},
},
RouteKey="GET /",
)["RouteId"]
request_params = client.get_route(ApiId=api_id, RouteId=route_id)[
"RequestParameters"
]
request_params.keys().should.have.length_of(3)
client.delete_route_request_parameter(
ApiId=api_id,
RouteId=route_id,
RequestParameterKey="route.request.header.authorization",
)
request_params = client.get_route(ApiId=api_id, RouteId=route_id)[
"RequestParameters"
]
request_params.keys().should.have.length_of(2)
request_params.should.have.key("action")
request_params.should.have.key("zparam")
@mock_apigatewayv2
def test_create_route_response_minimal():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
route_id = client.create_route(ApiId=api_id, RouteKey="GET /")["RouteId"]
resp = client.create_route_response(
ApiId=api_id, RouteId=route_id, RouteResponseKey="$default"
)
resp.should.have.key("RouteResponseId")
resp.should.have.key("RouteResponseKey").equals("$default")
@mock_apigatewayv2
def test_create_route_response():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
route_id = client.create_route(ApiId=api_id, RouteKey="GET /")["RouteId"]
resp = client.create_route_response(
ApiId=api_id,
ModelSelectionExpression="mse",
ResponseModels={"test": "tfacctest5832545056931060873"},
RouteId=route_id,
RouteResponseKey="$default",
)
resp.should.have.key("RouteResponseId")
resp.should.have.key("RouteResponseKey").equals("$default")
resp.should.have.key("ModelSelectionExpression").equals("mse")
resp.should.have.key("ResponseModels").equals(
{"test": "tfacctest5832545056931060873"}
)
@mock_apigatewayv2
def test_get_route_response():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
route_id = client.create_route(ApiId=api_id, RouteKey="GET /")["RouteId"]
route_response_id = client.create_route_response(
ApiId=api_id, RouteId=route_id, RouteResponseKey="$default"
)["RouteResponseId"]
resp = client.get_route_response(
ApiId=api_id, RouteId=route_id, RouteResponseId=route_response_id
)
resp.should.have.key("RouteResponseId")
resp.should.have.key("RouteResponseKey").equals("$default")
@mock_apigatewayv2
def test_get_route_response_unknown():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
route_id = client.create_route(ApiId=api_id, RouteKey="GET /")["RouteId"]
with pytest.raises(ClientError) as exc:
client.get_route_response(
ApiId=api_id, RouteId=route_id, RouteResponseId="unknown"
)
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
@mock_apigatewayv2
def test_delete_route_response_unknown():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
r_id = client.create_route(ApiId=api_id, RouteKey="GET /")["RouteId"]
rr_id = client.create_route_response(
ApiId=api_id, RouteId=r_id, RouteResponseKey="$default"
)["RouteResponseId"]
client.delete_route_response(ApiId=api_id, RouteId=r_id, RouteResponseId=rr_id)
with pytest.raises(ClientError) as exc:
client.get_route_response(ApiId=api_id, RouteId=r_id, RouteResponseId=rr_id)
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")

View File

@ -0,0 +1,62 @@
import boto3
from moto import mock_apigatewayv2
@mock_apigatewayv2
def test_create_api_with_tags():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
resp = client.create_api(
Name="test-api", ProtocolType="HTTP", Tags={"key1": "value1", "key2": "value2"}
)
resp.should.have.key("Tags").equals({"key1": "value1", "key2": "value2"})
@mock_apigatewayv2
def test_tag_resource():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
resource_arn = f"arn:aws:apigateway:eu-west-1::/apis/{api_id}"
client.tag_resource(
ResourceArn=resource_arn, Tags={"key1": "value1", "key2": "value2"}
)
resp = client.get_api(ApiId=api_id)
resp.should.have.key("Tags").equals({"key1": "value1", "key2": "value2"})
@mock_apigatewayv2
def test_get_tags():
client = boto3.client("apigatewayv2", region_name="eu-west-2")
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
resource_arn = f"arn:aws:apigateway:eu-west-2::/apis/{api_id}"
client.tag_resource(
ResourceArn=resource_arn, Tags={"key1": "value1", "key2": "value2"}
)
resp = client.get_tags(ResourceArn=resource_arn)
resp.should.have.key("Tags").equals({"key1": "value1", "key2": "value2"})
@mock_apigatewayv2
def test_untag_resource():
client = boto3.client("apigatewayv2", region_name="eu-west-2")
api_id = client.create_api(
Name="test-api", ProtocolType="HTTP", Tags={"key1": "value1"}
)["ApiId"]
resource_arn = f"arn:aws:apigateway:eu-west-2::/apis/{api_id}"
client.tag_resource(
ResourceArn=resource_arn, Tags={"key2": "value2", "key3": "value3"}
)
client.untag_resource(ResourceArn=resource_arn, TagKeys=["key2"])
resp = client.get_tags(ResourceArn=resource_arn)
resp.should.have.key("Tags").equals({"key1": "value1", "key3": "value3"})

View File

@ -0,0 +1,162 @@
import boto3
import pytest
from botocore.exceptions import ClientError
from moto import mock_apigatewayv2
@mock_apigatewayv2
def test_get_vpc_links_empty():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
resp = client.get_vpc_links()
resp.should.have.key("Items").equals([])
@mock_apigatewayv2
def test_create_vpc_links():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
resp = client.create_vpc_link(
Name="vpcl",
SecurityGroupIds=["sg1", "sg2"],
SubnetIds=["sid1", "sid2"],
Tags={"key1": "value1"},
)
resp.should.have.key("CreatedDate")
resp.should.have.key("Name").equals("vpcl")
resp.should.have.key("SecurityGroupIds").equals(["sg1", "sg2"])
resp.should.have.key("SubnetIds").equals(["sid1", "sid2"])
resp.should.have.key("Tags").equals({"key1": "value1"})
resp.should.have.key("VpcLinkId")
resp.should.have.key("VpcLinkStatus").equals("AVAILABLE")
resp.should.have.key("VpcLinkVersion").equals("V2")
@mock_apigatewayv2
def test_get_vpc_link():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
vpc_link_id = client.create_vpc_link(
Name="vpcl",
SecurityGroupIds=["sg1", "sg2"],
SubnetIds=["sid1", "sid2"],
Tags={"key1": "value1"},
)["VpcLinkId"]
resp = client.get_vpc_link(VpcLinkId=vpc_link_id)
resp.should.have.key("CreatedDate")
resp.should.have.key("Name").equals("vpcl")
resp.should.have.key("SecurityGroupIds").equals(["sg1", "sg2"])
resp.should.have.key("SubnetIds").equals(["sid1", "sid2"])
resp.should.have.key("Tags").equals({"key1": "value1"})
resp.should.have.key("VpcLinkId")
resp.should.have.key("VpcLinkStatus").equals("AVAILABLE")
resp.should.have.key("VpcLinkVersion").equals("V2")
@mock_apigatewayv2
def test_get_vpc_link_unknown():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
with pytest.raises(ClientError) as exc:
client.get_vpc_link(VpcLinkId="unknown")
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
err["Message"].should.equal("Invalid VpcLink identifier specified unknown")
@mock_apigatewayv2
def test_get_vpc_links():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
vpc_link_id = client.create_vpc_link(
Name="vpcl",
SecurityGroupIds=["sg1", "sg2"],
SubnetIds=["sid1", "sid2"],
Tags={"key1": "value1"},
)["VpcLinkId"]
links = client.get_vpc_links()["Items"]
links.should.have.length_of(1)
links[0]["VpcLinkId"].should.equal(vpc_link_id)
client.create_vpc_link(
Name="vpcl",
SecurityGroupIds=["sg1", "sg2"],
SubnetIds=["sid1", "sid2"],
Tags={"key1": "value1"},
)
links = client.get_vpc_links()["Items"]
links.should.have.length_of(2)
@mock_apigatewayv2
def test_delete_vpc_link():
client = boto3.client("apigatewayv2", region_name="eu-north-1")
vpc_link_id = client.create_vpc_link(
Name="vpcl",
SecurityGroupIds=["sg1", "sg2"],
SubnetIds=["sid1", "sid2"],
Tags={"key1": "value1"},
)["VpcLinkId"]
links = client.get_vpc_links()["Items"]
links.should.have.length_of(1)
client.delete_vpc_link(VpcLinkId=vpc_link_id)
links = client.get_vpc_links()["Items"]
links.should.have.length_of(0)
@mock_apigatewayv2
def test_update_vpc_link():
client = boto3.client("apigatewayv2", region_name="eu-north-1")
vpc_link_id = client.create_vpc_link(
Name="vpcl",
SecurityGroupIds=["sg1", "sg2"],
SubnetIds=["sid1", "sid2"],
Tags={"key1": "value1"},
)["VpcLinkId"]
resp = client.update_vpc_link(VpcLinkId=vpc_link_id, Name="vpcl2")
resp.should.have.key("CreatedDate")
resp.should.have.key("Name").equals("vpcl2")
resp.should.have.key("SecurityGroupIds").equals(["sg1", "sg2"])
resp.should.have.key("SubnetIds").equals(["sid1", "sid2"])
resp.should.have.key("Tags").equals({"key1": "value1"})
resp.should.have.key("VpcLinkId")
resp.should.have.key("VpcLinkStatus").equals("AVAILABLE")
resp.should.have.key("VpcLinkVersion").equals("V2")
@mock_apigatewayv2
def test_untag_vpc_link():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
vpc_link_id = client.create_vpc_link(
Name="vpcl",
SecurityGroupIds=["sg1", "sg2"],
SubnetIds=["sid1", "sid2"],
Tags={"Key1": "value1", "key2": "val2"},
)["VpcLinkId"]
arn = f"arn:aws:apigateway:eu-west-1::/vpclinks/{vpc_link_id}"
client.untag_resource(ResourceArn=arn, TagKeys=["Key1"])
resp = client.get_vpc_link(VpcLinkId=vpc_link_id)
resp.should.have.key("CreatedDate")
resp.should.have.key("Name").equals("vpcl")
resp.should.have.key("SecurityGroupIds").equals(["sg1", "sg2"])
resp.should.have.key("SubnetIds").equals(["sid1", "sid2"])
resp.should.have.key("Tags").equals({"key2": "val2"})
resp.should.have.key("VpcLinkId")
resp.should.have.key("VpcLinkStatus").equals("AVAILABLE")
resp.should.have.key("VpcLinkVersion").equals("V2")

View File

@ -0,0 +1,13 @@
import json
import sure # noqa # pylint: disable=unused-import
import moto.server as server
def test_apigatewayv2_list_apis():
backend = server.create_backend_app("apigatewayv2")
test_client = backend.test_client()
resp = test_client.get("/v2/apis")
resp.status_code.should.equal(200)
json.loads(resp.data).should.equal({"items": []})

View File

@ -117,6 +117,7 @@ def test_create_function_from_aws_bucket():
Description="test lambda function",
Timeout=3,
MemorySize=128,
PackageType="ZIP",
Publish=True,
VpcConfig={"SecurityGroupIds": ["sg-123abc"], "SubnetIds": ["subnet-123abc"]},
)
@ -139,6 +140,7 @@ def test_create_function_from_aws_bucket():
"Description": "test lambda function",
"Timeout": 3,
"MemorySize": 128,
"PackageType": "ZIP",
"Version": "1",
"VpcConfig": {
"SecurityGroupIds": ["sg-123abc"],
@ -338,6 +340,38 @@ def test_get_function_configuration(key):
conn.get_function_configuration(FunctionName="junk", Qualifier="$LATEST")
@pytest.mark.parametrize("key", ["FunctionName", "FunctionArn"])
@mock_lambda
@mock_s3
def test_get_function_code_signing_config(key):
bucket_name = str(uuid4())
s3_conn = boto3.client("s3", _lambda_region)
s3_conn.create_bucket(
Bucket=bucket_name,
CreateBucketConfiguration={"LocationConstraint": _lambda_region},
)
zip_content = get_test_zip_file1()
s3_conn.put_object(Bucket=bucket_name, Key="test.zip", Body=zip_content)
conn = boto3.client("lambda", _lambda_region)
function_name = str(uuid4())[0:6]
fxn = conn.create_function(
FunctionName=function_name,
Runtime="python3.7",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"S3Bucket": bucket_name, "S3Key": "test.zip"},
CodeSigningConfigArn="csc:arn",
)
name_or_arn = fxn[key]
result = conn.get_function_code_signing_config(FunctionName=name_or_arn)
result["FunctionName"].should.equal(function_name)
result["CodeSigningConfigArn"].should.equal("csc:arn")
@mock_lambda
@mock_s3
def test_get_function_by_arn():