APIGateway: Added usage plan keys API

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}"
 
@@ -322,6 +322,16 @@ class UsagePlan(BaseModel, dict):
         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):
@@ -422,6 +432,7 @@ class APIGatewayBackend(BaseBackend):
         self.apis = {}
         self.keys = {}
         self.usage_plans = {}
+        self.usage_plan_keys = {}
         self.region_name = region_name
 
     def reset(self):
@@ -605,6 +616,33 @@ class APIGatewayBackend(BaseBackend):
         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):
@@ -270,3 +270,34 @@ class APIGatewayResponse(BaseResponse):
         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

@@ -21,5 +21,7 @@ url_paths = {
     '{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>[^/]+)/?$': 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,
 }