import boto3 from moto import mock_autoscaling from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID from tests import EXAMPLE_AMI_ID from unittest import TestCase from .utils import setup_networking @mock_autoscaling class TestAutoScalingGroup(TestCase): def setUp(self) -> None: self.mocked_networking = setup_networking() self.as_client = boto3.client("autoscaling", region_name="us-east-1") self.lc_name = "tester_config" self.as_client.create_launch_configuration( LaunchConfigurationName=self.lc_name, ImageId=EXAMPLE_AMI_ID, InstanceType="t2.medium", ) def test_create_autoscaling_groups_defaults(self): """Test with the minimum inputs and check that all proper defaults are assigned for the other attributes""" self._create_group(name="tester_group") group = self.as_client.describe_auto_scaling_groups()["AutoScalingGroups"][0] assert group["AutoScalingGroupName"] == "tester_group" assert group["MaxSize"] == 2 assert group["MinSize"] == 1 assert group["LaunchConfigurationName"] == self.lc_name # Defaults assert group["AvailabilityZones"] == ["us-east-1a"] # subnet1 assert group["DesiredCapacity"] == 1 assert group["VPCZoneIdentifier"] == self.mocked_networking["subnet1"] assert group["DefaultCooldown"] == 300 assert group["HealthCheckGracePeriod"] == 300 assert group["HealthCheckType"] == "EC2" assert group["LoadBalancerNames"] == [] assert "PlacementGroup" not in group assert group["TerminationPolicies"] == ["Default"] assert group["Tags"] == [] def test_create_autoscaling_group__additional_params(self): self.as_client.create_auto_scaling_group( AutoScalingGroupName="tester_group", MinSize=1, MaxSize=2, LaunchConfigurationName=self.lc_name, CapacityRebalance=True, VPCZoneIdentifier=self.mocked_networking["subnet1"], ) group = self.as_client.describe_auto_scaling_groups()["AutoScalingGroups"][0] assert group["AutoScalingGroupName"] == "tester_group" assert group["CapacityRebalance"] is True def test_list_many_autoscaling_groups(self): for i in range(51): self._create_group(f"TestGroup{i}") response = self.as_client.describe_auto_scaling_groups() groups = response["AutoScalingGroups"] marker = response["NextToken"] assert len(groups) == 50 assert marker == groups[-1]["AutoScalingGroupName"] response2 = self.as_client.describe_auto_scaling_groups(NextToken=marker) groups.extend(response2["AutoScalingGroups"]) assert len(groups) == 51 assert "NextToken" not in response2.keys() def test_autoscaling_group_delete(self): self._create_group(name="tester_group") groups = self.as_client.describe_auto_scaling_groups()["AutoScalingGroups"] assert len(groups) == 1 self.as_client.delete_auto_scaling_group(AutoScalingGroupName="tester_group") groups = self.as_client.describe_auto_scaling_groups()["AutoScalingGroups"] assert len(groups) == 0 def test_describe_autoscaling_groups__instances(self): self._create_group(name="test_asg") response = self.as_client.describe_auto_scaling_groups( AutoScalingGroupNames=["test_asg"] ) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200 group = response["AutoScalingGroups"][0] assert ( f"arn:aws:autoscaling:us-east-1:{ACCOUNT_ID}:autoScalingGroup:" in group["AutoScalingGroupARN"] ) assert group["AutoScalingGroupName"] == "test_asg" assert group["LaunchConfigurationName"] == self.lc_name assert "LaunchTemplate" not in group assert group["AvailabilityZones"] == ["us-east-1a"] assert group["VPCZoneIdentifier"] == self.mocked_networking["subnet1"] for instance in group["Instances"]: assert instance["LaunchConfigurationName"] == self.lc_name assert "LaunchTemplate" not in instance assert instance["AvailabilityZone"] == "us-east-1a" assert instance["InstanceType"] == "t2.medium" def test_set_instance_health(self): self._create_group(name="test_asg") response = self.as_client.describe_auto_scaling_groups( AutoScalingGroupNames=["test_asg"] ) instance1 = response["AutoScalingGroups"][0]["Instances"][0] assert instance1["HealthStatus"] == "Healthy" self.as_client.set_instance_health( InstanceId=instance1["InstanceId"], HealthStatus="Unhealthy" ) response = self.as_client.describe_auto_scaling_groups( AutoScalingGroupNames=["test_asg"] ) instance1 = response["AutoScalingGroups"][0]["Instances"][0] assert instance1["HealthStatus"] == "Unhealthy" def test_suspend_processes(self): self._create_group(name="test-asg") # When we suspend the 'Launch' process on the ASG client self.as_client.suspend_processes( AutoScalingGroupName="test-asg", ScalingProcesses=["Launch"] ) res = self.as_client.describe_auto_scaling_groups( AutoScalingGroupNames=["test-asg"] ) # The 'Launch' process should, in fact, be suspended launch_suspended = False for proc in res["AutoScalingGroups"][0]["SuspendedProcesses"]: if proc.get("ProcessName") == "Launch": launch_suspended = True assert launch_suspended is True def test_suspend_processes_all_by_default(self): self._create_group(name="test-asg") # When we suspend with no processes specified self.as_client.suspend_processes(AutoScalingGroupName="test-asg") res = self.as_client.describe_auto_scaling_groups( AutoScalingGroupNames=["test-asg"] ) # All processes should be suspended all_proc_names = [ "Launch", "Terminate", "AddToLoadBalancer", "AlarmNotification", "AZRebalance", "HealthCheck", "InstanceRefresh", "ReplaceUnhealthy", "ScheduledActions", ] suspended_proc_names = [ proc["ProcessName"] for proc in res["AutoScalingGroups"][0]["SuspendedProcesses"] ] assert set(suspended_proc_names) == set(all_proc_names) def test_suspend_additional_processes(self): self._create_group(name="test-asg") # When we suspend the 'Launch' and 'Terminate' processes in separate calls self.as_client.suspend_processes( AutoScalingGroupName="test-asg", ScalingProcesses=["Launch"] ) self.as_client.suspend_processes( AutoScalingGroupName="test-asg", ScalingProcesses=["Terminate"] ) res = self.as_client.describe_auto_scaling_groups( AutoScalingGroupNames=["test-asg"] ) # Both 'Launch' and 'Terminate' should be suspended launch_suspended = False terminate_suspended = False for proc in res["AutoScalingGroups"][0]["SuspendedProcesses"]: if proc.get("ProcessName") == "Launch": launch_suspended = True if proc.get("ProcessName") == "Terminate": terminate_suspended = True assert launch_suspended is True assert terminate_suspended is True def test_resume_processes(self): self._create_group(name="test-asg") # When we suspect 'Launch' and 'Termiate' process then resume 'Launch' self.as_client.suspend_processes( AutoScalingGroupName="test-asg", ScalingProcesses=["Launch", "Terminate"] ) self.as_client.resume_processes( AutoScalingGroupName="test-asg", ScalingProcesses=["Launch"] ) res = self.as_client.describe_auto_scaling_groups( AutoScalingGroupNames=["test-asg"] ) # Only 'Terminate' should be suspended expected_suspended_processes = [ {"ProcessName": "Terminate", "SuspensionReason": ""} ] assert ( res["AutoScalingGroups"][0]["SuspendedProcesses"] == expected_suspended_processes ) def test_resume_processes_all_by_default(self): self._create_group(name="test-asg") # When we suspend two processes then resume with no process argument self.as_client.suspend_processes( AutoScalingGroupName="test-asg", ScalingProcesses=["Launch", "Terminate"] ) self.as_client.resume_processes(AutoScalingGroupName="test-asg") res = self.as_client.describe_auto_scaling_groups( AutoScalingGroupNames=["test-asg"] ) # No processes should be suspended assert res["AutoScalingGroups"][0]["SuspendedProcesses"] == [] def _create_group(self, name): self.as_client.create_auto_scaling_group( AutoScalingGroupName=name, MinSize=1, MaxSize=2, LaunchConfigurationName=self.lc_name, VPCZoneIdentifier=self.mocked_networking["subnet1"], )