Merge pull request #1717 from fewstera/usage-plans

APIGateway: Added API for usage plans and usage plan keys

moto/apigateway/exceptions.py

@@ -8,3 +8,11 @@ class StageNotFoundException(RESTError):
     def __init__(self):
         super(StageNotFoundException, self).__init__(
             "NotFoundException", "Invalid stage identifier specified")
+
+
+class ApiKeyNotFoundException(RESTError):
+    code = 404
+
+    def __init__(self):
+        super(ApiKeyNotFoundException, self).__init__(
+            "NotFoundException", "Invalid API Key identifier specified")

moto/apigateway/models.py

@@ -10,7 +10,7 @@ from boto3.session import Session
 import responses
 from moto.core import BaseBackend, BaseModel
 from .utils import create_id
-from .exceptions import StageNotFoundException
+from .exceptions import StageNotFoundException, ApiKeyNotFoundException
 
 STAGE_URL = "https://{api_id}.execute-api.{region_name}.amazonaws.com/{stage_name}"
 
@@ -309,6 +309,29 @@ class ApiKey(BaseModel, dict):
         self['stageKeys'] = stageKeys
 
 
+class UsagePlan(BaseModel, dict):
+
+    def __init__(self, name=None, description=None, apiStages=[],
+                 throttle=None, quota=None):
+        super(UsagePlan, self).__init__()
+        self['id'] = create_id()
+        self['name'] = name
+        self['description'] = description
+        self['apiStages'] = apiStages
+        self['throttle'] = throttle
+        self['quota'] = quota
+
+
+class UsagePlanKey(BaseModel, dict):
+
+    def __init__(self, id, type, name, value):
+        super(UsagePlanKey, self).__init__()
+        self['id'] = id
+        self['name'] = name
+        self['type'] = type
+        self['value'] = value
+
+
 class RestAPI(BaseModel):
 
     def __init__(self, id, region_name, name, description):
@@ -408,6 +431,8 @@ class APIGatewayBackend(BaseBackend):
         super(APIGatewayBackend, self).__init__()
         self.apis = {}
         self.keys = {}
+        self.usage_plans = {}
+        self.usage_plan_keys = {}
         self.region_name = region_name
 
     def reset(self):
@@ -576,6 +601,48 @@ class APIGatewayBackend(BaseBackend):
         self.keys.pop(api_key_id)
         return {}
 
+    def create_usage_plan(self, payload):
+        plan = UsagePlan(**payload)
+        self.usage_plans[plan['id']] = plan
+        return plan
+
+    def get_usage_plans(self):
+        return list(self.usage_plans.values())
+
+    def get_usage_plan(self, usage_plan_id):
+        return self.usage_plans[usage_plan_id]
+
+    def delete_usage_plan(self, usage_plan_id):
+        self.usage_plans.pop(usage_plan_id)
+        return {}
+
+    def create_usage_plan_key(self, usage_plan_id, payload):
+        if usage_plan_id not in self.usage_plan_keys:
+            self.usage_plan_keys[usage_plan_id] = {}
+
+        key_id = payload["keyId"]
+        if key_id not in self.keys:
+            raise ApiKeyNotFoundException()
+
+        api_key = self.keys[key_id]
+
+        usage_plan_key = UsagePlanKey(id=key_id, type=payload["keyType"], name=api_key["name"], value=api_key["value"])
+        self.usage_plan_keys[usage_plan_id][usage_plan_key['id']] = usage_plan_key
+        return usage_plan_key
+
+    def get_usage_plan_keys(self, usage_plan_id):
+        if usage_plan_id not in self.usage_plan_keys:
+            return []
+
+        return list(self.usage_plan_keys[usage_plan_id].values())
+
+    def get_usage_plan_key(self, usage_plan_id, key_id):
+        return self.usage_plan_keys[usage_plan_id][key_id]
+
+    def delete_usage_plan_key(self, usage_plan_id, key_id):
+        self.usage_plan_keys[usage_plan_id].pop(key_id)
+        return {}
+
 
 apigateway_backends = {}
 for region_name in Session().get_available_regions('apigateway'):

