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

create methods

Browse Source
This commit is contained in:
Steve Pulec 2016-03-01 18:50:06 -05:00
parent 45f92fb4c7
commit a737fbed48
5 changed files with 315 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

119
moto/apigateway/models.py
View File

@ -3,7 +3,67 @@ from __future__ import unicode_literals
import datetime
from moto.core import BaseBackend
from moto.core.utils import iso_8601_datetime_with_milliseconds
from .utils import create_rest_api_id
from .utils import create_id
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, path_part, parent_id):
self.id = id
self.path_part = path_part
self.parent_id = parent_id
self.resource_methods = {
'GET': {}
}
def to_dict(self):
response = {
"path": self.path_part,
"id": self.id,
"resourceMethods": self.resource_methods,
}
if self.parent_id:
response['parent_id'] = self.parent_id
return response
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]
class RestAPI(object):
@ -13,6 +73,9 @@ class RestAPI(object):
self.description = description
self.create_date = datetime.datetime.utcnow()
self.resources = {}
self.add_child('/') # Add default child
def to_dict(self):
return {
"id": self.id,
@ -21,6 +84,12 @@ class RestAPI(object):
"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, path_part=path, parent_id=parent_id)
self.resources[child_id] = child
return child
class APIGatewayBackend(BaseBackend):
def __init__(self, region_name):
@ -34,7 +103,7 @@ class APIGatewayBackend(BaseBackend):
self.__init__(region_name)
def create_rest_api(self, name, description):
api_id = create_rest_api_id()
api_id = create_id()
rest_api = RestAPI(api_id, name, description)
self.apis[api_id] = rest_api
return rest_api
@ -50,6 +119,52 @@ class APIGatewayBackend(BaseBackend):
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
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)

62
moto/apigateway/responses.py
View File

@ -31,10 +31,70 @@ class APIGatewayResponse(BaseResponse):
def restapis_individual(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
function_id = self.path.split("/")[-1]
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)
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())
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)
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)

6
moto/apigateway/urls.py
View File

@ -6,6 +6,10 @@ url_bases = [
]
url_paths = {
'{0}/restapis': APIGatewayResponse().restapis,
'{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>[^/]+)/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,
}

2
moto/apigateway/utils.py
View File

@ -3,7 +3,7 @@ import six
import random
def create_rest_api_id():
def create_id():
size = 10
chars = list(range(10)) + ['A-Z']
return ''.join(six.text_type(random.choice(chars)) for x in range(size))

131
tests/test_apigateway/test_apigateway.py
View File

@ -56,3 +56,134 @@ def test_list_and_delete_apis():
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_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}})