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

Added AutoScalingGroup LifeCycleHook support (fake) (#4259)

Browse Source
This commit is contained in:
mello7tre 2021-09-02 15:10:11 +02:00 committed by GitHub
parent 5044df98bc
commit edca235623
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 144 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
moto/autoscaling/models.py
View File

@ -45,6 +45,24 @@ class InstanceState(object):
self.instance.autoscaling_group = autoscaling_group
class FakeLifeCycleHook(BaseModel):
def __init__(
self, name, as_name, transition, timeout, result,
):
self.name = name
self.as_name = as_name
if transition:
self.transition = transition
if timeout:
self.timeout = timeout
else:
self.timeout = 3600
if result:
self.result = result
else:
self.result = "ABANDON"
class FakeScalingPolicy(BaseModel):
def __init__(
self,
@ -592,6 +610,7 @@ class AutoScalingBackend(BaseBackend):
self.autoscaling_groups = OrderedDict()
self.launch_configurations = OrderedDict()
self.policies = {}
self.lifecycle_hooks = {}
self.ec2_backend = ec2_backend
self.elb_backend = elb_backend
self.elbv2_backend = elbv2_backend
@ -897,6 +916,25 @@ class AutoScalingBackend(BaseBackend):
desired_capacity = int(desired_capacity)
self.set_desired_capacity(group_name, desired_capacity)
def create_lifecycle_hook(self, name, as_name, transition, timeout, result):
lifecycle_hook = FakeLifeCycleHook(name, as_name, transition, timeout, result,)
self.lifecycle_hooks["%s_%s" % (as_name, name)] = lifecycle_hook
return lifecycle_hook
def describe_lifecycle_hooks(self, as_name, lifecycle_hook_names=None):
return [
lifecycle_hook
for lifecycle_hook in self.lifecycle_hooks.values()
if (lifecycle_hook.as_name == as_name)
and (
not lifecycle_hook_names or lifecycle_hook.name in lifecycle_hook_names
)
]
def delete_lifecycle_hook(self, as_name, name):
self.lifecycle_hooks.pop("%s_%s" % (as_name, name), None)
def create_autoscaling_policy(
self, name, policy_type, adjustment_type, as_name, scaling_adjustment, cooldown
):

63
moto/autoscaling/responses.py
View File

@ -247,6 +247,32 @@ class AutoScalingResponse(BaseResponse):
template = self.response_template(DESCRIBE_AUTOSCALING_INSTANCES_TEMPLATE)
return template.render(instance_states=instance_states)
def put_lifecycle_hook(self):
lifecycle_hook = self.autoscaling_backend.create_lifecycle_hook(
name=self._get_param("LifecycleHookName"),
as_name=self._get_param("AutoScalingGroupName"),
transition=self._get_param("LifecycleTransition"),
timeout=self._get_int_param("HeartbeatTimeout"),
result=self._get_param("DefaultResult"),
)
template = self.response_template(CREATE_LIFECYLE_HOOK_TEMPLATE)
return template.render(lifecycle_hook=lifecycle_hook)
def describe_lifecycle_hooks(self):
lifecycle_hooks = self.autoscaling_backend.describe_lifecycle_hooks(
as_name=self._get_param("AutoScalingGroupName"),
lifecycle_hook_names=self._get_multi_param("LifecycleHookNames.member"),
)
template = self.response_template(DESCRIBE_LIFECYCLE_HOOKS_TEMPLATE)
return template.render(lifecycle_hooks=lifecycle_hooks)
def delete_lifecycle_hook(self):
as_name = self._get_param("AutoScalingGroupName")
name = self._get_param("LifecycleHookName")
self.autoscaling_backend.delete_lifecycle_hook(as_name, name)
template = self.response_template(DELETE_LIFECYCLE_HOOK_TEMPLATE)
return template.render()
def put_scaling_policy(self):
policy = self.autoscaling_backend.create_autoscaling_policy(
name=self._get_param("PolicyName"),
@ -723,6 +749,43 @@ DESCRIBE_AUTOSCALING_INSTANCES_TEMPLATE = """<DescribeAutoScalingInstancesRespon
</ResponseMetadata>
</DescribeAutoScalingInstancesResponse>"""
CREATE_LIFECYLE_HOOK_TEMPLATE = """<PutLifecycleHookResponse xmlns="http://autoscaling.amazonaws.com/doc/2011-01-01/">
<PutLifecycleHookResult/>
<ResponseMetadata>
<RequestId>3cfc6fef-c08b-11e2-a697-2922EXAMPLE</RequestId>
</ResponseMetadata>
</PutLifecycleHookResponse>"""
DESCRIBE_LIFECYCLE_HOOKS_TEMPLATE = """<DescribeLifecycleHooksResponse xmlns="http://autoscaling.amazonaws.com/doc/2011-01-01/">
<DescribeLifecycleHooksResult>
<LifecycleHooks>
{% for lifecycle_hook in lifecycle_hooks %}
<member>
<AutoScalingGroupName>{{ lifecycle_hook.as_name }}</AutoScalingGroupName>
<RoleARN>arn:aws:iam::1234567890:role/my-auto-scaling-role</RoleARN>
<LifecycleTransition>{{ lifecycle_hook.transition }}</LifecycleTransition>
<GlobalTimeout>172800</GlobalTimeout>
<LifecycleHookName>{{ lifecycle_hook.name }}</LifecycleHookName>
<HeartbeatTimeout>{{ lifecycle_hook.timeout }}</HeartbeatTimeout>
<DefaultResult>{{ lifecycle_hook.result }}</DefaultResult>
<NotificationTargetARN>arn:aws:sqs:us-east-1:123456789012:my-queue</NotificationTargetARN>
</member>
{% endfor %}
</LifecycleHooks>
</DescribeLifecycleHooksResult>
<ResponseMetadata>
<RequestId>ec3bffad-b739-11e2-b38d-15fbEXAMPLE</RequestId>
</ResponseMetadata>
</DescribeLifecycleHooksResponse>"""
DELETE_LIFECYCLE_HOOK_TEMPLATE = """<DeleteLifecycleHookResponse xmlns="http://autoscaling.amazonaws.com/doc/2011-01-01/">
<DeleteLifecycleHookResult>
</DeleteLifecycleHookResult>
<ResponseMetadata>
<RequestId>70a76d42-9665-11e2-9fdf-211deEXAMPLE</RequestId>
</ResponseMetadata>
</DeleteLifecycleHookResponse>"""
CREATE_SCALING_POLICY_TEMPLATE = """<PutScalingPolicyResponse xmlns="http://autoscaling.amazonaws.com/doc/2011-01-01/">
<PutScalingPolicyResult>
<PolicyARN>arn:aws:autoscaling:us-east-1:803981987763:scalingPolicy:b0dcf5e8

43
tests/test_autoscaling/test_autoscaling.py
View File

@ -2873,3 +2873,46 @@ def test_attach_instances():
instance["LaunchConfigurationName"].should.equal("test_launch_configuration")
instance["AutoScalingGroupName"].should.equal("test_asg")
instance["InstanceType"].should.equal("c4.2xlarge")
@mock_autoscaling
def test_autoscaling_lifecyclehook():
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
fake_lc = client.create_launch_configuration(
LaunchConfigurationName="test_launch_configuration",
ImageId="ami-pytest",
InstanceType="t3.micro",
KeyName="foobar",
)
fake_asg = client.create_auto_scaling_group(
AutoScalingGroupName="test_asg",
LaunchConfigurationName="test_launch_configuration",
MinSize=0,
MaxSize=1,
VPCZoneIdentifier=mocked_networking["subnet1"],
)
fake_lfh = client.put_lifecycle_hook(
LifecycleHookName="test-lifecyclehook",
AutoScalingGroupName="test_asg",
LifecycleTransition="autoscaling:EC2_INSTANCE_TERMINATING",
)
response = client.describe_lifecycle_hooks(
AutoScalingGroupName="test_asg", LifecycleHookNames=["test-lifecyclehook"]
)
len(response["LifecycleHooks"]).should.equal(1)
for hook in response["LifecycleHooks"]:
hook["LifecycleHookName"].should.equal("test-lifecyclehook")
hook["AutoScalingGroupName"].should.equal("test_asg")
hook["LifecycleTransition"].should.equal("autoscaling:EC2_INSTANCE_TERMINATING")
client.delete_lifecycle_hook(
LifecycleHookName="test-lifecyclehook", AutoScalingGroupName="test_asg",
)
response = client.describe_lifecycle_hooks(
AutoScalingGroupName="test_asg", LifecycleHookNames=["test-lifecyclehook"]
)
len(response["LifecycleHooks"]).should.equal(0)