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

added feature update_usage_plan and fixed some lint errors (#3727)

Browse Source 
Co-authored-by: rajinder saini <rajinder.saini@c02vt5k2htd6.corp.climate.com>
This commit is contained in:
rajinder 2021-02-24 23:46:11 -08:00 committed by GitHub
parent 0625bbfa11
commit 4b1c7225b6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 64 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
moto/apigateway/models.py
View File

@ -1218,6 +1218,25 @@ class APIGatewayBackend(BaseBackend):
return self.usage_plans[usage_plan_id]
def __apply_usage_plan_patch_operations(self, plan, patch_operations):
for op in patch_operations:
if op["op"] == "replace":
if "/quota/limit" in op["path"]:
plan["quota"]["limit"] = op["value"]
if "/quota/period" in op["path"]:
plan["quota"]["period"] = op["value"]
if "/throttle/rateLimit" in op["path"]:
plan["throttle"]["rateLimit"] = op["value"]
if "/throttle/burstLimit" in op["path"]:
plan["throttle"]["burstLimit"] = op["value"]
def update_usage_plan(self, usage_plan_id, patch_operations):
if usage_plan_id not in self.usage_plans:
raise UsagePlanNotFoundException()
plan = self.usage_plans[usage_plan_id]
self.__apply_usage_plan_patch_operations(plan, patch_operations)
return self.usage_plans[usage_plan_id]
def delete_usage_plan(self, usage_plan_id):
self.usage_plans.pop(usage_plan_id)
return {}

6
moto/apigateway/responses.py
View File

@ -477,7 +477,6 @@ class APIGatewayResponse(BaseResponse):
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":
@ -505,6 +504,11 @@ class APIGatewayResponse(BaseResponse):
)
elif self.method == "DELETE":
usage_plan_response = self.backend.delete_usage_plan(usage_plan)
elif self.method == "PATCH":
patch_operations = self._get_param("patchOperations")
usage_plan_response = self.backend.update_usage_plan(
usage_plan, patch_operations
)
return 200, {}, json.dumps(usage_plan_response)
def usage_plan_keys(self, request, full_url, headers):

2
moto/batch/models.py
View File

@ -762,7 +762,7 @@ class BatchBackend(BaseBackend):
if compute_resources is None and _type == "MANAGED":
raise InvalidParameterValueException(
"computeResources must be specified when creating a MANAGED environment".format(
"computeResources must be specified when creating a {0} environment".format(
state
)
)

6
moto/elb/models.py
View File

@ -474,9 +474,9 @@ class ELBBackend(BaseBackend):
):
load_balancer = self.get_load_balancer(load_balancer_name)
listener = [
l
for l in load_balancer.listeners
if int(l.load_balancer_port) == load_balancer_port
l_listener
for l_listener in load_balancer.listeners
if int(l_listener.load_balancer_port) == load_balancer_port
][0]
listener_idx = load_balancer.listeners.index(listener)
listener.policy_names = policies

6
moto/elb/responses.py
View File

@ -241,9 +241,9 @@ class ELBResponse(BaseResponse):
load_balancer_port = int(self._get_param("LoadBalancerPort"))
mb_listener = [
l
for l in load_balancer.listeners
if int(l.load_balancer_port) == load_balancer_port
listner
for listner in load_balancer.listeners
if int(listner.load_balancer_port) == load_balancer_port
]
if mb_listener:
policies = self._get_multi_param("PolicyNames.member")

4
moto/opsworks/models.py
View File

@ -518,11 +518,11 @@ class OpsWorksBackend(BaseBackend):
stackid = kwargs["stack_id"]
if stackid not in self.stacks:
raise ResourceNotFoundException(stackid)
if name in [l.name for l in self.stacks[stackid].layers]:
if name in [layer.name for layer in self.stacks[stackid].layers]:
raise ValidationException(
'There is already a layer named "{0}" ' "for this stack".format(name)
)
if shortname in [l.shortname for l in self.stacks[stackid].layers]:
if shortname in [layer.shortname for layer in self.stacks[stackid].layers]:
raise ValidationException(
'There is already a layer with shortname "{0}" '
"for this stack".format(shortname)

2
moto/server.py
View File

@ -113,7 +113,7 @@ class DomainDispatcherApplication(object):
if dynamo_api_version > "20111205":
host = "dynamodb2"
elif service == "sagemaker":
host = "api.sagemaker.{region}.amazonaws.com".format(
host = "api.{service}.{region}.amazonaws.com".format(
service=service, region=region
)
else:

30
tests/test_apigateway/test_apigateway.py
View File

@ -2058,6 +2058,36 @@ def test_usage_plans():
len(response["items"]).should.equal(1)
@mock_apigateway
def test_update_usage_plan():
region_name = "us-west-2"
client = boto3.client("apigateway", region_name=region_name)
payload = {
"name": "TEST-PLAN-2",
"description": "Description",
"quota": {"limit": 10, "period": "DAY", "offset": 0},
"throttle": {"rateLimit": 2, "burstLimit": 1},
"apiStages": [{"apiId": "foo", "stage": "bar"}],
"tags": {"tag_key": "tag_value"},
}
response = client.create_usage_plan(**payload)
usage_plan_id = response["id"]
response = client.update_usage_plan(
usagePlanId=usage_plan_id,
patchOperations=[
{"op": "replace", "path": "/quota/limit", "value": "1000"},
{"op": "replace", "path": "/quota/period", "value": "MONTH"},
{"op": "replace", "path": "/throttle/rateLimit", "value": "500"},
{"op": "replace", "path": "/throttle/burstLimit", "value": "1500"},
],
)
response["quota"]["limit"].should.equal("1000")
response["quota"]["period"].should.equal("MONTH")
response["quota"]["limit"].should.equal("1000")
response["quota"]["limit"].should.equal("1000")
@mock_apigateway
def test_usage_plan_keys():
region_name = "us-west-2"