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

Techdebt: Replace sure with regular asserts in Autoscaling (#6383)

Browse Source
This commit is contained in:
Bert Blommers 2023-06-09 11:16:59 +00:00 committed by GitHub
parent 3697019f01
commit 0b2e86ce68
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 679 additions and 809 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

321
tests/test_autoscaling/test_autoscaling.py
View File

@ -1,5 +1,4 @@
import boto3 import boto3
import sure # noqa # pylint: disable=unused-import
import pytest import pytest
from botocore.exceptions import ClientError from botocore.exceptions import ClientError
@ -42,8 +41,8 @@ def test_propogate_tags():
instances = ec2.describe_instances() instances = ec2.describe_instances()
tags = instances["Reservations"][0]["Instances"][0]["Tags"] tags = instances["Reservations"][0]["Instances"][0]["Tags"]
tags.should.contain({"Value": "TestTagValue1", "Key": "TestTagKey1"}) assert {"Value": "TestTagValue1", "Key": "TestTagKey1"} in tags
tags.should.contain({"Value": "TestGroup1", "Key": "aws:autoscaling:groupName"}) assert {"Value": "TestGroup1", "Key": "aws:autoscaling:groupName"} in tags
@mock_autoscaling @mock_autoscaling
@ -65,20 +64,14 @@ def test_create_autoscaling_group_from_instance():
VPCZoneIdentifier=mocked_instance_with_networking["subnet1"], VPCZoneIdentifier=mocked_instance_with_networking["subnet1"],
NewInstancesProtectedFromScaleIn=False, NewInstancesProtectedFromScaleIn=False,
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
describe_launch_configurations_response = client.describe_launch_configurations() describe_launch_configurations_response = client.describe_launch_configurations()
describe_launch_configurations_response[ assert len(describe_launch_configurations_response["LaunchConfigurations"]) == 1
"LaunchConfigurations" config = describe_launch_configurations_response["LaunchConfigurations"][0]
].should.have.length_of(1) assert config["LaunchConfigurationName"] == "test_asg"
launch_configuration_from_instance = describe_launch_configurations_response[ assert config["ImageId"] == image_id
"LaunchConfigurations" assert config["InstanceType"] == instance_type
][0]
launch_configuration_from_instance["LaunchConfigurationName"].should.equal(
"test_asg"
)
launch_configuration_from_instance["ImageId"].should.equal(image_id)
launch_configuration_from_instance["InstanceType"].should.equal(instance_type)
@mock_autoscaling @mock_autoscaling
@ -109,7 +102,7 @@ def test_create_autoscaling_group_from_instance_with_security_groups():
NewInstancesProtectedFromScaleIn=False, NewInstancesProtectedFromScaleIn=False,
) )
# Just verifying this works - used to throw an error when supplying a instance that belonged to an SG # Just verifying this works - used to throw an error when supplying a instance that belonged to an SG
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
@mock_autoscaling @mock_autoscaling
@ -128,11 +121,10 @@ def test_create_autoscaling_group_from_invalid_instance_id():
VPCZoneIdentifier=mocked_networking["subnet1"], VPCZoneIdentifier=mocked_networking["subnet1"],
NewInstancesProtectedFromScaleIn=False, NewInstancesProtectedFromScaleIn=False,
) )
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) err = ex.value.response["Error"]
ex.value.response["Error"]["Code"].should.equal("ValidationError") assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
ex.value.response["Error"]["Message"].should.equal( assert err["Code"] == "ValidationError"
f"Instance [{invalid_instance_id}] is invalid." assert err["Message"] == f"Instance [{invalid_instance_id}] is invalid."
)
@mock_autoscaling @mock_autoscaling
@ -158,7 +150,7 @@ def test_create_autoscaling_group_from_template():
VPCZoneIdentifier=mocked_networking["subnet1"], VPCZoneIdentifier=mocked_networking["subnet1"],
NewInstancesProtectedFromScaleIn=False, NewInstancesProtectedFromScaleIn=False,
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
@mock_ec2 @mock_ec2
@ -186,11 +178,9 @@ def test_create_auto_scaling_from_template_version__latest():
response = asg_client.describe_auto_scaling_groups(AutoScalingGroupNames=["name"])[ response = asg_client.describe_auto_scaling_groups(AutoScalingGroupNames=["name"])[
"AutoScalingGroups" "AutoScalingGroups"
][0] ][0]
response.should.have.key("LaunchTemplate") assert "LaunchTemplate" in response
response["LaunchTemplate"].should.have.key("LaunchTemplateName").equals( assert response["LaunchTemplate"]["LaunchTemplateName"] == launch_template_name
launch_template_name assert response["LaunchTemplate"]["Version"] == "$Latest"
)
response["LaunchTemplate"].should.have.key("Version").equals("$Latest")
@mock_ec2 @mock_ec2
@ -223,11 +213,9 @@ def test_create_auto_scaling_from_template_version__default():
response = asg_client.describe_auto_scaling_groups(AutoScalingGroupNames=["name"])[ response = asg_client.describe_auto_scaling_groups(AutoScalingGroupNames=["name"])[
"AutoScalingGroups" "AutoScalingGroups"
][0] ][0]
response.should.have.key("LaunchTemplate") assert "LaunchTemplate" in response
response["LaunchTemplate"].should.have.key("LaunchTemplateName").equals( assert response["LaunchTemplate"]["LaunchTemplateName"] == launch_template_name
launch_template_name assert response["LaunchTemplate"]["Version"] == "$Default"
)
response["LaunchTemplate"].should.have.key("Version").equals("$Default")
@mock_ec2 @mock_ec2
@ -252,9 +240,9 @@ def test_create_auto_scaling_from_template_version__no_version():
response = asg_client.describe_auto_scaling_groups(AutoScalingGroupNames=["name"])[ response = asg_client.describe_auto_scaling_groups(AutoScalingGroupNames=["name"])[
"AutoScalingGroups" "AutoScalingGroups"
][0] ][0]
response.should.have.key("LaunchTemplate") assert "LaunchTemplate" in response
# We never specified the version - and AWS will not return anything if we don't # We never specified the version - and AWS will not return anything if we don't
response["LaunchTemplate"].shouldnt.have.key("Version") assert "Version" not in response["LaunchTemplate"]
@mock_autoscaling @mock_autoscaling
@ -279,10 +267,12 @@ def test_create_autoscaling_group_no_template_ref():
VPCZoneIdentifier=mocked_networking["subnet1"], VPCZoneIdentifier=mocked_networking["subnet1"],
NewInstancesProtectedFromScaleIn=False, NewInstancesProtectedFromScaleIn=False,
) )
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) err = ex.value.response["Error"]
ex.value.response["Error"]["Code"].should.equal("ValidationError") assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
ex.value.response["Error"]["Message"].should.equal( assert err["Code"] == "ValidationError"
"Valid requests must contain either launchTemplateId or LaunchTemplateName" assert (
err["Message"]
== "Valid requests must contain either launchTemplateId or LaunchTemplateName"
) )
@ -312,10 +302,12 @@ def test_create_autoscaling_group_multiple_template_ref():
VPCZoneIdentifier=mocked_networking["subnet1"], VPCZoneIdentifier=mocked_networking["subnet1"],
NewInstancesProtectedFromScaleIn=False, NewInstancesProtectedFromScaleIn=False,
) )
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) err = ex.value.response["Error"]
ex.value.response["Error"]["Code"].should.equal("ValidationError") assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
ex.value.response["Error"]["Message"].should.equal( assert err["Code"] == "ValidationError"
"Valid requests must contain either launchTemplateId or LaunchTemplateName" assert (
err["Message"]
== "Valid requests must contain either launchTemplateId or LaunchTemplateName"
) )
@ -332,11 +324,12 @@ def test_create_autoscaling_group_no_launch_configuration():
VPCZoneIdentifier=mocked_networking["subnet1"], VPCZoneIdentifier=mocked_networking["subnet1"],
NewInstancesProtectedFromScaleIn=False, NewInstancesProtectedFromScaleIn=False,
) )
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) err = ex.value.response["Error"]
ex.value.response["Error"]["Code"].should.equal("ValidationError") assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
ex.value.response["Error"]["Message"].should.equal( assert err["Code"] == "ValidationError"
"Valid requests must contain either LaunchTemplate, LaunchConfigurationName, " assert (
"InstanceId or MixedInstancesPolicy parameter." err["Message"]
== "Valid requests must contain either LaunchTemplate, LaunchConfigurationName, InstanceId or MixedInstancesPolicy parameter."
) )
@ -371,11 +364,12 @@ def test_create_autoscaling_group_multiple_launch_configurations():
VPCZoneIdentifier=mocked_networking["subnet1"], VPCZoneIdentifier=mocked_networking["subnet1"],
NewInstancesProtectedFromScaleIn=False, NewInstancesProtectedFromScaleIn=False,
) )
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) err = ex.value.response["Error"]
ex.value.response["Error"]["Code"].should.equal("ValidationError") assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
ex.value.response["Error"]["Message"].should.equal( assert err["Code"] == "ValidationError"
"Valid requests must contain either LaunchTemplate, LaunchConfigurationName, " assert (
"InstanceId or MixedInstancesPolicy parameter." err["Message"]
== "Valid requests must contain either LaunchTemplate, LaunchConfigurationName, InstanceId or MixedInstancesPolicy parameter."
) )
@ -405,20 +399,20 @@ def test_describe_autoscaling_groups_launch_template():
} }
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
group = response["AutoScalingGroups"][0] group = response["AutoScalingGroups"][0]
group["AutoScalingGroupName"].should.equal("test_asg") assert group["AutoScalingGroupName"] == "test_asg"
group["LaunchTemplate"].should.equal(expected_launch_template) assert group["LaunchTemplate"] == expected_launch_template
group.should_not.have.key("LaunchConfigurationName") assert "LaunchConfigurationName" not in group
group["AvailabilityZones"].should.equal(["us-east-1a"]) assert group["AvailabilityZones"] == ["us-east-1a"]
group["VPCZoneIdentifier"].should.equal(mocked_networking["subnet1"]) assert group["VPCZoneIdentifier"] == mocked_networking["subnet1"]
group["NewInstancesProtectedFromScaleIn"].should.equal(True) assert group["NewInstancesProtectedFromScaleIn"] is True
for instance in group["Instances"]: for instance in group["Instances"]:
instance["LaunchTemplate"].should.equal(expected_launch_template) assert instance["LaunchTemplate"] == expected_launch_template
instance.should_not.have.key("LaunchConfigurationName") assert "LaunchConfigurationName" not in instance
instance["AvailabilityZone"].should.equal("us-east-1a") assert instance["AvailabilityZone"] == "us-east-1a"
instance["ProtectedFromScaleIn"].should.equal(True) assert instance["ProtectedFromScaleIn"] is True
instance["InstanceType"].should.equal("t2.micro") assert instance["InstanceType"] == "t2.micro"
@mock_autoscaling @mock_autoscaling
@ -441,14 +435,14 @@ def test_describe_autoscaling_instances_launch_config():
) )
response = client.describe_auto_scaling_instances() response = client.describe_auto_scaling_instances()
len(response["AutoScalingInstances"]).should.equal(5) assert len(response["AutoScalingInstances"]) == 5
for instance in response["AutoScalingInstances"]: for instance in response["AutoScalingInstances"]:
instance["LaunchConfigurationName"].should.equal("test_launch_configuration") assert instance["LaunchConfigurationName"] == "test_launch_configuration"
instance.should_not.have.key("LaunchTemplate") assert "LaunchTemplate" not in instance
instance["AutoScalingGroupName"].should.equal("test_asg") assert instance["AutoScalingGroupName"] == "test_asg"
instance["AvailabilityZone"].should.equal("us-east-1a") assert instance["AvailabilityZone"] == "us-east-1a"
instance["ProtectedFromScaleIn"].should.equal(True) assert instance["ProtectedFromScaleIn"] is True
instance["InstanceType"].should.equal("t2.micro") assert instance["InstanceType"] == "t2.micro"
@mock_autoscaling @mock_autoscaling
@ -477,14 +471,14 @@ def test_describe_autoscaling_instances_launch_template():
} }
response = client.describe_auto_scaling_instances() response = client.describe_auto_scaling_instances()
len(response["AutoScalingInstances"]).should.equal(5) assert len(response["AutoScalingInstances"]) == 5
for instance in response["AutoScalingInstances"]: for instance in response["AutoScalingInstances"]:
instance["LaunchTemplate"].should.equal(expected_launch_template) assert instance["LaunchTemplate"] == expected_launch_template
instance.should_not.have.key("LaunchConfigurationName") assert "LaunchConfigurationName" not in instance
instance["AutoScalingGroupName"].should.equal("test_asg") assert instance["AutoScalingGroupName"] == "test_asg"
instance["AvailabilityZone"].should.equal("us-east-1a") assert instance["AvailabilityZone"] == "us-east-1a"
instance["ProtectedFromScaleIn"].should.equal(True) assert instance["ProtectedFromScaleIn"] is True
instance["InstanceType"].should.equal("t2.micro") assert instance["InstanceType"] == "t2.micro"
@mock_autoscaling @mock_autoscaling
@ -515,11 +509,11 @@ def test_describe_autoscaling_instances_instanceid_filter():
response = client.describe_auto_scaling_instances( response = client.describe_auto_scaling_instances(
InstanceIds=instance_ids[0:2] InstanceIds=instance_ids[0:2]
) # Filter by first 2 of 5 ) # Filter by first 2 of 5
len(response["AutoScalingInstances"]).should.equal(2) assert len(response["AutoScalingInstances"]) == 2
for instance in response["AutoScalingInstances"]: for instance in response["AutoScalingInstances"]:
instance["AutoScalingGroupName"].should.equal("test_asg") assert instance["AutoScalingGroupName"] == "test_asg"
instance["AvailabilityZone"].should.equal("us-east-1a") assert instance["AvailabilityZone"] == "us-east-1a"
instance["ProtectedFromScaleIn"].should.equal(True) assert instance["ProtectedFromScaleIn"] is True
@mock_autoscaling @mock_autoscaling
@ -556,10 +550,10 @@ def test_update_autoscaling_group_launch_config():
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
group = response["AutoScalingGroups"][0] group = response["AutoScalingGroups"][0]
group["LaunchConfigurationName"].should.equal("test_launch_configuration_new") assert group["LaunchConfigurationName"] == "test_launch_configuration_new"
group["MinSize"].should.equal(1) assert group["MinSize"] == 1
set(group["AvailabilityZones"]).should.equal({"us-east-1a", "us-east-1b"}) assert set(group["AvailabilityZones"]) == {"us-east-1a", "us-east-1b"}
group["NewInstancesProtectedFromScaleIn"].should.equal(False) assert group["NewInstancesProtectedFromScaleIn"] is False
@mock_autoscaling @mock_autoscaling
@ -608,10 +602,10 @@ def test_update_autoscaling_group_launch_template():
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
group = response["AutoScalingGroups"][0] group = response["AutoScalingGroups"][0]
group["LaunchTemplate"].should.equal(expected_launch_template) assert group["LaunchTemplate"] == expected_launch_template
group["MinSize"].should.equal(1) assert group["MinSize"] == 1
set(group["AvailabilityZones"]).should.equal({"us-east-1a", "us-east-1b"}) assert set(group["AvailabilityZones"]) == {"us-east-1a", "us-east-1b"}
group["NewInstancesProtectedFromScaleIn"].should.equal(False) assert group["NewInstancesProtectedFromScaleIn"] is False
@mock_autoscaling @mock_autoscaling
@ -635,9 +629,9 @@ def test_update_autoscaling_group_min_size_desired_capacity_change():
client.update_auto_scaling_group(AutoScalingGroupName="test_asg", MinSize=5) client.update_auto_scaling_group(AutoScalingGroupName="test_asg", MinSize=5)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
group = response["AutoScalingGroups"][0] group = response["AutoScalingGroups"][0]
group["DesiredCapacity"].should.equal(5) assert group["DesiredCapacity"] == 5
group["MinSize"].should.equal(5) assert group["MinSize"] == 5
group["Instances"].should.have.length_of(5) assert len(group["Instances"]) == 5
@mock_autoscaling @mock_autoscaling
@ -661,9 +655,9 @@ def test_update_autoscaling_group_max_size_desired_capacity_change():
client.update_auto_scaling_group(AutoScalingGroupName="test_asg", MaxSize=5) client.update_auto_scaling_group(AutoScalingGroupName="test_asg", MaxSize=5)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
group = response["AutoScalingGroups"][0] group = response["AutoScalingGroups"][0]
group["DesiredCapacity"].should.equal(5) assert group["DesiredCapacity"] == 5
group["MaxSize"].should.equal(5) assert group["MaxSize"] == 5
group["Instances"].should.have.length_of(5) assert len(group["Instances"]) == 5
@mock_autoscaling @mock_autoscaling
@ -713,31 +707,32 @@ def test_autoscaling_describe_policies():
) )
response = client.describe_policies() response = client.describe_policies()
response["ScalingPolicies"].should.have.length_of(2) assert len(response["ScalingPolicies"]) == 2
response = client.describe_policies(AutoScalingGroupName="test_asg") response = client.describe_policies(AutoScalingGroupName="test_asg")
response["ScalingPolicies"].should.have.length_of(2) assert len(response["ScalingPolicies"]) == 2
response = client.describe_policies(PolicyTypes=["StepScaling"]) response = client.describe_policies(PolicyTypes=["StepScaling"])
response["ScalingPolicies"].should.have.length_of(0) assert len(response["ScalingPolicies"]) == 0
response = client.describe_policies( response = client.describe_policies(
AutoScalingGroupName="test_asg", AutoScalingGroupName="test_asg",
PolicyNames=["test_policy_down"], PolicyNames=["test_policy_down"],
PolicyTypes=["SimpleScaling"], PolicyTypes=["SimpleScaling"],
) )
response["ScalingPolicies"].should.have.length_of(1) assert len(response["ScalingPolicies"]) == 1
policy = response["ScalingPolicies"][0] policy = response["ScalingPolicies"][0]
policy["PolicyType"].should.equal("SimpleScaling") assert policy["PolicyType"] == "SimpleScaling"
policy["MetricAggregationType"].should.equal("Minimum") assert policy["MetricAggregationType"] == "Minimum"
policy["AdjustmentType"].should.equal("PercentChangeInCapacity") assert policy["AdjustmentType"] == "PercentChangeInCapacity"
policy["ScalingAdjustment"].should.equal(-10) assert policy["ScalingAdjustment"] == -10
policy["Cooldown"].should.equal(60) assert policy["Cooldown"] == 60
policy["PolicyARN"].should.equal( assert (
f"arn:aws:autoscaling:us-east-1:{ACCOUNT_ID}:scalingPolicy:c322761b-3172-4d56-9a21-0ed9d6161d67:autoScalingGroupName/test_asg:policyName/test_policy_down" policy["PolicyARN"]
== f"arn:aws:autoscaling:us-east-1:{ACCOUNT_ID}:scalingPolicy:c322761b-3172-4d56-9a21-0ed9d6161d67:autoScalingGroupName/test_asg:policyName/test_policy_down"
) )
policy["PolicyName"].should.equal("test_policy_down") assert policy["PolicyName"] == "test_policy_down"
policy.shouldnt.have.key("TargetTrackingConfiguration") assert "TargetTrackingConfiguration" not in policy
@mock_autoscaling @mock_autoscaling
@ -800,13 +795,13 @@ def test_create_autoscaling_policy_with_policytype__targettrackingscaling():
resp = client.describe_policies(AutoScalingGroupName=asg_name) resp = client.describe_policies(AutoScalingGroupName=asg_name)
policy = resp["ScalingPolicies"][0] policy = resp["ScalingPolicies"][0]
policy.should.have.key("PolicyName").equals(configuration_name) assert policy["PolicyName"] == configuration_name
policy.should.have.key("PolicyARN").equals( assert (
f"arn:aws:autoscaling:us-west-1:{ACCOUNT_ID}:scalingPolicy:c322761b-3172-4d56-9a21-0ed9d6161d67:autoScalingGroupName/{asg_name}:policyName/{configuration_name}" policy["PolicyARN"]
== f"arn:aws:autoscaling:us-west-1:{ACCOUNT_ID}:scalingPolicy:c322761b-3172-4d56-9a21-0ed9d6161d67:autoScalingGroupName/{asg_name}:policyName/{configuration_name}"
) )
policy.should.have.key("PolicyType").equals("TargetTrackingScaling") assert policy["PolicyType"] == "TargetTrackingScaling"
policy.should.have.key("TargetTrackingConfiguration").should.equal( assert policy["TargetTrackingConfiguration"] == {
{
"PredefinedMetricSpecification": { "PredefinedMetricSpecification": {
"PredefinedMetricType": "ASGAverageNetworkIn", "PredefinedMetricType": "ASGAverageNetworkIn",
}, },
@ -842,9 +837,8 @@ def test_create_autoscaling_policy_with_policytype__targettrackingscaling():
}, },
"TargetValue": 1000000.0, "TargetValue": 1000000.0,
} }
) assert "ScalingAdjustment" not in policy
policy.shouldnt.have.key("ScalingAdjustment") assert "Cooldown" not in policy
policy.shouldnt.have.key("Cooldown")
@mock_autoscaling @mock_autoscaling
@ -883,23 +877,22 @@ def test_create_autoscaling_policy_with_policytype__stepscaling():
resp = client.describe_policies(AutoScalingGroupName=asg_name) resp = client.describe_policies(AutoScalingGroupName=asg_name)
policy = resp["ScalingPolicies"][0] policy = resp["ScalingPolicies"][0]
policy.should.have.key("PolicyName").equals(launch_config_name) assert policy["PolicyName"] == launch_config_name
policy.should.have.key("PolicyARN").equals( assert (
f"arn:aws:autoscaling:eu-west-1:{ACCOUNT_ID}:scalingPolicy:c322761b-3172-4d56-9a21-0ed9d6161d67:autoScalingGroupName/{asg_name}:policyName/{launch_config_name}" policy["PolicyARN"]
== f"arn:aws:autoscaling:eu-west-1:{ACCOUNT_ID}:scalingPolicy:c322761b-3172-4d56-9a21-0ed9d6161d67:autoScalingGroupName/{asg_name}:policyName/{launch_config_name}"
) )
policy.should.have.key("PolicyType").equals("StepScaling") assert policy["PolicyType"] == "StepScaling"
policy.should.have.key("StepAdjustments").equal( assert policy["StepAdjustments"] == [
[
{ {
"MetricIntervalLowerBound": 2, "MetricIntervalLowerBound": 2,
"MetricIntervalUpperBound": 8, "MetricIntervalUpperBound": 8,
"ScalingAdjustment": 1, "ScalingAdjustment": 1,
} }
] ]
) assert "TargetTrackingConfiguration" not in policy
policy.shouldnt.have.key("TargetTrackingConfiguration") assert "ScalingAdjustment" not in policy
policy.shouldnt.have.key("ScalingAdjustment") assert "Cooldown" not in policy
policy.shouldnt.have.key("Cooldown")
@mock_autoscaling @mock_autoscaling
@ -935,9 +928,10 @@ def test_create_autoscaling_policy_with_predictive_scaling_config():
resp = client.describe_policies(AutoScalingGroupName=asg_name) resp = client.describe_policies(AutoScalingGroupName=asg_name)
policy = resp["ScalingPolicies"][0] policy = resp["ScalingPolicies"][0]
policy.should.have.key("PredictiveScalingConfiguration").equals( assert policy["PredictiveScalingConfiguration"] == {
{"MetricSpecifications": [{"TargetValue": 5.0}], "SchedulingBufferTime": 7} "MetricSpecifications": [{"TargetValue": 5.0}],
) "SchedulingBufferTime": 7,
}
@mock_autoscaling @mock_autoscaling
@ -970,8 +964,7 @@ def test_create_auto_scaling_group_with_mixed_instances_policy():
# Assert we can describe MixedInstancesPolicy # Assert we can describe MixedInstancesPolicy
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=[asg_name]) response = client.describe_auto_scaling_groups(AutoScalingGroupNames=[asg_name])
group = response["AutoScalingGroups"][0] group = response["AutoScalingGroups"][0]
group.should.have.key("MixedInstancesPolicy").equals( assert group["MixedInstancesPolicy"] == {
{
"LaunchTemplate": { "LaunchTemplate": {
"LaunchTemplateSpecification": { "LaunchTemplateSpecification": {
"LaunchTemplateId": lt["LaunchTemplateId"], "LaunchTemplateId": lt["LaunchTemplateId"],
@ -980,19 +973,16 @@ def test_create_auto_scaling_group_with_mixed_instances_policy():
} }
} }
} }
)
# Assert the LaunchTemplate is known for the resulting instances # Assert the LaunchTemplate is known for the resulting instances
response = client.describe_auto_scaling_instances() response = client.describe_auto_scaling_instances()
len(response["AutoScalingInstances"]).should.equal(2) assert len(response["AutoScalingInstances"]) == 2
for instance in response["AutoScalingInstances"]: for instance in response["AutoScalingInstances"]:
instance["LaunchTemplate"].should.equal( assert instance["LaunchTemplate"] == {
{
"LaunchTemplateId": lt["LaunchTemplateId"], "LaunchTemplateId": lt["LaunchTemplateId"],
"LaunchTemplateName": "launchie", "LaunchTemplateName": "launchie",
"Version": "$DEFAULT", "Version": "$DEFAULT",
} }
)
@mock_autoscaling @mock_autoscaling
@ -1031,8 +1021,7 @@ def test_create_auto_scaling_group_with_mixed_instances_policy_overrides():
# Assert we can describe MixedInstancesPolicy # Assert we can describe MixedInstancesPolicy
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=[asg_name]) response = client.describe_auto_scaling_groups(AutoScalingGroupNames=[asg_name])
group = response["AutoScalingGroups"][0] group = response["AutoScalingGroups"][0]
group.should.have.key("MixedInstancesPolicy").equals( assert group["MixedInstancesPolicy"] == {
{
"LaunchTemplate": { "LaunchTemplate": {
"LaunchTemplateSpecification": { "LaunchTemplateSpecification": {
"LaunchTemplateId": lt["LaunchTemplateId"], "LaunchTemplateId": lt["LaunchTemplateId"],
@ -1047,7 +1036,6 @@ def test_create_auto_scaling_group_with_mixed_instances_policy_overrides():
], ],
} }
} }
)
@mock_autoscaling @mock_autoscaling
@ -1084,9 +1072,7 @@ def test_set_instance_protection():
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
for instance in response["AutoScalingGroups"][0]["Instances"]: for instance in response["AutoScalingGroups"][0]["Instances"]:
instance["ProtectedFromScaleIn"].should.equal( assert instance["ProtectedFromScaleIn"] is (instance["InstanceId"] in protected)
instance["InstanceId"] in protected
)
@mock_autoscaling @mock_autoscaling
@ -1112,9 +1098,9 @@ def test_set_desired_capacity_up():
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
instances = response["AutoScalingGroups"][0]["Instances"] instances = response["AutoScalingGroups"][0]["Instances"]
instances.should.have.length_of(10) assert len(instances) == 10
for instance in instances: for instance in instances:
instance["ProtectedFromScaleIn"].should.equal(True) assert instance["ProtectedFromScaleIn"] is True
@mock_autoscaling @mock_autoscaling
@ -1153,10 +1139,11 @@ def test_set_desired_capacity_down():
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
group = response["AutoScalingGroups"][0] group = response["AutoScalingGroups"][0]
group["DesiredCapacity"].should.equal(1) assert group["DesiredCapacity"] == 1
instance_ids = {instance["InstanceId"] for instance in group["Instances"]} instance_ids = {instance["InstanceId"] for instance in group["Instances"]}
set(protected).should.equal(instance_ids) assert set(protected) == instance_ids
set(unprotected).should_not.be.within(instance_ids) # only unprotected killed for x in unprotected:
assert x not in instance_ids # only unprotected killed
@mock_autoscaling @mock_autoscaling
@ -1191,7 +1178,7 @@ def test_terminate_instance_via_ec2_in_autoscaling_group():
instance["InstanceId"] instance["InstanceId"]
for instance in response["AutoScalingGroups"][0]["Instances"] for instance in response["AutoScalingGroups"][0]["Instances"]
) )
replaced_instance_id.should_not.equal(original_instance_id) assert replaced_instance_id != original_instance_id
@mock_ec2 @mock_ec2
@ -1228,11 +1215,11 @@ def test_attach_instances():
InstanceIds=[fake_instance["InstanceId"]], AutoScalingGroupName="test_asg" InstanceIds=[fake_instance["InstanceId"]], AutoScalingGroupName="test_asg"
) )
response = asg_client.describe_auto_scaling_instances() response = asg_client.describe_auto_scaling_instances()
len(response["AutoScalingInstances"]).should.equal(1) assert len(response["AutoScalingInstances"]) == 1
for instance in response["AutoScalingInstances"]: for instance in response["AutoScalingInstances"]:
instance["LaunchConfigurationName"].should.equal("test_launch_configuration") assert instance["LaunchConfigurationName"] == "test_launch_configuration"
instance["AutoScalingGroupName"].should.equal("test_asg") assert instance["AutoScalingGroupName"] == "test_asg"
instance["InstanceType"].should.equal("c4.2xlarge") assert instance["InstanceType"] == "c4.2xlarge"
@mock_autoscaling @mock_autoscaling
@ -1261,11 +1248,11 @@ def test_autoscaling_lifecyclehook():
response = client.describe_lifecycle_hooks( response = client.describe_lifecycle_hooks(
AutoScalingGroupName="test_asg", LifecycleHookNames=["test-lifecyclehook"] AutoScalingGroupName="test_asg", LifecycleHookNames=["test-lifecyclehook"]
) )
len(response["LifecycleHooks"]).should.equal(1) assert len(response["LifecycleHooks"]) == 1
for hook in response["LifecycleHooks"]: for hook in response["LifecycleHooks"]:
hook["LifecycleHookName"].should.equal("test-lifecyclehook") assert hook["LifecycleHookName"] == "test-lifecyclehook"
hook["AutoScalingGroupName"].should.equal("test_asg") assert hook["AutoScalingGroupName"] == "test_asg"
hook["LifecycleTransition"].should.equal("autoscaling:EC2_INSTANCE_TERMINATING") assert hook["LifecycleTransition"] == "autoscaling:EC2_INSTANCE_TERMINATING"
client.delete_lifecycle_hook( client.delete_lifecycle_hook(
LifecycleHookName="test-lifecyclehook", AutoScalingGroupName="test_asg" LifecycleHookName="test-lifecyclehook", AutoScalingGroupName="test_asg"
@ -1275,7 +1262,7 @@ def test_autoscaling_lifecyclehook():
AutoScalingGroupName="test_asg", LifecycleHookNames=["test-lifecyclehook"] AutoScalingGroupName="test_asg", LifecycleHookNames=["test-lifecyclehook"]
) )
len(response["LifecycleHooks"]).should.equal(0) assert len(response["LifecycleHooks"]) == 0
@pytest.mark.parametrize("original,new", [(2, 1), (2, 3), (1, 5), (1, 1)]) @pytest.mark.parametrize("original,new", [(2, 1), (2, 3), (1, 5), (1, 1)])
@ -1300,18 +1287,18 @@ def test_set_desired_capacity_without_protection(original, new):
) )
group = client.describe_auto_scaling_groups()["AutoScalingGroups"][0] group = client.describe_auto_scaling_groups()["AutoScalingGroups"][0]
group["DesiredCapacity"].should.equal(original) assert group["DesiredCapacity"] == original
instances = client.describe_auto_scaling_instances()["AutoScalingInstances"] instances = client.describe_auto_scaling_instances()["AutoScalingInstances"]
instances.should.have.length_of(original) assert len(instances) == original
client.update_auto_scaling_group( client.update_auto_scaling_group(
AutoScalingGroupName="tester_group", DesiredCapacity=new AutoScalingGroupName="tester_group", DesiredCapacity=new
) )
group = client.describe_auto_scaling_groups()["AutoScalingGroups"][0] group = client.describe_auto_scaling_groups()["AutoScalingGroups"][0]
group["DesiredCapacity"].should.equal(new) assert group["DesiredCapacity"] == new
instances = client.describe_auto_scaling_instances()["AutoScalingInstances"] instances = client.describe_auto_scaling_instances()["AutoScalingInstances"]
instances.should.have.length_of(new) assert len(instances) == new
@mock_autoscaling @mock_autoscaling
@ -1342,8 +1329,8 @@ def test_create_template_with_block_device():
ec2_client = boto3.client("ec2", region_name="ap-southeast-2") ec2_client = boto3.client("ec2", region_name="ap-southeast-2")
volumes = ec2_client.describe_volumes()["Volumes"] volumes = ec2_client.describe_volumes()["Volumes"]
# The standard root volume # The standard root volume
volumes[0]["VolumeType"].should.equal("gp2") assert volumes[0]["VolumeType"] == "gp2"
volumes[0]["Size"].should.equal(8) assert volumes[0]["Size"] == 8
# Our Ebs-volume # Our Ebs-volume
volumes[1]["VolumeType"].should.equal("gp3") assert volumes[1]["VolumeType"] == "gp3"
volumes[1]["Size"].should.equal(20) assert volumes[1]["Size"] == 20

