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

ApplicationAutoscaling: support autoscaling policies, deregister_scalable_target (#3350)

Browse source 
* ApplicationAutoscaling: support autoscaling policies, deregister_scalable_target.

* PR3350 comment changes: drop unnecessary pass statements, unit test three exception cases.

Co-authored-by: Joseph Weitekamp <jweite@amazon.com>
This commit is contained in:
jweite 2020-10-30 11:21:34 -04:00 • committed by GitHub
commit cbd4efb42d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 439 additions and 21 deletions
Download patch file Download diff file Expand all files Collapse all files

96
moto/applicationautoscaling/responses.py
View file

@ -15,10 +15,7 @@ class ApplicationAutoScalingResponse(BaseResponse):
return applicationautoscaling_backends[self.region]
def describe_scalable_targets(self):
try:
self._validate_params()
except AWSValidationException as e:
return e.response()
self._validate_params()
service_namespace = self._get_param("ServiceNamespace")
resource_ids = self._get_param("ResourceIds")
scalable_dimension = self._get_param("ScalableDimension")
@ -37,19 +34,65 @@ class ApplicationAutoScalingResponse(BaseResponse):
def register_scalable_target(self):
""" Registers or updates a scalable target. """
try:
self._validate_params()
self.applicationautoscaling_backend.register_scalable_target(
self._get_param("ServiceNamespace"),
self._get_param("ResourceId"),
self._get_param("ScalableDimension"),
min_capacity=self._get_int_param("MinCapacity"),
max_capacity=self._get_int_param("MaxCapacity"),
role_arn=self._get_param("RoleARN"),
suspended_state=self._get_param("SuspendedState"),
)
except AWSValidationException as e:
return e.response()
self._validate_params()
self.applicationautoscaling_backend.register_scalable_target(
self._get_param("ServiceNamespace"),
self._get_param("ResourceId"),
self._get_param("ScalableDimension"),
min_capacity=self._get_int_param("MinCapacity"),
max_capacity=self._get_int_param("MaxCapacity"),
role_arn=self._get_param("RoleARN"),
suspended_state=self._get_param("SuspendedState"),
)
return json.dumps({})
def deregister_scalable_target(self):
""" Deregisters a scalable target. """
self._validate_params()
self.applicationautoscaling_backend.deregister_scalable_target(
self._get_param("ServiceNamespace"),
self._get_param("ResourceId"),
self._get_param("ScalableDimension"),
)
return json.dumps({})
def put_scaling_policy(self):
policy = self.applicationautoscaling_backend.put_scaling_policy(
policy_name=self._get_param("PolicyName"),
service_namespace=self._get_param("ServiceNamespace"),
resource_id=self._get_param("ResourceId"),
scalable_dimension=self._get_param("ScalableDimension"),
policy_type=self._get_param("PolicyType"),
policy_body=self._get_param(
"StepScalingPolicyConfiguration",
self._get_param("TargetTrackingScalingPolicyConfiguration"),
),
)
return json.dumps({"PolicyARN": policy.policy_arn, "Alarms": []}) # ToDo
def describe_scaling_policies(self):
(
next_token,
policy_page,
) = self.applicationautoscaling_backend.describe_scaling_policies(
service_namespace=self._get_param("ServiceNamespace"),
resource_id=self._get_param("ResourceId"),
scalable_dimension=self._get_param("ScalableDimension"),
max_results=self._get_param("MaxResults"),
next_token=self._get_param("NextToken"),
)
response_obj = {"ScalingPolicies": [_build_policy(p) for p in policy_page]}
if next_token:
response_obj["NextToken"] = next_token
return json.dumps(response_obj)
def delete_scaling_policy(self):
self.applicationautoscaling_backend.delete_scaling_policy(
policy_name=self._get_param("PolicyName"),
service_namespace=self._get_param("ServiceNamespace"),
resource_id=self._get_param("ResourceId"),
scalable_dimension=self._get_param("ScalableDimension"),
)
return json.dumps({})
def _validate_params(self):
@ -95,3 +138,22 @@ def _build_target(t):
"MinCapacity": t.min_capacity,
"SuspendedState": t.suspended_state,
}
def _build_policy(p):
response = {
"PolicyARN": p.policy_arn,
"PolicyName": p.policy_name,
"ServiceNamespace": p.service_namespace,
"ResourceId": p.resource_id,
"ScalableDimension": p.scalable_dimension,
"PolicyType": p.policy_type,
"CreationTime": p.creation_time,
}
if p.policy_type == "StepScaling":
response["StepScalingPolicyConfiguration"] = p.step_scaling_policy_configuration
elif p.policy_type == "TargetTrackingScaling":
response[
"TargetTrackingScalingPolicyConfiguration"
] = p.target_tracking_scaling_policy_configuration
return response