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

add integration responses.

Browse source
This commit is contained in:
Steve Pulec 2016-03-05 09:48:37 -05:00
commit 95f9b3fb35
4 changed files with 194 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

42
moto/apigateway/models.py
View file

@ -13,11 +13,34 @@ class Deployment(dict):
self['stageName'] = name
class IntegrationResponse(dict):
def __init__(self, status_code, selection_pattern=None):
self['responseTemplates'] = {"application/json": None}
self['statusCode'] = status_code
if selection_pattern:
self['selectionPattern'] = selection_pattern
class Integration(dict):
def __init__(self, integration_type, uri):
def __init__(self, integration_type, uri, http_method):
super(Integration, self).__init__()
self['type'] = integration_type
self['uri'] = uri
self['httpMethod'] = http_method
self["integrationResponses"] = {
"200": IntegrationResponse(200)
}
def create_integration_response(self, status_code, selection_pattern):
integration_response = IntegrationResponse(status_code, selection_pattern)
self["integrationResponses"][status_code] = integration_response
return integration_response
def get_integration_response(self, status_code):
return self["integrationResponses"][status_code]
def delete_integration_response(self, status_code):
return self["integrationResponses"].pop(status_code)
class MethodResponse(dict):
@ -80,7 +103,7 @@ class Resource(object):
return self.resource_methods[method_type]
def add_integration(self, method_type, integration_type, uri):
integration = Integration(integration_type, uri)
integration = Integration(integration_type, uri, method_type)
self.resource_methods[method_type]['methodIntegration'] = integration
return integration
@ -220,6 +243,21 @@ class APIGatewayBackend(BaseBackend):
resource = self.get_resource(function_id, resource_id)
return resource.delete_integration(method_type)
def create_integration_response(self, function_id, resource_id, method_type, status_code, selection_pattern):
integration = self.get_integration(function_id, resource_id, method_type)
integration_response = integration.create_integration_response(status_code, selection_pattern)
return integration_response
def get_integration_response(self, function_id, resource_id, method_type, status_code):
integration = self.get_integration(function_id, resource_id, method_type)
integration_response = integration.get_integration_response(status_code)
return integration_response
def delete_integration_response(self, function_id, resource_id, method_type, status_code):
integration = self.get_integration(function_id, resource_id, method_type)
integration_response = integration.delete_integration_response(status_code)
return integration_response
def create_deployment(self, function_id, name):
api = self.get_rest_api(function_id)
deployment = api.create_deployment(name)

38
moto/apigateway/responses.py
View file

@ -57,14 +57,12 @@ class APIGatewayResponse(BaseResponse):
if self.method == 'GET':
resource = self.backend.get_resource(function_id, resource_id)
return 200, headers, json.dumps(resource.to_dict())
elif self.method == 'POST':
path_part = self._get_param("pathPart")
resource = self.backend.create_resource(function_id, resource_id, path_part)
return 200, headers, json.dumps(resource.to_dict())
elif self.method == 'DELETE':
resource = self.backend.delete_resource(function_id, resource_id)
return 200, headers, json.dumps(resource.to_dict())
return 200, headers, json.dumps(resource.to_dict())
def resource_methods(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
@ -91,13 +89,11 @@ class APIGatewayResponse(BaseResponse):
if self.method == 'GET':
method_response = self.backend.get_method_response(function_id, resource_id, method_type, response_code)
return 200, headers, json.dumps(method_response)
elif self.method == 'PUT':
method_response = self.backend.create_method_response(function_id, resource_id, method_type, response_code)
return 200, headers, json.dumps(method_response)
elif self.method == 'DELETE':
method_response = self.backend.delete_method_response(function_id, resource_id, method_type, response_code)
return 200, headers, json.dumps(method_response)
return 200, headers, json.dumps(method_response)
def integrations(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
@ -108,15 +104,36 @@ class APIGatewayResponse(BaseResponse):
if self.method == 'GET':
integration_response = self.backend.get_integration(function_id, resource_id, method_type)
return 200, headers, json.dumps(integration_response)
elif self.method == 'PUT':
integration_type = self._get_param('type')
uri = self._get_param('uri')
integration_response = self.backend.create_integration(function_id, resource_id, method_type, integration_type, uri)
return 200, headers, json.dumps(integration_response)
elif self.method == 'DELETE':
integration_response = self.backend.delete_integration(function_id, resource_id, method_type)
return 200, headers, json.dumps(integration_response)
return 200, headers, json.dumps(integration_response)
def integration_responses(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
function_id = url_path_parts[2]
resource_id = url_path_parts[4]
method_type = url_path_parts[6]
status_code = url_path_parts[9]
if self.method == 'GET':
integration_response = self.backend.get_integration_response(
function_id, resource_id, method_type, status_code
)
elif self.method == 'PUT':
selection_pattern = self._get_param("selectionPattern")
integration_response = self.backend.create_integration_response(
function_id, resource_id, method_type, status_code, selection_pattern
)
elif self.method == 'DELETE':
integration_response = self.backend.delete_integration_response(
function_id, resource_id, method_type, status_code
)
return 200, headers, json.dumps(integration_response)
def deployments(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
@ -138,7 +155,6 @@ class APIGatewayResponse(BaseResponse):
if self.method == 'GET':
deployment = self.backend.get_deployment(function_id, deployment_id)
return 200, headers, json.dumps(deployment)
elif self.method == 'DELETE':
deployment = self.backend.delete_deployment(function_id, deployment_id)
return 200, headers, json.dumps(deployment)
return 200, headers, json.dumps(deployment)

1
moto/apigateway/urls.py
View file

@ -15,4 +15,5 @@ url_paths = {
'{0}/restapis/(?P<function_id>[^/]+)/resources/(?P<resource_id>[^/]+)/methods/(?P<method_name>[^/]+)/?$': APIGatewayResponse().resource_methods,
'{0}/restapis/(?P<function_id>[^/]+)/resources/(?P<resource_id>[^/]+)/methods/(?P<method_name>[^/]+)/responses/200$': APIGatewayResponse().resource_method_responses,
'{0}/restapis/(?P<function_id>[^/]+)/resources/(?P<resource_id>[^/]+)/methods/(?P<method_name>[^/]+)/integration/?$': APIGatewayResponse().integrations,
'{0}/restapis/(?P<function_id>[^/]+)/resources/(?P<resource_id>[^/]+)/methods/(?P<method_name>[^/]+)/integration/responses/(?P<status_code>\d+)/?$': APIGatewayResponse().integration_responses,
}