114
tests/test_autoscaling/test_autoscaling_cloudformation.py
View File

@ -1,5 +1,4 @@
import boto3 import boto3
import sure # noqa # pylint: disable=unused-import
import json import json
from moto import mock_autoscaling, mock_cloudformation, mock_ec2, mock_elb from moto import mock_autoscaling, mock_cloudformation, mock_ec2, mock_elb
@ -33,12 +32,12 @@ Outputs:
cf_client.create_stack(StackName=stack_name, TemplateBody=cf_template) cf_client.create_stack(StackName=stack_name, TemplateBody=cf_template)
stack = cf_client.describe_stacks(StackName=stack_name)["Stacks"][0] stack = cf_client.describe_stacks(StackName=stack_name)["Stacks"][0]
stack["Outputs"][0]["OutputValue"].should.be.equal("test_launch_configuration") assert stack["Outputs"][0]["OutputValue"] == "test_launch_configuration"
lc = client.describe_launch_configurations()["LaunchConfigurations"][0] lc = client.describe_launch_configurations()["LaunchConfigurations"][0]
lc["LaunchConfigurationName"].should.be.equal("test_launch_configuration") assert lc["LaunchConfigurationName"] == "test_launch_configuration"
lc["ImageId"].should.be.equal(EXAMPLE_AMI_ID) assert lc["ImageId"] == EXAMPLE_AMI_ID
lc["InstanceType"].should.be.equal("t2.micro") assert lc["InstanceType"] == "t2.micro"
cf_template = """ cf_template = """
Resources: Resources:
@ -57,12 +56,12 @@ Outputs:
cf_client.update_stack(StackName=stack_name, TemplateBody=cf_template) cf_client.update_stack(StackName=stack_name, TemplateBody=cf_template)
stack = cf_client.describe_stacks(StackName=stack_name)["Stacks"][0] stack = cf_client.describe_stacks(StackName=stack_name)["Stacks"][0]
stack["Outputs"][0]["OutputValue"].should.be.equal("test_launch_configuration") assert stack["Outputs"][0]["OutputValue"] == "test_launch_configuration"
lc = client.describe_launch_configurations()["LaunchConfigurations"][0] lc = client.describe_launch_configurations()["LaunchConfigurations"][0]
lc["LaunchConfigurationName"].should.be.equal("test_launch_configuration") assert lc["LaunchConfigurationName"] == "test_launch_configuration"
lc["ImageId"].should.be.equal(EXAMPLE_AMI_ID) assert lc["ImageId"] == EXAMPLE_AMI_ID
lc["InstanceType"].should.be.equal("m5.large") assert lc["InstanceType"] == "m5.large"
@mock_autoscaling @mock_autoscaling
@ -107,13 +106,13 @@ Outputs:
Parameters=[{"ParameterKey": "SubnetId", "ParameterValue": subnet_id}], Parameters=[{"ParameterKey": "SubnetId", "ParameterValue": subnet_id}],
) )
stack = cf_client.describe_stacks(StackName=stack_name)["Stacks"][0] stack = cf_client.describe_stacks(StackName=stack_name)["Stacks"][0]
stack["Outputs"][0]["OutputValue"].should.be.equal("test_auto_scaling_group") assert stack["Outputs"][0]["OutputValue"] == "test_auto_scaling_group"
asg = client.describe_auto_scaling_groups()["AutoScalingGroups"][0] asg = client.describe_auto_scaling_groups()["AutoScalingGroups"][0]
asg["AutoScalingGroupName"].should.be.equal("test_auto_scaling_group") assert asg["AutoScalingGroupName"] == "test_auto_scaling_group"
asg["MinSize"].should.be.equal(1) assert asg["MinSize"] == 1
asg["MaxSize"].should.be.equal(5) assert asg["MaxSize"] == 5
asg["LaunchConfigurationName"].should.be.equal("test_launch_configuration") assert asg["LaunchConfigurationName"] == "test_launch_configuration"
client.create_launch_configuration( client.create_launch_configuration(
LaunchConfigurationName="test_launch_configuration_new", LaunchConfigurationName="test_launch_configuration_new",
@ -148,13 +147,13 @@ Outputs:
Parameters=[{"ParameterKey": "SubnetId", "ParameterValue": subnet_id}], Parameters=[{"ParameterKey": "SubnetId", "ParameterValue": subnet_id}],
) )
stack = cf_client.describe_stacks(StackName=stack_name)["Stacks"][0] stack = cf_client.describe_stacks(StackName=stack_name)["Stacks"][0]
stack["Outputs"][0]["OutputValue"].should.be.equal("test_auto_scaling_group") assert stack["Outputs"][0]["OutputValue"] == "test_auto_scaling_group"
asg = client.describe_auto_scaling_groups()["AutoScalingGroups"][0] asg = client.describe_auto_scaling_groups()["AutoScalingGroups"][0]
asg["AutoScalingGroupName"].should.be.equal("test_auto_scaling_group") assert asg["AutoScalingGroupName"] == "test_auto_scaling_group"
asg["MinSize"].should.be.equal(2) assert asg["MinSize"] == 2
asg["MaxSize"].should.be.equal(6) assert asg["MaxSize"] == 6
asg["LaunchConfigurationName"].should.be.equal("test_launch_configuration_new") assert asg["LaunchConfigurationName"] == "test_launch_configuration_new"
@mock_autoscaling @mock_autoscaling
@ -208,16 +207,16 @@ Outputs:
], ],
) )
stack = cf_client.describe_stacks(StackName=stack_name)["Stacks"][0] stack = cf_client.describe_stacks(StackName=stack_name)["Stacks"][0]
stack["Outputs"][0]["OutputValue"].should.be.equal("test_auto_scaling_group") assert stack["Outputs"][0]["OutputValue"] == "test_auto_scaling_group"
asg = client.describe_auto_scaling_groups()["AutoScalingGroups"][0] asg = client.describe_auto_scaling_groups()["AutoScalingGroups"][0]
asg["AutoScalingGroupName"].should.be.equal("test_auto_scaling_group") assert asg["AutoScalingGroupName"] == "test_auto_scaling_group"
asg["MinSize"].should.be.equal(1) assert asg["MinSize"] == 1
asg["MaxSize"].should.be.equal(5) assert asg["MaxSize"] == 5
lt = asg["LaunchTemplate"] lt = asg["LaunchTemplate"]
lt["LaunchTemplateId"].should.be.equal(launch_template_id) assert lt["LaunchTemplateId"] == launch_template_id
lt["LaunchTemplateName"].should.be.equal("test_launch_template") assert lt["LaunchTemplateName"] == "test_launch_template"
lt["Version"].should.be.equal("1") assert lt["Version"] == "1"
template_response = ec2_client.create_launch_template( template_response = ec2_client.create_launch_template(
LaunchTemplateName="test_launch_template_new", LaunchTemplateName="test_launch_template_new",
@ -259,16 +258,16 @@ Outputs:
], ],
) )
stack = cf_client.describe_stacks(StackName=stack_name)["Stacks"][0] stack = cf_client.describe_stacks(StackName=stack_name)["Stacks"][0]
stack["Outputs"][0]["OutputValue"].should.be.equal("test_auto_scaling_group") assert stack["Outputs"][0]["OutputValue"] == "test_auto_scaling_group"
asg = client.describe_auto_scaling_groups()["AutoScalingGroups"][0] asg = client.describe_auto_scaling_groups()["AutoScalingGroups"][0]
asg["AutoScalingGroupName"].should.be.equal("test_auto_scaling_group") assert asg["AutoScalingGroupName"] == "test_auto_scaling_group"
asg["MinSize"].should.be.equal(2) assert asg["MinSize"] == 2
asg["MaxSize"].should.be.equal(6) assert asg["MaxSize"] == 6
lt = asg["LaunchTemplate"] lt = asg["LaunchTemplate"]
lt["LaunchTemplateId"].should.be.equal(launch_template_id) assert lt["LaunchTemplateId"] == launch_template_id
lt["LaunchTemplateName"].should.be.equal("test_launch_template_new") assert lt["LaunchTemplateName"] == "test_launch_template_new"
lt["Version"].should.be.equal("1") assert lt["Version"] == "1"
@mock_autoscaling @mock_autoscaling
@ -356,16 +355,14 @@ def test_autoscaling_group_with_elb():
cf.create_stack(StackName="web_stack", TemplateBody=web_setup_template_json) cf.create_stack(StackName="web_stack", TemplateBody=web_setup_template_json)
autoscale_group = client.describe_auto_scaling_groups()["AutoScalingGroups"][0] autoscale_group = client.describe_auto_scaling_groups()["AutoScalingGroups"][0]
autoscale_group["LaunchConfigurationName"].should.contain("my-launch-config") assert "my-launch-config" in autoscale_group["LaunchConfigurationName"]
autoscale_group["LoadBalancerNames"].should.equal(["my-elb"]) assert autoscale_group["LoadBalancerNames"] == ["my-elb"]
# Confirm the Launch config was actually created # Confirm the Launch config was actually created
client.describe_launch_configurations()[ assert len(client.describe_launch_configurations()["LaunchConfigurations"]) == 1
"LaunchConfigurations"
].should.have.length_of(1)
# Confirm the ELB was actually created # Confirm the ELB was actually created
elb.describe_load_balancers()["LoadBalancerDescriptions"].should.have.length_of(1) assert len(elb.describe_load_balancers()["LoadBalancerDescriptions"]) == 1
resources = cf.list_stack_resources(StackName="web_stack")["StackResourceSummaries"] resources = cf.list_stack_resources(StackName="web_stack")["StackResourceSummaries"]
as_group_resource = [ as_group_resource = [
@ -373,37 +370,37 @@ def test_autoscaling_group_with_elb():
for resource in resources for resource in resources
if resource["ResourceType"] == "AWS::AutoScaling::AutoScalingGroup" if resource["ResourceType"] == "AWS::AutoScaling::AutoScalingGroup"
][0] ][0]
as_group_resource["PhysicalResourceId"].should.contain("my-as-group") assert "my-as-group" in as_group_resource["PhysicalResourceId"]
launch_config_resource = [ launch_config_resource = [
resource resource
for resource in resources for resource in resources
if resource["ResourceType"] == "AWS::AutoScaling::LaunchConfiguration" if resource["ResourceType"] == "AWS::AutoScaling::LaunchConfiguration"
][0] ][0]
launch_config_resource["PhysicalResourceId"].should.contain("my-launch-config") assert "my-launch-config" in launch_config_resource["PhysicalResourceId"]
elb_resource = [ elb_resource = [
resource resource
for resource in resources for resource in resources
if resource["ResourceType"] == "AWS::ElasticLoadBalancing::LoadBalancer" if resource["ResourceType"] == "AWS::ElasticLoadBalancing::LoadBalancer"
][0] ][0]
elb_resource["PhysicalResourceId"].should.contain("my-elb") assert "my-elb" in elb_resource["PhysicalResourceId"]
# confirm the instances were created with the right tags # confirm the instances were created with the right tags
reservations = ec2.describe_instances()["Reservations"] reservations = ec2.describe_instances()["Reservations"]
reservations.should.have.length_of(1) assert len(reservations) == 1
reservations[0]["Instances"].should.have.length_of(2) assert len(reservations[0]["Instances"]) == 2
for instance in reservations[0]["Instances"]: for instance in reservations[0]["Instances"]:
tag_keys = [t["Key"] for t in instance["Tags"]] tag_keys = [t["Key"] for t in instance["Tags"]]
tag_keys.should.contain("propagated-test-tag") assert "propagated-test-tag" in tag_keys
tag_keys.should_not.contain("not-propagated-test-tag") assert "not-propagated-test-tag" not in tag_keys
# confirm scheduled scaling action was created # confirm scheduled scaling action was created
response = client.describe_scheduled_actions( response = client.describe_scheduled_actions(
AutoScalingGroupName="test-scaling-group" AutoScalingGroupName="test-scaling-group"
)["ScheduledUpdateGroupActions"] )["ScheduledUpdateGroupActions"]
response.should.have.length_of(1) assert len(response) == 1
@mock_autoscaling @mock_autoscaling
@ -441,9 +438,9 @@ def test_autoscaling_group_update():
cf.create_stack(StackName="asg_stack", TemplateBody=asg_template_json) cf.create_stack(StackName="asg_stack", TemplateBody=asg_template_json)
asg = client.describe_auto_scaling_groups()["AutoScalingGroups"][0] asg = client.describe_auto_scaling_groups()["AutoScalingGroups"][0]
asg["MinSize"].should.equal(2) assert asg["MinSize"] == 2
asg["MaxSize"].should.equal(2) assert asg["MaxSize"] == 2
asg["DesiredCapacity"].should.equal(2) assert asg["DesiredCapacity"] == 2
asg_template["Resources"]["my-as-group"]["Properties"]["MaxSize"] = 3 asg_template["Resources"]["my-as-group"]["Properties"]["MaxSize"] = 3
asg_template["Resources"]["my-as-group"]["Properties"]["Tags"] = [ asg_template["Resources"]["my-as-group"]["Properties"]["Tags"] = [
@ -461,9 +458,9 @@ def test_autoscaling_group_update():
asg_template_json = json.dumps(asg_template) asg_template_json = json.dumps(asg_template)
cf.update_stack(StackName="asg_stack", TemplateBody=asg_template_json) cf.update_stack(StackName="asg_stack", TemplateBody=asg_template_json)
asg = client.describe_auto_scaling_groups()["AutoScalingGroups"][0] asg = client.describe_auto_scaling_groups()["AutoScalingGroups"][0]
asg["MinSize"].should.equal(2) assert asg["MinSize"] == 2
asg["MaxSize"].should.equal(3) assert asg["MaxSize"] == 3
asg["DesiredCapacity"].should.equal(2) assert asg["DesiredCapacity"] == 2
# confirm the instances were created with the right tags # confirm the instances were created with the right tags
reservations = ec2.describe_instances()["Reservations"] reservations = ec2.describe_instances()["Reservations"]
@ -472,9 +469,10 @@ def test_autoscaling_group_update():
for instance in res["Instances"]: for instance in res["Instances"]:
if instance["State"]["Name"] == "running": if instance["State"]["Name"] == "running":
running_instance_count += 1 running_instance_count += 1
instance["Tags"].should.contain( assert {
{"Key": "propagated-test-tag", "Value": "propagated-test-tag-value"} "Key": "propagated-test-tag",
) "Value": "propagated-test-tag-value",
} in instance["Tags"]
tag_keys = [t["Key"] for t in instance["Tags"]] tag_keys = [t["Key"] for t in instance["Tags"]]
tag_keys.should_not.contain("not-propagated-test-tag") assert "not-propagated-test-tag" not in tag_keys
running_instance_count.should.equal(2) assert running_instance_count == 2

80
tests/test_autoscaling/test_autoscaling_groups.py
View File

@ -1,5 +1,4 @@
import boto3 import boto3
import sure # noqa # pylint: disable=unused-import
from moto import mock_autoscaling from moto import mock_autoscaling
from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
@ -27,22 +26,22 @@ class TestAutoScalingGroup(TestCase):
self._create_group(name="tester_group") self._create_group(name="tester_group")
group = self.as_client.describe_auto_scaling_groups()["AutoScalingGroups"][0] group = self.as_client.describe_auto_scaling_groups()["AutoScalingGroups"][0]
group["AutoScalingGroupName"].should.equal("tester_group") assert group["AutoScalingGroupName"] == "tester_group"
group["MaxSize"].should.equal(2) assert group["MaxSize"] == 2
group["MinSize"].should.equal(1) assert group["MinSize"] == 1
group["LaunchConfigurationName"].should.equal(self.lc_name) assert group["LaunchConfigurationName"] == self.lc_name
# Defaults # Defaults
group["AvailabilityZones"].should.equal(["us-east-1a"]) # subnet1 assert group["AvailabilityZones"] == ["us-east-1a"] # subnet1
group["DesiredCapacity"].should.equal(1) assert group["DesiredCapacity"] == 1
group["VPCZoneIdentifier"].should.equal(self.mocked_networking["subnet1"]) assert group["VPCZoneIdentifier"] == self.mocked_networking["subnet1"]
group["DefaultCooldown"].should.equal(300) assert group["DefaultCooldown"] == 300
group["HealthCheckGracePeriod"].should.equal(300) assert group["HealthCheckGracePeriod"] == 300
group["HealthCheckType"].should.equal("EC2") assert group["HealthCheckType"] == "EC2"
group["LoadBalancerNames"].should.equal([]) assert group["LoadBalancerNames"] == []
group.shouldnt.have.key("PlacementGroup") assert "PlacementGroup" not in group
group["TerminationPolicies"].should.equal(["Default"]) assert group["TerminationPolicies"] == ["Default"]
group["Tags"].should.equal([]) assert group["Tags"] == []
def test_create_autoscaling_group__additional_params(self): def test_create_autoscaling_group__additional_params(self):
self.as_client.create_auto_scaling_group( self.as_client.create_auto_scaling_group(
@ -55,36 +54,35 @@ class TestAutoScalingGroup(TestCase):
) )
group = self.as_client.describe_auto_scaling_groups()["AutoScalingGroups"][0] group = self.as_client.describe_auto_scaling_groups()["AutoScalingGroups"][0]
group["AutoScalingGroupName"].should.equal("tester_group") assert group["AutoScalingGroupName"] == "tester_group"
group["CapacityRebalance"].should.equal(True) assert group["CapacityRebalance"] is True
def test_list_many_autoscaling_groups(self): def test_list_many_autoscaling_groups(self):
for i in range(51): for i in range(51):
self._create_group(f"TestGroup{i}") self._create_group(f"TestGroup{i}")
response = self.as_client.describe_auto_scaling_groups() response = self.as_client.describe_auto_scaling_groups()
groups = response["AutoScalingGroups"] groups = response["AutoScalingGroups"]
marker = response["NextToken"] marker = response["NextToken"]
groups.should.have.length_of(50) assert len(groups) == 50
marker.should.equal(groups[-1]["AutoScalingGroupName"]) assert marker == groups[-1]["AutoScalingGroupName"]
response2 = self.as_client.describe_auto_scaling_groups(NextToken=marker) response2 = self.as_client.describe_auto_scaling_groups(NextToken=marker)
groups.extend(response2["AutoScalingGroups"]) groups.extend(response2["AutoScalingGroups"])
groups.should.have.length_of(51) assert len(groups) == 51
assert "NextToken" not in response2.keys() assert "NextToken" not in response2.keys()
def test_autoscaling_group_delete(self): def test_autoscaling_group_delete(self):
self._create_group(name="tester_group") self._create_group(name="tester_group")
groups = self.as_client.describe_auto_scaling_groups()["AutoScalingGroups"] groups = self.as_client.describe_auto_scaling_groups()["AutoScalingGroups"]
groups.should.have.length_of(1) assert len(groups) == 1
self.as_client.delete_auto_scaling_group(AutoScalingGroupName="tester_group") self.as_client.delete_auto_scaling_group(AutoScalingGroupName="tester_group")
groups = self.as_client.describe_auto_scaling_groups()["AutoScalingGroups"] groups = self.as_client.describe_auto_scaling_groups()["AutoScalingGroups"]
groups.should.have.length_of(0) assert len(groups) == 0
def test_describe_autoscaling_groups__instances(self): def test_describe_autoscaling_groups__instances(self):
self._create_group(name="test_asg") self._create_group(name="test_asg")
@ -92,21 +90,22 @@ class TestAutoScalingGroup(TestCase):
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
AutoScalingGroupNames=["test_asg"] AutoScalingGroupNames=["test_asg"]
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
group = response["AutoScalingGroups"][0] group = response["AutoScalingGroups"][0]
group["AutoScalingGroupARN"].should.match( assert (
f"arn:aws:autoscaling:us-east-1:{ACCOUNT_ID}:autoScalingGroup:" f"arn:aws:autoscaling:us-east-1:{ACCOUNT_ID}:autoScalingGroup:"
in group["AutoScalingGroupARN"]
) )
group["AutoScalingGroupName"].should.equal("test_asg") assert group["AutoScalingGroupName"] == "test_asg"
group["LaunchConfigurationName"].should.equal(self.lc_name) assert group["LaunchConfigurationName"] == self.lc_name
group.should_not.have.key("LaunchTemplate") assert "LaunchTemplate" not in group
group["AvailabilityZones"].should.equal(["us-east-1a"]) assert group["AvailabilityZones"] == ["us-east-1a"]
group["VPCZoneIdentifier"].should.equal(self.mocked_networking["subnet1"]) assert group["VPCZoneIdentifier"] == self.mocked_networking["subnet1"]
for instance in group["Instances"]: for instance in group["Instances"]:
instance["LaunchConfigurationName"].should.equal(self.lc_name) assert instance["LaunchConfigurationName"] == self.lc_name
instance.should_not.have.key("LaunchTemplate") assert "LaunchTemplate" not in instance
instance["AvailabilityZone"].should.equal("us-east-1a") assert instance["AvailabilityZone"] == "us-east-1a"
instance["InstanceType"].should.equal("t2.medium") assert instance["InstanceType"] == "t2.medium"
def test_set_instance_health(self): def test_set_instance_health(self):
self._create_group(name="test_asg") self._create_group(name="test_asg")
@ -116,7 +115,7 @@ class TestAutoScalingGroup(TestCase):
) )
instance1 = response["AutoScalingGroups"][0]["Instances"][0] instance1 = response["AutoScalingGroups"][0]["Instances"][0]
instance1["HealthStatus"].should.equal("Healthy") assert instance1["HealthStatus"] == "Healthy"
self.as_client.set_instance_health( self.as_client.set_instance_health(
InstanceId=instance1["InstanceId"], HealthStatus="Unhealthy" InstanceId=instance1["InstanceId"], HealthStatus="Unhealthy"
@ -127,7 +126,7 @@ class TestAutoScalingGroup(TestCase):
) )
instance1 = response["AutoScalingGroups"][0]["Instances"][0] instance1 = response["AutoScalingGroups"][0]["Instances"][0]
instance1["HealthStatus"].should.equal("Unhealthy") assert instance1["HealthStatus"] == "Unhealthy"
def test_suspend_processes(self): def test_suspend_processes(self):
self._create_group(name="test-asg") self._create_group(name="test-asg")
@ -175,7 +174,7 @@ class TestAutoScalingGroup(TestCase):
proc["ProcessName"] proc["ProcessName"]
for proc in res["AutoScalingGroups"][0]["SuspendedProcesses"] for proc in res["AutoScalingGroups"][0]["SuspendedProcesses"]
] ]
set(suspended_proc_names).should.equal(set(all_proc_names)) assert set(suspended_proc_names) == set(all_proc_names)
def test_suspend_additional_processes(self): def test_suspend_additional_processes(self):
self._create_group(name="test-asg") self._create_group(name="test-asg")
@ -224,8 +223,9 @@ class TestAutoScalingGroup(TestCase):
expected_suspended_processes = [ expected_suspended_processes = [
{"ProcessName": "Terminate", "SuspensionReason": ""} {"ProcessName": "Terminate", "SuspensionReason": ""}
] ]
res["AutoScalingGroups"][0]["SuspendedProcesses"].should.equal( assert (
expected_suspended_processes res["AutoScalingGroups"][0]["SuspendedProcesses"]
== expected_suspended_processes
) )
def test_resume_processes_all_by_default(self): def test_resume_processes_all_by_default(self):
@ -243,7 +243,7 @@ class TestAutoScalingGroup(TestCase):
) )
# No processes should be suspended # No processes should be suspended
res["AutoScalingGroups"][0]["SuspendedProcesses"].should.equal([]) assert res["AutoScalingGroups"][0]["SuspendedProcesses"] == []
def _create_group(self, name): def _create_group(self, name):
self.as_client.create_auto_scaling_group( self.as_client.create_auto_scaling_group(

10
tests/test_autoscaling/test_autoscaling_metrics.py
View File

@ -1,5 +1,4 @@
import boto3 import boto3
import sure # noqa # pylint: disable=unused-import
from moto import mock_autoscaling, mock_elb, mock_ec2 from moto import mock_autoscaling, mock_elb, mock_ec2
@ -43,7 +42,8 @@ def test_enable_metrics_collection():
resp = as_client.describe_auto_scaling_groups( resp = as_client.describe_auto_scaling_groups(
AutoScalingGroupNames=["tester_group"] AutoScalingGroupNames=["tester_group"]
)["AutoScalingGroups"][0] )["AutoScalingGroups"][0]
resp.should.have.key("EnabledMetrics").length_of(1) assert len(resp["EnabledMetrics"]) == 1
resp["EnabledMetrics"][0].should.equal( assert resp["EnabledMetrics"][0] == {
{"Metric": "GroupMinSize", "Granularity": "1Minute"} "Metric": "GroupMinSize",
) "Granularity": "1Minute",
}

11
tests/test_autoscaling/test_autoscaling_scheduledactions.py
View File

@ -1,5 +1,4 @@
import boto3 import boto3
import sure # noqa # pylint: disable=unused-import
from moto import mock_autoscaling from moto import mock_autoscaling
from unittest import TestCase from unittest import TestCase
@ -19,7 +18,7 @@ class TestAutoScalingScheduledActions(TestCase):
AutoScalingGroupName=self.asg_name AutoScalingGroupName=self.asg_name
) )
actions = response["ScheduledUpdateGroupActions"] actions = response["ScheduledUpdateGroupActions"]
actions.should.have.length_of(30) assert len(actions) == 30
def test_non_existing_group_name(self): def test_non_existing_group_name(self):
self._create_scheduled_action(name="my-scheduled-action", idx=1) self._create_scheduled_action(name="my-scheduled-action", idx=1)
@ -29,7 +28,7 @@ class TestAutoScalingScheduledActions(TestCase):
) )
actions = response["ScheduledUpdateGroupActions"] actions = response["ScheduledUpdateGroupActions"]
# since there is no such group name, no actions have been returned # since there is no such group name, no actions have been returned
actions.should.have.length_of(0) assert len(actions) == 0
def test_describe_scheduled_actions_returns_all_actions_when_no_argument_is_passed( def test_describe_scheduled_actions_returns_all_actions_when_no_argument_is_passed(
self, self,
@ -46,7 +45,7 @@ class TestAutoScalingScheduledActions(TestCase):
actions = response["ScheduledUpdateGroupActions"] actions = response["ScheduledUpdateGroupActions"]
# Since no argument is passed describe_scheduled_actions, all scheduled actions are returned # Since no argument is passed describe_scheduled_actions, all scheduled actions are returned
actions.should.have.length_of(40) assert len(actions) == 40
def test_scheduled_action_delete(self): def test_scheduled_action_delete(self):
for i in range(3): for i in range(3):
@ -56,7 +55,7 @@ class TestAutoScalingScheduledActions(TestCase):
AutoScalingGroupName=self.asg_name AutoScalingGroupName=self.asg_name
) )
actions = response["ScheduledUpdateGroupActions"] actions = response["ScheduledUpdateGroupActions"]
actions.should.have.length_of(3) assert len(actions) == 3
self.client.delete_scheduled_action( self.client.delete_scheduled_action(
AutoScalingGroupName=self.asg_name, AutoScalingGroupName=self.asg_name,
@ -70,7 +69,7 @@ class TestAutoScalingScheduledActions(TestCase):
AutoScalingGroupName=self.asg_name AutoScalingGroupName=self.asg_name
) )
actions = response["ScheduledUpdateGroupActions"] actions = response["ScheduledUpdateGroupActions"]
actions.should.have.length_of(1) assert len(actions) == 1
def _create_scheduled_action(self, name, idx, asg_name=None): def _create_scheduled_action(self, name, idx, asg_name=None):
self.client.put_scheduled_update_group_action( self.client.put_scheduled_update_group_action(

82
tests/test_autoscaling/test_autoscaling_tags.py
View File

@ -50,7 +50,7 @@ def test_autoscaling_tags_update():
) )
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"]) response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
response["AutoScalingGroups"][0]["Tags"].should.have.length_of(2) assert len(response["AutoScalingGroups"][0]["Tags"]) == 2
@mock_autoscaling @mock_autoscaling
@ -101,16 +101,16 @@ def test_delete_tags_by_key():
) )
group = response["AutoScalingGroups"][0] group = response["AutoScalingGroups"][0]
tags = group["Tags"] tags = group["Tags"]
tags.should.contain(tag_to_keep) assert tag_to_keep in tags
tags.should_not.contain(tag_to_delete) assert tag_to_delete not in tags
@mock_autoscaling @mock_autoscaling
def test_describe_tags_without_resources(): def test_describe_tags_without_resources():
client = boto3.client("autoscaling", region_name="us-east-2") client = boto3.client("autoscaling", region_name="us-east-2")
resp = client.describe_tags() resp = client.describe_tags()
resp.should.have.key("Tags").equals([]) assert resp["Tags"] == []
resp.shouldnt.have.key("NextToken") assert "NextToken" not in resp
@mock_autoscaling @mock_autoscaling
@ -120,43 +120,35 @@ def test_describe_tags_no_filter():
create_asgs(client, subnet) create_asgs(client, subnet)
response = client.describe_tags() response = client.describe_tags()
response.should.have.key("Tags").length_of(4) len(response["Tags"]) == 4
response["Tags"].should.contain( assert {
{
"ResourceId": "test_asg", "ResourceId": "test_asg",
"ResourceType": "auto-scaling-group", "ResourceType": "auto-scaling-group",
"Key": "test_key", "Key": "test_key",
"Value": "updated_test_value", "Value": "updated_test_value",
"PropagateAtLaunch": True, "PropagateAtLaunch": True,
} } in response["Tags"]
) assert {
response["Tags"].should.contain(
{
"ResourceId": "test_asg", "ResourceId": "test_asg",
"ResourceType": "auto-scaling-group", "ResourceType": "auto-scaling-group",
"Key": "test_key2", "Key": "test_key2",
"Value": "test_value2", "Value": "test_value2",
"PropagateAtLaunch": False, "PropagateAtLaunch": False,
} } in response["Tags"]
) assert {
response["Tags"].should.contain(
{
"ResourceId": "test_asg2", "ResourceId": "test_asg2",
"ResourceType": "auto-scaling-group", "ResourceType": "auto-scaling-group",
"Key": "asg2tag1", "Key": "asg2tag1",
"Value": "val", "Value": "val",
"PropagateAtLaunch": False, "PropagateAtLaunch": False,
} } in response["Tags"]
) assert {
response["Tags"].should.contain(
{
"ResourceId": "test_asg2", "ResourceId": "test_asg2",
"ResourceType": "auto-scaling-group", "ResourceType": "auto-scaling-group",
"Key": "asg2tag2", "Key": "asg2tag2",
"Value": "diff", "Value": "diff",
"PropagateAtLaunch": False, "PropagateAtLaunch": False,
} } in response["Tags"]
)
@mock_autoscaling @mock_autoscaling
@ -168,66 +160,54 @@ def test_describe_tags_filter_by_name():
response = client.describe_tags( response = client.describe_tags(
Filters=[{"Name": "auto-scaling-group", "Values": ["test_asg"]}] Filters=[{"Name": "auto-scaling-group", "Values": ["test_asg"]}]
) )
response.should.have.key("Tags").length_of(2) assert len(response["Tags"]) == 2
response["Tags"].should.contain( assert {
{
"ResourceId": "test_asg", "ResourceId": "test_asg",
"ResourceType": "auto-scaling-group", "ResourceType": "auto-scaling-group",
"Key": "test_key", "Key": "test_key",
"Value": "updated_test_value", "Value": "updated_test_value",
"PropagateAtLaunch": True, "PropagateAtLaunch": True,
} } in response["Tags"]
) assert {
response["Tags"].should.contain(
{
"ResourceId": "test_asg", "ResourceId": "test_asg",
"ResourceType": "auto-scaling-group", "ResourceType": "auto-scaling-group",
"Key": "test_key2", "Key": "test_key2",
"Value": "test_value2", "Value": "test_value2",
"PropagateAtLaunch": False, "PropagateAtLaunch": False,
} } in response["Tags"]
)
response = client.describe_tags( response = client.describe_tags(
Filters=[{"Name": "auto-scaling-group", "Values": ["test_asg", "test_asg2"]}] Filters=[{"Name": "auto-scaling-group", "Values": ["test_asg", "test_asg2"]}]
) )
response.should.have.key("Tags").length_of(4) assert len(response["Tags"]) == 4
response["Tags"].should.contain( assert {
{
"ResourceId": "test_asg", "ResourceId": "test_asg",
"ResourceType": "auto-scaling-group", "ResourceType": "auto-scaling-group",
"Key": "test_key", "Key": "test_key",
"Value": "updated_test_value", "Value": "updated_test_value",
"PropagateAtLaunch": True, "PropagateAtLaunch": True,
} } in response["Tags"]
) assert {
response["Tags"].should.contain(
{
"ResourceId": "test_asg", "ResourceId": "test_asg",
"ResourceType": "auto-scaling-group", "ResourceType": "auto-scaling-group",
"Key": "test_key2", "Key": "test_key2",
"Value": "test_value2", "Value": "test_value2",
"PropagateAtLaunch": False, "PropagateAtLaunch": False,
} } in response["Tags"]
) assert {
response["Tags"].should.contain(
{
"ResourceId": "test_asg2", "ResourceId": "test_asg2",
"ResourceType": "auto-scaling-group", "ResourceType": "auto-scaling-group",
"Key": "asg2tag1", "Key": "asg2tag1",
"Value": "val", "Value": "val",
"PropagateAtLaunch": False, "PropagateAtLaunch": False,
} } in response["Tags"]
) assert {
response["Tags"].should.contain(
{
"ResourceId": "test_asg2", "ResourceId": "test_asg2",
"ResourceType": "auto-scaling-group", "ResourceType": "auto-scaling-group",
"Key": "asg2tag2", "Key": "asg2tag2",
"Value": "diff", "Value": "diff",
"PropagateAtLaunch": False, "PropagateAtLaunch": False,
} } in response["Tags"]
)
@mock_autoscaling @mock_autoscaling
@ -239,8 +219,8 @@ def test_describe_tags_filter_by_propgateatlaunch():
response = client.describe_tags( response = client.describe_tags(
Filters=[{"Name": "propagate-at-launch", "Values": ["True"]}] Filters=[{"Name": "propagate-at-launch", "Values": ["True"]}]
) )
response.should.have.key("Tags").length_of(1) len(response["Tags"]) == 1
response["Tags"].should.contain( assert response["Tags"] == [
{ {
"ResourceId": "test_asg", "ResourceId": "test_asg",
"ResourceType": "auto-scaling-group", "ResourceType": "auto-scaling-group",
@ -248,7 +228,7 @@ def test_describe_tags_filter_by_propgateatlaunch():
"Value": "updated_test_value", "Value": "updated_test_value",
"PropagateAtLaunch": True, "PropagateAtLaunch": True,
} }
) ]
def create_asgs(client, subnet): def create_asgs(client, subnet):

