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

Merge pull request #558 from spulec/apigateway

Browse Source 
Apigateway
This commit is contained in:
Steve Pulec 2016-03-05 18:07:20 -05:00
commit aebc7c8ea3
10 changed files with 1081 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
moto/__init__.py
View File

@ -5,6 +5,7 @@ logging.getLogger('boto').setLevel(logging.CRITICAL)
__title__ = 'moto'
__version__ = '0.4.23'
from .apigateway import mock_apigateway # flake8: noqa
from .autoscaling import mock_autoscaling # flake8: noqa
from .awslambda import mock_lambda # flake8: noqa
from .cloudformation import mock_cloudformation # flake8: noqa

12
moto/apigateway/__init__.py Normal file
View File

@ -0,0 +1,12 @@
from __future__ import unicode_literals
from .models import apigateway_backends
from ..core.models import MockAWS
apigateway_backend = apigateway_backends['us-east-1']
def mock_apigateway(func=None):
if func:
return MockAWS(apigateway_backends)(func)
else:
return MockAWS(apigateway_backends)

340
moto/apigateway/models.py Normal file
View File

@ -0,0 +1,340 @@
from __future__ import unicode_literals
import datetime
import httpretty
import requests
from moto.core import BaseBackend
from moto.core.utils import iso_8601_datetime_with_milliseconds
from .utils import create_id
STAGE_URL = "https://{api_id}.execute-api.{region_name}.amazonaws.com/{stage_name}"
class Deployment(dict):
def __init__(self, deployment_id, name):
super(Deployment, self).__init__()
self['id'] = deployment_id
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, 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):
def __init__(self, status_code):
super(MethodResponse, self).__init__()
self['statusCode'] = status_code
class Method(dict):
def __init__(self, method_type, authorization_type):
super(Method, self).__init__()
self.update(dict(
httpMethod=method_type,
authorizationType=authorization_type,
authorizerId=None,
apiKeyRequired=None,
requestParameters=None,
requestModels=None,
methodIntegration=None,
))
self.method_responses = {}
def create_response(self, response_code):
method_response = MethodResponse(response_code)
self.method_responses[response_code] = method_response
return method_response
def get_response(self, response_code):
return self.method_responses[response_code]
def delete_response(self, response_code):
return self.method_responses.pop(response_code)
class Resource(object):
def __init__(self, id, region_name, api_id, path_part, parent_id):
self.id = id
self.region_name = region_name
self.api_id = api_id
self.path_part = path_part
self.parent_id = parent_id
self.resource_methods = {
'GET': {}
}
def to_dict(self):
response = {
"path": self.get_path(),
"id": self.id,
"resourceMethods": self.resource_methods,
}
if self.parent_id:
response['parentId'] = self.parent_id
response['pathPart'] = self.path_part
return response
def get_path(self):
return self.get_parent_path() + self.path_part
def get_parent_path(self):
if self.parent_id:
backend = apigateway_backends[self.region_name]
parent = backend.get_resource(self.api_id, self.parent_id)
parent_path = parent.get_path()
if parent_path != '/': # Root parent
parent_path += '/'
return parent_path
else:
return ''
def get_response(self, request):
integration = self.get_integration(request.method)
integration_type = integration['type']
if integration_type == 'HTTP':
uri = integration['uri']
requests_func = getattr(requests, integration['httpMethod'].lower())
response = requests_func(uri)
else:
raise NotImplementedError("The {0} type has not been implemented".format(integration_type))
return response.status_code, response.text
def add_method(self, method_type, authorization_type):
method = Method(method_type=method_type, authorization_type=authorization_type)
self.resource_methods[method_type] = method
return method
def get_method(self, method_type):
return self.resource_methods[method_type]
def add_integration(self, method_type, integration_type, uri):
integration = Integration(integration_type, uri, method_type)
self.resource_methods[method_type]['methodIntegration'] = integration
return integration
def get_integration(self, method_type):
return self.resource_methods[method_type]['methodIntegration']
def delete_integration(self, method_type):
return self.resource_methods[method_type].pop('methodIntegration')
class RestAPI(object):
def __init__(self, id, region_name, name, description):
self.id = id
self.region_name = region_name
self.name = name
self.description = description
self.create_date = datetime.datetime.utcnow()
self.deployments = {}
self.resources = {}
self.add_child('/') # Add default child
def to_dict(self):
return {
"id": self.id,
"name": self.name,
"description": self.description,
"createdDate": iso_8601_datetime_with_milliseconds(self.create_date),
}
def add_child(self, path, parent_id=None):
child_id = create_id()
child = Resource(id=child_id, region_name=self.region_name, api_id=self.id, path_part=path, parent_id=parent_id)
self.resources[child_id] = child
return child
def get_resource_for_path(self, path_after_stage_name):
for resource in self.resources.values():
if resource.get_path() == path_after_stage_name:
return resource
# TODO deal with no matching resource
def resource_callback(self, request, full_url, headers):
path_after_stage_name = '/'.join(request.path.split("/")[2:])
if not path_after_stage_name:
path_after_stage_name = '/'
resource = self.get_resource_for_path(path_after_stage_name)
status_code, response = resource.get_response(request)
return status_code, headers, response
def update_integration_mocks(self, stage_name):
httpretty.enable()
stage_url = STAGE_URL.format(api_id=self.id, region_name=self.region_name, stage_name=stage_name)
for method in httpretty.httpretty.METHODS:
httpretty.register_uri(method, stage_url, body=self.resource_callback)
def create_deployment(self, name):
deployment_id = create_id()
deployment = Deployment(deployment_id, name)
self.deployments[deployment_id] = deployment
self.update_integration_mocks(name)
return deployment
def get_deployment(self, deployment_id):
return self.deployments[deployment_id]
def get_deployments(self):
return list(self.deployments.values())
def delete_deployment(self, deployment_id):
return self.deployments.pop(deployment_id)
class APIGatewayBackend(BaseBackend):
def __init__(self, region_name):
super(APIGatewayBackend, self).__init__()
self.apis = {}
self.region_name = region_name
def reset(self):
region_name = self.region_name
self.__dict__ = {}
self.__init__(region_name)
def create_rest_api(self, name, description):
api_id = create_id()
rest_api = RestAPI(api_id, self.region_name, name, description)
self.apis[api_id] = rest_api
return rest_api
def get_rest_api(self, function_id):
rest_api = self.apis[function_id]
return rest_api
def list_apis(self):
return self.apis.values()
def delete_rest_api(self, function_id):
rest_api = self.apis.pop(function_id)
return rest_api
def list_resources(self, function_id):
api = self.get_rest_api(function_id)
return api.resources.values()
def get_resource(self, function_id, resource_id):
api = self.get_rest_api(function_id)
resource = api.resources[resource_id]
return resource
def create_resource(self, function_id, parent_resource_id, path_part):
api = self.get_rest_api(function_id)
child = api.add_child(
path=path_part,
parent_id=parent_resource_id,
)
return child
def delete_resource(self, function_id, resource_id):
api = self.get_rest_api(function_id)
resource = api.resources.pop(resource_id)
return resource
def get_method(self, function_id, resource_id, method_type):
resource = self.get_resource(function_id, resource_id)
return resource.get_method(method_type)
def create_method(self, function_id, resource_id, method_type, authorization_type):
resource = self.get_resource(function_id, resource_id)
method = resource.add_method(method_type, authorization_type)
return method
def get_method_response(self, function_id, resource_id, method_type, response_code):
method = self.get_method(function_id, resource_id, method_type)
method_response = method.get_response(response_code)
return method_response
def create_method_response(self, function_id, resource_id, method_type, response_code):
method = self.get_method(function_id, resource_id, method_type)
method_response = method.create_response(response_code)
return method_response
def delete_method_response(self, function_id, resource_id, method_type, response_code):
method = self.get_method(function_id, resource_id, method_type)
method_response = method.delete_response(response_code)
return method_response
def create_integration(self, function_id, resource_id, method_type, integration_type, uri):
resource = self.get_resource(function_id, resource_id)
integration = resource.add_integration(method_type, integration_type, uri)
return integration
def get_integration(self, function_id, resource_id, method_type):
resource = self.get_resource(function_id, resource_id)
return resource.get_integration(method_type)
def delete_integration(self, function_id, resource_id, method_type):
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)
return deployment
def get_deployment(self, function_id, deployment_id):
api = self.get_rest_api(function_id)
return api.get_deployment(deployment_id)
def get_deployments(self, function_id):
api = self.get_rest_api(function_id)
return api.get_deployments()
def delete_deployment(self, function_id, deployment_id):
api = self.get_rest_api(function_id)
return api.delete_deployment(deployment_id)
apigateway_backends = {}
for region_name in ['us-east-1', 'us-west-2', 'eu-west-1', 'ap-northeast-1']: # Not available in boto yet
apigateway_backends[region_name] = APIGatewayBackend(region_name)

