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

added more api gateway coverage

Browse source 
added: get_stages, stage not found exception, update stage configuration, descriptions on deployments, setting stage variables on deployments and stage creating
This commit is contained in:
dominictootell 2016-09-13 12:44:17 +01:00
commit 1aaafc5f07
5 changed files with 427 additions and 19 deletions
Download patch file Download diff file Expand all files Collapse all files

40
moto/apigateway/responses.py
View file

@ -4,6 +4,7 @@ import json
from moto.core.responses import BaseResponse
from .models import apigateway_backends
from .exceptions import StageNonFoundException
class APIGatewayResponse(BaseResponse):
@ -11,6 +12,15 @@ class APIGatewayResponse(BaseResponse):
def _get_param(self, key):
return json.loads(self.body.decode("ascii")).get(key)
def _get_param_with_default_value(self, key, default):
jsonbody = json.loads(self.body.decode("ascii"))
if key in jsonbody:
return jsonbody.get(key)
else:
return default
@property
def backend(self):
return apigateway_backends[self.region]
@ -95,6 +105,23 @@ class APIGatewayResponse(BaseResponse):
method_response = self.backend.delete_method_response(function_id, resource_id, method_type, response_code)
return 200, headers, json.dumps(method_response)
def restapis_stages(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
function_id = url_path_parts[2]
if self.method == 'POST':
stage_name = self._get_param("stageName")
deployment_id = self._get_param("deploymentId")
stage_variables = self._get_param_with_default_value('variables',{})
stage_response = self.backend.create_stage(function_id, stage_name, deployment_id,variables=stage_variables)
elif self.method == 'GET':
stages = self.backend.get_stages(function_id)
return 200, headers, json.dumps({"item": stages})
return 200, headers, json.dumps(stage_response)
def stages(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/")
@ -102,10 +129,13 @@ class APIGatewayResponse(BaseResponse):
stage_name = url_path_parts[4]
if self.method == 'GET':
stage_response = self.backend.get_stage(function_id, stage_name)
try:
stage_response = self.backend.get_stage(function_id, stage_name)
except StageNonFoundException as error:
return error.code, headers,'{{"message":"{0}","code":"{1}"}}'.format(error.message,error.error_type)
elif self.method == 'PATCH':
path_operations = self._get_param('patchOperations')
stage_response = self.backend.update_stage(function_id, stage_name, path_operations)
patch_operations = self._get_param('patchOperations')
stage_response = self.backend.update_stage(function_id, stage_name, patch_operations)
return 200, headers, json.dumps(stage_response)
def integrations(self, request, full_url, headers):
@ -158,7 +188,9 @@ class APIGatewayResponse(BaseResponse):
return 200, headers, json.dumps({"item": deployments})
elif self.method == 'POST':
name = self._get_param("stageName")
deployment = self.backend.create_deployment(function_id, name)
description = self._get_param_with_default_value("description","")
stage_variables = self._get_param_with_default_value('variables',{})
deployment = self.backend.create_deployment(function_id, name, description,stage_variables)
return 200, headers, json.dumps(deployment)
def individual_deployment(self, request, full_url, headers):