426
tests/test_autoscaling/test_elb.py
View File

@ -1,5 +1,4 @@
import boto3 import boto3
import sure # noqa # pylint: disable=unused-import
from moto import mock_autoscaling, mock_elb, mock_ec2 from moto import mock_autoscaling, mock_elb, mock_ec2
from .utils import setup_networking from .utils import setup_networking
@ -57,8 +56,8 @@ class TestAutoScalingELB(TestCase):
AutoScalingGroupName=self.asg_name AutoScalingGroupName=self.asg_name
) )
assert response["ResponseMetadata"]["RequestId"] assert response["ResponseMetadata"]["RequestId"]
list(response["LoadBalancers"]).should.have.length_of(1) assert len(list(response["LoadBalancers"])) == 1
response["LoadBalancers"][0]["LoadBalancerName"].should.equal(self.lb_name) assert response["LoadBalancers"][0]["LoadBalancerName"] == self.lb_name
def test_create_elb_and_autoscaling_group_no_relationship(self): def test_create_elb_and_autoscaling_group_no_relationship(self):
self.as_client.create_auto_scaling_group( self.as_client.create_auto_scaling_group(
@ -74,13 +73,11 @@ class TestAutoScalingELB(TestCase):
response = self.as_client.describe_load_balancers( response = self.as_client.describe_load_balancers(
AutoScalingGroupName=self.asg_name AutoScalingGroupName=self.asg_name
) )
list(response["LoadBalancers"]).should.have.length_of(0) assert len(list(response["LoadBalancers"])) == 0
response = self.elb_client.describe_load_balancers( response = self.elb_client.describe_load_balancers(
LoadBalancerNames=[self.lb_name] LoadBalancerNames=[self.lb_name]
) )
list( assert len(list(response["LoadBalancerDescriptions"][0]["Instances"])) == 0
response["LoadBalancerDescriptions"][0]["Instances"]
).should.have.length_of(0)
def test_attach_load_balancer(self): def test_attach_load_balancer(self):
self.as_client.create_auto_scaling_group( self.as_client.create_auto_scaling_group(
@ -103,21 +100,20 @@ class TestAutoScalingELB(TestCase):
response = self.as_client.attach_load_balancers( response = self.as_client.attach_load_balancers(
AutoScalingGroupName=self.asg_name, LoadBalancerNames=[self.lb_name] AutoScalingGroupName=self.asg_name, LoadBalancerNames=[self.lb_name]
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
response = self.elb_client.describe_load_balancers( response = self.elb_client.describe_load_balancers(
LoadBalancerNames=[self.lb_name] LoadBalancerNames=[self.lb_name]
) )
list( assert (
response["LoadBalancerDescriptions"][0]["Instances"] len(list(response["LoadBalancerDescriptions"][0]["Instances"]))
).should.have.length_of(self.instance_count) == self.instance_count
)
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
AutoScalingGroupNames=[self.asg_name] AutoScalingGroupNames=[self.asg_name]
) )
list( assert len(list(response["AutoScalingGroups"][0]["LoadBalancerNames"])) == 1
response["AutoScalingGroups"][0]["LoadBalancerNames"]
).should.have.length_of(1)
def test_detach_load_balancer(self): def test_detach_load_balancer(self):
self.as_client.create_auto_scaling_group( self.as_client.create_auto_scaling_group(
@ -141,19 +137,17 @@ class TestAutoScalingELB(TestCase):
response = self.as_client.detach_load_balancers( response = self.as_client.detach_load_balancers(
AutoScalingGroupName=self.asg_name, LoadBalancerNames=[self.lb_name] AutoScalingGroupName=self.asg_name, LoadBalancerNames=[self.lb_name]
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
response = self.elb_client.describe_load_balancers( response = self.elb_client.describe_load_balancers(
LoadBalancerNames=[self.lb_name] LoadBalancerNames=[self.lb_name]
) )
list( assert len(list(response["LoadBalancerDescriptions"][0]["Instances"])) == 0
response["LoadBalancerDescriptions"][0]["Instances"]
).should.have.length_of(0)
response = self.as_client.describe_load_balancers( response = self.as_client.describe_load_balancers(
AutoScalingGroupName=self.asg_name AutoScalingGroupName=self.asg_name
) )
list(response["LoadBalancers"]).should.have.length_of(0) assert len(list(response["LoadBalancers"])) == 0
def test_create_autoscaling_group_within_elb(self): def test_create_autoscaling_group_within_elb(self):
# we attach a couple of machines to the load balancer # we attach a couple of machines to the load balancer
@ -221,53 +215,53 @@ class TestAutoScalingELB(TestCase):
AutoScalingGroupNames=[self.asg_name] AutoScalingGroupNames=[self.asg_name]
) )
group = resp["AutoScalingGroups"][0] group = resp["AutoScalingGroups"][0]
group["AutoScalingGroupName"].should.equal(self.asg_name) assert group["AutoScalingGroupName"] == self.asg_name
set(group["AvailabilityZones"]).should.equal(set(["us-east-1a", "us-east-1b"])) assert set(group["AvailabilityZones"]) == set(["us-east-1a", "us-east-1b"])
group["DesiredCapacity"].should.equal(2) assert group["DesiredCapacity"] == 2
group["MaxSize"].should.equal(INSTANCE_COUNT_GROUP) assert group["MaxSize"] == INSTANCE_COUNT_GROUP
group["MinSize"].should.equal(INSTANCE_COUNT_GROUP) assert group["MinSize"] == INSTANCE_COUNT_GROUP
group["Instances"].should.have.length_of(INSTANCE_COUNT_GROUP) assert len(group["Instances"]) == INSTANCE_COUNT_GROUP
group["VPCZoneIdentifier"].should.equal( assert (
f"{self.mocked_networking['subnet1']},{self.mocked_networking['subnet2']}" group["VPCZoneIdentifier"]
== f"{self.mocked_networking['subnet1']},{self.mocked_networking['subnet2']}"
) )
group["LaunchConfigurationName"].should.equal(self.lc_name) assert group["LaunchConfigurationName"] == self.lc_name
group["DefaultCooldown"].should.equal(60) assert group["DefaultCooldown"] == 60
group["HealthCheckGracePeriod"].should.equal(100) assert group["HealthCheckGracePeriod"] == 100
group["HealthCheckType"].should.equal("EC2") assert group["HealthCheckType"] == "EC2"
group["LoadBalancerNames"].should.equal([self.lb_name]) assert group["LoadBalancerNames"] == [self.lb_name]
group["PlacementGroup"].should.equal("test_placement") assert group["PlacementGroup"] == "test_placement"
list(group["TerminationPolicies"]).should.equal( assert list(group["TerminationPolicies"]) == [
["OldestInstance", "NewestInstance"] "OldestInstance",
) "NewestInstance",
list(group["Tags"]).should.have.length_of(1) ]
assert len(list(group["Tags"])) == 1
tag = group["Tags"][0] tag = group["Tags"][0]
tag["ResourceId"].should.equal(self.asg_name) assert tag["ResourceId"] == self.asg_name
tag["Key"].should.equal("test_key") assert tag["Key"] == "test_key"
tag["Value"].should.equal("test_value") assert tag["Value"] == "test_value"
tag["PropagateAtLaunch"].should.equal(True) assert tag["PropagateAtLaunch"] is True
instances_attached = self.elb_client.describe_instance_health( instances_attached = self.elb_client.describe_instance_health(
LoadBalancerName=self.lb_name LoadBalancerName=self.lb_name
)["InstanceStates"] )["InstanceStates"]
instances_attached.should.have.length_of( assert len(instances_attached) == INSTANCE_COUNT_START + INSTANCE_COUNT_GROUP
INSTANCE_COUNT_START + INSTANCE_COUNT_GROUP
)
attached_ids = [i["InstanceId"] for i in instances_attached] attached_ids = [i["InstanceId"] for i in instances_attached]
for ec2_instance_id in instances_ids: for ec2_instance_id in instances_ids:
attached_ids.should.contain(ec2_instance_id) assert ec2_instance_id in attached_ids
scheduled_actions = self.as_client.describe_scheduled_actions( scheduled_actions = self.as_client.describe_scheduled_actions(
AutoScalingGroupName=self.asg_name AutoScalingGroupName=self.asg_name
) )
scheduled_action_1 = scheduled_actions["ScheduledUpdateGroupActions"][0] scheduled_action_1 = scheduled_actions["ScheduledUpdateGroupActions"][0]
scheduled_action_1["AutoScalingGroupName"].should.equal(self.asg_name) assert scheduled_action_1["AutoScalingGroupName"] == self.asg_name
scheduled_action_1["DesiredCapacity"].should.equal(9) assert scheduled_action_1["DesiredCapacity"] == 9
scheduled_action_1["MaxSize"].should.equal(12) assert scheduled_action_1["MaxSize"] == 12
scheduled_action_1["MinSize"].should.equal(5) assert scheduled_action_1["MinSize"] == 5
scheduled_action_1.should.contain("StartTime") assert "StartTime" in scheduled_action_1
scheduled_action_1.should.contain("EndTime") assert "EndTime" in scheduled_action_1
scheduled_action_1["Recurrence"].should.equal("* * * * *") assert scheduled_action_1["Recurrence"] == "* * * * *"
scheduled_action_1["ScheduledActionName"].should.equal("my-scheduled-action") assert scheduled_action_1["ScheduledActionName"] == "my-scheduled-action"
@mock_autoscaling @mock_autoscaling
@ -316,7 +310,7 @@ class TestAutoScalingInstances(TestCase):
response = self.as_client.attach_load_balancers( response = self.as_client.attach_load_balancers(
AutoScalingGroupName=self.asg_name, LoadBalancerNames=[self.lb_name] AutoScalingGroupName=self.asg_name, LoadBalancerNames=[self.lb_name]
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
def test_detach_one_instance_decrement(self): def test_detach_one_instance_decrement(self):
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
@ -336,38 +330,34 @@ class TestAutoScalingInstances(TestCase):
InstanceIds=[instance_to_detach], InstanceIds=[instance_to_detach],
ShouldDecrementDesiredCapacity=True, ShouldDecrementDesiredCapacity=True,
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
AutoScalingGroupNames=[self.asg_name] AutoScalingGroupNames=[self.asg_name]
) )
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(1) assert len(response["AutoScalingGroups"][0]["Instances"]) == 1
instance_to_detach.shouldnt.be.within( assert instance_to_detach not in [
[x["InstanceId"] for x in response["AutoScalingGroups"][0]["Instances"]] x["InstanceId"] for x in response["AutoScalingGroups"][0]["Instances"]
) ]
# test to ensure tag has been removed # test to ensure tag has been removed
response = ec2_client.describe_instances(InstanceIds=[instance_to_detach]) response = ec2_client.describe_instances(InstanceIds=[instance_to_detach])
tags = response["Reservations"][0]["Instances"][0]["Tags"] tags = response["Reservations"][0]["Instances"][0]["Tags"]
tags.should.have.length_of(1) assert len(tags) == 1
# test to ensure tag is present on other instance # test to ensure tag is present on other instance
response = ec2_client.describe_instances(InstanceIds=[instance_to_keep]) response = ec2_client.describe_instances(InstanceIds=[instance_to_keep])
tags = response["Reservations"][0]["Instances"][0]["Tags"] tags = response["Reservations"][0]["Instances"][0]["Tags"]
tags.should.have.length_of(2) assert len(tags) == 2
response = self.elb_client.describe_load_balancers( response = self.elb_client.describe_load_balancers(
LoadBalancerNames=[self.lb_name] LoadBalancerNames=[self.lb_name]
) )
list( assert len(list(response["LoadBalancerDescriptions"][0]["Instances"])) == 1
response["LoadBalancerDescriptions"][0]["Instances"] assert instance_to_detach not in [
).should.have.length_of(1)
instance_to_detach.shouldnt.be.within(
[
x["InstanceId"] x["InstanceId"]
for x in response["LoadBalancerDescriptions"][0]["Instances"] for x in response["LoadBalancerDescriptions"][0]["Instances"]
] ]
)
def test_detach_one_instance(self): def test_detach_one_instance(self):
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
@ -387,34 +377,30 @@ class TestAutoScalingInstances(TestCase):
InstanceIds=[instance_to_detach], InstanceIds=[instance_to_detach],
ShouldDecrementDesiredCapacity=False, ShouldDecrementDesiredCapacity=False,
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
AutoScalingGroupNames=[self.asg_name] AutoScalingGroupNames=[self.asg_name]
) )
# test to ensure instance was replaced # test to ensure instance was replaced
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(2) assert len(response["AutoScalingGroups"][0]["Instances"]) == 2
response = ec2_client.describe_instances(InstanceIds=[instance_to_detach]) response = ec2_client.describe_instances(InstanceIds=[instance_to_detach])
tags = response["Reservations"][0]["Instances"][0]["Tags"] tags = response["Reservations"][0]["Instances"][0]["Tags"]
tags.should.have.length_of(1) assert len(tags) == 1
response = ec2_client.describe_instances(InstanceIds=[instance_to_keep]) response = ec2_client.describe_instances(InstanceIds=[instance_to_keep])
tags = response["Reservations"][0]["Instances"][0]["Tags"] tags = response["Reservations"][0]["Instances"][0]["Tags"]
tags.should.have.length_of(2) assert len(tags) == 2
response = self.elb_client.describe_load_balancers( response = self.elb_client.describe_load_balancers(
LoadBalancerNames=[self.lb_name] LoadBalancerNames=[self.lb_name]
) )
list( assert len(list(response["LoadBalancerDescriptions"][0]["Instances"])) == 2
response["LoadBalancerDescriptions"][0]["Instances"] assert instance_to_detach not in [
).should.have.length_of(2)
instance_to_detach.shouldnt.be.within(
[
x["InstanceId"] x["InstanceId"]
for x in response["LoadBalancerDescriptions"][0]["Instances"] for x in response["LoadBalancerDescriptions"][0]["Instances"]
] ]
)
def test_standby_one_instance_decrement(self): def test_standby_one_instance_decrement(self):
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
@ -431,38 +417,32 @@ class TestAutoScalingInstances(TestCase):
InstanceIds=[instance_to_standby], InstanceIds=[instance_to_standby],
ShouldDecrementDesiredCapacity=True, ShouldDecrementDesiredCapacity=True,
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
AutoScalingGroupNames=[self.asg_name] AutoScalingGroupNames=[self.asg_name]
) )
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(2) assert len(response["AutoScalingGroups"][0]["Instances"]) == 2
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(1) assert response["AutoScalingGroups"][0]["DesiredCapacity"] == 1
response = self.as_client.describe_auto_scaling_instances( response = self.as_client.describe_auto_scaling_instances(
InstanceIds=[instance_to_standby] InstanceIds=[instance_to_standby]
) )
response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") assert response["AutoScalingInstances"][0]["LifecycleState"] == "Standby"
# test to ensure tag has been retained (standby instance is still part of the ASG) # test to ensure tag has been retained (standby instance is still part of the ASG)
response = ec2_client.describe_instances(InstanceIds=[instance_to_standby]) response = ec2_client.describe_instances(InstanceIds=[instance_to_standby])
for reservation in response["Reservations"]: for reservation in response["Reservations"]:
for instance in reservation["Instances"]: for instance in reservation["Instances"]:
tags = instance["Tags"] tags = instance["Tags"]
tags.should.have.length_of(2) assert len(tags) == 2
response = self.elb_client.describe_load_balancers( response = self.elb_client.describe_load_balancers(
LoadBalancerNames=[self.lb_name] LoadBalancerNames=[self.lb_name]
) )
list( assert len(response["LoadBalancerDescriptions"][0]["Instances"]) == 1
response["LoadBalancerDescriptions"][0]["Instances"] for x in response["LoadBalancerDescriptions"][0]["Instances"]:
).should.have.length_of(1) assert x["InstanceId"] not in instance_to_standby
instance_to_standby.shouldnt.be.within(
[
x["InstanceId"]
for x in response["LoadBalancerDescriptions"][0]["Instances"]
]
)
def test_standby_one_instance(self): def test_standby_one_instance(self):
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
@ -479,38 +459,32 @@ class TestAutoScalingInstances(TestCase):
InstanceIds=[instance_to_standby], InstanceIds=[instance_to_standby],
ShouldDecrementDesiredCapacity=False, ShouldDecrementDesiredCapacity=False,
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
AutoScalingGroupNames=[self.asg_name] AutoScalingGroupNames=[self.asg_name]
) )
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3) assert len(response["AutoScalingGroups"][0]["Instances"]) == 3
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2) assert response["AutoScalingGroups"][0]["DesiredCapacity"] == 2
response = self.as_client.describe_auto_scaling_instances( response = self.as_client.describe_auto_scaling_instances(
InstanceIds=[instance_to_standby] InstanceIds=[instance_to_standby]
) )
response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") assert response["AutoScalingInstances"][0]["LifecycleState"] == "Standby"
# test to ensure tag has been retained (standby instance is still part of the ASG) # test to ensure tag has been retained (standby instance is still part of the ASG)
response = ec2_client.describe_instances(InstanceIds=[instance_to_standby]) response = ec2_client.describe_instances(InstanceIds=[instance_to_standby])
for reservation in response["Reservations"]: for reservation in response["Reservations"]:
for instance in reservation["Instances"]: for instance in reservation["Instances"]:
tags = instance["Tags"] tags = instance["Tags"]
tags.should.have.length_of(2) assert len(tags) == 2
response = self.elb_client.describe_load_balancers( response = self.elb_client.describe_load_balancers(
LoadBalancerNames=[self.lb_name] LoadBalancerNames=[self.lb_name]
) )
list( assert len(response["LoadBalancerDescriptions"][0]["Instances"]) == 2
response["LoadBalancerDescriptions"][0]["Instances"] for x in response["LoadBalancerDescriptions"][0]["Instances"]:
).should.have.length_of(2) assert x["InstanceId"] not in instance_to_standby
instance_to_standby.shouldnt.be.within(
[
x["InstanceId"]
for x in response["LoadBalancerDescriptions"][0]["Instances"]
]
)
def test_standby_elb_update(self): def test_standby_elb_update(self):
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
@ -525,31 +499,25 @@ class TestAutoScalingInstances(TestCase):
InstanceIds=[instance_to_standby], InstanceIds=[instance_to_standby],
ShouldDecrementDesiredCapacity=False, ShouldDecrementDesiredCapacity=False,
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
AutoScalingGroupNames=[self.asg_name] AutoScalingGroupNames=[self.asg_name]
) )
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3) assert len(response["AutoScalingGroups"][0]["Instances"]) == 3
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2) assert response["AutoScalingGroups"][0]["DesiredCapacity"] == 2
response = self.as_client.describe_auto_scaling_instances( response = self.as_client.describe_auto_scaling_instances(
InstanceIds=[instance_to_standby] InstanceIds=[instance_to_standby]
) )
response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") assert response["AutoScalingInstances"][0]["LifecycleState"] == "Standby"
response = self.elb_client.describe_load_balancers( response = self.elb_client.describe_load_balancers(
LoadBalancerNames=[self.lb_name] LoadBalancerNames=[self.lb_name]
) )
list( assert len(response["LoadBalancerDescriptions"][0]["Instances"]) == 2
response["LoadBalancerDescriptions"][0]["Instances"] for x in response["LoadBalancerDescriptions"][0]["Instances"]:
).should.have.length_of(2) assert x["InstanceId"] not in instance_to_standby
instance_to_standby.shouldnt.be.within(
[
x["InstanceId"]
for x in response["LoadBalancerDescriptions"][0]["Instances"]
]
)
def test_standby_terminate_instance_decrement(self): def test_standby_terminate_instance_decrement(self):
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
@ -566,54 +534,49 @@ class TestAutoScalingInstances(TestCase):
InstanceIds=[instance_to_standby_terminate], InstanceIds=[instance_to_standby_terminate],
ShouldDecrementDesiredCapacity=False, ShouldDecrementDesiredCapacity=False,
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
AutoScalingGroupNames=[self.asg_name] AutoScalingGroupNames=[self.asg_name]
) )
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3) assert len(response["AutoScalingGroups"][0]["Instances"]) == 3
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2) assert response["AutoScalingGroups"][0]["DesiredCapacity"] == 2
response = self.as_client.describe_auto_scaling_instances( response = self.as_client.describe_auto_scaling_instances(
InstanceIds=[instance_to_standby_terminate] InstanceIds=[instance_to_standby_terminate]
) )
response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") assert response["AutoScalingInstances"][0]["LifecycleState"] == "Standby"
response = self.as_client.terminate_instance_in_auto_scaling_group( response = self.as_client.terminate_instance_in_auto_scaling_group(
InstanceId=instance_to_standby_terminate, InstanceId=instance_to_standby_terminate,
ShouldDecrementDesiredCapacity=True, ShouldDecrementDesiredCapacity=True,
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
# AWS still decrements desired capacity ASG if requested, even if the terminated instance is in standby # AWS still decrements desired capacity ASG if requested, even if the terminated instance is in standby
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
AutoScalingGroupNames=[self.asg_name] AutoScalingGroupNames=[self.asg_name]
) )
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(1) assert len(response["AutoScalingGroups"][0]["Instances"]) == 1
response["AutoScalingGroups"][0]["Instances"][0]["InstanceId"].should_not.equal( assert (
instance_to_standby_terminate response["AutoScalingGroups"][0]["Instances"][0]["InstanceId"]
!= instance_to_standby_terminate
) )
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(1) assert response["AutoScalingGroups"][0]["DesiredCapacity"] == 1
response = ec2_client.describe_instances( response = ec2_client.describe_instances(
InstanceIds=[instance_to_standby_terminate] InstanceIds=[instance_to_standby_terminate]
) )
response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal( assert (
"terminated" response["Reservations"][0]["Instances"][0]["State"]["Name"] == "terminated"
) )
response = self.elb_client.describe_load_balancers( response = self.elb_client.describe_load_balancers(
LoadBalancerNames=[self.lb_name] LoadBalancerNames=[self.lb_name]
) )
list( assert len(list(response["LoadBalancerDescriptions"][0]["Instances"])) == 1
response["LoadBalancerDescriptions"][0]["Instances"] for x in response["LoadBalancerDescriptions"][0]["Instances"]:
).should.have.length_of(1) assert x["InstanceId"] not in instance_to_standby_terminate
instance_to_standby_terminate.shouldnt.be.within(
[
x["InstanceId"]
for x in response["LoadBalancerDescriptions"][0]["Instances"]
]
)
def test_standby_terminate_instance_no_decrement(self): def test_standby_terminate_instance_no_decrement(self):
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
@ -630,54 +593,48 @@ class TestAutoScalingInstances(TestCase):
InstanceIds=[instance_to_standby_terminate], InstanceIds=[instance_to_standby_terminate],
ShouldDecrementDesiredCapacity=False, ShouldDecrementDesiredCapacity=False,
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
AutoScalingGroupNames=[self.asg_name] AutoScalingGroupNames=[self.asg_name]
) )
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3) assert len(response["AutoScalingGroups"][0]["Instances"]) == 3
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2) assert response["AutoScalingGroups"][0]["DesiredCapacity"] == 2
response = self.as_client.describe_auto_scaling_instances( response = self.as_client.describe_auto_scaling_instances(
InstanceIds=[instance_to_standby_terminate] InstanceIds=[instance_to_standby_terminate]
) )
response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") assert response["AutoScalingInstances"][0]["LifecycleState"] == "Standby"
response = self.as_client.terminate_instance_in_auto_scaling_group( response = self.as_client.terminate_instance_in_auto_scaling_group(
InstanceId=instance_to_standby_terminate, InstanceId=instance_to_standby_terminate,
ShouldDecrementDesiredCapacity=False, ShouldDecrementDesiredCapacity=False,
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
AutoScalingGroupNames=[self.asg_name] AutoScalingGroupNames=[self.asg_name]
) )
group = response["AutoScalingGroups"][0] group = response["AutoScalingGroups"][0]
group["Instances"].should.have.length_of(2) assert len(group["Instances"]) == 2
instance_to_standby_terminate.shouldnt.be.within( assert instance_to_standby_terminate not in [
[x["InstanceId"] for x in group["Instances"]] x["InstanceId"] for x in group["Instances"]
) ]
group["DesiredCapacity"].should.equal(2) assert group["DesiredCapacity"] == 2
response = ec2_client.describe_instances( response = ec2_client.describe_instances(
InstanceIds=[instance_to_standby_terminate] InstanceIds=[instance_to_standby_terminate]
) )
response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal( assert (
"terminated" response["Reservations"][0]["Instances"][0]["State"]["Name"] == "terminated"
) )
response = self.elb_client.describe_load_balancers( response = self.elb_client.describe_load_balancers(
LoadBalancerNames=[self.lb_name] LoadBalancerNames=[self.lb_name]
) )
list( assert len(response["LoadBalancerDescriptions"][0]["Instances"]) == 2
response["LoadBalancerDescriptions"][0]["Instances"] for x in response["LoadBalancerDescriptions"][0]["Instances"]:
).should.have.length_of(2) assert x["InstanceId"] not in instance_to_standby_terminate
instance_to_standby_terminate.shouldnt.be.within(
[
x["InstanceId"]
for x in response["LoadBalancerDescriptions"][0]["Instances"]
]
)
def test_standby_detach_instance_decrement(self): def test_standby_detach_instance_decrement(self):
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
@ -694,55 +651,45 @@ class TestAutoScalingInstances(TestCase):
InstanceIds=[instance_to_standby_detach], InstanceIds=[instance_to_standby_detach],
ShouldDecrementDesiredCapacity=False, ShouldDecrementDesiredCapacity=False,
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
AutoScalingGroupNames=[self.asg_name] AutoScalingGroupNames=[self.asg_name]
) )
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3) assert len(response["AutoScalingGroups"][0]["Instances"]) == 3
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2) assert response["AutoScalingGroups"][0]["DesiredCapacity"] == 2
response = self.as_client.describe_auto_scaling_instances( response = self.as_client.describe_auto_scaling_instances(
InstanceIds=[instance_to_standby_detach] InstanceIds=[instance_to_standby_detach]
) )
response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") assert response["AutoScalingInstances"][0]["LifecycleState"] == "Standby"
response = self.as_client.detach_instances( response = self.as_client.detach_instances(
AutoScalingGroupName=self.asg_name, AutoScalingGroupName=self.asg_name,
InstanceIds=[instance_to_standby_detach], InstanceIds=[instance_to_standby_detach],
ShouldDecrementDesiredCapacity=True, ShouldDecrementDesiredCapacity=True,
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
# AWS still decrements desired capacity ASG if requested, even if the detached instance was in standby # AWS still decrements desired capacity ASG if requested, even if the detached instance was in standby
response = self.as_client.describe_auto_scaling_groups( group = self.as_client.describe_auto_scaling_groups(
AutoScalingGroupNames=[self.asg_name] AutoScalingGroupNames=[self.asg_name]
) )["AutoScalingGroups"][0]
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(1) assert len(group["Instances"]) == 1
response["AutoScalingGroups"][0]["Instances"][0]["InstanceId"].should_not.equal( assert group["Instances"][0]["InstanceId"] != instance_to_standby_detach
instance_to_standby_detach assert group["DesiredCapacity"] == 1
)
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(1)
response = ec2_client.describe_instances( instance = ec2_client.describe_instances(
InstanceIds=[instance_to_standby_detach] InstanceIds=[instance_to_standby_detach]
) )["Reservations"][0]["Instances"][0]
response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal( assert instance["State"]["Name"] == "running"
"running"
)
response = self.elb_client.describe_load_balancers( response = self.elb_client.describe_load_balancers(
LoadBalancerNames=[self.lb_name] LoadBalancerNames=[self.lb_name]
) )
list( assert len(response["LoadBalancerDescriptions"][0]["Instances"]) == 1
response["LoadBalancerDescriptions"][0]["Instances"] for x in response["LoadBalancerDescriptions"][0]["Instances"]:
).should.have.length_of(1) assert x["InstanceId"] not in instance_to_standby_detach
instance_to_standby_detach.shouldnt.be.within(
[
x["InstanceId"]
for x in response["LoadBalancerDescriptions"][0]["Instances"]
]
)
def test_standby_detach_instance_no_decrement(self): def test_standby_detach_instance_no_decrement(self):
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
@ -759,55 +706,47 @@ class TestAutoScalingInstances(TestCase):
InstanceIds=[instance_to_standby_detach], InstanceIds=[instance_to_standby_detach],
ShouldDecrementDesiredCapacity=False, ShouldDecrementDesiredCapacity=False,
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
AutoScalingGroupNames=[self.asg_name] AutoScalingGroupNames=[self.asg_name]
) )
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3) assert len(response["AutoScalingGroups"][0]["Instances"]) == 3
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2) assert response["AutoScalingGroups"][0]["DesiredCapacity"] == 2
response = self.as_client.describe_auto_scaling_instances( response = self.as_client.describe_auto_scaling_instances(
InstanceIds=[instance_to_standby_detach] InstanceIds=[instance_to_standby_detach]
) )
response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") assert response["AutoScalingInstances"][0]["LifecycleState"] == "Standby"
response = self.as_client.detach_instances( response = self.as_client.detach_instances(
AutoScalingGroupName=self.asg_name, AutoScalingGroupName=self.asg_name,
InstanceIds=[instance_to_standby_detach], InstanceIds=[instance_to_standby_detach],
ShouldDecrementDesiredCapacity=False, ShouldDecrementDesiredCapacity=False,
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
AutoScalingGroupNames=[self.asg_name] AutoScalingGroupNames=[self.asg_name]
) )
group = response["AutoScalingGroups"][0] group = response["AutoScalingGroups"][0]
group["Instances"].should.have.length_of(2) assert len(group["Instances"]) == 2
instance_to_standby_detach.shouldnt.be.within( assert instance_to_standby_detach not in [
[x["InstanceId"] for x in group["Instances"]] x["InstanceId"] for x in group["Instances"]
) ]
group["DesiredCapacity"].should.equal(2) assert group["DesiredCapacity"] == 2
response = ec2_client.describe_instances( instance = ec2_client.describe_instances(
InstanceIds=[instance_to_standby_detach] InstanceIds=[instance_to_standby_detach]
) )["Reservations"][0]["Instances"][0]
response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal( assert instance["State"]["Name"] == "running"
"running"
)
response = self.elb_client.describe_load_balancers( response = self.elb_client.describe_load_balancers(
LoadBalancerNames=[self.lb_name] LoadBalancerNames=[self.lb_name]
) )
list( assert len(list(response["LoadBalancerDescriptions"][0]["Instances"])) == 2
response["LoadBalancerDescriptions"][0]["Instances"] for x in response["LoadBalancerDescriptions"][0]["Instances"]:
).should.have.length_of(2) assert x["InstanceId"] not in instance_to_standby_detach
instance_to_standby_detach.shouldnt.be.within(
[
x["InstanceId"]
for x in response["LoadBalancerDescriptions"][0]["Instances"]
]
)
def test_standby_exit_standby(self): def test_standby_exit_standby(self):
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
@ -824,54 +763,48 @@ class TestAutoScalingInstances(TestCase):
InstanceIds=[instance_to_standby_exit_standby], InstanceIds=[instance_to_standby_exit_standby],
ShouldDecrementDesiredCapacity=False, ShouldDecrementDesiredCapacity=False,
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
AutoScalingGroupNames=[self.asg_name] AutoScalingGroupNames=[self.asg_name]
) )
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3) assert len(response["AutoScalingGroups"][0]["Instances"]) == 3
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2) assert response["AutoScalingGroups"][0]["DesiredCapacity"] == 2
response = self.as_client.describe_auto_scaling_instances( response = self.as_client.describe_auto_scaling_instances(
InstanceIds=[instance_to_standby_exit_standby] InstanceIds=[instance_to_standby_exit_standby]
) )
response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby") assert response["AutoScalingInstances"][0]["LifecycleState"] == "Standby"
response = self.as_client.exit_standby( response = self.as_client.exit_standby(
AutoScalingGroupName=self.asg_name, AutoScalingGroupName=self.asg_name,
InstanceIds=[instance_to_standby_exit_standby], InstanceIds=[instance_to_standby_exit_standby],
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
AutoScalingGroupNames=[self.asg_name] AutoScalingGroupNames=[self.asg_name]
) )
group = response["AutoScalingGroups"][0] group = response["AutoScalingGroups"][0]
group["Instances"].should.have.length_of(3) assert len(group["Instances"]) == 3
instance_to_standby_exit_standby.should.be.within( assert instance_to_standby_exit_standby in [
[x["InstanceId"] for x in group["Instances"]] x["InstanceId"] for x in group["Instances"]
) ]
group["DesiredCapacity"].should.equal(3) assert group["DesiredCapacity"] == 3
response = ec2_client.describe_instances( instance = ec2_client.describe_instances(
InstanceIds=[instance_to_standby_exit_standby] InstanceIds=[instance_to_standby_exit_standby]
) )["Reservations"][0]["Instances"][0]
response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal( assert instance["State"]["Name"] == "running"
"running"
)
response = self.elb_client.describe_load_balancers( response = self.elb_client.describe_load_balancers(
LoadBalancerNames=[self.lb_name] LoadBalancerNames=[self.lb_name]
) )
list( assert len(list(response["LoadBalancerDescriptions"][0]["Instances"])) == 3
response["LoadBalancerDescriptions"][0]["Instances"] assert instance_to_standby_exit_standby in [
).should.have.length_of(3)
instance_to_standby_exit_standby.should.be.within(
[
x["InstanceId"] x["InstanceId"]
for x in response["LoadBalancerDescriptions"][0]["Instances"] for x in response["LoadBalancerDescriptions"][0]["Instances"]
] ]
)
@mock_autoscaling @mock_autoscaling
@ -921,7 +854,7 @@ class TestAutoScalingInstancesProtected(TestCase):
response = self.as_client.attach_load_balancers( response = self.as_client.attach_load_balancers(
AutoScalingGroupName=self.asg_name, LoadBalancerNames=[self.lb_name] AutoScalingGroupName=self.asg_name, LoadBalancerNames=[self.lb_name]
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
def test_attach_one_instance(self): def test_attach_one_instance(self):
ec2 = boto3.resource("ec2", "us-east-1") ec2 = boto3.resource("ec2", "us-east-1")
@ -935,22 +868,20 @@ class TestAutoScalingInstancesProtected(TestCase):
response = self.as_client.attach_instances( response = self.as_client.attach_instances(
AutoScalingGroupName=self.asg_name, InstanceIds=instances_to_add AutoScalingGroupName=self.asg_name, InstanceIds=instances_to_add
) )
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
AutoScalingGroupNames=[self.asg_name] AutoScalingGroupNames=[self.asg_name]
) )
instances = response["AutoScalingGroups"][0]["Instances"] instances = response["AutoScalingGroups"][0]["Instances"]
instances.should.have.length_of(3) assert len(instances) == 3
for instance in instances: for instance in instances:
instance["ProtectedFromScaleIn"].should.equal(True) assert instance["ProtectedFromScaleIn"] is True
response = self.elb_client.describe_load_balancers( response = self.elb_client.describe_load_balancers(
LoadBalancerNames=[self.lb_name] LoadBalancerNames=[self.lb_name]
) )
list( assert len(list(response["LoadBalancerDescriptions"][0]["Instances"])) == 3
response["LoadBalancerDescriptions"][0]["Instances"]
).should.have.length_of(3)
@mock_autoscaling @mock_autoscaling
@ -991,7 +922,6 @@ class TestAutoScalingTerminateInstances(TestCase):
) )
def test_terminate_instance_in_auto_scaling_group_decrement(self): def test_terminate_instance_in_auto_scaling_group_decrement(self):
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
AutoScalingGroupNames=[self.asg_name] AutoScalingGroupNames=[self.asg_name]
) )
@ -1006,15 +936,13 @@ class TestAutoScalingTerminateInstances(TestCase):
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
AutoScalingGroupNames=[self.asg_name] AutoScalingGroupNames=[self.asg_name]
) )
response["AutoScalingGroups"][0]["Instances"].should.equal([]) assert response["AutoScalingGroups"][0]["Instances"] == []
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(0) assert response["AutoScalingGroups"][0]["DesiredCapacity"] == 0
response = self.elb_client.describe_load_balancers( response = self.elb_client.describe_load_balancers(
LoadBalancerNames=[self.lb_name] LoadBalancerNames=[self.lb_name]
) )
list( assert len(list(response["LoadBalancerDescriptions"][0]["Instances"])) == 0
response["LoadBalancerDescriptions"][0]["Instances"]
).should.have.length_of(0)
def test_terminate_instance_in_auto_scaling_group_no_decrement(self): def test_terminate_instance_in_auto_scaling_group_no_decrement(self):
response = self.as_client.describe_auto_scaling_groups( response = self.as_client.describe_auto_scaling_groups(
@ -1035,18 +963,14 @@ class TestAutoScalingTerminateInstances(TestCase):
instance["InstanceId"] instance["InstanceId"]
for instance in response["AutoScalingGroups"][0]["Instances"] for instance in response["AutoScalingGroups"][0]["Instances"]
) )
replaced_instance_id.should_not.equal(original_instance_id) assert replaced_instance_id != original_instance_id
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(1) assert response["AutoScalingGroups"][0]["DesiredCapacity"] == 1
response = self.elb_client.describe_load_balancers( response = self.elb_client.describe_load_balancers(
LoadBalancerNames=[self.lb_name] LoadBalancerNames=[self.lb_name]
) )
list( assert len(list(response["LoadBalancerDescriptions"][0]["Instances"])) == 1
response["LoadBalancerDescriptions"][0]["Instances"] assert original_instance_id not in [
).should.have.length_of(1)
original_instance_id.shouldnt.be.within(
[
x["InstanceId"] x["InstanceId"]
for x in response["LoadBalancerDescriptions"][0]["Instances"] for x in response["LoadBalancerDescriptions"][0]["Instances"]
] ]
)

