diff --git a/tests/test_autoscaling/test_autoscaling.py b/tests/test_autoscaling/test_autoscaling.py index de37760dd..9438f9575 100644 --- a/tests/test_autoscaling/test_autoscaling.py +++ b/tests/test_autoscaling/test_autoscaling.py @@ -1,141 +1,12 @@ import boto3 import sure # noqa # pylint: disable=unused-import -from botocore.exceptions import ClientError import pytest -from moto import mock_autoscaling, mock_elb, mock_ec2 +from botocore.exceptions import ClientError +from moto import mock_autoscaling, mock_ec2 from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID - -from .utils import setup_networking, setup_instance_with_networking from tests import EXAMPLE_AMI_ID - - -@mock_autoscaling -@mock_ec2 -@mock_elb -def test_create_autoscaling_group_within_elb(): - mocked_networking = setup_networking() - elb_client = boto3.client("elb", region_name="us-east-1") - elb_client.create_load_balancer( - LoadBalancerName="test_lb", - Listeners=[{"Protocol": "http", "LoadBalancerPort": 80, "InstancePort": 8080}], - AvailabilityZones=[], - ) - - # we attach a couple of machines to the load balancer - # that are not managed by the auto scaling group - INSTANCE_COUNT_START = 3 - INSTANCE_COUNT_GROUP = 2 - ec2 = boto3.resource("ec2", region_name="us-east-1") - instances = ec2.create_instances( - ImageId=EXAMPLE_AMI_ID, - InstanceType="t1.micro", - MaxCount=INSTANCE_COUNT_START, - MinCount=INSTANCE_COUNT_START, - SubnetId=mocked_networking["subnet1"], - ) - - instances_ids = [_.id for _ in instances] - elb_client.register_instances_with_load_balancer( - LoadBalancerName="test_lb", - Instances=[{"InstanceId": id} for id in instances_ids], - ) - - as_client = boto3.client("autoscaling", region_name="us-east-1") - as_client.create_launch_configuration( - LaunchConfigurationName="tester", - ImageId=EXAMPLE_AMI_ID, - InstanceType="t2.medium", - ) - - as_client.create_auto_scaling_group( - AutoScalingGroupName="tester_group", - AvailabilityZones=["us-east-1a", "us-east-1b"], - DefaultCooldown=60, - DesiredCapacity=INSTANCE_COUNT_GROUP, - HealthCheckGracePeriod=100, - HealthCheckType="EC2", - LaunchConfigurationName="tester", - LoadBalancerNames=["test_lb"], - MinSize=INSTANCE_COUNT_GROUP, - MaxSize=INSTANCE_COUNT_GROUP, - PlacementGroup="test_placement", - Tags=[ - { - "ResourceId": "tester_group", - "ResourceType": "group", - "Key": "test_key", - "Value": "test_value", - "PropagateAtLaunch": True, - } - ], - TerminationPolicies=["OldestInstance", "NewestInstance"], - VPCZoneIdentifier="{subnet1},{subnet2}".format( - subnet1=mocked_networking["subnet1"], subnet2=mocked_networking["subnet2"] - ), - ) - - as_client.put_scheduled_update_group_action( - AutoScalingGroupName="tester_group", - ScheduledActionName="my-scheduled-action", - StartTime="2022-07-01T00:00:00Z", - EndTime="2022-09-01T00:00:00Z", - Recurrence="* * * * *", - MinSize=5, - MaxSize=12, - DesiredCapacity=9, - ) - - group = as_client.describe_auto_scaling_groups()["AutoScalingGroups"][0] - group["AutoScalingGroupName"].should.equal("tester_group") - set(group["AvailabilityZones"]).should.equal(set(["us-east-1a", "us-east-1b"])) - group["DesiredCapacity"].should.equal(2) - group["MaxSize"].should.equal(INSTANCE_COUNT_GROUP) - group["MinSize"].should.equal(INSTANCE_COUNT_GROUP) - group["Instances"].should.have.length_of(INSTANCE_COUNT_GROUP) - group["VPCZoneIdentifier"].should.equal( - "{subnet1},{subnet2}".format( - subnet1=mocked_networking["subnet1"], subnet2=mocked_networking["subnet2"] - ) - ) - group["LaunchConfigurationName"].should.equal("tester") - group["DefaultCooldown"].should.equal(60) - group["HealthCheckGracePeriod"].should.equal(100) - group["HealthCheckType"].should.equal("EC2") - group["LoadBalancerNames"].should.equal(["test_lb"]) - group["PlacementGroup"].should.equal("test_placement") - list(group["TerminationPolicies"]).should.equal( - ["OldestInstance", "NewestInstance"] - ) - list(group["Tags"]).should.have.length_of(1) - tag = group["Tags"][0] - tag["ResourceId"].should.equal("tester_group") - tag["Key"].should.equal("test_key") - tag["Value"].should.equal("test_value") - tag["PropagateAtLaunch"].should.equal(True) - - instances_attached = elb_client.describe_instance_health( - LoadBalancerName="test_lb" - )["InstanceStates"] - instances_attached.should.have.length_of( - INSTANCE_COUNT_START + INSTANCE_COUNT_GROUP - ) - attached_ids = [i["InstanceId"] for i in instances_attached] - for ec2_instance_id in instances_ids: - attached_ids.should.contain(ec2_instance_id) - - scheduled_actions = as_client.describe_scheduled_actions( - AutoScalingGroupName="tester_group" - ) - scheduled_action_1 = scheduled_actions["ScheduledUpdateGroupActions"][0] - scheduled_action_1["AutoScalingGroupName"].should.equal("tester_group") - scheduled_action_1["DesiredCapacity"].should.equal(9) - scheduled_action_1["MaxSize"].should.equal(12) - scheduled_action_1["MinSize"].should.equal(5) - scheduled_action_1.should.contain("StartTime") - scheduled_action_1.should.contain("EndTime") - scheduled_action_1["Recurrence"].should.equal("* * * * *") - scheduled_action_1["ScheduledActionName"].should.equal("my-scheduled-action") +from .utils import setup_networking, setup_instance_with_networking @mock_autoscaling @@ -386,185 +257,6 @@ def test_scheduled_action_delete(): actions.should.have.length_of(1) -@mock_autoscaling -@mock_elb -def test_describe_load_balancers(): - mocked_networking = setup_networking() - INSTANCE_COUNT = 2 - - elb_client = boto3.client("elb", region_name="us-east-1") - elb_client.create_load_balancer( - LoadBalancerName="my-lb", - Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}], - AvailabilityZones=["us-east-1a", "us-east-1b"], - ) - - client = boto3.client("autoscaling", region_name="us-east-1") - client.create_launch_configuration( - LaunchConfigurationName="test_launch_configuration", - ImageId=EXAMPLE_AMI_ID, - InstanceType="t2.medium", - ) - client.create_auto_scaling_group( - AutoScalingGroupName="test_asg", - LaunchConfigurationName="test_launch_configuration", - LoadBalancerNames=["my-lb"], - MinSize=0, - MaxSize=INSTANCE_COUNT, - DesiredCapacity=INSTANCE_COUNT, - Tags=[ - { - "ResourceId": "test_asg", - "Key": "test_key", - "Value": "test_value", - "PropagateAtLaunch": True, - } - ], - VPCZoneIdentifier=mocked_networking["subnet1"], - ) - - response = client.describe_load_balancers(AutoScalingGroupName="test_asg") - assert response["ResponseMetadata"]["RequestId"] - list(response["LoadBalancers"]).should.have.length_of(1) - response["LoadBalancers"][0]["LoadBalancerName"].should.equal("my-lb") - - -@mock_autoscaling -@mock_elb -def test_create_elb_and_autoscaling_group_no_relationship(): - mocked_networking = setup_networking() - INSTANCE_COUNT = 2 - ELB_NAME = "my-elb" - - elb_client = boto3.client("elb", region_name="us-east-1") - elb_client.create_load_balancer( - LoadBalancerName=ELB_NAME, - Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}], - AvailabilityZones=["us-east-1a", "us-east-1b"], - ) - - client = boto3.client("autoscaling", region_name="us-east-1") - client.create_launch_configuration( - LaunchConfigurationName="test_launch_configuration", - ImageId=EXAMPLE_AMI_ID, - InstanceType="t2.medium", - ) - - client.create_auto_scaling_group( - AutoScalingGroupName="test_asg", - LaunchConfigurationName="test_launch_configuration", - MinSize=0, - MaxSize=INSTANCE_COUNT, - DesiredCapacity=INSTANCE_COUNT, - VPCZoneIdentifier=mocked_networking["subnet1"], - ) - - # autoscaling group and elb should have no relationship - response = client.describe_load_balancers(AutoScalingGroupName="test_asg") - list(response["LoadBalancers"]).should.have.length_of(0) - response = elb_client.describe_load_balancers(LoadBalancerNames=[ELB_NAME]) - list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(0) - - -@mock_autoscaling -@mock_elb -def test_attach_load_balancer(): - mocked_networking = setup_networking() - INSTANCE_COUNT = 2 - - elb_client = boto3.client("elb", region_name="us-east-1") - elb_client.create_load_balancer( - LoadBalancerName="my-lb", - Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}], - AvailabilityZones=["us-east-1a", "us-east-1b"], - ) - - client = boto3.client("autoscaling", region_name="us-east-1") - client.create_launch_configuration( - LaunchConfigurationName="test_launch_configuration", - ImageId=EXAMPLE_AMI_ID, - InstanceType="t2.medium", - ) - client.create_auto_scaling_group( - AutoScalingGroupName="test_asg", - LaunchConfigurationName="test_launch_configuration", - MinSize=0, - MaxSize=INSTANCE_COUNT, - DesiredCapacity=INSTANCE_COUNT, - Tags=[ - { - "ResourceId": "test_asg", - "Key": "test_key", - "Value": "test_value", - "PropagateAtLaunch": True, - } - ], - VPCZoneIdentifier=mocked_networking["subnet1"], - ) - - response = client.attach_load_balancers( - AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"] - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"]) - list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of( - INSTANCE_COUNT - ) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - list(response["AutoScalingGroups"][0]["LoadBalancerNames"]).should.have.length_of(1) - - -@mock_autoscaling -@mock_elb -def test_detach_load_balancer(): - mocked_networking = setup_networking() - INSTANCE_COUNT = 2 - - elb_client = boto3.client("elb", region_name="us-east-1") - elb_client.create_load_balancer( - LoadBalancerName="my-lb", - Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}], - AvailabilityZones=["us-east-1a", "us-east-1b"], - ) - - client = boto3.client("autoscaling", region_name="us-east-1") - client.create_launch_configuration( - LaunchConfigurationName="test_launch_configuration", - ImageId=EXAMPLE_AMI_ID, - InstanceType="t2.medium", - ) - client.create_auto_scaling_group( - AutoScalingGroupName="test_asg", - LaunchConfigurationName="test_launch_configuration", - LoadBalancerNames=["my-lb"], - MinSize=0, - MaxSize=INSTANCE_COUNT, - DesiredCapacity=INSTANCE_COUNT, - Tags=[ - { - "ResourceId": "test_asg", - "Key": "test_key", - "Value": "test_value", - "PropagateAtLaunch": True, - } - ], - VPCZoneIdentifier=mocked_networking["subnet1"], - ) - - response = client.detach_load_balancers( - AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"] - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"]) - list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(0) - - response = client.describe_load_balancers(AutoScalingGroupName="test_asg") - list(response["LoadBalancers"]).should.have.length_of(0) - - @mock_autoscaling def test_create_autoscaling_group(): mocked_networking = setup_networking() @@ -1324,83 +1016,6 @@ def test_autoscaling_describe_policies(): policy.shouldnt.have.key("TargetTrackingConfiguration") -@mock_elb -@mock_autoscaling -@mock_ec2 -def test_detach_one_instance_decrement(): - mocked_networking = setup_networking() - client = boto3.client("autoscaling", region_name="us-east-1") - _ = client.create_launch_configuration( - LaunchConfigurationName="test_launch_configuration", - ImageId=EXAMPLE_AMI_ID, - InstanceType="t2.medium", - ) - client.create_auto_scaling_group( - AutoScalingGroupName="test_asg", - LaunchConfigurationName="test_launch_configuration", - MinSize=0, - MaxSize=2, - DesiredCapacity=2, - Tags=[ - { - "ResourceId": "test_asg", - "ResourceType": "auto-scaling-group", - "Key": "propogated-tag-key", - "Value": "propagate-tag-value", - "PropagateAtLaunch": True, - } - ], - VPCZoneIdentifier=mocked_networking["subnet1"], - ) - - elb_client = boto3.client("elb", region_name="us-east-1") - elb_client.create_load_balancer( - LoadBalancerName="my-lb", - Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}], - AvailabilityZones=["us-east-1a", "us-east-1b"], - ) - - response = client.attach_load_balancers( - AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"] - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - instance_to_detach = response["AutoScalingGroups"][0]["Instances"][0]["InstanceId"] - instance_to_keep = response["AutoScalingGroups"][0]["Instances"][1]["InstanceId"] - - ec2_client = boto3.client("ec2", region_name="us-east-1") - - response = client.detach_instances( - AutoScalingGroupName="test_asg", - InstanceIds=[instance_to_detach], - ShouldDecrementDesiredCapacity=True, - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - response["AutoScalingGroups"][0]["Instances"].should.have.length_of(1) - instance_to_detach.shouldnt.be.within( - [x["InstanceId"] for x in response["AutoScalingGroups"][0]["Instances"]] - ) - - # test to ensure tag has been removed - response = ec2_client.describe_instances(InstanceIds=[instance_to_detach]) - tags = response["Reservations"][0]["Instances"][0]["Tags"] - tags.should.have.length_of(1) - - # test to ensure tag is present on other instance - response = ec2_client.describe_instances(InstanceIds=[instance_to_keep]) - tags = response["Reservations"][0]["Instances"][0]["Tags"] - tags.should.have.length_of(2) - - response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"]) - list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(1) - instance_to_detach.shouldnt.be.within( - [x["InstanceId"] for x in response["LoadBalancerDescriptions"][0]["Instances"]] - ) - - @mock_autoscaling @mock_ec2 def test_create_autoscaling_policy_with_policytype__targettrackingscaling(): @@ -1547,804 +1162,6 @@ def test_create_autoscaling_policy_with_predictive_scaling_config(): ) -@mock_elb -@mock_autoscaling -@mock_ec2 -def test_detach_one_instance(): - mocked_networking = setup_networking() - client = boto3.client("autoscaling", region_name="us-east-1") - _ = client.create_launch_configuration( - LaunchConfigurationName="test_launch_configuration", - ImageId=EXAMPLE_AMI_ID, - InstanceType="t2.medium", - ) - client.create_auto_scaling_group( - AutoScalingGroupName="test_asg", - LaunchConfigurationName="test_launch_configuration", - MinSize=0, - MaxSize=2, - DesiredCapacity=2, - Tags=[ - { - "ResourceId": "test_asg", - "ResourceType": "auto-scaling-group", - "Key": "propogated-tag-key", - "Value": "propagate-tag-value", - "PropagateAtLaunch": True, - } - ], - VPCZoneIdentifier=mocked_networking["subnet1"], - ) - - elb_client = boto3.client("elb", region_name="us-east-1") - elb_client.create_load_balancer( - LoadBalancerName="my-lb", - Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}], - AvailabilityZones=["us-east-1a", "us-east-1b"], - ) - - response = client.attach_load_balancers( - AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"] - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - instance_to_detach = response["AutoScalingGroups"][0]["Instances"][0]["InstanceId"] - instance_to_keep = response["AutoScalingGroups"][0]["Instances"][1]["InstanceId"] - - ec2_client = boto3.client("ec2", region_name="us-east-1") - - response = client.detach_instances( - AutoScalingGroupName="test_asg", - InstanceIds=[instance_to_detach], - ShouldDecrementDesiredCapacity=False, - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - # test to ensure instance was replaced - response["AutoScalingGroups"][0]["Instances"].should.have.length_of(2) - - response = ec2_client.describe_instances(InstanceIds=[instance_to_detach]) - tags = response["Reservations"][0]["Instances"][0]["Tags"] - tags.should.have.length_of(1) - - response = ec2_client.describe_instances(InstanceIds=[instance_to_keep]) - tags = response["Reservations"][0]["Instances"][0]["Tags"] - tags.should.have.length_of(2) - - response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"]) - list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(2) - instance_to_detach.shouldnt.be.within( - [x["InstanceId"] for x in response["LoadBalancerDescriptions"][0]["Instances"]] - ) - - -@mock_elb -@mock_autoscaling -@mock_ec2 -def test_standby_one_instance_decrement(): - mocked_networking = setup_networking() - client = boto3.client("autoscaling", region_name="us-east-1") - _ = client.create_launch_configuration( - LaunchConfigurationName="test_launch_configuration", - ImageId=EXAMPLE_AMI_ID, - InstanceType="t2.medium", - ) - client.create_auto_scaling_group( - AutoScalingGroupName="test_asg", - LaunchConfigurationName="test_launch_configuration", - MinSize=0, - MaxSize=2, - DesiredCapacity=2, - Tags=[ - { - "ResourceId": "test_asg", - "ResourceType": "auto-scaling-group", - "Key": "propogated-tag-key", - "Value": "propagate-tag-value", - "PropagateAtLaunch": True, - } - ], - VPCZoneIdentifier=mocked_networking["subnet1"], - ) - - elb_client = boto3.client("elb", region_name="us-east-1") - elb_client.create_load_balancer( - LoadBalancerName="my-lb", - Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}], - AvailabilityZones=["us-east-1a", "us-east-1b"], - ) - - response = client.attach_load_balancers( - AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"] - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - instance_to_standby = response["AutoScalingGroups"][0]["Instances"][0]["InstanceId"] - - ec2_client = boto3.client("ec2", region_name="us-east-1") - - response = client.enter_standby( - AutoScalingGroupName="test_asg", - InstanceIds=[instance_to_standby], - ShouldDecrementDesiredCapacity=True, - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - response["AutoScalingGroups"][0]["Instances"].should.have.length_of(2) - response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(1) - - response = client.describe_auto_scaling_instances(InstanceIds=[instance_to_standby]) - response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") - - # test to ensure tag has been retained (standby instance is still part of the ASG) - response = ec2_client.describe_instances() - for reservation in response["Reservations"]: - for instance in reservation["Instances"]: - tags = instance["Tags"] - tags.should.have.length_of(2) - - response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"]) - list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(1) - instance_to_standby.shouldnt.be.within( - [x["InstanceId"] for x in response["LoadBalancerDescriptions"][0]["Instances"]] - ) - - -@mock_elb -@mock_autoscaling -@mock_ec2 -def test_standby_one_instance(): - mocked_networking = setup_networking() - client = boto3.client("autoscaling", region_name="us-east-1") - _ = client.create_launch_configuration( - LaunchConfigurationName="test_launch_configuration", - ImageId=EXAMPLE_AMI_ID, - InstanceType="t2.medium", - ) - client.create_auto_scaling_group( - AutoScalingGroupName="test_asg", - LaunchConfigurationName="test_launch_configuration", - MinSize=0, - MaxSize=2, - DesiredCapacity=2, - Tags=[ - { - "ResourceId": "test_asg", - "ResourceType": "auto-scaling-group", - "Key": "propogated-tag-key", - "Value": "propagate-tag-value", - "PropagateAtLaunch": True, - } - ], - VPCZoneIdentifier=mocked_networking["subnet1"], - ) - - elb_client = boto3.client("elb", region_name="us-east-1") - elb_client.create_load_balancer( - LoadBalancerName="my-lb", - Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}], - AvailabilityZones=["us-east-1a", "us-east-1b"], - ) - - response = client.attach_load_balancers( - AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"] - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - instance_to_standby = response["AutoScalingGroups"][0]["Instances"][0]["InstanceId"] - - ec2_client = boto3.client("ec2", region_name="us-east-1") - - response = client.enter_standby( - AutoScalingGroupName="test_asg", - InstanceIds=[instance_to_standby], - ShouldDecrementDesiredCapacity=False, - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3) - response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2) - - response = client.describe_auto_scaling_instances(InstanceIds=[instance_to_standby]) - response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") - - # test to ensure tag has been retained (standby instance is still part of the ASG) - response = ec2_client.describe_instances() - for reservation in response["Reservations"]: - for instance in reservation["Instances"]: - tags = instance["Tags"] - tags.should.have.length_of(2) - - response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"]) - list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(2) - instance_to_standby.shouldnt.be.within( - [x["InstanceId"] for x in response["LoadBalancerDescriptions"][0]["Instances"]] - ) - - -@mock_elb -@mock_autoscaling -@mock_ec2 -def test_standby_elb_update(): - mocked_networking = setup_networking() - client = boto3.client("autoscaling", region_name="us-east-1") - _ = client.create_launch_configuration( - LaunchConfigurationName="test_launch_configuration", - ImageId=EXAMPLE_AMI_ID, - InstanceType="t2.medium", - ) - client.create_auto_scaling_group( - AutoScalingGroupName="test_asg", - LaunchConfigurationName="test_launch_configuration", - MinSize=0, - MaxSize=2, - DesiredCapacity=2, - Tags=[ - { - "ResourceId": "test_asg", - "ResourceType": "auto-scaling-group", - "Key": "propogated-tag-key", - "Value": "propagate-tag-value", - "PropagateAtLaunch": True, - } - ], - VPCZoneIdentifier=mocked_networking["subnet1"], - ) - - elb_client = boto3.client("elb", region_name="us-east-1") - elb_client.create_load_balancer( - LoadBalancerName="my-lb", - Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}], - AvailabilityZones=["us-east-1a", "us-east-1b"], - ) - - response = client.attach_load_balancers( - AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"] - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - instance_to_standby = response["AutoScalingGroups"][0]["Instances"][0]["InstanceId"] - - response = client.enter_standby( - AutoScalingGroupName="test_asg", - InstanceIds=[instance_to_standby], - ShouldDecrementDesiredCapacity=False, - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3) - response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2) - - response = client.describe_auto_scaling_instances(InstanceIds=[instance_to_standby]) - response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") - - response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"]) - list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(2) - instance_to_standby.shouldnt.be.within( - [x["InstanceId"] for x in response["LoadBalancerDescriptions"][0]["Instances"]] - ) - - -@mock_elb -@mock_autoscaling -@mock_ec2 -def test_standby_terminate_instance_decrement(): - mocked_networking = setup_networking() - client = boto3.client("autoscaling", region_name="us-east-1") - _ = client.create_launch_configuration( - LaunchConfigurationName="test_launch_configuration", - ImageId=EXAMPLE_AMI_ID, - InstanceType="t2.medium", - ) - client.create_auto_scaling_group( - AutoScalingGroupName="test_asg", - LaunchConfigurationName="test_launch_configuration", - MinSize=0, - MaxSize=3, - DesiredCapacity=2, - Tags=[ - { - "ResourceId": "test_asg", - "ResourceType": "auto-scaling-group", - "Key": "propogated-tag-key", - "Value": "propagate-tag-value", - "PropagateAtLaunch": True, - } - ], - VPCZoneIdentifier=mocked_networking["subnet1"], - ) - - elb_client = boto3.client("elb", region_name="us-east-1") - elb_client.create_load_balancer( - LoadBalancerName="my-lb", - Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}], - AvailabilityZones=["us-east-1a", "us-east-1b"], - ) - - response = client.attach_load_balancers( - AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"] - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - instance_to_standby_terminate = response["AutoScalingGroups"][0]["Instances"][0][ - "InstanceId" - ] - - ec2_client = boto3.client("ec2", region_name="us-east-1") - - response = client.enter_standby( - AutoScalingGroupName="test_asg", - InstanceIds=[instance_to_standby_terminate], - ShouldDecrementDesiredCapacity=False, - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3) - response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2) - - response = client.describe_auto_scaling_instances( - InstanceIds=[instance_to_standby_terminate] - ) - response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") - - response = client.terminate_instance_in_auto_scaling_group( - InstanceId=instance_to_standby_terminate, ShouldDecrementDesiredCapacity=True - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - # AWS still decrements desired capacity ASG if requested, even if the terminated instance is in standby - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - response["AutoScalingGroups"][0]["Instances"].should.have.length_of(1) - response["AutoScalingGroups"][0]["Instances"][0]["InstanceId"].should_not.equal( - instance_to_standby_terminate - ) - response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(1) - - response = ec2_client.describe_instances( - InstanceIds=[instance_to_standby_terminate] - ) - response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal( - "terminated" - ) - - response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"]) - list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(1) - instance_to_standby_terminate.shouldnt.be.within( - [x["InstanceId"] for x in response["LoadBalancerDescriptions"][0]["Instances"]] - ) - - -@mock_elb -@mock_autoscaling -@mock_ec2 -def test_standby_terminate_instance_no_decrement(): - mocked_networking = setup_networking() - client = boto3.client("autoscaling", region_name="us-east-1") - _ = client.create_launch_configuration( - LaunchConfigurationName="test_launch_configuration", - ImageId=EXAMPLE_AMI_ID, - InstanceType="t2.medium", - ) - client.create_auto_scaling_group( - AutoScalingGroupName="test_asg", - LaunchConfigurationName="test_launch_configuration", - MinSize=0, - MaxSize=3, - DesiredCapacity=2, - Tags=[ - { - "ResourceId": "test_asg", - "ResourceType": "auto-scaling-group", - "Key": "propogated-tag-key", - "Value": "propagate-tag-value", - "PropagateAtLaunch": True, - } - ], - VPCZoneIdentifier=mocked_networking["subnet1"], - ) - - elb_client = boto3.client("elb", region_name="us-east-1") - elb_client.create_load_balancer( - LoadBalancerName="my-lb", - Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}], - AvailabilityZones=["us-east-1a", "us-east-1b"], - ) - - response = client.attach_load_balancers( - AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"] - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - instance_to_standby_terminate = response["AutoScalingGroups"][0]["Instances"][0][ - "InstanceId" - ] - - ec2_client = boto3.client("ec2", region_name="us-east-1") - - response = client.enter_standby( - AutoScalingGroupName="test_asg", - InstanceIds=[instance_to_standby_terminate], - ShouldDecrementDesiredCapacity=False, - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3) - response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2) - - response = client.describe_auto_scaling_instances( - InstanceIds=[instance_to_standby_terminate] - ) - response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") - - response = client.terminate_instance_in_auto_scaling_group( - InstanceId=instance_to_standby_terminate, ShouldDecrementDesiredCapacity=False - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - group = response["AutoScalingGroups"][0] - group["Instances"].should.have.length_of(2) - instance_to_standby_terminate.shouldnt.be.within( - [x["InstanceId"] for x in group["Instances"]] - ) - group["DesiredCapacity"].should.equal(2) - - response = ec2_client.describe_instances( - InstanceIds=[instance_to_standby_terminate] - ) - response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal( - "terminated" - ) - - response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"]) - list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(2) - instance_to_standby_terminate.shouldnt.be.within( - [x["InstanceId"] for x in response["LoadBalancerDescriptions"][0]["Instances"]] - ) - - -@mock_elb -@mock_autoscaling -@mock_ec2 -def test_standby_detach_instance_decrement(): - mocked_networking = setup_networking() - client = boto3.client("autoscaling", region_name="us-east-1") - _ = client.create_launch_configuration( - LaunchConfigurationName="test_launch_configuration", - ImageId=EXAMPLE_AMI_ID, - InstanceType="t2.medium", - ) - client.create_auto_scaling_group( - AutoScalingGroupName="test_asg", - LaunchConfigurationName="test_launch_configuration", - MinSize=0, - MaxSize=3, - DesiredCapacity=2, - Tags=[ - { - "ResourceId": "test_asg", - "ResourceType": "auto-scaling-group", - "Key": "propogated-tag-key", - "Value": "propagate-tag-value", - "PropagateAtLaunch": True, - } - ], - VPCZoneIdentifier=mocked_networking["subnet1"], - ) - - elb_client = boto3.client("elb", region_name="us-east-1") - elb_client.create_load_balancer( - LoadBalancerName="my-lb", - Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}], - AvailabilityZones=["us-east-1a", "us-east-1b"], - ) - - response = client.attach_load_balancers( - AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"] - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - instance_to_standby_detach = response["AutoScalingGroups"][0]["Instances"][0][ - "InstanceId" - ] - - ec2_client = boto3.client("ec2", region_name="us-east-1") - - response = client.enter_standby( - AutoScalingGroupName="test_asg", - InstanceIds=[instance_to_standby_detach], - ShouldDecrementDesiredCapacity=False, - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3) - response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2) - - response = client.describe_auto_scaling_instances( - InstanceIds=[instance_to_standby_detach] - ) - response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") - - response = client.detach_instances( - AutoScalingGroupName="test_asg", - InstanceIds=[instance_to_standby_detach], - ShouldDecrementDesiredCapacity=True, - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - # AWS still decrements desired capacity ASG if requested, even if the detached instance was in standby - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - response["AutoScalingGroups"][0]["Instances"].should.have.length_of(1) - response["AutoScalingGroups"][0]["Instances"][0]["InstanceId"].should_not.equal( - instance_to_standby_detach - ) - response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(1) - - response = ec2_client.describe_instances(InstanceIds=[instance_to_standby_detach]) - response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal("running") - - response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"]) - list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(1) - instance_to_standby_detach.shouldnt.be.within( - [x["InstanceId"] for x in response["LoadBalancerDescriptions"][0]["Instances"]] - ) - - -@mock_elb -@mock_autoscaling -@mock_ec2 -def test_standby_detach_instance_no_decrement(): - mocked_networking = setup_networking() - client = boto3.client("autoscaling", region_name="us-east-1") - _ = client.create_launch_configuration( - LaunchConfigurationName="test_launch_configuration", - ImageId=EXAMPLE_AMI_ID, - InstanceType="t2.medium", - ) - client.create_auto_scaling_group( - AutoScalingGroupName="test_asg", - LaunchConfigurationName="test_launch_configuration", - MinSize=0, - MaxSize=3, - DesiredCapacity=2, - Tags=[ - { - "ResourceId": "test_asg", - "ResourceType": "auto-scaling-group", - "Key": "propogated-tag-key", - "Value": "propagate-tag-value", - "PropagateAtLaunch": True, - } - ], - VPCZoneIdentifier=mocked_networking["subnet1"], - ) - - elb_client = boto3.client("elb", region_name="us-east-1") - elb_client.create_load_balancer( - LoadBalancerName="my-lb", - Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}], - AvailabilityZones=["us-east-1a", "us-east-1b"], - ) - - response = client.attach_load_balancers( - AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"] - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - instance_to_standby_detach = response["AutoScalingGroups"][0]["Instances"][0][ - "InstanceId" - ] - - ec2_client = boto3.client("ec2", region_name="us-east-1") - - response = client.enter_standby( - AutoScalingGroupName="test_asg", - InstanceIds=[instance_to_standby_detach], - ShouldDecrementDesiredCapacity=False, - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3) - response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2) - - response = client.describe_auto_scaling_instances( - InstanceIds=[instance_to_standby_detach] - ) - response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") - - response = client.detach_instances( - AutoScalingGroupName="test_asg", - InstanceIds=[instance_to_standby_detach], - ShouldDecrementDesiredCapacity=False, - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - group = response["AutoScalingGroups"][0] - group["Instances"].should.have.length_of(2) - instance_to_standby_detach.shouldnt.be.within( - [x["InstanceId"] for x in group["Instances"]] - ) - group["DesiredCapacity"].should.equal(2) - - response = ec2_client.describe_instances(InstanceIds=[instance_to_standby_detach]) - response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal("running") - - response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"]) - list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(2) - instance_to_standby_detach.shouldnt.be.within( - [x["InstanceId"] for x in response["LoadBalancerDescriptions"][0]["Instances"]] - ) - - -@mock_elb -@mock_autoscaling -@mock_ec2 -def test_standby_exit_standby(): - mocked_networking = setup_networking() - client = boto3.client("autoscaling", region_name="us-east-1") - _ = client.create_launch_configuration( - LaunchConfigurationName="test_launch_configuration", - ImageId=EXAMPLE_AMI_ID, - InstanceType="t2.medium", - ) - client.create_auto_scaling_group( - AutoScalingGroupName="test_asg", - LaunchConfigurationName="test_launch_configuration", - MinSize=0, - MaxSize=3, - DesiredCapacity=2, - Tags=[ - { - "ResourceId": "test_asg", - "ResourceType": "auto-scaling-group", - "Key": "propogated-tag-key", - "Value": "propagate-tag-value", - "PropagateAtLaunch": True, - } - ], - VPCZoneIdentifier=mocked_networking["subnet1"], - ) - - elb_client = boto3.client("elb", region_name="us-east-1") - elb_client.create_load_balancer( - LoadBalancerName="my-lb", - Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}], - AvailabilityZones=["us-east-1a", "us-east-1b"], - ) - - response = client.attach_load_balancers( - AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"] - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - instance_to_standby_exit_standby = response["AutoScalingGroups"][0]["Instances"][0][ - "InstanceId" - ] - - ec2_client = boto3.client("ec2", region_name="us-east-1") - - response = client.enter_standby( - AutoScalingGroupName="test_asg", - InstanceIds=[instance_to_standby_exit_standby], - ShouldDecrementDesiredCapacity=False, - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3) - response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2) - - response = client.describe_auto_scaling_instances( - InstanceIds=[instance_to_standby_exit_standby] - ) - response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") - - response = client.exit_standby( - AutoScalingGroupName="test_asg", InstanceIds=[instance_to_standby_exit_standby] - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - group = response["AutoScalingGroups"][0] - group["Instances"].should.have.length_of(3) - instance_to_standby_exit_standby.should.be.within( - [x["InstanceId"] for x in group["Instances"]] - ) - group["DesiredCapacity"].should.equal(3) - - response = ec2_client.describe_instances( - InstanceIds=[instance_to_standby_exit_standby] - ) - response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal("running") - - response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"]) - list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(3) - instance_to_standby_exit_standby.should.be.within( - [x["InstanceId"] for x in response["LoadBalancerDescriptions"][0]["Instances"]] - ) - - -@mock_elb -@mock_autoscaling -@mock_ec2 -def test_attach_one_instance(): - mocked_networking = setup_networking() - client = boto3.client("autoscaling", region_name="us-east-1") - _ = client.create_launch_configuration( - LaunchConfigurationName="test_launch_configuration", - ImageId=EXAMPLE_AMI_ID, - InstanceType="t2.medium", - ) - client.create_auto_scaling_group( - AutoScalingGroupName="test_asg", - LaunchConfigurationName="test_launch_configuration", - MinSize=0, - MaxSize=4, - DesiredCapacity=2, - Tags=[ - { - "ResourceId": "test_asg", - "ResourceType": "auto-scaling-group", - "Key": "propogated-tag-key", - "Value": "propagate-tag-value", - "PropagateAtLaunch": True, - } - ], - VPCZoneIdentifier=mocked_networking["subnet1"], - NewInstancesProtectedFromScaleIn=True, - ) - - elb_client = boto3.client("elb", region_name="us-east-1") - elb_client.create_load_balancer( - LoadBalancerName="my-lb", - Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}], - AvailabilityZones=["us-east-1a", "us-east-1b"], - ) - - response = client.attach_load_balancers( - AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"] - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - ec2 = boto3.resource("ec2", "us-east-1") - instances_to_add = [ - x.id - for x in ec2.create_instances(ImageId=EXAMPLE_AMI_ID, MinCount=1, MaxCount=1) - ] - - response = client.attach_instances( - AutoScalingGroupName="test_asg", InstanceIds=instances_to_add - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - instances = response["AutoScalingGroups"][0]["Instances"] - instances.should.have.length_of(3) - for instance in instances: - instance["ProtectedFromScaleIn"].should.equal(True) - - response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"]) - list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(3) - - @mock_autoscaling @mock_ec2 def test_describe_instance_health(): @@ -2719,113 +1536,6 @@ def test_terminate_instance_via_ec2_in_autoscaling_group(): replaced_instance_id.should_not.equal(original_instance_id) -@mock_elb -@mock_autoscaling -@mock_ec2 -def test_terminate_instance_in_auto_scaling_group_decrement(): - mocked_networking = setup_networking() - client = boto3.client("autoscaling", region_name="us-east-1") - _ = client.create_launch_configuration( - LaunchConfigurationName="test_launch_configuration", - ImageId=EXAMPLE_AMI_ID, - InstanceType="t2.medium", - ) - _ = client.create_auto_scaling_group( - AutoScalingGroupName="test_asg", - LaunchConfigurationName="test_launch_configuration", - MinSize=0, - DesiredCapacity=1, - MaxSize=2, - VPCZoneIdentifier=mocked_networking["subnet1"], - NewInstancesProtectedFromScaleIn=False, - ) - - elb_client = boto3.client("elb", region_name="us-east-1") - elb_client.create_load_balancer( - LoadBalancerName="my-lb", - Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}], - AvailabilityZones=["us-east-1a", "us-east-1b"], - ) - - response = client.attach_load_balancers( - AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"] - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - original_instance_id = next( - instance["InstanceId"] - for instance in response["AutoScalingGroups"][0]["Instances"] - ) - client.terminate_instance_in_auto_scaling_group( - InstanceId=original_instance_id, ShouldDecrementDesiredCapacity=True - ) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - response["AutoScalingGroups"][0]["Instances"].should.equal([]) - response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(0) - - response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"]) - list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(0) - - -@mock_elb -@mock_autoscaling -@mock_ec2 -def test_terminate_instance_in_auto_scaling_group_no_decrement(): - mocked_networking = setup_networking() - client = boto3.client("autoscaling", region_name="us-east-1") - _ = client.create_launch_configuration( - LaunchConfigurationName="test_launch_configuration", - ImageId=EXAMPLE_AMI_ID, - InstanceType="t2.medium", - ) - _ = client.create_auto_scaling_group( - AutoScalingGroupName="test_asg", - LaunchConfigurationName="test_launch_configuration", - MinSize=0, - DesiredCapacity=1, - MaxSize=2, - VPCZoneIdentifier=mocked_networking["subnet1"], - NewInstancesProtectedFromScaleIn=False, - ) - - elb_client = boto3.client("elb", region_name="us-east-1") - elb_client.create_load_balancer( - LoadBalancerName="my-lb", - Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}], - AvailabilityZones=["us-east-1a", "us-east-1b"], - ) - - response = client.attach_load_balancers( - AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"] - ) - response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - original_instance_id = next( - instance["InstanceId"] - for instance in response["AutoScalingGroups"][0]["Instances"] - ) - client.terminate_instance_in_auto_scaling_group( - InstanceId=original_instance_id, ShouldDecrementDesiredCapacity=False - ) - - response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) - replaced_instance_id = next( - instance["InstanceId"] - for instance in response["AutoScalingGroups"][0]["Instances"] - ) - replaced_instance_id.should_not.equal(original_instance_id) - response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(1) - - response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"]) - list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(1) - original_instance_id.shouldnt.be.within( - [x["InstanceId"] for x in response["LoadBalancerDescriptions"][0]["Instances"]] - ) - - @mock_ec2 @mock_autoscaling def test_attach_instances(): diff --git a/tests/test_autoscaling/test_elb.py b/tests/test_autoscaling/test_elb.py new file mode 100644 index 000000000..15e632789 --- /dev/null +++ b/tests/test_autoscaling/test_elb.py @@ -0,0 +1,1058 @@ +import boto3 +import sure # noqa # pylint: disable=unused-import + +from moto import mock_autoscaling, mock_elb, mock_ec2 +from .utils import setup_networking +from tests import EXAMPLE_AMI_ID +from unittest import TestCase +from uuid import uuid4 + + +@mock_autoscaling +@mock_ec2 +@mock_elb +class TestAutoScalingELB(TestCase): + def setUp(self): + self.mocked_networking = setup_networking() + self.instance_count = 2 + self.elb_client = boto3.client("elb", region_name="us-east-1") + self.as_client = boto3.client("autoscaling", region_name="us-east-1") + self.lb_name = str(uuid4()) + self.asg_name = str(uuid4()) + self.lc_name = str(uuid4()) + + self.elb_client.create_load_balancer( + LoadBalancerName=self.lb_name, + Listeners=[ + {"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080} + ], + AvailabilityZones=["us-east-1a", "us-east-1b"], + ) + self.as_client.create_launch_configuration( + LaunchConfigurationName=self.lc_name, + ImageId=EXAMPLE_AMI_ID, + InstanceType="t2.medium", + ) + + def test_describe_load_balancers(self): + self.as_client.create_auto_scaling_group( + AutoScalingGroupName=self.asg_name, + LaunchConfigurationName=self.lc_name, + LoadBalancerNames=[self.lb_name], + MinSize=0, + MaxSize=self.instance_count, + DesiredCapacity=self.instance_count, + Tags=[ + { + "ResourceId": self.asg_name, + "Key": "test_key", + "Value": "test_value", + "PropagateAtLaunch": True, + } + ], + VPCZoneIdentifier=self.mocked_networking["subnet1"], + ) + + response = self.as_client.describe_load_balancers( + AutoScalingGroupName=self.asg_name + ) + assert response["ResponseMetadata"]["RequestId"] + list(response["LoadBalancers"]).should.have.length_of(1) + response["LoadBalancers"][0]["LoadBalancerName"].should.equal(self.lb_name) + + def test_create_elb_and_autoscaling_group_no_relationship(self): + self.as_client.create_auto_scaling_group( + AutoScalingGroupName=self.asg_name, + LaunchConfigurationName=self.lc_name, + MinSize=0, + MaxSize=self.instance_count, + DesiredCapacity=self.instance_count, + VPCZoneIdentifier=self.mocked_networking["subnet1"], + ) + + # autoscaling group and elb should have no relationship + response = self.as_client.describe_load_balancers( + AutoScalingGroupName=self.asg_name + ) + list(response["LoadBalancers"]).should.have.length_of(0) + response = self.elb_client.describe_load_balancers( + LoadBalancerNames=[self.lb_name] + ) + list( + response["LoadBalancerDescriptions"][0]["Instances"] + ).should.have.length_of(0) + + def test_attach_load_balancer(self): + self.as_client.create_auto_scaling_group( + AutoScalingGroupName=self.asg_name, + LaunchConfigurationName=self.lc_name, + MinSize=0, + MaxSize=self.instance_count, + DesiredCapacity=self.instance_count, + Tags=[ + { + "ResourceId": self.asg_name, + "Key": "test_key", + "Value": "test_value", + "PropagateAtLaunch": True, + } + ], + VPCZoneIdentifier=self.mocked_networking["subnet1"], + ) + + response = self.as_client.attach_load_balancers( + AutoScalingGroupName=self.asg_name, LoadBalancerNames=[self.lb_name] + ) + response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + + response = self.elb_client.describe_load_balancers( + LoadBalancerNames=[self.lb_name] + ) + list( + response["LoadBalancerDescriptions"][0]["Instances"] + ).should.have.length_of(self.instance_count) + + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + list( + response["AutoScalingGroups"][0]["LoadBalancerNames"] + ).should.have.length_of(1) + + def test_detach_load_balancer(self): + self.as_client.create_auto_scaling_group( + AutoScalingGroupName=self.asg_name, + LaunchConfigurationName=self.lc_name, + LoadBalancerNames=[self.lb_name], + MinSize=0, + MaxSize=self.instance_count, + DesiredCapacity=self.instance_count, + Tags=[ + { + "ResourceId": self.asg_name, + "Key": "test_key", + "Value": "test_value", + "PropagateAtLaunch": True, + } + ], + VPCZoneIdentifier=self.mocked_networking["subnet1"], + ) + + response = self.as_client.detach_load_balancers( + AutoScalingGroupName=self.asg_name, LoadBalancerNames=[self.lb_name] + ) + response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + + response = self.elb_client.describe_load_balancers( + LoadBalancerNames=[self.lb_name] + ) + list( + response["LoadBalancerDescriptions"][0]["Instances"] + ).should.have.length_of(0) + + response = self.as_client.describe_load_balancers( + AutoScalingGroupName=self.asg_name + ) + list(response["LoadBalancers"]).should.have.length_of(0) + + def test_create_autoscaling_group_within_elb(self): + # we attach a couple of machines to the load balancer + # that are not managed by the auto scaling group + INSTANCE_COUNT_START = 3 + INSTANCE_COUNT_GROUP = 2 + ec2 = boto3.resource("ec2", region_name="us-east-1") + instances = ec2.create_instances( + ImageId=EXAMPLE_AMI_ID, + InstanceType="t1.micro", + MaxCount=INSTANCE_COUNT_START, + MinCount=INSTANCE_COUNT_START, + SubnetId=self.mocked_networking["subnet1"], + ) + + instances_ids = [_.id for _ in instances] + self.elb_client.register_instances_with_load_balancer( + LoadBalancerName=self.lb_name, + Instances=[{"InstanceId": id} for id in instances_ids], + ) + + self.as_client.create_launch_configuration( + LaunchConfigurationName=self.lc_name, + ImageId=EXAMPLE_AMI_ID, + InstanceType="t2.medium", + ) + + self.as_client.create_auto_scaling_group( + AutoScalingGroupName=self.asg_name, + AvailabilityZones=["us-east-1a", "us-east-1b"], + DefaultCooldown=60, + DesiredCapacity=INSTANCE_COUNT_GROUP, + HealthCheckGracePeriod=100, + HealthCheckType="EC2", + LaunchConfigurationName=self.lc_name, + LoadBalancerNames=[self.lb_name], + MinSize=INSTANCE_COUNT_GROUP, + MaxSize=INSTANCE_COUNT_GROUP, + PlacementGroup="test_placement", + Tags=[ + { + "ResourceId": self.asg_name, + "ResourceType": "group", + "Key": "test_key", + "Value": "test_value", + "PropagateAtLaunch": True, + } + ], + TerminationPolicies=["OldestInstance", "NewestInstance"], + VPCZoneIdentifier="{subnet1},{subnet2}".format( + subnet1=self.mocked_networking["subnet1"], + subnet2=self.mocked_networking["subnet2"], + ), + ) + + self.as_client.put_scheduled_update_group_action( + AutoScalingGroupName=self.asg_name, + ScheduledActionName="my-scheduled-action", + StartTime="2022-07-01T00:00:00Z", + EndTime="2022-09-01T00:00:00Z", + Recurrence="* * * * *", + MinSize=5, + MaxSize=12, + DesiredCapacity=9, + ) + + resp = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + group = resp["AutoScalingGroups"][0] + group["AutoScalingGroupName"].should.equal(self.asg_name) + set(group["AvailabilityZones"]).should.equal(set(["us-east-1a", "us-east-1b"])) + group["DesiredCapacity"].should.equal(2) + group["MaxSize"].should.equal(INSTANCE_COUNT_GROUP) + group["MinSize"].should.equal(INSTANCE_COUNT_GROUP) + group["Instances"].should.have.length_of(INSTANCE_COUNT_GROUP) + group["VPCZoneIdentifier"].should.equal( + "{subnet1},{subnet2}".format( + subnet1=self.mocked_networking["subnet1"], + subnet2=self.mocked_networking["subnet2"], + ) + ) + group["LaunchConfigurationName"].should.equal(self.lc_name) + group["DefaultCooldown"].should.equal(60) + group["HealthCheckGracePeriod"].should.equal(100) + group["HealthCheckType"].should.equal("EC2") + group["LoadBalancerNames"].should.equal([self.lb_name]) + group["PlacementGroup"].should.equal("test_placement") + list(group["TerminationPolicies"]).should.equal( + ["OldestInstance", "NewestInstance"] + ) + list(group["Tags"]).should.have.length_of(1) + tag = group["Tags"][0] + tag["ResourceId"].should.equal(self.asg_name) + tag["Key"].should.equal("test_key") + tag["Value"].should.equal("test_value") + tag["PropagateAtLaunch"].should.equal(True) + + instances_attached = self.elb_client.describe_instance_health( + LoadBalancerName=self.lb_name + )["InstanceStates"] + instances_attached.should.have.length_of( + INSTANCE_COUNT_START + INSTANCE_COUNT_GROUP + ) + attached_ids = [i["InstanceId"] for i in instances_attached] + for ec2_instance_id in instances_ids: + attached_ids.should.contain(ec2_instance_id) + + scheduled_actions = self.as_client.describe_scheduled_actions( + AutoScalingGroupName=self.asg_name + ) + scheduled_action_1 = scheduled_actions["ScheduledUpdateGroupActions"][0] + scheduled_action_1["AutoScalingGroupName"].should.equal(self.asg_name) + scheduled_action_1["DesiredCapacity"].should.equal(9) + scheduled_action_1["MaxSize"].should.equal(12) + scheduled_action_1["MinSize"].should.equal(5) + scheduled_action_1.should.contain("StartTime") + scheduled_action_1.should.contain("EndTime") + scheduled_action_1["Recurrence"].should.equal("* * * * *") + scheduled_action_1["ScheduledActionName"].should.equal("my-scheduled-action") + + +@mock_autoscaling +@mock_elb +@mock_ec2 +class TestAutoScalingInstances(TestCase): + def setUp(self) -> None: + self.mocked_networking = setup_networking() + self.as_client = boto3.client("autoscaling", region_name="us-east-1") + self.elb_client = boto3.client("elb", region_name="us-east-1") + + self.asg_name = str(uuid4()) + self.lb_name = str(uuid4()) + self.lc_name = str(uuid4()) + self.as_client.create_launch_configuration( + LaunchConfigurationName=self.lc_name, + ImageId=EXAMPLE_AMI_ID, + InstanceType="t2.medium", + ) + + self.as_client.create_auto_scaling_group( + AutoScalingGroupName=self.asg_name, + LaunchConfigurationName=self.lc_name, + MinSize=0, + MaxSize=2, + DesiredCapacity=2, + Tags=[ + { + "ResourceId": self.asg_name, + "Key": "test_key", + "Value": "test_value", + "PropagateAtLaunch": True, + } + ], + VPCZoneIdentifier=self.mocked_networking["subnet1"], + ) + + self.elb_client.create_load_balancer( + LoadBalancerName=self.lb_name, + Listeners=[ + {"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080} + ], + AvailabilityZones=["us-east-1a", "us-east-1b"], + ) + + response = self.as_client.attach_load_balancers( + AutoScalingGroupName=self.asg_name, LoadBalancerNames=[self.lb_name] + ) + response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + + def test_detach_one_instance_decrement(self): + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + instance_to_detach = response["AutoScalingGroups"][0]["Instances"][0][ + "InstanceId" + ] + instance_to_keep = response["AutoScalingGroups"][0]["Instances"][1][ + "InstanceId" + ] + + ec2_client = boto3.client("ec2", region_name="us-east-1") + + response = self.as_client.detach_instances( + AutoScalingGroupName=self.asg_name, + InstanceIds=[instance_to_detach], + ShouldDecrementDesiredCapacity=True, + ) + response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + response["AutoScalingGroups"][0]["Instances"].should.have.length_of(1) + instance_to_detach.shouldnt.be.within( + [x["InstanceId"] for x in response["AutoScalingGroups"][0]["Instances"]] + ) + + # test to ensure tag has been removed + response = ec2_client.describe_instances(InstanceIds=[instance_to_detach]) + tags = response["Reservations"][0]["Instances"][0]["Tags"] + tags.should.have.length_of(1) + + # test to ensure tag is present on other instance + response = ec2_client.describe_instances(InstanceIds=[instance_to_keep]) + tags = response["Reservations"][0]["Instances"][0]["Tags"] + tags.should.have.length_of(2) + + response = self.elb_client.describe_load_balancers( + LoadBalancerNames=[self.lb_name] + ) + list( + response["LoadBalancerDescriptions"][0]["Instances"] + ).should.have.length_of(1) + instance_to_detach.shouldnt.be.within( + [ + x["InstanceId"] + for x in response["LoadBalancerDescriptions"][0]["Instances"] + ] + ) + + def test_detach_one_instance(self): + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + instance_to_detach = response["AutoScalingGroups"][0]["Instances"][0][ + "InstanceId" + ] + instance_to_keep = response["AutoScalingGroups"][0]["Instances"][1][ + "InstanceId" + ] + + ec2_client = boto3.client("ec2", region_name="us-east-1") + + response = self.as_client.detach_instances( + AutoScalingGroupName=self.asg_name, + InstanceIds=[instance_to_detach], + ShouldDecrementDesiredCapacity=False, + ) + response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + # test to ensure instance was replaced + response["AutoScalingGroups"][0]["Instances"].should.have.length_of(2) + + response = ec2_client.describe_instances(InstanceIds=[instance_to_detach]) + tags = response["Reservations"][0]["Instances"][0]["Tags"] + tags.should.have.length_of(1) + + response = ec2_client.describe_instances(InstanceIds=[instance_to_keep]) + tags = response["Reservations"][0]["Instances"][0]["Tags"] + tags.should.have.length_of(2) + + response = self.elb_client.describe_load_balancers( + LoadBalancerNames=[self.lb_name] + ) + list( + response["LoadBalancerDescriptions"][0]["Instances"] + ).should.have.length_of(2) + instance_to_detach.shouldnt.be.within( + [ + x["InstanceId"] + for x in response["LoadBalancerDescriptions"][0]["Instances"] + ] + ) + + def test_standby_one_instance_decrement(self): + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + instance_to_standby = response["AutoScalingGroups"][0]["Instances"][0][ + "InstanceId" + ] + + ec2_client = boto3.client("ec2", region_name="us-east-1") + + response = self.as_client.enter_standby( + AutoScalingGroupName=self.asg_name, + InstanceIds=[instance_to_standby], + ShouldDecrementDesiredCapacity=True, + ) + response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + response["AutoScalingGroups"][0]["Instances"].should.have.length_of(2) + response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(1) + + response = self.as_client.describe_auto_scaling_instances( + InstanceIds=[instance_to_standby] + ) + response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") + + # test to ensure tag has been retained (standby instance is still part of the ASG) + response = ec2_client.describe_instances(InstanceIds=[instance_to_standby]) + for reservation in response["Reservations"]: + for instance in reservation["Instances"]: + tags = instance["Tags"] + tags.should.have.length_of(2) + + response = self.elb_client.describe_load_balancers( + LoadBalancerNames=[self.lb_name] + ) + list( + response["LoadBalancerDescriptions"][0]["Instances"] + ).should.have.length_of(1) + instance_to_standby.shouldnt.be.within( + [ + x["InstanceId"] + for x in response["LoadBalancerDescriptions"][0]["Instances"] + ] + ) + + def test_standby_one_instance(self): + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + instance_to_standby = response["AutoScalingGroups"][0]["Instances"][0][ + "InstanceId" + ] + + ec2_client = boto3.client("ec2", region_name="us-east-1") + + response = self.as_client.enter_standby( + AutoScalingGroupName=self.asg_name, + InstanceIds=[instance_to_standby], + ShouldDecrementDesiredCapacity=False, + ) + response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3) + response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2) + + response = self.as_client.describe_auto_scaling_instances( + InstanceIds=[instance_to_standby] + ) + response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") + + # test to ensure tag has been retained (standby instance is still part of the ASG) + response = ec2_client.describe_instances(InstanceIds=[instance_to_standby]) + for reservation in response["Reservations"]: + for instance in reservation["Instances"]: + tags = instance["Tags"] + tags.should.have.length_of(2) + + response = self.elb_client.describe_load_balancers( + LoadBalancerNames=[self.lb_name] + ) + list( + response["LoadBalancerDescriptions"][0]["Instances"] + ).should.have.length_of(2) + instance_to_standby.shouldnt.be.within( + [ + x["InstanceId"] + for x in response["LoadBalancerDescriptions"][0]["Instances"] + ] + ) + + def test_standby_elb_update(self): + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + instance_to_standby = response["AutoScalingGroups"][0]["Instances"][0][ + "InstanceId" + ] + + response = self.as_client.enter_standby( + AutoScalingGroupName=self.asg_name, + InstanceIds=[instance_to_standby], + ShouldDecrementDesiredCapacity=False, + ) + response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3) + response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2) + + response = self.as_client.describe_auto_scaling_instances( + InstanceIds=[instance_to_standby] + ) + response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") + + response = self.elb_client.describe_load_balancers( + LoadBalancerNames=[self.lb_name] + ) + list( + response["LoadBalancerDescriptions"][0]["Instances"] + ).should.have.length_of(2) + instance_to_standby.shouldnt.be.within( + [ + x["InstanceId"] + for x in response["LoadBalancerDescriptions"][0]["Instances"] + ] + ) + + def test_standby_terminate_instance_decrement(self): + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + instance_to_standby_terminate = response["AutoScalingGroups"][0]["Instances"][ + 0 + ]["InstanceId"] + + ec2_client = boto3.client("ec2", region_name="us-east-1") + + response = self.as_client.enter_standby( + AutoScalingGroupName=self.asg_name, + InstanceIds=[instance_to_standby_terminate], + ShouldDecrementDesiredCapacity=False, + ) + response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3) + response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2) + + response = self.as_client.describe_auto_scaling_instances( + InstanceIds=[instance_to_standby_terminate] + ) + response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") + + response = self.as_client.terminate_instance_in_auto_scaling_group( + InstanceId=instance_to_standby_terminate, + ShouldDecrementDesiredCapacity=True, + ) + response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + + # AWS still decrements desired capacity ASG if requested, even if the terminated instance is in standby + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + response["AutoScalingGroups"][0]["Instances"].should.have.length_of(1) + response["AutoScalingGroups"][0]["Instances"][0]["InstanceId"].should_not.equal( + instance_to_standby_terminate + ) + response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(1) + + response = ec2_client.describe_instances( + InstanceIds=[instance_to_standby_terminate] + ) + response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal( + "terminated" + ) + + response = self.elb_client.describe_load_balancers( + LoadBalancerNames=[self.lb_name] + ) + list( + response["LoadBalancerDescriptions"][0]["Instances"] + ).should.have.length_of(1) + instance_to_standby_terminate.shouldnt.be.within( + [ + x["InstanceId"] + for x in response["LoadBalancerDescriptions"][0]["Instances"] + ] + ) + + def test_standby_terminate_instance_no_decrement(self): + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + instance_to_standby_terminate = response["AutoScalingGroups"][0]["Instances"][ + 0 + ]["InstanceId"] + + ec2_client = boto3.client("ec2", region_name="us-east-1") + + response = self.as_client.enter_standby( + AutoScalingGroupName=self.asg_name, + InstanceIds=[instance_to_standby_terminate], + ShouldDecrementDesiredCapacity=False, + ) + response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3) + response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2) + + response = self.as_client.describe_auto_scaling_instances( + InstanceIds=[instance_to_standby_terminate] + ) + response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") + + response = self.as_client.terminate_instance_in_auto_scaling_group( + InstanceId=instance_to_standby_terminate, + ShouldDecrementDesiredCapacity=False, + ) + response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + group = response["AutoScalingGroups"][0] + group["Instances"].should.have.length_of(2) + instance_to_standby_terminate.shouldnt.be.within( + [x["InstanceId"] for x in group["Instances"]] + ) + group["DesiredCapacity"].should.equal(2) + + response = ec2_client.describe_instances( + InstanceIds=[instance_to_standby_terminate] + ) + response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal( + "terminated" + ) + + response = self.elb_client.describe_load_balancers( + LoadBalancerNames=[self.lb_name] + ) + list( + response["LoadBalancerDescriptions"][0]["Instances"] + ).should.have.length_of(2) + instance_to_standby_terminate.shouldnt.be.within( + [ + x["InstanceId"] + for x in response["LoadBalancerDescriptions"][0]["Instances"] + ] + ) + + def test_standby_detach_instance_decrement(self): + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + instance_to_standby_detach = response["AutoScalingGroups"][0]["Instances"][0][ + "InstanceId" + ] + + ec2_client = boto3.client("ec2", region_name="us-east-1") + + response = self.as_client.enter_standby( + AutoScalingGroupName=self.asg_name, + InstanceIds=[instance_to_standby_detach], + ShouldDecrementDesiredCapacity=False, + ) + response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3) + response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2) + + response = self.as_client.describe_auto_scaling_instances( + InstanceIds=[instance_to_standby_detach] + ) + response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") + + response = self.as_client.detach_instances( + AutoScalingGroupName=self.asg_name, + InstanceIds=[instance_to_standby_detach], + ShouldDecrementDesiredCapacity=True, + ) + response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + + # AWS still decrements desired capacity ASG if requested, even if the detached instance was in standby + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + response["AutoScalingGroups"][0]["Instances"].should.have.length_of(1) + response["AutoScalingGroups"][0]["Instances"][0]["InstanceId"].should_not.equal( + instance_to_standby_detach + ) + response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(1) + + response = ec2_client.describe_instances( + InstanceIds=[instance_to_standby_detach] + ) + response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal( + "running" + ) + + response = self.elb_client.describe_load_balancers( + LoadBalancerNames=[self.lb_name] + ) + list( + response["LoadBalancerDescriptions"][0]["Instances"] + ).should.have.length_of(1) + instance_to_standby_detach.shouldnt.be.within( + [ + x["InstanceId"] + for x in response["LoadBalancerDescriptions"][0]["Instances"] + ] + ) + + def test_standby_detach_instance_no_decrement(self): + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + instance_to_standby_detach = response["AutoScalingGroups"][0]["Instances"][0][ + "InstanceId" + ] + + ec2_client = boto3.client("ec2", region_name="us-east-1") + + response = self.as_client.enter_standby( + AutoScalingGroupName=self.asg_name, + InstanceIds=[instance_to_standby_detach], + ShouldDecrementDesiredCapacity=False, + ) + response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3) + response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2) + + response = self.as_client.describe_auto_scaling_instances( + InstanceIds=[instance_to_standby_detach] + ) + response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") + + response = self.as_client.detach_instances( + AutoScalingGroupName=self.asg_name, + InstanceIds=[instance_to_standby_detach], + ShouldDecrementDesiredCapacity=False, + ) + response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + group = response["AutoScalingGroups"][0] + group["Instances"].should.have.length_of(2) + instance_to_standby_detach.shouldnt.be.within( + [x["InstanceId"] for x in group["Instances"]] + ) + group["DesiredCapacity"].should.equal(2) + + response = ec2_client.describe_instances( + InstanceIds=[instance_to_standby_detach] + ) + response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal( + "running" + ) + + response = self.elb_client.describe_load_balancers( + LoadBalancerNames=[self.lb_name] + ) + list( + response["LoadBalancerDescriptions"][0]["Instances"] + ).should.have.length_of(2) + instance_to_standby_detach.shouldnt.be.within( + [ + x["InstanceId"] + for x in response["LoadBalancerDescriptions"][0]["Instances"] + ] + ) + + def test_standby_exit_standby(self): + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + instance_to_standby_exit_standby = response["AutoScalingGroups"][0][ + "Instances" + ][0]["InstanceId"] + + ec2_client = boto3.client("ec2", region_name="us-east-1") + + response = self.as_client.enter_standby( + AutoScalingGroupName=self.asg_name, + InstanceIds=[instance_to_standby_exit_standby], + ShouldDecrementDesiredCapacity=False, + ) + response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3) + response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2) + + response = self.as_client.describe_auto_scaling_instances( + InstanceIds=[instance_to_standby_exit_standby] + ) + response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") + + response = self.as_client.exit_standby( + AutoScalingGroupName=self.asg_name, + InstanceIds=[instance_to_standby_exit_standby], + ) + response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + group = response["AutoScalingGroups"][0] + group["Instances"].should.have.length_of(3) + instance_to_standby_exit_standby.should.be.within( + [x["InstanceId"] for x in group["Instances"]] + ) + group["DesiredCapacity"].should.equal(3) + + response = ec2_client.describe_instances( + InstanceIds=[instance_to_standby_exit_standby] + ) + response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal( + "running" + ) + + response = self.elb_client.describe_load_balancers( + LoadBalancerNames=[self.lb_name] + ) + list( + response["LoadBalancerDescriptions"][0]["Instances"] + ).should.have.length_of(3) + instance_to_standby_exit_standby.should.be.within( + [ + x["InstanceId"] + for x in response["LoadBalancerDescriptions"][0]["Instances"] + ] + ) + + +@mock_autoscaling +@mock_elb +@mock_ec2 +class TestAutoScalingInstancesProtected(TestCase): + def setUp(self) -> None: + self.mocked_networking = setup_networking() + self.as_client = boto3.client("autoscaling", region_name="us-east-1") + self.elb_client = boto3.client("elb", region_name="us-east-1") + + self.asg_name = str(uuid4()) + self.lb_name = str(uuid4()) + self.lc_name = str(uuid4()) + self.as_client.create_launch_configuration( + LaunchConfigurationName=self.lc_name, + ImageId=EXAMPLE_AMI_ID, + InstanceType="t2.medium", + ) + + self.as_client.create_auto_scaling_group( + AutoScalingGroupName=self.asg_name, + LaunchConfigurationName=self.lc_name, + MinSize=0, + MaxSize=4, + DesiredCapacity=2, + Tags=[ + { + "ResourceId": self.asg_name, + "Key": "test_key", + "Value": "test_value", + "PropagateAtLaunch": True, + } + ], + VPCZoneIdentifier=self.mocked_networking["subnet1"], + NewInstancesProtectedFromScaleIn=True, + ) + + self.elb_client.create_load_balancer( + LoadBalancerName=self.lb_name, + Listeners=[ + {"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080} + ], + AvailabilityZones=["us-east-1a", "us-east-1b"], + ) + + response = self.as_client.attach_load_balancers( + AutoScalingGroupName=self.asg_name, LoadBalancerNames=[self.lb_name] + ) + response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + + def test_attach_one_instance(self): + ec2 = boto3.resource("ec2", "us-east-1") + instances_to_add = [ + x.id + for x in ec2.create_instances( + ImageId=EXAMPLE_AMI_ID, MinCount=1, MaxCount=1 + ) + ] + + response = self.as_client.attach_instances( + AutoScalingGroupName=self.asg_name, InstanceIds=instances_to_add + ) + response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + instances = response["AutoScalingGroups"][0]["Instances"] + instances.should.have.length_of(3) + for instance in instances: + instance["ProtectedFromScaleIn"].should.equal(True) + + response = self.elb_client.describe_load_balancers( + LoadBalancerNames=[self.lb_name] + ) + list( + response["LoadBalancerDescriptions"][0]["Instances"] + ).should.have.length_of(3) + + +@mock_autoscaling +@mock_ec2 +@mock_elb +class TestAutoScalingTerminateInstances(TestCase): + def setUp(self) -> None: + self.mocked_networking = setup_networking() + self.as_client = boto3.client("autoscaling", region_name="us-east-1") + self.elb_client = boto3.client("elb", region_name="us-east-1") + self.lb_name = str(uuid4()) + self.lc_name = str(uuid4()) + self.asg_name = str(uuid4()) + + self.as_client.create_launch_configuration( + LaunchConfigurationName=self.lc_name, + ImageId=EXAMPLE_AMI_ID, + InstanceType="t2.medium", + ) + self.as_client.create_auto_scaling_group( + AutoScalingGroupName=self.asg_name, + LaunchConfigurationName=self.lc_name, + MinSize=0, + DesiredCapacity=1, + MaxSize=2, + VPCZoneIdentifier=self.mocked_networking["subnet1"], + NewInstancesProtectedFromScaleIn=False, + ) + self.elb_client.create_load_balancer( + LoadBalancerName=self.lb_name, + Listeners=[ + {"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080} + ], + AvailabilityZones=["us-east-1a", "us-east-1b"], + ) + self.as_client.attach_load_balancers( + AutoScalingGroupName=self.asg_name, LoadBalancerNames=[self.lb_name] + ) + + def test_terminate_instance_in_auto_scaling_group_decrement(self): + + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + original_instance_id = next( + instance["InstanceId"] + for instance in response["AutoScalingGroups"][0]["Instances"] + ) + self.as_client.terminate_instance_in_auto_scaling_group( + InstanceId=original_instance_id, ShouldDecrementDesiredCapacity=True + ) + + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + response["AutoScalingGroups"][0]["Instances"].should.equal([]) + response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(0) + + response = self.elb_client.describe_load_balancers( + LoadBalancerNames=[self.lb_name] + ) + list( + response["LoadBalancerDescriptions"][0]["Instances"] + ).should.have.length_of(0) + + def test_terminate_instance_in_auto_scaling_group_no_decrement(self): + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + original_instance_id = next( + instance["InstanceId"] + for instance in response["AutoScalingGroups"][0]["Instances"] + ) + self.as_client.terminate_instance_in_auto_scaling_group( + InstanceId=original_instance_id, ShouldDecrementDesiredCapacity=False + ) + + response = self.as_client.describe_auto_scaling_groups( + AutoScalingGroupNames=[self.asg_name] + ) + replaced_instance_id = next( + instance["InstanceId"] + for instance in response["AutoScalingGroups"][0]["Instances"] + ) + replaced_instance_id.should_not.equal(original_instance_id) + response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(1) + + response = self.elb_client.describe_load_balancers( + LoadBalancerNames=[self.lb_name] + ) + list( + response["LoadBalancerDescriptions"][0]["Instances"] + ).should.have.length_of(1) + original_instance_id.shouldnt.be.within( + [ + x["InstanceId"] + for x in response["LoadBalancerDescriptions"][0]["Instances"] + ] + ) diff --git a/tests/test_autoscaling/test_elbv2.py b/tests/test_autoscaling/test_elbv2.py index c5e2da16e..5edbcd7e3 100644 --- a/tests/test_autoscaling/test_elbv2.py +++ b/tests/test_autoscaling/test_elbv2.py @@ -1,139 +1,110 @@ import boto3 - import sure # noqa # pylint: disable=unused-import +import unittest from moto import mock_autoscaling, mock_elbv2 - -from .utils import setup_networking from tests import EXAMPLE_AMI_ID +from uuid import uuid4 +from .utils import setup_networking @mock_elbv2 @mock_autoscaling -def test_attach_detach_target_groups(): - mocked_networking = setup_networking() - INSTANCE_COUNT = 2 - client = boto3.client("autoscaling", region_name="us-east-1") - elbv2_client = boto3.client("elbv2", region_name="us-east-1") +class TestAutoscalignELBv2(unittest.TestCase): + def setUp(self) -> None: + self.mocked_networking = setup_networking() + self.instance_count = 2 + self.as_client = boto3.client("autoscaling", region_name="us-east-1") + self.elbv2_client = boto3.client("elbv2", region_name="us-east-1") - response = elbv2_client.create_target_group( - Name="a-target", - Protocol="HTTP", - Port=8080, - VpcId=mocked_networking["vpc"], - HealthCheckProtocol="HTTP", - HealthCheckPort="8080", - HealthCheckPath="/", - HealthCheckIntervalSeconds=5, - HealthCheckTimeoutSeconds=5, - HealthyThresholdCount=5, - UnhealthyThresholdCount=2, - Matcher={"HttpCode": "200"}, - ) - target_group_arn = response["TargetGroups"][0]["TargetGroupArn"] + self.target_name = str(uuid4())[0:6] + self.lc_name = str(uuid4())[0:6] + self.asg_name = str(uuid4())[0:6] + response = self.elbv2_client.create_target_group( + Name=self.target_name, + Protocol="HTTP", + Port=8080, + VpcId=self.mocked_networking["vpc"], + HealthCheckProtocol="HTTP", + HealthCheckPort="8080", + HealthCheckPath="/", + HealthCheckIntervalSeconds=5, + HealthCheckTimeoutSeconds=5, + HealthyThresholdCount=5, + UnhealthyThresholdCount=2, + Matcher={"HttpCode": "200"}, + ) + self.target_group_arn = response["TargetGroups"][0]["TargetGroupArn"] - client.create_launch_configuration( - LaunchConfigurationName="test_launch_configuration", - ImageId=EXAMPLE_AMI_ID, - InstanceType="t2.medium", - ) + self.as_client.create_launch_configuration( + LaunchConfigurationName=self.lc_name, + ImageId=EXAMPLE_AMI_ID, + InstanceType="t2.medium", + ) + self.as_client.create_auto_scaling_group( + AutoScalingGroupName=self.asg_name, + LaunchConfigurationName=self.lc_name, + MinSize=0, + MaxSize=self.instance_count, + DesiredCapacity=self.instance_count, + TargetGroupARNs=[self.target_group_arn], + VPCZoneIdentifier=self.mocked_networking["subnet1"], + ) - # create asg, attach to target group on create - client.create_auto_scaling_group( - AutoScalingGroupName="test_asg", - LaunchConfigurationName="test_launch_configuration", - MinSize=0, - MaxSize=INSTANCE_COUNT, - DesiredCapacity=INSTANCE_COUNT, - TargetGroupARNs=[target_group_arn], - VPCZoneIdentifier=mocked_networking["subnet1"], - ) - # create asg without attaching to target group - client.create_auto_scaling_group( - AutoScalingGroupName="test_asg2", - LaunchConfigurationName="test_launch_configuration", - MinSize=0, - MaxSize=INSTANCE_COUNT, - DesiredCapacity=INSTANCE_COUNT, - VPCZoneIdentifier=mocked_networking["subnet2"], - ) + response = self.as_client.describe_load_balancer_target_groups( + AutoScalingGroupName=self.asg_name + ) + list(response["LoadBalancerTargetGroups"]).should.have.length_of(1) - response = client.describe_load_balancer_target_groups( - AutoScalingGroupName="test_asg" - ) - list(response["LoadBalancerTargetGroups"]).should.have.length_of(1) + response = self.elbv2_client.describe_target_health( + TargetGroupArn=self.target_group_arn + ) + list(response["TargetHealthDescriptions"]).should.have.length_of( + self.instance_count + ) - response = elbv2_client.describe_target_health(TargetGroupArn=target_group_arn) - list(response["TargetHealthDescriptions"]).should.have.length_of(INSTANCE_COUNT) + def test_attach_detach_target_groups(self): + # create asg without attaching to target group + asg_name2 = str(uuid4()) + self.as_client.create_auto_scaling_group( + AutoScalingGroupName=asg_name2, + LaunchConfigurationName=self.lc_name, + MinSize=0, + MaxSize=self.instance_count, + DesiredCapacity=self.instance_count, + VPCZoneIdentifier=self.mocked_networking["subnet2"], + ) - client.attach_load_balancer_target_groups( - AutoScalingGroupName="test_asg2", TargetGroupARNs=[target_group_arn] - ) + self.as_client.attach_load_balancer_target_groups( + AutoScalingGroupName=asg_name2, TargetGroupARNs=[self.target_group_arn] + ) - response = elbv2_client.describe_target_health(TargetGroupArn=target_group_arn) - list(response["TargetHealthDescriptions"]).should.have.length_of(INSTANCE_COUNT * 2) + response = self.elbv2_client.describe_target_health( + TargetGroupArn=self.target_group_arn + ) + list(response["TargetHealthDescriptions"]).should.have.length_of( + self.instance_count * 2 + ) - response = client.detach_load_balancer_target_groups( - AutoScalingGroupName="test_asg2", TargetGroupARNs=[target_group_arn] - ) - response = elbv2_client.describe_target_health(TargetGroupArn=target_group_arn) - list(response["TargetHealthDescriptions"]).should.have.length_of(INSTANCE_COUNT) + response = self.as_client.detach_load_balancer_target_groups( + AutoScalingGroupName=asg_name2, TargetGroupARNs=[self.target_group_arn] + ) + response = self.elbv2_client.describe_target_health( + TargetGroupArn=self.target_group_arn + ) + list(response["TargetHealthDescriptions"]).should.have.length_of( + self.instance_count + ) + def test_detach_all_target_groups(self): + response = self.as_client.detach_load_balancer_target_groups( + AutoScalingGroupName=self.asg_name, TargetGroupARNs=[self.target_group_arn] + ) -@mock_elbv2 -@mock_autoscaling -def test_detach_all_target_groups(): - mocked_networking = setup_networking() - INSTANCE_COUNT = 2 - client = boto3.client("autoscaling", region_name="us-east-1") - elbv2_client = boto3.client("elbv2", region_name="us-east-1") - - response = elbv2_client.create_target_group( - Name="a-target", - Protocol="HTTP", - Port=8080, - VpcId=mocked_networking["vpc"], - HealthCheckProtocol="HTTP", - HealthCheckPort="8080", - HealthCheckPath="/", - HealthCheckIntervalSeconds=5, - HealthCheckTimeoutSeconds=5, - HealthyThresholdCount=5, - UnhealthyThresholdCount=2, - Matcher={"HttpCode": "200"}, - ) - target_group_arn = response["TargetGroups"][0]["TargetGroupArn"] - - client.create_launch_configuration( - LaunchConfigurationName="test_launch_configuration", - ImageId=EXAMPLE_AMI_ID, - InstanceType="t2.medium", - ) - - client.create_auto_scaling_group( - AutoScalingGroupName="test_asg", - LaunchConfigurationName="test_launch_configuration", - MinSize=0, - MaxSize=INSTANCE_COUNT, - DesiredCapacity=INSTANCE_COUNT, - TargetGroupARNs=[target_group_arn], - VPCZoneIdentifier=mocked_networking["subnet1"], - ) - - response = client.describe_load_balancer_target_groups( - AutoScalingGroupName="test_asg" - ) - list(response["LoadBalancerTargetGroups"]).should.have.length_of(1) - - response = elbv2_client.describe_target_health(TargetGroupArn=target_group_arn) - list(response["TargetHealthDescriptions"]).should.have.length_of(INSTANCE_COUNT) - - response = client.detach_load_balancer_target_groups( - AutoScalingGroupName="test_asg", TargetGroupARNs=[target_group_arn] - ) - - response = elbv2_client.describe_target_health(TargetGroupArn=target_group_arn) - list(response["TargetHealthDescriptions"]).should.have.length_of(0) - response = client.describe_load_balancer_target_groups( - AutoScalingGroupName="test_asg" - ) - list(response["LoadBalancerTargetGroups"]).should.have.length_of(0) + response = self.elbv2_client.describe_target_health( + TargetGroupArn=self.target_group_arn + ) + list(response["TargetHealthDescriptions"]).should.have.length_of(0) + response = self.as_client.describe_load_balancer_target_groups( + AutoScalingGroupName=self.asg_name + ) + list(response["LoadBalancerTargetGroups"]).should.have.length_of(0)