Add instance protection support in autoscaling, with tests

tests/test_autoscaling/test_autoscaling.py

@@ -710,6 +710,7 @@ def test_create_autoscaling_group_boto3():
              'PropagateAtLaunch': False
              }],
         VPCZoneIdentifier=mocked_networking['subnet1'],
+        NewInstancesProtectedFromScaleIn=False,
     )
     response['ResponseMetadata']['HTTPStatusCode'].should.equal(200)
 
@@ -728,13 +729,48 @@ def test_describe_autoscaling_groups_boto3():
         MaxSize=20,
         DesiredCapacity=5,
         VPCZoneIdentifier=mocked_networking['subnet1'],
+        NewInstancesProtectedFromScaleIn=True,
     )
+
     response = client.describe_auto_scaling_groups(
         AutoScalingGroupNames=["test_asg"]
     )
     response['ResponseMetadata']['HTTPStatusCode'].should.equal(200)
-    response['AutoScalingGroups'][0][
-        'AutoScalingGroupName'].should.equal('test_asg')
+    group = response['AutoScalingGroups'][0]
+    group['AutoScalingGroupName'].should.equal('test_asg')
+    group['NewInstancesProtectedFromScaleIn'].should.equal(True)
+    group['Instances'][0]['ProtectedFromScaleIn'].should.equal(True)
+
+
+@mock_autoscaling
+def test_describe_autoscaling_instances_boto3():
+    mocked_networking = setup_networking()
+    client = boto3.client('autoscaling', region_name='us-east-1')
+    _ = client.create_launch_configuration(
+        LaunchConfigurationName='test_launch_configuration'
+    )
+    _ = client.create_auto_scaling_group(
+        AutoScalingGroupName='test_asg',
+        LaunchConfigurationName='test_launch_configuration',
+        MinSize=0,
+        MaxSize=20,
+        DesiredCapacity=5,
+        VPCZoneIdentifier=mocked_networking['subnet1'],
+        NewInstancesProtectedFromScaleIn=True,
+    )
+
+    response = client.describe_auto_scaling_groups(
+        AutoScalingGroupNames=["test_asg"]
+    )
+    instance_ids = [
+        instance['InstanceId']
+        for instance in response['AutoScalingGroups'][0]['Instances']
+    ]
+
+    response = client.describe_auto_scaling_instances(InstanceIds=instance_ids)
+    for instance in response['AutoScalingInstances']:
+        instance['AutoScalingGroupName'].should.equal('test_asg')
+        instance['ProtectedFromScaleIn'].should.equal(True)
 
 
 @mock_autoscaling
@@ -751,17 +787,21 @@ def test_update_autoscaling_group_boto3():
         MaxSize=20,
         DesiredCapacity=5,
         VPCZoneIdentifier=mocked_networking['subnet1'],
+        NewInstancesProtectedFromScaleIn=True,
     )
 
-    response = client.update_auto_scaling_group(
+    _ = client.update_auto_scaling_group(
         AutoScalingGroupName='test_asg',
         MinSize=1,
+        NewInstancesProtectedFromScaleIn=False,
     )
 
     response = client.describe_auto_scaling_groups(
         AutoScalingGroupNames=["test_asg"]
     )
-    response['AutoScalingGroups'][0]['MinSize'].should.equal(1)
+    group = response['AutoScalingGroups'][0]
+    group['MinSize'].should.equal(1)
+    group['NewInstancesProtectedFromScaleIn'].should.equal(False)
 
 
 @mock_autoscaling
@@ -992,9 +1032,7 @@ def test_attach_one_instance():
             'PropagateAtLaunch': True
         }],
         VPCZoneIdentifier=mocked_networking['subnet1'],
-    )
-    response = client.describe_auto_scaling_groups(
-        AutoScalingGroupNames=['test_asg']
+        NewInstancesProtectedFromScaleIn=True,
     )
 
     ec2 = boto3.resource('ec2', 'us-east-1')
@@ -1009,7 +1047,11 @@ def test_attach_one_instance():
     response = client.describe_auto_scaling_groups(
         AutoScalingGroupNames=['test_asg']
     )
-    response['AutoScalingGroups'][0]['Instances'].should.have.length_of(3)
+    instances = response['AutoScalingGroups'][0]['Instances']
+    instances.should.have.length_of(3)
+    for instance in instances:
+        instance['ProtectedFromScaleIn'].should.equal(True)
+
 
 @mock_autoscaling
 @mock_ec2