19
tests/test_autoscaling/test_elbv2.py
View File

@ -1,5 +1,4 @@
import boto3 import boto3
import sure # noqa # pylint: disable=unused-import
import unittest import unittest
from moto import mock_autoscaling, mock_elbv2 from moto import mock_autoscaling, mock_elbv2
from tests import EXAMPLE_AMI_ID from tests import EXAMPLE_AMI_ID
@ -53,14 +52,12 @@ class TestAutoscalignELBv2(unittest.TestCase):
response = self.as_client.describe_load_balancer_target_groups( response = self.as_client.describe_load_balancer_target_groups(
AutoScalingGroupName=self.asg_name AutoScalingGroupName=self.asg_name
) )
list(response["LoadBalancerTargetGroups"]).should.have.length_of(1) assert len(list(response["LoadBalancerTargetGroups"])) == 1
response = self.elbv2_client.describe_target_health( response = self.elbv2_client.describe_target_health(
TargetGroupArn=self.target_group_arn TargetGroupArn=self.target_group_arn
) )
list(response["TargetHealthDescriptions"]).should.have.length_of( assert len(list(response["TargetHealthDescriptions"])) == self.instance_count
self.instance_count
)
def test_attach_detach_target_groups(self): def test_attach_detach_target_groups(self):
# create asg without attaching to target group # create asg without attaching to target group
@ -81,8 +78,8 @@ class TestAutoscalignELBv2(unittest.TestCase):
response = self.elbv2_client.describe_target_health( response = self.elbv2_client.describe_target_health(
TargetGroupArn=self.target_group_arn TargetGroupArn=self.target_group_arn
) )
list(response["TargetHealthDescriptions"]).should.have.length_of( assert (
self.instance_count * 2 len(list(response["TargetHealthDescriptions"])) == self.instance_count * 2
) )
response = self.as_client.detach_load_balancer_target_groups( response = self.as_client.detach_load_balancer_target_groups(
@ -91,9 +88,7 @@ class TestAutoscalignELBv2(unittest.TestCase):
response = self.elbv2_client.describe_target_health( response = self.elbv2_client.describe_target_health(
TargetGroupArn=self.target_group_arn TargetGroupArn=self.target_group_arn
) )
list(response["TargetHealthDescriptions"]).should.have.length_of( assert len(list(response["TargetHealthDescriptions"])) == self.instance_count
self.instance_count
)
def test_detach_all_target_groups(self): def test_detach_all_target_groups(self):
response = self.as_client.detach_load_balancer_target_groups( response = self.as_client.detach_load_balancer_target_groups(
@ -103,8 +98,8 @@ class TestAutoscalignELBv2(unittest.TestCase):
response = self.elbv2_client.describe_target_health( response = self.elbv2_client.describe_target_health(
TargetGroupArn=self.target_group_arn TargetGroupArn=self.target_group_arn
) )
list(response["TargetHealthDescriptions"]).should.have.length_of(0) assert len(list(response["TargetHealthDescriptions"])) == 0
response = self.as_client.describe_load_balancer_target_groups( response = self.as_client.describe_load_balancer_target_groups(
AutoScalingGroupName=self.asg_name AutoScalingGroupName=self.asg_name
) )
list(response["LoadBalancerTargetGroups"]).should.have.length_of(0) assert len(list(response["LoadBalancerTargetGroups"])) == 0

126
tests/test_autoscaling/test_launch_configurations.py
View File

@ -3,7 +3,6 @@ import boto3
from botocore.exceptions import ClientError from botocore.exceptions import ClientError
import pytest import pytest
import sure # noqa # pylint: disable=unused-import
from moto import mock_autoscaling, mock_ec2 from moto import mock_autoscaling, mock_ec2
from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
@ -26,21 +25,22 @@ def test_create_launch_configuration():
) )
launch_config = client.describe_launch_configurations()["LaunchConfigurations"][0] launch_config = client.describe_launch_configurations()["LaunchConfigurations"][0]
launch_config["LaunchConfigurationName"].should.equal("tester") assert launch_config["LaunchConfigurationName"] == "tester"
launch_config.should.have.key("LaunchConfigurationARN") assert "LaunchConfigurationARN" in launch_config
launch_config["ImageId"].should.equal(EXAMPLE_AMI_ID) assert launch_config["ImageId"] == EXAMPLE_AMI_ID
launch_config["InstanceType"].should.equal("t1.micro") assert launch_config["InstanceType"] == "t1.micro"
launch_config["KeyName"].should.equal("the_keys") assert launch_config["KeyName"] == "the_keys"
set(launch_config["SecurityGroups"]).should.equal(set(["default", "default2"])) assert set(launch_config["SecurityGroups"]) == set(["default", "default2"])
userdata = launch_config["UserData"] userdata = launch_config["UserData"]
userdata = base64.b64decode(userdata) userdata = base64.b64decode(userdata)
userdata.should.equal(b"This is some user_data") assert userdata == b"This is some user_data"
launch_config["InstanceMonitoring"].should.equal({"Enabled": True}) assert launch_config["InstanceMonitoring"] == {"Enabled": True}
launch_config["IamInstanceProfile"].should.equal( assert (
f"arn:aws:iam::{ACCOUNT_ID}:instance-profile/testing" launch_config["IamInstanceProfile"]
== f"arn:aws:iam::{ACCOUNT_ID}:instance-profile/testing"
) )
launch_config["SpotPrice"].should.equal("0.1") assert launch_config["SpotPrice"] == "0.1"
launch_config["BlockDeviceMappings"].should.equal([]) assert launch_config["BlockDeviceMappings"] == []
@mock_autoscaling @mock_autoscaling
@ -75,29 +75,29 @@ def test_create_launch_configuration_with_block_device_mappings():
) )
launch_config = client.describe_launch_configurations()["LaunchConfigurations"][0] launch_config = client.describe_launch_configurations()["LaunchConfigurations"][0]
launch_config["LaunchConfigurationName"].should.equal("tester") assert launch_config["LaunchConfigurationName"] == "tester"
mappings = launch_config["BlockDeviceMappings"] mappings = launch_config["BlockDeviceMappings"]
mappings.should.have.length_of(3) assert len(mappings) == 3
xvdh = [m for m in mappings if m["DeviceName"] == "/dev/xvdh"][0] xvdh = [m for m in mappings if m["DeviceName"] == "/dev/xvdh"][0]
xvdp = [m for m in mappings if m["DeviceName"] == "/dev/xvdp"][0] xvdp = [m for m in mappings if m["DeviceName"] == "/dev/xvdp"][0]
xvdb = [m for m in mappings if m["DeviceName"] == "/dev/xvdb"][0] xvdb = [m for m in mappings if m["DeviceName"] == "/dev/xvdb"][0]
xvdh.shouldnt.have.key("VirtualName") assert "VirtualName" not in xvdh
xvdh.should.have.key("Ebs") assert "Ebs" in xvdh
xvdh["Ebs"]["VolumeSize"].should.equal(100) assert xvdh["Ebs"]["VolumeSize"] == 100
xvdh["Ebs"]["VolumeType"].should.equal("io1") assert xvdh["Ebs"]["VolumeType"] == "io1"
xvdh["Ebs"]["DeleteOnTermination"].should.equal(False) assert xvdh["Ebs"]["DeleteOnTermination"] is False
xvdh["Ebs"]["Iops"].should.equal(1000) assert xvdh["Ebs"]["Iops"] == 1000
xvdp.shouldnt.have.key("VirtualName") assert "VirtualName" not in xvdp
xvdp.should.have.key("Ebs") assert "Ebs" in xvdp
xvdp["Ebs"]["SnapshotId"].should.equal("snap-1234abcd") assert xvdp["Ebs"]["SnapshotId"] == "snap-1234abcd"
xvdp["Ebs"]["VolumeType"].should.equal("standard") assert xvdp["Ebs"]["VolumeType"] == "standard"
xvdb["VirtualName"].should.equal("ephemeral0") assert xvdb["VirtualName"] == "ephemeral0"
xvdb.shouldnt.have.key("Ebs") assert "Ebs" not in xvdb
@mock_autoscaling @mock_autoscaling
@ -119,17 +119,15 @@ def test_create_launch_configuration_additional_parameters():
) )
launch_config = client.describe_launch_configurations()["LaunchConfigurations"][0] launch_config = client.describe_launch_configurations()["LaunchConfigurations"][0]
launch_config["ClassicLinkVPCId"].should.equal("vpc_id") assert launch_config["ClassicLinkVPCId"] == "vpc_id"
launch_config["ClassicLinkVPCSecurityGroups"].should.equal(["classic_sg1"]) assert launch_config["ClassicLinkVPCSecurityGroups"] == ["classic_sg1"]
launch_config["EbsOptimized"].should.equal(True) assert launch_config["EbsOptimized"] is True
launch_config["AssociatePublicIpAddress"].should.equal(True) assert launch_config["AssociatePublicIpAddress"] is True
launch_config["MetadataOptions"].should.equal( assert launch_config["MetadataOptions"] == {
{
"HttpTokens": "optional", "HttpTokens": "optional",
"HttpPutResponseHopLimit": 123, "HttpPutResponseHopLimit": 123,
"HttpEndpoint": "disabled", "HttpEndpoint": "disabled",
} }
)
@mock_autoscaling @mock_autoscaling
@ -151,7 +149,7 @@ def test_create_launch_configuration_without_public_ip():
) )
launch_config = client.describe_launch_configurations()["LaunchConfigurations"][0] launch_config = client.describe_launch_configurations()["LaunchConfigurations"][0]
launch_config["AssociatePublicIpAddress"].should.equal(False) assert launch_config["AssociatePublicIpAddress"] is False
asg_name = f"asg-{random_image_id}" asg_name = f"asg-{random_image_id}"
client.create_auto_scaling_group( client.create_auto_scaling_group(
@ -169,7 +167,7 @@ def test_create_launch_configuration_without_public_ip():
instance = ec2_client.describe_instances(InstanceIds=[instance_id])["Reservations"][ instance = ec2_client.describe_instances(InstanceIds=[instance_id])["Reservations"][
0 0
]["Instances"][0] ]["Instances"][0]
instance.shouldnt.have.key("PublicIpAddress") assert "PublicIpAddress" not in instance
@mock_autoscaling @mock_autoscaling
@ -182,8 +180,8 @@ def test_create_launch_configuration_additional_params_default_to_false():
) )
launch_config = client.describe_launch_configurations()["LaunchConfigurations"][0] launch_config = client.describe_launch_configurations()["LaunchConfigurations"][0]
launch_config["EbsOptimized"].should.equal(False) assert launch_config["EbsOptimized"] is False
launch_config["AssociatePublicIpAddress"].should.equal(False) assert launch_config["AssociatePublicIpAddress"] is False
@mock_autoscaling @mock_autoscaling
@ -200,12 +198,12 @@ def test_create_launch_configuration_defaults():
launch_config = client.describe_launch_configurations()["LaunchConfigurations"][0] launch_config = client.describe_launch_configurations()["LaunchConfigurations"][0]
# Defaults # Defaults
launch_config["KeyName"].should.equal("") assert launch_config["KeyName"] == ""
launch_config["SecurityGroups"].should.equal([]) assert launch_config["SecurityGroups"] == []
launch_config["UserData"].should.equal("") assert launch_config["UserData"] == ""
launch_config["InstanceMonitoring"].should.equal({"Enabled": False}) assert launch_config["InstanceMonitoring"] == {"Enabled": False}
launch_config.shouldnt.have.key("IamInstanceProfile") assert "IamInstanceProfile" not in launch_config
launch_config.shouldnt.have.key("SpotPrice") assert "SpotPrice" not in launch_config
@mock_autoscaling @mock_autoscaling
@ -221,10 +219,8 @@ def test_launch_configuration_describe_filter():
configs = client.describe_launch_configurations( configs = client.describe_launch_configurations(
LaunchConfigurationNames=["tester", "tester2"] LaunchConfigurationNames=["tester", "tester2"]
) )
configs["LaunchConfigurations"].should.have.length_of(2) assert len(configs["LaunchConfigurations"]) == 2
client.describe_launch_configurations()[ assert len(client.describe_launch_configurations()["LaunchConfigurations"]) == 3
"LaunchConfigurations"
].should.have.length_of(3)
@mock_autoscaling @mock_autoscaling
@ -240,13 +236,13 @@ def test_launch_configuration_describe_paginated():
response = conn.describe_launch_configurations() response = conn.describe_launch_configurations()
lcs = response["LaunchConfigurations"] lcs = response["LaunchConfigurations"]
marker = response["NextToken"] marker = response["NextToken"]
lcs.should.have.length_of(50) assert len(lcs) == 50
marker.should.equal(lcs[-1]["LaunchConfigurationName"]) assert marker == lcs[-1]["LaunchConfigurationName"]
response2 = conn.describe_launch_configurations(NextToken=marker) response2 = conn.describe_launch_configurations(NextToken=marker)
lcs.extend(response2["LaunchConfigurations"]) lcs.extend(response2["LaunchConfigurations"])
lcs.should.have.length_of(51) assert len(lcs) == 51
assert "NextToken" not in response2.keys() assert "NextToken" not in response2.keys()
@ -259,15 +255,11 @@ def test_launch_configuration_delete():
InstanceType="m1.small", InstanceType="m1.small",
) )
client.describe_launch_configurations()[ assert len(client.describe_launch_configurations()["LaunchConfigurations"]) == 1
"LaunchConfigurations"
].should.have.length_of(1)
client.delete_launch_configuration(LaunchConfigurationName="tester") client.delete_launch_configuration(LaunchConfigurationName="tester")
client.describe_launch_configurations()[ assert len(client.describe_launch_configurations()["LaunchConfigurations"]) == 0
"LaunchConfigurations"
].should.have.length_of(0)
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -292,10 +284,8 @@ def test_invalid_launch_configuration_request_raises_error(request_params):
client = boto3.client("autoscaling", region_name="us-east-1") client = boto3.client("autoscaling", region_name="us-east-1")
with pytest.raises(ClientError) as ex: with pytest.raises(ClientError) as ex:
client.create_launch_configuration(**request_params) client.create_launch_configuration(**request_params)
ex.value.response["Error"]["Code"].should.equal("ValidationError") assert ex.value.response["Error"]["Code"] == "ValidationError"
ex.value.response["Error"]["Message"].should.match( assert "Valid requests must contain" in ex.value.response["Error"]["Message"]
r"^Valid requests must contain.*"
)
@mock_autoscaling @mock_autoscaling
@ -338,10 +328,10 @@ def test_launch_config_with_block_device_mappings__volumes_are_created():
volumes = ec2_client.describe_volumes( volumes = ec2_client.describe_volumes(
Filters=[{"Name": "attachment.instance-id", "Values": [instance_id]}] Filters=[{"Name": "attachment.instance-id", "Values": [instance_id]}]
)["Volumes"] )["Volumes"]
volumes.should.have.length_of(2) assert len(volumes) == 2
volumes[0].should.have.key("Size").equals(8) assert volumes[0]["Size"] == 8
volumes[0].should.have.key("Encrypted").equals(False) assert volumes[0]["Encrypted"] is False
volumes[0].should.have.key("VolumeType").equals("gp2") assert volumes[0]["VolumeType"] == "gp2"
volumes[1].should.have.key("Size").equals(10) assert volumes[1]["Size"] == 10
volumes[1].should.have.key("Encrypted").equals(False) assert volumes[1]["Encrypted"] is False
volumes[1].should.have.key("VolumeType").equals("standard") assert volumes[1]["VolumeType"] == "standard"

