Techdebt: Replace sure with regular asserts in ApplicationAutoscaling tests (#6378)

This commit is contained in:
Bert Blommers 2023-06-08 11:32:46 +00:00 committed by GitHub
parent 7c702ee33f
commit 08e509506a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 122 additions and 125 deletions

View File

@ -1,6 +1,7 @@
import re
import boto3
import pytest
import sure # noqa # pylint: disable=unused-import
from moto import mock_applicationautoscaling, mock_ecs
from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
@ -62,15 +63,13 @@ def test_describe_scalable_targets_one_basic_ecs_success():
response = client.describe_scalable_targets(
ServiceNamespace=DEFAULT_SERVICE_NAMESPACE
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
len(response["ScalableTargets"]).should.equal(1)
assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
assert len(response["ScalableTargets"]) == 1
t = response["ScalableTargets"][0]
t.should.have.key("ServiceNamespace").which.should.equal(DEFAULT_SERVICE_NAMESPACE)
t.should.have.key("ResourceId").which.should.equal(DEFAULT_RESOURCE_ID)
t.should.have.key("ScalableDimension").which.should.equal(
DEFAULT_SCALABLE_DIMENSION
)
t.should.have.key("CreationTime").which.should.be.a("datetime.datetime")
assert t["ServiceNamespace"] == DEFAULT_SERVICE_NAMESPACE
assert t["ResourceId"] == DEFAULT_RESOURCE_ID
assert t["ScalableDimension"] == DEFAULT_SCALABLE_DIMENSION
assert "CreationTime" in t
@mock_ecs
@ -83,21 +82,19 @@ def test_describe_scalable_targets_one_full_ecs_success():
response = client.describe_scalable_targets(
ServiceNamespace=DEFAULT_SERVICE_NAMESPACE
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
len(response["ScalableTargets"]).should.equal(1)
assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
assert len(response["ScalableTargets"]) == 1
t = response["ScalableTargets"][0]
t.should.have.key("ServiceNamespace").which.should.equal(DEFAULT_SERVICE_NAMESPACE)
t.should.have.key("ResourceId").which.should.equal(DEFAULT_RESOURCE_ID)
t.should.have.key("ScalableDimension").which.should.equal(
DEFAULT_SCALABLE_DIMENSION
)
t.should.have.key("MinCapacity").which.should.equal(DEFAULT_MIN_CAPACITY)
t.should.have.key("MaxCapacity").which.should.equal(DEFAULT_MAX_CAPACITY)
t.should.have.key("RoleARN").which.should.equal(DEFAULT_ROLE_ARN)
t.should.have.key("CreationTime").which.should.be.a("datetime.datetime")
t.should.have.key("SuspendedState")
t["SuspendedState"]["DynamicScalingInSuspended"].should.equal(
DEFAULT_SUSPENDED_STATE["DynamicScalingInSuspended"]
assert t["ServiceNamespace"] == DEFAULT_SERVICE_NAMESPACE
assert t["ResourceId"] == DEFAULT_RESOURCE_ID
assert t["ScalableDimension"] == DEFAULT_SCALABLE_DIMENSION
assert t["MinCapacity"] == DEFAULT_MIN_CAPACITY
assert t["MaxCapacity"] == DEFAULT_MAX_CAPACITY
assert t["RoleARN"] == DEFAULT_ROLE_ARN
assert "CreationTime" in t
assert (
t["SuspendedState"]["DynamicScalingInSuspended"]
== DEFAULT_SUSPENDED_STATE["DynamicScalingInSuspended"]
)
@ -138,8 +135,8 @@ def test_describe_scalable_targets_only_return_ecs_targets():
response = client.describe_scalable_targets(
ServiceNamespace=DEFAULT_SERVICE_NAMESPACE
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
len(response["ScalableTargets"]).should.equal(2)
assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
assert len(response["ScalableTargets"]) == 2
@mock_ecs
@ -163,17 +160,17 @@ def test_describe_scalable_targets_next_token_success():
response = client.describe_scalable_targets(
ServiceNamespace=DEFAULT_SERVICE_NAMESPACE
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
len(response["ScalableTargets"]).should.equal(50)
response["ScalableTargets"][0]["ResourceId"].should.equal("service/default/0")
response.should.have.key("NextToken").which.should.equal("49")
assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
assert len(response["ScalableTargets"]) == 50
assert response["ScalableTargets"][0]["ResourceId"] == "service/default/0"
assert response["NextToken"] == "49"
response = client.describe_scalable_targets(
ServiceNamespace=DEFAULT_SERVICE_NAMESPACE, NextToken=str(response["NextToken"])
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
len(response["ScalableTargets"]).should.equal(50)
response["ScalableTargets"][0]["ResourceId"].should.equal("service/default/50")
response.should_not.have.key("NextToken")
assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
assert len(response["ScalableTargets"]) == 50
assert response["ScalableTargets"][0]["ResourceId"] == "service/default/50"
assert "NextToken" not in response
def register_scalable_target(client, **kwargs):
@ -192,7 +189,6 @@ def register_scalable_target(client, **kwargs):
@mock_ecs
@mock_applicationautoscaling
def test_register_scalable_target_resource_id_variations():
# Required to register an ECS target in moto
ecs = boto3.client("ecs", region_name=DEFAULT_REGION)
_create_ecs_defaults(ecs)
@ -259,14 +255,14 @@ def test_register_scalable_target_resource_id_variations():
MaxCapacity=8,
)
response = client.describe_scalable_targets(ServiceNamespace=namespace)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
num_targets = 2 if namespace == "dynamodb" and "index" in resource_id else 1
len(response["ScalableTargets"]).should.equal(num_targets)
assert len(response["ScalableTargets"]) == num_targets
t = response["ScalableTargets"][-1]
t.should.have.key("ServiceNamespace").which.should.equal(namespace)
t.should.have.key("ResourceId").which.should.equal(resource_id)
t.should.have.key("ScalableDimension").which.should.equal(scalable_dimension)
t.should.have.key("CreationTime").which.should.be.a("datetime.datetime")
assert t["ServiceNamespace"] == namespace
assert t["ResourceId"] == resource_id
assert t["ScalableDimension"] == scalable_dimension
assert "CreationTime" in t
@mock_ecs
@ -297,19 +293,21 @@ def test_register_scalable_target_updates_existing_target():
ServiceNamespace=DEFAULT_SERVICE_NAMESPACE
)
len(response["ScalableTargets"]).should.equal(1)
assert len(response["ScalableTargets"]) == 1
t = response["ScalableTargets"][0]
t.should.have.key("MinCapacity").which.should.equal(updated_min_capacity)
t.should.have.key("MaxCapacity").which.should.equal(updated_max_capacity)
t.should.have.key("SuspendedState")
t["SuspendedState"]["DynamicScalingInSuspended"].should.equal(
updated_suspended_state["DynamicScalingInSuspended"]
assert t["MinCapacity"] == updated_min_capacity
assert t["MaxCapacity"] == updated_max_capacity
assert (
t["SuspendedState"]["DynamicScalingInSuspended"]
== updated_suspended_state["DynamicScalingInSuspended"]
)
t["SuspendedState"]["DynamicScalingOutSuspended"].should.equal(
updated_suspended_state["DynamicScalingOutSuspended"]
assert (
t["SuspendedState"]["DynamicScalingOutSuspended"]
== updated_suspended_state["DynamicScalingOutSuspended"]
)
t["SuspendedState"]["ScheduledScalingSuspended"].should.equal(
updated_suspended_state["ScheduledScalingSuspended"]
assert (
t["SuspendedState"]["ScheduledScalingSuspended"]
== updated_suspended_state["ScheduledScalingSuspended"]
)
@ -365,8 +363,8 @@ def test_put_scaling_policy(policy_type, policy_body_kwargs):
PolicyType="ABCDEFG",
**policy_body_kwargs,
)
e.value.response["Error"]["Message"].should.match(
r"Unknown policy type .* specified."
assert (
e.value.response["Error"]["Message"] == "Unknown policy type ABCDEFG specified."
)
response = client.put_scaling_policy(
@ -377,9 +375,13 @@ def test_put_scaling_policy(policy_type, policy_body_kwargs):
PolicyType=policy_type,
**policy_body_kwargs,
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response["PolicyARN"].should.match(
rf"arn:aws:autoscaling:.*1:{ACCOUNT_ID}:scalingPolicy:.*:resource/{namespace}/{resource_id}:policyName/{policy_name}"
assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
assert (
re.match(
pattern=rf"arn:aws:autoscaling:{DEFAULT_REGION}:{ACCOUNT_ID}:scalingPolicy:.*:resource/{namespace}/{resource_id}:policyName/{policy_name}",
string=response["PolicyARN"],
)
is not None
)
@ -415,7 +417,7 @@ def test_describe_scaling_policies():
PolicyType=policy_type,
TargetTrackingScalingPolicyConfiguration=policy_body,
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
response = client.describe_scaling_policies(
PolicyNames=[policy_name],
@ -423,18 +425,22 @@ def test_describe_scaling_policies():
ResourceId=resource_id,
ScalableDimension=scalable_dimension,
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
policy = response["ScalingPolicies"][0]
policy["PolicyName"].should.equal(policy_name)
policy["ServiceNamespace"].should.equal(namespace)
policy["ResourceId"].should.equal(resource_id)
policy["ScalableDimension"].should.equal(scalable_dimension)
policy["PolicyType"].should.equal(policy_type)
policy["TargetTrackingScalingPolicyConfiguration"].should.equal(policy_body)
policy["PolicyARN"].should.match(
rf"arn:aws:autoscaling:.*1:{ACCOUNT_ID}:scalingPolicy:.*:resource/{namespace}/{resource_id}:policyName/{policy_name}"
assert policy["PolicyName"] == policy_name
assert policy["ServiceNamespace"] == namespace
assert policy["ResourceId"] == resource_id
assert policy["ScalableDimension"] == scalable_dimension
assert policy["PolicyType"] == policy_type
assert policy["TargetTrackingScalingPolicyConfiguration"] == policy_body
assert (
re.match(
pattern=rf"arn:aws:autoscaling:{DEFAULT_REGION}:{ACCOUNT_ID}:scalingPolicy:.*:resource/{namespace}/{resource_id}:policyName/{policy_name}",
string=policy["PolicyARN"],
)
is not None
)
policy.should.have.key("CreationTime").which.should.be.a("datetime.datetime")
assert "CreationTime" in policy
@mock_applicationautoscaling
@ -468,7 +474,7 @@ def test_delete_scaling_policies():
ResourceId=resource_id,
ScalableDimension=scalable_dimension,
)
e.value.response["Error"]["Message"].should.match(r"No scaling policy found .*")
assert "No scaling policy found" in e.value.response["Error"]["Message"]
response = client.put_scaling_policy(
PolicyName=policy_name,
@ -478,7 +484,7 @@ def test_delete_scaling_policies():
PolicyType=policy_type,
TargetTrackingScalingPolicyConfiguration=policy_body,
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
response = client.delete_scaling_policy(
PolicyName=policy_name,
@ -486,7 +492,7 @@ def test_delete_scaling_policies():
ResourceId=resource_id,
ScalableDimension=scalable_dimension,
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
response = client.describe_scaling_policies(
PolicyNames=[policy_name],
@ -494,8 +500,8 @@ def test_delete_scaling_policies():
ResourceId=resource_id,
ScalableDimension=scalable_dimension,
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
len(response["ScalingPolicies"]).should.equal(0)
assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
assert len(response["ScalingPolicies"]) == 0
@mock_applicationautoscaling
@ -514,7 +520,7 @@ def test_deregister_scalable_target():
)
response = client.describe_scalable_targets(ServiceNamespace=namespace)
len(response["ScalableTargets"]).should.equal(1)
assert len(response["ScalableTargets"]) == 1
client.deregister_scalable_target(
ServiceNamespace=namespace,
@ -523,7 +529,7 @@ def test_deregister_scalable_target():
)
response = client.describe_scalable_targets(ServiceNamespace=namespace)
len(response["ScalableTargets"]).should.equal(0)
assert len(response["ScalableTargets"]) == 0
with pytest.raises(client.exceptions.ValidationException) as e:
client.deregister_scalable_target(
@ -531,14 +537,14 @@ def test_deregister_scalable_target():
ResourceId=resource_id,
ScalableDimension=scalable_dimension,
)
e.value.response["Error"]["Message"].should.match(r"No scalable target found .*")
assert "No scalable target found" in e.value.response["Error"]["Message"]
@mock_applicationautoscaling
def test_delete_scheduled_action():
client = boto3.client("application-autoscaling", region_name="eu-west-1")
resp = client.describe_scheduled_actions(ServiceNamespace="ecs")
resp.should.have.key("ScheduledActions").length_of(0)
assert len(resp["ScheduledActions"]) == 0
for i in range(3):
client.put_scheduled_action(
@ -549,7 +555,7 @@ def test_delete_scheduled_action():
)
resp = client.describe_scheduled_actions(ServiceNamespace="ecs")
resp.should.have.key("ScheduledActions").length_of(3)
assert len(resp["ScheduledActions"]) == 3
client.delete_scheduled_action(
ServiceNamespace="ecs",
@ -559,14 +565,14 @@ def test_delete_scheduled_action():
)
resp = client.describe_scheduled_actions(ServiceNamespace="ecs")
resp.should.have.key("ScheduledActions").length_of(2)
assert len(resp["ScheduledActions"]) == 2
@mock_applicationautoscaling
def test_describe_scheduled_actions():
client = boto3.client("application-autoscaling", region_name="eu-west-1")
resp = client.describe_scheduled_actions(ServiceNamespace="ecs")
resp.should.have.key("ScheduledActions").length_of(0)
assert len(resp["ScheduledActions"]) == 0
for i in range(3):
client.put_scheduled_action(
@ -583,29 +589,29 @@ def test_describe_scheduled_actions():
)
resp = client.describe_scheduled_actions(ServiceNamespace="ecs")
resp.should.have.key("ScheduledActions").length_of(3)
assert len(resp["ScheduledActions"]) == 3
resp = client.describe_scheduled_actions(ServiceNamespace="ec2")
resp.should.have.key("ScheduledActions").length_of(0)
assert len(resp["ScheduledActions"]) == 0
resp = client.describe_scheduled_actions(
ServiceNamespace="dynamodb", ScheduledActionNames=["ddb_action_0"]
)
resp.should.have.key("ScheduledActions").length_of(1)
assert len(resp["ScheduledActions"]) == 1
resp = client.describe_scheduled_actions(
ServiceNamespace="dynamodb",
ResourceId="table/table_0",
ScalableDimension="dynamodb:table:ReadCapacityUnits",
)
resp.should.have.key("ScheduledActions").length_of(1)
assert len(resp["ScheduledActions"]) == 1
resp = client.describe_scheduled_actions(
ServiceNamespace="dynamodb",
ResourceId="table/table_0",
ScalableDimension="dynamodb:table:WriteCapacityUnits",
)
resp.should.have.key("ScheduledActions").length_of(0)
assert len(resp["ScheduledActions"]) == 0
@mock_applicationautoscaling
@ -619,18 +625,19 @@ def test_put_scheduled_action():
)
resp = client.describe_scheduled_actions(ServiceNamespace="ecs")
resp.should.have.key("ScheduledActions").length_of(1)
assert len(resp["ScheduledActions"]) == 1
action = resp["ScheduledActions"][0]
action.should.have.key("ScheduledActionName").equals("action_name")
action.should.have.key("ScheduledActionARN").equals(
f"arn:aws:autoscaling:ap-southeast-1:{ACCOUNT_ID}:scheduledAction:ecs:scheduledActionName/action_name"
assert action["ScheduledActionName"] == "action_name"
assert (
action["ScheduledActionARN"]
== f"arn:aws:autoscaling:ap-southeast-1:{ACCOUNT_ID}:scheduledAction:ecs:scheduledActionName/action_name"
)
action.should.have.key("ServiceNamespace").equals("ecs")
action.should.have.key("ResourceId").equals("ecs:cluster:x")
action.should.have.key("ScalableDimension").equals("ecs:service:DesiredCount")
action.should.have.key("CreationTime")
action.shouldnt.have.key("ScalableTargetAction")
assert action["ServiceNamespace"] == "ecs"
assert action["ResourceId"] == "ecs:cluster:x"
assert action["ScalableDimension"] == "ecs:service:DesiredCount"
assert "CreationTime" in action
assert "ScalableTargetAction" not in action
@mock_applicationautoscaling
@ -654,17 +661,16 @@ def test_put_scheduled_action__use_update():
)
resp = client.describe_scheduled_actions(ServiceNamespace="ecs")
resp.should.have.key("ScheduledActions").length_of(1)
assert len(resp["ScheduledActions"]) == 1
action = resp["ScheduledActions"][0]
action.should.have.key("ScheduledActionName").equals("action_name_updated")
action.should.have.key("ScheduledActionARN").equals(
f"arn:aws:autoscaling:ap-southeast-1:{ACCOUNT_ID}:scheduledAction:ecs:scheduledActionName/action_name"
)
action.should.have.key("ServiceNamespace").equals("ecs")
action.should.have.key("ResourceId").equals("ecs:cluster:x")
action.should.have.key("ScalableDimension").equals("ecs:service:DesiredCount")
action.should.have.key("CreationTime")
action.should.have.key("ScalableTargetAction").equals(
{"MaxCapacity": 23, "MinCapacity": 12}
assert action["ScheduledActionName"] == "action_name_updated"
assert (
action["ScheduledActionARN"]
== f"arn:aws:autoscaling:ap-southeast-1:{ACCOUNT_ID}:scheduledAction:ecs:scheduledActionName/action_name"
)
assert action["ServiceNamespace"] == "ecs"
assert action["ResourceId"] == "ecs:cluster:x"
assert action["ScalableDimension"] == "ecs:service:DesiredCount"
assert "CreationTime" in action
assert action["ScalableTargetAction"] == {"MaxCapacity": 23, "MinCapacity": 12}

View File

@ -3,7 +3,6 @@ from moto import mock_applicationautoscaling, mock_ecs
from moto.applicationautoscaling import models
from moto.applicationautoscaling.exceptions import AWSValidationException
import pytest
import sure # noqa # pylint: disable=unused-import
from botocore.exceptions import ClientError
from .test_applicationautoscaling import register_scalable_target
@ -28,11 +27,9 @@ def test_describe_scalable_targets_with_invalid_scalable_dimension_should_return
ServiceNamespace=DEFAULT_SERVICE_NAMESPACE, ScalableDimension="foo"
)
err = ex.value.response
err["Error"]["Code"].should.equal("ValidationException")
err["Error"]["Message"].split(":")[0].should.look_like(
"1 validation error detected"
)
err["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
assert err["Error"]["Code"] == "ValidationException"
assert "1 validation error detected" in err["Error"]["Message"]
assert err["ResponseMetadata"]["HTTPStatusCode"] == 400
@mock_applicationautoscaling
@ -44,11 +41,9 @@ def test_describe_scalable_targets_with_invalid_service_namespace_should_return_
ServiceNamespace="foo", ScalableDimension=DEFAULT_SCALABLE_DIMENSION
)
err = ex.value.response
err["Error"]["Code"].should.equal("ValidationException")
err["Error"]["Message"].split(":")[0].should.look_like(
"1 validation error detected"
)
err["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
assert err["Error"]["Code"] == "ValidationException"
assert "1 validation error detected" in err["Error"]["Message"]
assert err["ResponseMetadata"]["HTTPStatusCode"] == 400
@mock_applicationautoscaling
@ -60,11 +55,9 @@ def test_describe_scalable_targets_with_multiple_invalid_parameters_should_retur
ServiceNamespace="foo", ScalableDimension="bar"
)
err = ex.value.response
err["Error"]["Code"].should.equal("ValidationException")
err["Error"]["Message"].split(":")[0].should.look_like(
"2 validation errors detected"
)
err["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
assert err["Error"]["Code"] == "ValidationException"
assert "2 validation errors detected" in err["Error"]["Message"]
assert err["ResponseMetadata"]["HTTPStatusCode"] == 400
@mock_ecs
@ -76,9 +69,9 @@ def test_register_scalable_target_ecs_with_non_existent_service_should_return_cl
with pytest.raises(ClientError) as ex:
register_scalable_target(client, ServiceNamespace="ecs", ResourceId=resource_id)
err = ex.value.response
err["Error"]["Code"].should.equal("ClusterNotFoundException")
err["Error"]["Message"].should.equal("Cluster not found.")
err["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
assert err["Error"]["Code"] == "ClusterNotFoundException"
assert err["Error"]["Message"] == "Cluster not found."
assert err["ResponseMetadata"]["HTTPStatusCode"] == 400
@pytest.mark.parametrize(
@ -91,9 +84,7 @@ def test_register_scalable_target_ecs_with_non_existent_service_should_return_cl
)
def test_target_params_are_valid_success(namespace, r_id, dimension, expected):
if expected is True:
models._target_params_are_valid(namespace, r_id, dimension).should.equal(
expected
)
assert models._target_params_are_valid(namespace, r_id, dimension) == expected
else:
with pytest.raises(AWSValidationException):
models._target_params_are_valid(namespace, r_id, dimension)