moto/apigateway/responses.py

@@ -4,7 +4,7 @@ import json
 
 from moto.core.responses import BaseResponse
 from .models import apigateway_backends
-from .exceptions import StageNotFoundException
+from .exceptions import StageNotFoundException, ApiKeyNotFoundException
 
 
 class APIGatewayResponse(BaseResponse):
@@ -248,3 +248,56 @@ class APIGatewayResponse(BaseResponse):
         elif self.method == 'DELETE':
             apikey_response = self.backend.delete_apikey(apikey)
         return 200, {}, json.dumps(apikey_response)
+
+    def usage_plans(self, request, full_url, headers):
+        self.setup_class(request, full_url, headers)
+
+        if self.method == 'POST':
+            usage_plan_response = self.backend.create_usage_plan(json.loads(self.body))
+        elif self.method == 'GET':
+            usage_plans_response = self.backend.get_usage_plans()
+            return 200, {}, json.dumps({"item": usage_plans_response})
+        return 200, {}, json.dumps(usage_plan_response)
+
+    def usage_plan_individual(self, request, full_url, headers):
+        self.setup_class(request, full_url, headers)
+
+        url_path_parts = self.path.split("/")
+        usage_plan = url_path_parts[2]
+
+        if self.method == 'GET':
+            usage_plan_response = self.backend.get_usage_plan(usage_plan)
+        elif self.method == 'DELETE':
+            usage_plan_response = self.backend.delete_usage_plan(usage_plan)
+        return 200, {}, json.dumps(usage_plan_response)
+
+    def usage_plan_keys(self, request, full_url, headers):
+        self.setup_class(request, full_url, headers)
+
+        url_path_parts = self.path.split("/")
+        usage_plan_id = url_path_parts[2]
+
+        if self.method == 'POST':
+            try:
+                usage_plan_response = self.backend.create_usage_plan_key(usage_plan_id, json.loads(self.body))
+            except ApiKeyNotFoundException as error:
+                return error.code, {}, '{{"message":"{0}","code":"{1}"}}'.format(error.message, error.error_type)
+
+        elif self.method == 'GET':
+            usage_plans_response = self.backend.get_usage_plan_keys(usage_plan_id)
+            return 200, {}, json.dumps({"item": usage_plans_response})
+
+        return 200, {}, json.dumps(usage_plan_response)
+
+    def usage_plan_key_individual(self, request, full_url, headers):
+        self.setup_class(request, full_url, headers)
+
+        url_path_parts = self.path.split("/")
+        usage_plan_id = url_path_parts[2]
+        key_id = url_path_parts[4]
+
+        if self.method == 'GET':
+            usage_plan_response = self.backend.get_usage_plan_key(usage_plan_id, key_id)
+        elif self.method == 'DELETE':
+            usage_plan_response = self.backend.delete_usage_plan_key(usage_plan_id, key_id)
+        return 200, {}, json.dumps(usage_plan_response)

moto/apigateway/urls.py

@@ -20,4 +20,8 @@ url_paths = {
     '{0}/restapis/(?P<function_id>[^/]+)/resources/(?P<resource_id>[^/]+)/methods/(?P<method_name>[^/]+)/integration/responses/(?P<status_code>\d+)/?$': APIGatewayResponse().integration_responses,
     '{0}/apikeys$': APIGatewayResponse().apikeys,
     '{0}/apikeys/(?P<apikey>[^/]+)': APIGatewayResponse().apikey_individual,
+    '{0}/usageplans$': APIGatewayResponse().usage_plans,
+    '{0}/usageplans/(?P<usage_plan_id>[^/]+)/?$': APIGatewayResponse().usage_plan_individual,
+    '{0}/usageplans/(?P<usage_plan_id>[^/]+)/keys$': APIGatewayResponse().usage_plan_keys,
+    '{0}/usageplans/(?P<usage_plan_id>[^/]+)/keys/(?P<api_key_id>[^/]+)/?$': APIGatewayResponse().usage_plan_key_individual,
 }