@@ -1100,3 +1142,111 @@ def test_suspend_processes():
             launch_suspended = True
 
     assert launch_suspended is True
+
+@mock_autoscaling
+def test_set_instance_protection():
+    mocked_networking = setup_networking()
+    client = boto3.client('autoscaling', region_name='us-east-1')
+    _ = client.create_launch_configuration(
+        LaunchConfigurationName='test_launch_configuration'
+    )
+    _ = client.create_auto_scaling_group(
+        AutoScalingGroupName='test_asg',
+        LaunchConfigurationName='test_launch_configuration',
+        MinSize=0,
+        MaxSize=20,
+        DesiredCapacity=5,
+        VPCZoneIdentifier=mocked_networking['subnet1'],
+        NewInstancesProtectedFromScaleIn=False,
+    )
+
+    response = client.describe_auto_scaling_groups(AutoScalingGroupNames=['test_asg'])
+    instance_ids = [
+        instance['InstanceId']
+        for instance in response['AutoScalingGroups'][0]['Instances']
+    ]
+    protected = instance_ids[:3]
+
+    _ = client.set_instance_protection(
+        AutoScalingGroupName='test_asg',
+        InstanceIds=protected,
+        ProtectedFromScaleIn=True,
+    )
+
+    response = client.describe_auto_scaling_groups(AutoScalingGroupNames=['test_asg'])
+    for instance in response['AutoScalingGroups'][0]['Instances']:
+        instance['ProtectedFromScaleIn'].should.equal(
+            instance['InstanceId'] in protected
+        )
+
+
+@mock_autoscaling
+def test_set_desired_capacity_up_boto3():
+    mocked_networking = setup_networking()
+    client = boto3.client('autoscaling', region_name='us-east-1')
+    _ = client.create_launch_configuration(
+        LaunchConfigurationName='test_launch_configuration'
+    )
+    _ = client.create_auto_scaling_group(
+        AutoScalingGroupName='test_asg',
+        LaunchConfigurationName='test_launch_configuration',
+        MinSize=0,
+        MaxSize=20,
+        DesiredCapacity=5,
+        VPCZoneIdentifier=mocked_networking['subnet1'],
+        NewInstancesProtectedFromScaleIn=True,
+    )
+
+    _ = client.set_desired_capacity(
+        AutoScalingGroupName='test_asg',
+        DesiredCapacity=10,
+    )
+
+    response = client.describe_auto_scaling_groups(AutoScalingGroupNames=['test_asg'])
+    instances = response['AutoScalingGroups'][0]['Instances']
+    instances.should.have.length_of(10)
+    for instance in instances:
+        instance['ProtectedFromScaleIn'].should.equal(True)
+
+
+@mock_autoscaling
+def test_set_desired_capacity_down_boto3():
+    mocked_networking = setup_networking()
+    client = boto3.client('autoscaling', region_name='us-east-1')
+    _ = client.create_launch_configuration(
+        LaunchConfigurationName='test_launch_configuration'
+    )
+    _ = client.create_auto_scaling_group(
+        AutoScalingGroupName='test_asg',
+        LaunchConfigurationName='test_launch_configuration',
+        MinSize=0,
+        MaxSize=20,
+        DesiredCapacity=5,
+        VPCZoneIdentifier=mocked_networking['subnet1'],
+        NewInstancesProtectedFromScaleIn=True,
+    )
+
+    response = client.describe_auto_scaling_groups(AutoScalingGroupNames=['test_asg'])
+    instance_ids = [
+        instance['InstanceId']
+        for instance in response['AutoScalingGroups'][0]['Instances']
+    ]
+    unprotected, protected = instance_ids[:2], instance_ids[2:]
+
+    _ = client.set_instance_protection(
+        AutoScalingGroupName='test_asg',
+        InstanceIds=unprotected,
+        ProtectedFromScaleIn=False,
+    )
+
+    _ = client.set_desired_capacity(
+        AutoScalingGroupName='test_asg',
+        DesiredCapacity=1,
+    )
+
+    response = client.describe_auto_scaling_groups(AutoScalingGroupNames=['test_asg'])
+    group = response['AutoScalingGroups'][0]
+    group['DesiredCapacity'].should.equal(1)
+    instance_ids = {instance['InstanceId'] for instance in group['Instances']}
+    set(protected).should.equal(instance_ids)
+    set(unprotected).should_not.be.within(instance_ids)  # only unprotected killed