160
moto/apigateway/responses.py Normal file
View File

@ -0,0 +1,160 @@
from __future__ import unicode_literals
import json
from moto.core.responses import BaseResponse
from .models import apigateway_backends
class APIGatewayResponse(BaseResponse):
def _get_param(self, key):
return json.loads(self.body.decode("ascii")).get(key)
@property
def backend(self):
return apigateway_backends[self.region]
def restapis(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == 'GET':
apis = self.backend.list_apis()
return 200, headers, json.dumps({"item": [
api.to_dict() for api in apis
]})
elif self.method == 'POST':
name = self._get_param('name')
description = self._get_param('description')
rest_api = self.backend.create_rest_api(name, description)
return 200, headers, json.dumps(rest_api.to_dict())
def restapis_individual(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
function_id = self.path.replace("/restapis/", "", 1).split("/")[0]
if self.method == 'GET':
rest_api = self.backend.get_rest_api(function_id)
return 200, headers, json.dumps(rest_api.to_dict())
elif self.method == 'DELETE':
rest_api = self.backend.delete_rest_api(function_id)
return 200, headers, json.dumps(rest_api.to_dict())
def resources(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
function_id = self.path.replace("/restapis/", "", 1).split("/")[0]
if self.method == 'GET':
resources = self.backend.list_resources(function_id)
return 200, headers, json.dumps({"item": [
resource.to_dict() for resource in resources
]})
def resource_individual(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
function_id = self.path.replace("/restapis/", "", 1).split("/")[0]
resource_id = self.path.split("/")[-1]
if self.method == 'GET':
resource = self.backend.get_resource(function_id, resource_id)
elif self.method == 'POST':
path_part = self._get_param("pathPart")
resource = self.backend.create_resource(function_id, resource_id, path_part)
elif self.method == 'DELETE':
resource = self.backend.delete_resource(function_id, resource_id)
return 200, headers, json.dumps(resource.to_dict())
def resource_methods(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]
if self.method == 'GET':
method = self.backend.get_method(function_id, resource_id, method_type)
return 200, headers, json.dumps(method)
elif self.method == 'PUT':
authorization_type = self._get_param("authorizationType")
method = self.backend.create_method(function_id, resource_id, method_type, authorization_type)
return 200, headers, json.dumps(method)
def resource_method_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]
response_code = url_path_parts[8]
if self.method == 'GET':
method_response = self.backend.get_method_response(function_id, resource_id, method_type, response_code)
elif self.method == 'PUT':
method_response = self.backend.create_method_response(function_id, resource_id, method_type, response_code)
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)
def integrations(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]
if self.method == 'GET':
integration_response = self.backend.get_integration(function_id, resource_id, method_type)
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)
elif self.method == 'DELETE':
integration_response = self.backend.delete_integration(function_id, resource_id, method_type)
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)
function_id = self.path.replace("/restapis/", "", 1).split("/")[0]
if self.method == 'GET':
deployments = self.backend.get_deployments(function_id)
return 200, headers, json.dumps({"item": deployments})
elif self.method == 'POST':
name = self._get_param("stageName")
deployment = self.backend.create_deployment(function_id, name)
return 200, headers, json.dumps(deployment)
def individual_deployment(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
function_id = url_path_parts[2]
deployment_id = url_path_parts[4]
if self.method == 'GET':
deployment = self.backend.get_deployment(function_id, deployment_id)
elif self.method == 'DELETE':
deployment = self.backend.delete_deployment(function_id, deployment_id)
return 200, headers, json.dumps(deployment)

19
moto/apigateway/urls.py Normal file
View File

@ -0,0 +1,19 @@
from __future__ import unicode_literals
from .responses import APIGatewayResponse
url_bases = [
"https?://apigateway.(.+).amazonaws.com"
]
url_paths = {
'{0}/restapis$': APIGatewayResponse().restapis,
'{0}/restapis/(?P<function_id>[^/]+)/?$': APIGatewayResponse().restapis_individual,
'{0}/restapis/(?P<function_id>[^/]+)/resources$': APIGatewayResponse().resources,
'{0}/restapis/(?P<function_id>[^/]+)/deployments$': APIGatewayResponse().deployments,
'{0}/restapis/(?P<function_id>[^/]+)/deployments/(?P<deployment_id>[^/]+)/?$': APIGatewayResponse().individual_deployment,
'{0}/restapis/(?P<function_id>[^/]+)/resources/(?P<resource_id>[^/]+)/?$': APIGatewayResponse().resource_individual,
'{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,
}

9
moto/apigateway/utils.py Normal file
View File

@ -0,0 +1,9 @@
from __future__ import unicode_literals
import six
import random
def create_id():
size = 10
chars = list(range(10)) + ['A-Z']
return ''.join(six.text_type(random.choice(chars)) for x in range(size))

2
moto/backends.py
View File

@ -1,4 +1,5 @@
from __future__ import unicode_literals
from moto.apigateway import apigateway_backend
from moto.autoscaling import autoscaling_backend
from moto.awslambda import lambda_backend
from moto.cloudwatch import cloudwatch_backend
@ -23,6 +24,7 @@ from moto.sts import sts_backend
from moto.route53 import route53_backend
BACKENDS = {
'apigateway': apigateway_backend,
'autoscaling': autoscaling_backend,
'cloudformation': cloudformation_backend,
'cloudwatch': cloudwatch_backend,

6
moto/core/responses.py
View File

@ -94,9 +94,8 @@ class BaseResponse(_TemplateEnvironmentMixin):
def dispatch(cls, *args, **kwargs):
return cls()._dispatch(*args, **kwargs)
def _dispatch(self, request, full_url, headers):
def setup_class(self, request, full_url, headers):
querystring = {}
if hasattr(request, 'body'):
# Boto
self.body = request.body
@ -133,6 +132,9 @@ class BaseResponse(_TemplateEnvironmentMixin):
self.headers = request.headers
self.response_headers = headers
def _dispatch(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
return self.call_action()
def call_action(self):

518
tests/test_apigateway/test_apigateway.py Normal file
View File

@ -0,0 +1,518 @@
from __future__ import unicode_literals
from datetime import datetime
from dateutil.tz import tzutc
import boto3
from freezegun import freeze_time
import httpretty
import requests
import sure # noqa
from moto import mock_apigateway
@freeze_time("2015-01-01")
@mock_apigateway
def test_create_and_get_rest_api():
client = boto3.client('apigateway', region_name='us-west-2')
response = client.create_rest_api(
name='my_api',
description='this is my api',
)
api_id = response['id']
response = client.get_rest_api(
restApiId=api_id
)
response.pop('ResponseMetadata')
response.should.equal({
'id': api_id,
'name': 'my_api',
'description': 'this is my api',
'createdDate': datetime(2015, 1, 1, tzinfo=tzutc())
})
@mock_apigateway
def test_list_and_delete_apis():
client = boto3.client('apigateway', region_name='us-west-2')
response = client.create_rest_api(
name='my_api',
description='this is my api',
)
api_id = response['id']
client.create_rest_api(
name='my_api2',
description='this is my api2',
)
response = client.get_rest_apis()
len(response['items']).should.equal(2)
client.delete_rest_api(
restApiId=api_id
)
response = client.get_rest_apis()
len(response['items']).should.equal(1)
@mock_apigateway
def test_create_resource():
client = boto3.client('apigateway', region_name='us-west-2')
response = client.create_rest_api(
name='my_api',
description='this is my api',
)
api_id = response['id']
resources = client.get_resources(restApiId=api_id)
root_id = [resource for resource in resources['items'] if resource['path'] == '/'][0]['id']
root_resource = client.get_resource(
restApiId=api_id,
resourceId=root_id,
)
root_resource.should.equal({
'path': '/',
'id': root_id,
'ResponseMetadata': {'HTTPStatusCode': 200},
'resourceMethods': {
'GET': {}
}
})
response = client.create_resource(
restApiId=api_id,
parentId=root_id,
pathPart='/users',
)
resources = client.get_resources(restApiId=api_id)['items']
len(resources).should.equal(2)
non_root_resource = [resource for resource in resources if resource['path'] != '/'][0]
response = client.delete_resource(
restApiId=api_id,
resourceId=non_root_resource['id']
)
len(client.get_resources(restApiId=api_id)['items']).should.equal(1)
@mock_apigateway
def test_child_resource():
client = boto3.client('apigateway', region_name='us-west-2')
response = client.create_rest_api(
name='my_api',
description='this is my api',
)
api_id = response['id']
resources = client.get_resources(restApiId=api_id)
root_id = [resource for resource in resources['items'] if resource['path'] == '/'][0]['id']
response = client.create_resource(
restApiId=api_id,
parentId=root_id,
pathPart='users',
)
users_id = response['id']
response = client.create_resource(
restApiId=api_id,
parentId=users_id,
pathPart='tags',
)
tags_id = response['id']
child_resource = client.get_resource(
restApiId=api_id,
resourceId=tags_id,
)
child_resource.should.equal({
'path': '/users/tags',
'pathPart': 'tags',
'parentId': users_id,
'id': tags_id,
'ResponseMetadata': {'HTTPStatusCode': 200},
'resourceMethods': {'GET': {}},
})
@mock_apigateway
def test_create_method():
client = boto3.client('apigateway', region_name='us-west-2')
response = client.create_rest_api(
name='my_api',
description='this is my api',
)
api_id = response['id']
resources = client.get_resources(restApiId=api_id)
root_id = [resource for resource in resources['items'] if resource['path'] == '/'][0]['id']
client.put_method(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET',
authorizationType='none',
)
response = client.get_method(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET'
)
response.should.equal({
'httpMethod': 'GET',
'authorizationType': 'none',
'ResponseMetadata': {'HTTPStatusCode': 200}
})
@mock_apigateway
def test_create_method_response():
client = boto3.client('apigateway', region_name='us-west-2')
response = client.create_rest_api(
name='my_api',
description='this is my api',
)
api_id = response['id']
resources = client.get_resources(restApiId=api_id)
root_id = [resource for resource in resources['items'] if resource['path'] == '/'][0]['id']
client.put_method(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET',
authorizationType='none',
)
response = client.get_method(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET'
)
response = client.put_method_response(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET',
statusCode='200',
)
response.should.equal({
'ResponseMetadata': {'HTTPStatusCode': 200},
'statusCode': '200'
})
response = client.get_method_response(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET',
statusCode='200',
)
response.should.equal({
'ResponseMetadata': {'HTTPStatusCode': 200},
'statusCode': '200'
})
response = client.delete_method_response(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET',
statusCode='200',
)
response.should.equal({'ResponseMetadata': {'HTTPStatusCode': 200}})
@mock_apigateway
def test_integrations():
client = boto3.client('apigateway', region_name='us-west-2')
response = client.create_rest_api(
name='my_api',
description='this is my api',
)
api_id = response['id']
resources = client.get_resources(restApiId=api_id)
root_id = [resource for resource in resources['items'] if resource['path'] == '/'][0]['id']
client.put_method(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET',
authorizationType='none',
)
client.put_method_response(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET',
statusCode='200',
)
response = client.put_integration(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET',
type='HTTP',
uri='http://httpbin.org/robots.txt',
)
response.should.equal({
'ResponseMetadata': {'HTTPStatusCode': 200},
'httpMethod': 'GET',
'integrationResponses': {
'200': {
'responseTemplates': {
'application/json': None
},
'statusCode': 200
}
},
'type': 'HTTP',
'uri': 'http://httpbin.org/robots.txt'
})
response = client.get_integration(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET'
)
response.should.equal({
'ResponseMetadata': {'HTTPStatusCode': 200},
'httpMethod': 'GET',
'integrationResponses': {
'200': {
'responseTemplates': {
'application/json': None
},
'statusCode': 200
}
},
'type': 'HTTP',
'uri': 'http://httpbin.org/robots.txt'
})
response = client.get_resource(
restApiId=api_id,
resourceId=root_id,
)
response['resourceMethods']['GET']['methodIntegration'].should.equal({
'httpMethod': 'GET',
'integrationResponses': {
'200': {
'responseTemplates': {
'application/json': None
},
'statusCode': 200
}
},
'type': 'HTTP',
'uri': 'http://httpbin.org/robots.txt'
})
client.delete_integration(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET'
)
response = client.get_resource(
restApiId=api_id,
resourceId=root_id,
)
response['resourceMethods']['GET'].shouldnt.contain("methodIntegration")
@mock_apigateway
def test_integration_response():
client = boto3.client('apigateway', region_name='us-west-2')
response = client.create_rest_api(
name='my_api',
description='this is my api',
)
api_id = response['id']
resources = client.get_resources(restApiId=api_id)
root_id = [resource for resource in resources['items'] if resource['path'] == '/'][0]['id']
client.put_method(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET',
authorizationType='none',
)
client.put_method_response(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET',
statusCode='200',
)
response = client.put_integration(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET',
type='HTTP',
uri='http://httpbin.org/robots.txt',
)
response = client.put_integration_response(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET',
statusCode='200',
selectionPattern='foobar',
)
response.should.equal({
'statusCode': '200',
'selectionPattern': 'foobar',
'ResponseMetadata': {'HTTPStatusCode': 200},
'responseTemplates': {
'application/json': None
}
})
response = client.get_integration_response(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET',
statusCode='200',
)
response.should.equal({
'statusCode': '200',
'selectionPattern': 'foobar',
'ResponseMetadata': {'HTTPStatusCode': 200},
'responseTemplates': {
'application/json': None
}
})
response = client.get_method(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET',
)
response['methodIntegration']['integrationResponses'].should.equal({
'200': {
'responseTemplates': {
'application/json': None
},
'selectionPattern': 'foobar',
'statusCode': '200'
}
})
response = client.delete_integration_response(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET',
statusCode='200',
)
response = client.get_method(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET',
)
response['methodIntegration']['integrationResponses'].should.equal({})
@mock_apigateway
def test_deployment():
client = boto3.client('apigateway', region_name='us-west-2')
response = client.create_rest_api(
name='my_api',
description='this is my api',
)
api_id = response['id']
response = client.create_deployment(
restApiId=api_id,
stageName='staging',
)
deployment_id = response['id']
response = client.get_deployment(
restApiId=api_id,
deploymentId=deployment_id,
)
response.should.equal({
'id': deployment_id,
'ResponseMetadata': {'HTTPStatusCode': 200}
})
response = client.get_deployments(
restApiId=api_id,
)
response['items'].should.equal([
{'id': deployment_id}
])
response = client.delete_deployment(
restApiId=api_id,
deploymentId=deployment_id,
)
response = client.get_deployments(
restApiId=api_id,
)
len(response['items']).should.equal(0)
@httpretty.activate
@mock_apigateway
def test_http_proxying_integration():
httpretty.register_uri(
httpretty.GET, "http://httpbin.org/robots.txt", body='a fake response'
)
region_name = 'us-west-2'
client = boto3.client('apigateway', region_name=region_name)
response = client.create_rest_api(
name='my_api',
description='this is my api',
)
api_id = response['id']
resources = client.get_resources(restApiId=api_id)
root_id = [resource for resource in resources['items'] if resource['path'] == '/'][0]['id']
client.put_method(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET',
authorizationType='none',
)
client.put_method_response(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET',
statusCode='200',
)
response = client.put_integration(
restApiId=api_id,
resourceId=root_id,
httpMethod='GET',
type='HTTP',
uri='http://httpbin.org/robots.txt',
)
stage_name = 'staging'
client.create_deployment(
restApiId=api_id,
stageName=stage_name,
)
deploy_url = "https://{api_id}.execute-api.{region_name}.amazonaws.com/{stage_name}".format(api_id=api_id, region_name=region_name, stage_name=stage_name)
requests.get(deploy_url).content.should.equal(b"a fake response")

16
tests/test_apigateway/test_server.py Normal file
View File

@ -0,0 +1,16 @@
from __future__ import unicode_literals
import sure # noqa
import moto.server as server
'''
Test the different server responses
'''
def test_list_apis():
backend = server.create_backend_app("apigateway")
test_client = backend.test_client()
res = test_client.get('/restapis')
res.data.should.equal(b'{"item": []}')