33
tests/test_autoscaling/test_policies.py
View File

@ -1,5 +1,4 @@
import boto3 import boto3
import sure # noqa # pylint: disable=unused-import
import pytest import pytest
from moto import mock_autoscaling from moto import mock_autoscaling
@ -39,11 +38,11 @@ def test_create_policy_boto3():
) )
policy = client.describe_policies()["ScalingPolicies"][0] policy = client.describe_policies()["ScalingPolicies"][0]
policy["PolicyName"].should.equal("ScaleUp") assert policy["PolicyName"] == "ScaleUp"
policy["AdjustmentType"].should.equal("ExactCapacity") assert policy["AdjustmentType"] == "ExactCapacity"
policy["AutoScalingGroupName"].should.equal("tester_group") assert policy["AutoScalingGroupName"] == "tester_group"
policy["ScalingAdjustment"].should.equal(3) assert policy["ScalingAdjustment"] == 3
policy["Cooldown"].should.equal(60) assert policy["Cooldown"] == 60
@mock_autoscaling @mock_autoscaling
@ -58,10 +57,10 @@ def test_create_policy_default_values_boto3():
) )
policy = client.describe_policies()["ScalingPolicies"][0] policy = client.describe_policies()["ScalingPolicies"][0]
policy["PolicyName"].should.equal("ScaleUp") assert policy["PolicyName"] == "ScaleUp"
# Defaults # Defaults
policy["Cooldown"].should.equal(300) assert policy["Cooldown"] == 300
@mock_autoscaling @mock_autoscaling
@ -75,9 +74,9 @@ def test_update_policy_boto3():
ScalingAdjustment=3, ScalingAdjustment=3,
) )
client.describe_policies()["ScalingPolicies"].should.have.length_of(1) assert len(client.describe_policies()["ScalingPolicies"]) == 1
policy = client.describe_policies()["ScalingPolicies"][0] policy = client.describe_policies()["ScalingPolicies"][0]
policy["ScalingAdjustment"].should.equal(3) assert policy["ScalingAdjustment"] == 3
# Now update it by creating another with the same name # Now update it by creating another with the same name
client.put_scaling_policy( client.put_scaling_policy(
@ -86,9 +85,9 @@ def test_update_policy_boto3():
AutoScalingGroupName="tester_group", AutoScalingGroupName="tester_group",
ScalingAdjustment=2, ScalingAdjustment=2,
) )
client.describe_policies()["ScalingPolicies"].should.have.length_of(1) assert len(client.describe_policies()["ScalingPolicies"]) == 1
policy = client.describe_policies()["ScalingPolicies"][0] policy = client.describe_policies()["ScalingPolicies"][0]
policy["ScalingAdjustment"].should.equal(2) assert policy["ScalingAdjustment"] == 2
@mock_autoscaling @mock_autoscaling
@ -102,10 +101,10 @@ def test_delete_policy_boto3():
ScalingAdjustment=3, ScalingAdjustment=3,
) )
client.describe_policies()["ScalingPolicies"].should.have.length_of(1) assert len(client.describe_policies()["ScalingPolicies"]) == 1
client.delete_policy(PolicyName="ScaleUp") client.delete_policy(PolicyName="ScaleUp")
client.describe_policies()["ScalingPolicies"].should.have.length_of(0) assert len(client.describe_policies()["ScalingPolicies"]) == 0
@mock_autoscaling @mock_autoscaling
@ -122,7 +121,7 @@ def test_execute_policy_exact_capacity_boto3():
client.execute_policy(PolicyName="ScaleUp") client.execute_policy(PolicyName="ScaleUp")
instances = client.describe_auto_scaling_instances() instances = client.describe_auto_scaling_instances()
instances["AutoScalingInstances"].should.have.length_of(3) assert len(instances["AutoScalingInstances"]) == 3
@mock_autoscaling @mock_autoscaling
@ -139,7 +138,7 @@ def test_execute_policy_positive_change_in_capacity_boto3():
client.execute_policy(PolicyName="ScaleUp") client.execute_policy(PolicyName="ScaleUp")
instances = client.describe_auto_scaling_instances() instances = client.describe_auto_scaling_instances()
instances["AutoScalingInstances"].should.have.length_of(5) assert len(instances["AutoScalingInstances"]) == 5
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -162,4 +161,4 @@ def test_execute_policy_percent_change_in_capacity_boto3(adjustment, nr_of_insta
client.execute_policy(PolicyName="ScaleUp") client.execute_policy(PolicyName="ScaleUp")
instances = client.describe_auto_scaling_instances() instances = client.describe_auto_scaling_instances()
instances["AutoScalingInstances"].should.have.length_of(nr_of_instances) assert len(instances["AutoScalingInstances"]) == nr_of_instances

6
tests/test_autoscaling/test_server.py
View File

@ -1,5 +1,3 @@
import sure # noqa # pylint: disable=unused-import
import moto.server as server import moto.server as server
""" """
@ -13,5 +11,5 @@ def test_describe_autoscaling_groups():
res = test_client.get("/?Action=DescribeLaunchConfigurations") res = test_client.get("/?Action=DescribeLaunchConfigurations")
res.data.should.contain(b"<DescribeLaunchConfigurationsResponse") assert b"<DescribeLaunchConfigurationsResponse" in res.data
res.data.should.contain(b"<LaunchConfigurations>") assert b"<LaunchConfigurations>" in res.data