Fixed apigateway usage plan api when dealing with non existing usage plans and non existing api keys

This commit is contained in:
Alessandro Palumbo 2020-04-27 19:39:33 +02:00
commit 05860fcdd1
5 changed files with 101 additions and 2 deletions

View file

@ -112,6 +112,15 @@ class ApiKeyNotFoundException(RESTError):
)
class UsagePlanNotFoundException(RESTError):
code = 404
def __init__(self):
super(UsagePlanNotFoundException, self).__init__(
"NotFoundException", "Invalid Usage Plan ID specified"
)
class ApiKeyAlreadyExists(RESTError):
code = 409

View file

@ -20,6 +20,7 @@ from moto.core.utils import path_url
from moto.sts.models import ACCOUNT_ID
from .exceptions import (
ApiKeyNotFoundException,
UsagePlanNotFoundException,
AwsProxyNotAllowed,
CrossAccountNotAllowed,
IntegrationMethodNotDefined,
@ -1045,6 +1046,9 @@ class APIGatewayBackend(BaseBackend):
return plans
def get_usage_plan(self, usage_plan_id):
if usage_plan_id not in self.usage_plans:
raise UsagePlanNotFoundException()
return self.usage_plans[usage_plan_id]
def delete_usage_plan(self, usage_plan_id):
@ -1077,6 +1081,17 @@ class APIGatewayBackend(BaseBackend):
return list(self.usage_plan_keys[usage_plan_id].values())
def get_usage_plan_key(self, usage_plan_id, key_id):
# first check if is a valid api key
if key_id not in self.keys:
raise ApiKeyNotFoundException()
# then check if is a valid api key and that the key is in the plan
if (
usage_plan_id not in self.usage_plan_keys
or key_id not in self.usage_plan_keys[usage_plan_id]
):
raise UsagePlanNotFoundException()
return self.usage_plan_keys[usage_plan_id][key_id]
def delete_usage_plan_key(self, usage_plan_id, key_id):

View file

@ -6,6 +6,7 @@ from moto.core.responses import BaseResponse
from .models import apigateway_backends
from .exceptions import (
ApiKeyNotFoundException,
UsagePlanNotFoundException,
BadRequestException,
CrossAccountNotAllowed,
AuthorizerNotFoundException,
@ -490,7 +491,16 @@ class APIGatewayResponse(BaseResponse):
usage_plan = url_path_parts[2]
if self.method == "GET":
usage_plan_response = self.backend.get_usage_plan(usage_plan)
try:
usage_plan_response = self.backend.get_usage_plan(usage_plan)
except (UsagePlanNotFoundException) as error:
return (
error.code,
{},
'{{"message":"{0}","code":"{1}"}}'.format(
error.message, error.error_type
),
)
elif self.method == "DELETE":
usage_plan_response = self.backend.delete_usage_plan(usage_plan)
return 200, {}, json.dumps(usage_plan_response)
@ -529,7 +539,18 @@ class APIGatewayResponse(BaseResponse):
key_id = url_path_parts[4]
if self.method == "GET":
usage_plan_response = self.backend.get_usage_plan_key(usage_plan_id, key_id)
try:
usage_plan_response = self.backend.get_usage_plan_key(
usage_plan_id, key_id
)
except (UsagePlanNotFoundException, ApiKeyNotFoundException) as error:
return (
error.code,
{},
'{{"message":"{0}","code":"{1}"}}'.format(
error.message, error.error_type
),
)
elif self.method == "DELETE":
usage_plan_response = self.backend.delete_usage_plan_key(
usage_plan_id, key_id