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

Techdebt: Replace sure with regular asserts in CF (#6481)

Browse Source
This commit is contained in:
Bert Blommers 2023-07-04 15:53:58 +00:00 committed by GitHub
parent 4011a68f06
commit d98d8cf0d5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 1015 additions and 1132 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
tests/test_cloudformation/test_cloudformation_custom_resources.py
View File

@ -1,7 +1,6 @@
import boto3 import boto3
import json import json
import requests import requests
import sure # noqa # pylint: disable=unused-import
import time import time
import pytest import pytest
@ -69,9 +68,9 @@ def test_create_custom_lambda_resource():
# Verify the correct Output was returned # Verify the correct Output was returned
outputs = get_outputs(cf, stack_name) outputs = get_outputs(cf, stack_name)
outputs.should.have.length_of(1) assert len(outputs) == 1
outputs[0].should.have.key("OutputKey").equals("infokey") assert outputs[0]["OutputKey"] == "infokey"
outputs[0].should.have.key("OutputValue").equals("special value") assert outputs[0]["OutputValue"] == "special value"
@mock_cloudformation @mock_cloudformation
@ -104,25 +103,25 @@ def test_create_custom_lambda_resource__verify_cfnresponse_failed():
execution_failed, logs = wait_for_log_msg( execution_failed, logs = wait_for_log_msg(
expected_msg="failed executing http.request", log_group=log_group_name expected_msg="failed executing http.request", log_group=log_group_name
) )
execution_failed.should.equal(True) assert execution_failed is True
printed_events = [ printed_events = [
line for line in logs if line.startswith("{'RequestType': 'Create'") line for line in logs if line.startswith("{'RequestType': 'Create'")
] ]
printed_events.should.have.length_of(1) assert len(printed_events) == 1
original_event = json.loads(printed_events[0].replace("'", '"')) original_event = json.loads(printed_events[0].replace("'", '"'))
original_event.should.have.key("RequestType").equals("Create") assert original_event["RequestType"] == "Create"
original_event.should.have.key("ServiceToken") # Should equal Lambda ARN assert "ServiceToken" in original_event # Should equal Lambda ARN
original_event.should.have.key("ResponseURL") assert "ResponseURL" in original_event
original_event.should.have.key("StackId") assert "StackId" in original_event
original_event.should.have.key("RequestId") # type UUID assert "RequestId" in original_event # type UUID
original_event.should.have.key("LogicalResourceId").equals("CustomInfo") assert original_event["LogicalResourceId"] == "CustomInfo"
original_event.should.have.key("ResourceType").equals("Custom::Info") assert original_event["ResourceType"] == "Custom::Info"
original_event.should.have.key("ResourceProperties") assert "ResourceProperties" in original_event
original_event["ResourceProperties"].should.have.key( assert (
"ServiceToken" "ServiceToken" in original_event["ResourceProperties"]
) # Should equal Lambda ARN ) # Should equal Lambda ARN
original_event["ResourceProperties"].should.have.key("MyProperty").equals("stuff") assert original_event["ResourceProperties"]["MyProperty"] == "stuff"
@mock_cloudformation @mock_cloudformation
@ -155,8 +154,8 @@ def test_create_custom_lambda_resource__verify_manual_request():
) )
stack_id = stack["StackId"] stack_id = stack["StackId"]
stack = cf.describe_stacks(StackName=stack_id)["Stacks"][0] stack = cf.describe_stacks(StackName=stack_id)["Stacks"][0]
stack["Outputs"].should.equal([]) assert stack["Outputs"] == []
stack["StackStatus"].should.equal("CREATE_IN_PROGRESS") assert stack["StackStatus"] == "CREATE_IN_PROGRESS"
callback_url = f"http://cloudformation.{region_name}.amazonaws.com/cloudformation_{region_name}/cfnresponse?stack={stack_id}" callback_url = f"http://cloudformation.{region_name}.amazonaws.com/cloudformation_{region_name}/cfnresponse?stack={stack_id}"
data = { data = {
@ -168,10 +167,10 @@ def test_create_custom_lambda_resource__verify_manual_request():
requests.post(callback_url, json=data) requests.post(callback_url, json=data)
stack = cf.describe_stacks(StackName=stack_id)["Stacks"][0] stack = cf.describe_stacks(StackName=stack_id)["Stacks"][0]
stack["StackStatus"].should.equal("CREATE_COMPLETE") assert stack["StackStatus"] == "CREATE_COMPLETE"
stack["Outputs"].should.equal( assert stack["Outputs"] == [
[{"OutputKey": "infokey", "OutputValue": "resultfromthirdpartysystem"}] {"OutputKey": "infokey", "OutputValue": "resultfromthirdpartysystem"}
) ]
@mock_cloudformation @mock_cloudformation
@ -186,9 +185,10 @@ def test_create_custom_lambda_resource__unknown_arn():
Capabilities=["CAPABILITY_IAM"], Capabilities=["CAPABILITY_IAM"],
) )
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("ValidationError") assert err["Code"] == "ValidationError"
err["Message"].should.equal( assert (
"Template error: instance of Fn::GetAtt references undefined resource InfoFunction" err["Message"]
== "Template error: instance of Fn::GetAtt references undefined resource InfoFunction"
) )

61
tests/test_cloudformation/test_cloudformation_multi_accounts.py
View File

@ -94,11 +94,11 @@ class TestStackSetMultipleAccounts(TestCase):
StackInstanceAccount=accnt, StackInstanceAccount=accnt,
StackInstanceRegion=region or "us-east-1", StackInstanceRegion=region or "us-east-1",
)["StackInstance"] )["StackInstance"]
resp.should.have.key("Account").equals(accnt) assert resp["Account"] == accnt
def _verify_queues(self, accnt, expected, region=None): def _verify_queues(self, accnt, expected, region=None):
list(sqs_backends[accnt][region or "us-east-1"].queues.keys()).should.equal( assert (
expected list(sqs_backends[accnt][region or "us-east-1"].queues.keys()) == expected
) )
@ -122,9 +122,10 @@ class TestServiceManagedStacks(TestStackSetMultipleAccounts):
StackSetName=self.name, Accounts=["888781156701"], Regions=["us-east-1"] StackSetName=self.name, Accounts=["888781156701"], Regions=["us-east-1"]
) )
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("ValidationError") assert err["Code"] == "ValidationError"
err["Message"].should.equal( assert (
"StackSets with SERVICE_MANAGED permission model can only have OrganizationalUnit as target" err["Message"]
== "StackSets with SERVICE_MANAGED permission model can only have OrganizationalUnit as target"
) )
def test_create_instances__specifying_only_accounts_in_deployment_targets(self): def test_create_instances__specifying_only_accounts_in_deployment_targets(self):
@ -137,8 +138,8 @@ class TestServiceManagedStacks(TestStackSetMultipleAccounts):
Regions=["us-east-1"], Regions=["us-east-1"],
) )
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("ValidationError") assert err["Code"] == "ValidationError"
err["Message"].should.equal("OrganizationalUnitIds are required") assert err["Message"] == "OrganizationalUnitIds are required"
def test_create_instances___with_invalid_ou(self): def test_create_instances___with_invalid_ou(self):
with pytest.raises(ClientError) as exc: with pytest.raises(ClientError) as exc:
@ -148,9 +149,10 @@ class TestServiceManagedStacks(TestStackSetMultipleAccounts):
Regions=["us-east-1"], Regions=["us-east-1"],
) )
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("ValidationError") assert err["Code"] == "ValidationError"
err["Message"].should.equal( assert (
"1 validation error detected: Value '[unknown]' at 'deploymentTargets.organizationalUnitIds' failed to satisfy constraint: Member must satisfy constraint: [Member must have length less than or equal to 68, Member must have length greater than or equal to 6, Member must satisfy regular expression pattern: ^(ou-[a-z0-9]{4,32}-[a-z0-9]{8,32}|r-[a-z0-9]{4,32})$]" err["Message"]
== "1 validation error detected: Value '[unknown]' at 'deploymentTargets.organizationalUnitIds' failed to satisfy constraint: Member must satisfy constraint: [Member must have length less than or equal to 68, Member must have length greater than or equal to 6, Member must satisfy regular expression pattern: ^(ou-[a-z0-9]{4,32}-[a-z0-9]{8,32}|r-[a-z0-9]{4,32})$]"
) )
def test_create_instances__single_ou(self): def test_create_instances__single_ou(self):
@ -210,12 +212,12 @@ class TestSelfManagedStacks(TestStackSetMultipleAccounts):
self._verify_stack_instance(self.acct01, region="us-east-2") self._verify_stack_instance(self.acct01, region="us-east-2")
# acct01 has a Stack # acct01 has a Stack
cf_backends[self.acct01]["us-east-2"].stacks.should.have.length_of(1) assert len(cf_backends[self.acct01]["us-east-2"].stacks) == 1
# Other acounts do not # Other acounts do not
cf_backends[self.acct02]["us-east-2"].stacks.should.have.length_of(0) assert len(cf_backends[self.acct02]["us-east-2"].stacks) == 0
cf_backends[self.acct21]["us-east-2"].stacks.should.have.length_of(0) assert len(cf_backends[self.acct21]["us-east-2"].stacks) == 0
cf_backends[self.acct22]["us-east-2"].stacks.should.have.length_of(0) assert len(cf_backends[self.acct22]["us-east-2"].stacks) == 0
cf_backends[DEFAULT_ACCOUNT_ID]["us-east-2"].stacks.should.have.length_of(0) assert len(cf_backends[DEFAULT_ACCOUNT_ID]["us-east-2"].stacks) == 0
# acct01 has a queue # acct01 has a queue
self._verify_queues(self.acct01, ["testqueue"], region="us-east-2") self._verify_queues(self.acct01, ["testqueue"], region="us-east-2")
@ -233,12 +235,12 @@ class TestSelfManagedStacks(TestStackSetMultipleAccounts):
) )
# acct01 and 02 have a Stack # acct01 and 02 have a Stack
cf_backends[self.acct01]["us-east-2"].stacks.should.have.length_of(1) assert len(cf_backends[self.acct01]["us-east-2"].stacks) == 1
cf_backends[self.acct02]["us-east-2"].stacks.should.have.length_of(1) assert len(cf_backends[self.acct02]["us-east-2"].stacks) == 1
# Other acounts do not # Other acounts do not
cf_backends[self.acct21]["us-east-2"].stacks.should.have.length_of(0) assert len(cf_backends[self.acct21]["us-east-2"].stacks) == 0
cf_backends[self.acct22]["us-east-2"].stacks.should.have.length_of(0) assert len(cf_backends[self.acct22]["us-east-2"].stacks) == 0
cf_backends[DEFAULT_ACCOUNT_ID]["us-east-2"].stacks.should.have.length_of(0) assert len(cf_backends[DEFAULT_ACCOUNT_ID]["us-east-2"].stacks) == 0
# acct01 and 02 have the queue # acct01 and 02 have the queue
self._verify_queues(self.acct01, ["testqueue"], region="us-east-2") self._verify_queues(self.acct01, ["testqueue"], region="us-east-2")
@ -265,12 +267,12 @@ class TestSelfManagedStacks(TestStackSetMultipleAccounts):
) )
# Act02 still has it's stack # Act02 still has it's stack
cf_backends[self.acct02]["us-east-2"].stacks.should.have.length_of(1) assert len(cf_backends[self.acct02]["us-east-2"].stacks) == 1
# Other Stacks are removed # Other Stacks are removed
cf_backends[self.acct01]["us-east-2"].stacks.should.have.length_of(0) assert len(cf_backends[self.acct01]["us-east-2"].stacks) == 0
cf_backends[self.acct21]["us-east-2"].stacks.should.have.length_of(0) assert len(cf_backends[self.acct21]["us-east-2"].stacks) == 0
cf_backends[self.acct22]["us-east-2"].stacks.should.have.length_of(0) assert len(cf_backends[self.acct22]["us-east-2"].stacks) == 0
cf_backends[DEFAULT_ACCOUNT_ID]["us-east-2"].stacks.should.have.length_of(0) assert len(cf_backends[DEFAULT_ACCOUNT_ID]["us-east-2"].stacks) == 0
# Acct02 still has it's queue as well # Acct02 still has it's queue as well
self._verify_queues(self.acct02, ["testqueue"], region="us-east-2") self._verify_queues(self.acct02, ["testqueue"], region="us-east-2")
@ -288,7 +290,8 @@ class TestSelfManagedStacks(TestStackSetMultipleAccounts):
Regions=["us-east-1"], Regions=["us-east-1"],
) )
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("ValidationError") assert err["Code"] == "ValidationError"
err["Message"].should.equal( assert (
"StackSets with SELF_MANAGED permission model can only have accounts as target" err["Message"]
== "StackSets with SELF_MANAGED permission model can only have accounts as target"
) )

1206
tests/test_cloudformation/test_cloudformation_stack_crud_boto3.py
View File

File diff suppressed because it is too large Load Diff

477
tests/test_cloudformation/test_cloudformation_stack_integration.py
View File

@ -1,13 +1,11 @@
import boto3
import json import json
import io import io
import pytest
import zipfile import zipfile
from decimal import Decimal
from botocore.exceptions import ClientError from botocore.exceptions import ClientError
import boto3 from decimal import Decimal
import sure # noqa # pylint: disable=unused-import
import pytest
from string import Template from string import Template
from moto import ( from moto import (
@ -29,9 +27,6 @@ from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
from tests import EXAMPLE_AMI_ID, EXAMPLE_AMI_ID2 from tests import EXAMPLE_AMI_ID, EXAMPLE_AMI_ID2
from tests.markers import requires_docker from tests.markers import requires_docker
from tests.test_cloudformation.fixtures import fn_join, single_instance_with_ebs_volume from tests.test_cloudformation.fixtures import fn_join, single_instance_with_ebs_volume
from tests.helpers import ( # noqa # pylint: disable=unused-import
containing_item_with_attributes,
)
@mock_cloudformation @mock_cloudformation
@ -41,8 +36,8 @@ def test_create_template_without_required_param_boto3():
with pytest.raises(ClientError) as ex: with pytest.raises(ClientError) as ex:
cf.create_stack(StackName="test_stack", TemplateBody=template_json) cf.create_stack(StackName="test_stack", TemplateBody=template_json)
err = ex.value.response["Error"] err = ex.value.response["Error"]
err.should.have.key("Code").equal("Missing Parameter") assert err["Code"] == "Missing Parameter"
err.should.have.key("Message").equal("Missing parameter KeyName") assert err["Message"] == "Missing parameter KeyName"
@mock_ec2 @mock_ec2
@ -56,7 +51,7 @@ def test_fn_join_boto3():
stack = cf.describe_stacks()["Stacks"][0] stack = cf.describe_stacks()["Stacks"][0]
fn_join_output = stack["Outputs"][0] fn_join_output = stack["Outputs"][0]
fn_join_output["OutputValue"].should.equal(f"test eip:{eip['PublicIp']}") assert fn_join_output["OutputValue"] == f"test eip:{eip['PublicIp']}"
@mock_cloudformation @mock_cloudformation
@ -85,14 +80,14 @@ def test_conditional_resources_boto3():
Parameters=[{"ParameterKey": "EnvType", "ParameterValue": "staging"}], Parameters=[{"ParameterKey": "EnvType", "ParameterValue": "staging"}],
) )
sqs = boto3.client("sqs", region_name="us-west-1") sqs = boto3.client("sqs", region_name="us-west-1")
sqs.list_queues().shouldnt.have.key("QueueUrls") assert "QueueUrls" not in sqs.list_queues()
cf.create_stack( cf.create_stack(
StackName="test_stack_with_queue", StackName="test_stack_with_queue",
TemplateBody=sqs_template_json, TemplateBody=sqs_template_json,
Parameters=[{"ParameterKey": "EnvType", "ParameterValue": "prod"}], Parameters=[{"ParameterKey": "EnvType", "ParameterValue": "prod"}],
) )
sqs.list_queues()["QueueUrls"].should.have.length_of(1) assert len(sqs.list_queues()["QueueUrls"]) == 1
@mock_cloudformation @mock_cloudformation
@ -126,7 +121,7 @@ def test_conditional_if_handling_boto3():
cf.create_stack(StackName="test_stack", TemplateBody=dummy_template_json) cf.create_stack(StackName="test_stack", TemplateBody=dummy_template_json)
ec2 = boto3.client("ec2", region_name="us-west-1") ec2 = boto3.client("ec2", region_name="us-west-1")
ec2_instance = ec2.describe_instances()["Reservations"][0]["Instances"][0] ec2_instance = ec2.describe_instances()["Reservations"][0]["Instances"][0]
ec2_instance["ImageId"].should.equal(EXAMPLE_AMI_ID2) assert ec2_instance["ImageId"] == EXAMPLE_AMI_ID2
cf = boto3.client("cloudformation", region_name="us-west-2") cf = boto3.client("cloudformation", region_name="us-west-2")
cf.create_stack( cf.create_stack(
@ -136,7 +131,7 @@ def test_conditional_if_handling_boto3():
) )
ec2 = boto3.client("ec2", region_name="us-west-2") ec2 = boto3.client("ec2", region_name="us-west-2")
ec2_instance = ec2.describe_instances()["Reservations"][0]["Instances"][0] ec2_instance = ec2.describe_instances()["Reservations"][0]["Instances"][0]
ec2_instance["ImageId"].should.equal(EXAMPLE_AMI_ID) assert ec2_instance["ImageId"] == EXAMPLE_AMI_ID
@mock_cloudformation @mock_cloudformation
@ -172,13 +167,13 @@ def test_cloudformation_mapping_boto3():
cf.create_stack(StackName="test_stack1", TemplateBody=dummy_template_json) cf.create_stack(StackName="test_stack1", TemplateBody=dummy_template_json)
ec2 = boto3.client("ec2", region_name="us-east-1") ec2 = boto3.client("ec2", region_name="us-east-1")
ec2_instance = ec2.describe_instances()["Reservations"][0]["Instances"][0] ec2_instance = ec2.describe_instances()["Reservations"][0]["Instances"][0]
ec2_instance["ImageId"].should.equal(EXAMPLE_AMI_ID) assert ec2_instance["ImageId"] == EXAMPLE_AMI_ID
cf = boto3.client("cloudformation", region_name="us-west-1") cf = boto3.client("cloudformation", region_name="us-west-1")
cf.create_stack(StackName="test_stack1", TemplateBody=dummy_template_json) cf.create_stack(StackName="test_stack1", TemplateBody=dummy_template_json)
ec2 = boto3.client("ec2", region_name="us-west-1") ec2 = boto3.client("ec2", region_name="us-west-1")
ec2_instance = ec2.describe_instances()["Reservations"][0]["Instances"][0] ec2_instance = ec2.describe_instances()["Reservations"][0]["Instances"][0]
ec2_instance["ImageId"].should.equal(EXAMPLE_AMI_ID2) assert ec2_instance["ImageId"] == EXAMPLE_AMI_ID2
@mock_cloudformation @mock_cloudformation
@ -232,23 +227,23 @@ def lambda_handler(event, context):
conn = boto3.client("lambda", "us-east-1") conn = boto3.client("lambda", "us-east-1")
result = conn.list_functions() result = conn.list_functions()
result["Functions"].should.have.length_of(1) assert len(result["Functions"]) == 1
result["Functions"][0]["Description"].should.equal("Test function") assert result["Functions"][0]["Description"] == "Test function"
result["Functions"][0]["Handler"].should.equal("index.lambda_handler") assert result["Functions"][0]["Handler"] == "index.lambda_handler"
result["Functions"][0]["MemorySize"].should.equal(128) assert result["Functions"][0]["MemorySize"] == 128
result["Functions"][0]["Runtime"].should.equal("python2.7") assert result["Functions"][0]["Runtime"] == "python2.7"
result["Functions"][0]["Environment"].should.equal( assert result["Functions"][0]["Environment"] == {
{"Variables": {"TEST_ENV_KEY": "test-env-val"}} "Variables": {"TEST_ENV_KEY": "test-env-val"}
) }
function_name = result["Functions"][0]["FunctionName"] function_name = result["Functions"][0]["FunctionName"]
result = conn.get_function(FunctionName=function_name) result = conn.get_function(FunctionName=function_name)
result["Concurrency"]["ReservedConcurrentExecutions"].should.equal(10) assert result["Concurrency"]["ReservedConcurrentExecutions"] == 10
response = conn.invoke(FunctionName=function_name) response = conn.invoke(FunctionName=function_name)
result = json.loads(response["Payload"].read()) result = json.loads(response["Payload"].read())
result.should.equal({"event": "{}"}) assert result == {"event": "{}"}
def _make_zipfile(func_str): def _make_zipfile(func_str):
@ -302,18 +297,16 @@ def lambda_handler(event, context):
layer_name = result["Layers"][0]["LayerName"] layer_name = result["Layers"][0]["LayerName"]
result = lambda_conn.list_layer_versions(LayerName=layer_name) result = lambda_conn.list_layer_versions(LayerName=layer_name)
result["LayerVersions"][0].pop("CreatedDate") result["LayerVersions"][0].pop("CreatedDate")
result["LayerVersions"].should.equal( assert result["LayerVersions"] == [
[ {
{ "Version": 1,
"Version": 1, "LayerVersionArn": f"arn:aws:lambda:{region}:{ACCOUNT_ID}:layer:{layer_name}:1",
"LayerVersionArn": f"arn:aws:lambda:{region}:{ACCOUNT_ID}:layer:{layer_name}:1", "CompatibleRuntimes": ["python2.7", "python3.6"],
"CompatibleRuntimes": ["python2.7", "python3.6"], "Description": "Test Layer",
"Description": "Test Layer", "LicenseInfo": "MIT",
"LicenseInfo": "MIT", "CompatibleArchitectures": [],
"CompatibleArchitectures": [], }
} ]
]
)
@mock_cloudformation @mock_cloudformation
@ -370,14 +363,13 @@ def test_nat_gateway():
route_resource = resource route_resource = resource
result = ec2_conn.describe_nat_gateways() result = ec2_conn.describe_nat_gateways()
result["NatGateways"].should.have.length_of(1) assert len(result["NatGateways"]) == 1
result["NatGateways"][0]["VpcId"].should.equal(vpc_id) assert result["NatGateways"][0]["VpcId"] == vpc_id
result["NatGateways"][0]["SubnetId"].should.equal(subnet_id) assert result["NatGateways"][0]["SubnetId"] == subnet_id
result["NatGateways"][0]["State"].should.equal("available") assert result["NatGateways"][0]["State"] == "available"
result["NatGateways"][0]["NatGatewayId"].should.equal( physical_id = nat_gateway_resource.get("PhysicalResourceId")
nat_gateway_resource.get("PhysicalResourceId") assert result["NatGateways"][0]["NatGatewayId"] == physical_id
) assert "rtb-" in route_resource["PhysicalResourceId"]
route_resource.get("PhysicalResourceId").should.contain("rtb-")
@mock_cloudformation() @mock_cloudformation()
@ -403,11 +395,11 @@ def test_stack_kms():
kms_conn = boto3.client("kms", "us-east-1") kms_conn = boto3.client("kms", "us-east-1")
keys = kms_conn.list_keys()["Keys"] keys = kms_conn.list_keys()["Keys"]
len(keys).should.equal(1) assert len(keys) == 1
result = kms_conn.describe_key(KeyId=keys[0]["KeyId"]) result = kms_conn.describe_key(KeyId=keys[0]["KeyId"])
result["KeyMetadata"]["Enabled"].should.equal(True) assert result["KeyMetadata"]["Enabled"] is True
result["KeyMetadata"]["KeyUsage"].should.equal("ENCRYPT_DECRYPT") assert result["KeyMetadata"]["KeyUsage"] == "ENCRYPT_DECRYPT"
@mock_cloudformation() @mock_cloudformation()
@ -467,34 +459,32 @@ def test_stack_spot_fleet():
)["StackId"] )["StackId"]
stack_resources = cf_conn.list_stack_resources(StackName=stack_id) stack_resources = cf_conn.list_stack_resources(StackName=stack_id)
stack_resources["StackResourceSummaries"].should.have.length_of(1) assert len(stack_resources["StackResourceSummaries"]) == 1
spot_fleet_id = stack_resources["StackResourceSummaries"][0]["PhysicalResourceId"] spot_fleet_id = stack_resources["StackResourceSummaries"][0]["PhysicalResourceId"]
spot_fleet_requests = conn.describe_spot_fleet_requests( spot_fleet_requests = conn.describe_spot_fleet_requests(
SpotFleetRequestIds=[spot_fleet_id] SpotFleetRequestIds=[spot_fleet_id]
)["SpotFleetRequestConfigs"] )["SpotFleetRequestConfigs"]
len(spot_fleet_requests).should.equal(1) assert len(spot_fleet_requests) == 1
spot_fleet_request = spot_fleet_requests[0] spot_fleet_request = spot_fleet_requests[0]
spot_fleet_request["SpotFleetRequestState"].should.equal("active") assert spot_fleet_request["SpotFleetRequestState"] == "active"
spot_fleet_config = spot_fleet_request["SpotFleetRequestConfig"] spot_fleet_config = spot_fleet_request["SpotFleetRequestConfig"]
spot_fleet_config["SpotPrice"].should.equal("0.12") assert spot_fleet_config["SpotPrice"] == "0.12"
spot_fleet_config["TargetCapacity"].should.equal(6) assert spot_fleet_config["TargetCapacity"] == 6
spot_fleet_config["IamFleetRole"].should.equal( assert spot_fleet_config["IamFleetRole"] == f"arn:aws:iam::{ACCOUNT_ID}:role/fleet"
f"arn:aws:iam::{ACCOUNT_ID}:role/fleet" assert spot_fleet_config["AllocationStrategy"] == "diversified"
) assert spot_fleet_config["FulfilledCapacity"] == 6.0
spot_fleet_config["AllocationStrategy"].should.equal("diversified")
spot_fleet_config["FulfilledCapacity"].should.equal(6.0)
len(spot_fleet_config["LaunchSpecifications"]).should.equal(2) assert len(spot_fleet_config["LaunchSpecifications"]) == 2
launch_spec = spot_fleet_config["LaunchSpecifications"][0] launch_spec = spot_fleet_config["LaunchSpecifications"][0]
launch_spec["EbsOptimized"].should.equal(False) assert launch_spec["EbsOptimized"] is False
launch_spec["ImageId"].should.equal(EXAMPLE_AMI_ID) assert launch_spec["ImageId"] == EXAMPLE_AMI_ID
launch_spec["InstanceType"].should.equal("t2.small") assert launch_spec["InstanceType"] == "t2.small"
launch_spec["SubnetId"].should.equal(subnet_id) assert launch_spec["SubnetId"] == subnet_id
launch_spec["SpotPrice"].should.equal("0.13") assert launch_spec["SpotPrice"] == "0.13"
launch_spec["WeightedCapacity"].should.equal(2.0) assert launch_spec["WeightedCapacity"] == 2.0
@mock_cloudformation() @mock_cloudformation()
@ -551,19 +541,19 @@ def test_stack_spot_fleet_should_figure_out_default_price():
)["StackId"] )["StackId"]
stack_resources = cf_conn.list_stack_resources(StackName=stack_id) stack_resources = cf_conn.list_stack_resources(StackName=stack_id)
stack_resources["StackResourceSummaries"].should.have.length_of(1) assert len(stack_resources["StackResourceSummaries"]) == 1
spot_fleet_id = stack_resources["StackResourceSummaries"][0]["PhysicalResourceId"] spot_fleet_id = stack_resources["StackResourceSummaries"][0]["PhysicalResourceId"]
spot_fleet_requests = conn.describe_spot_fleet_requests( spot_fleet_requests = conn.describe_spot_fleet_requests(
SpotFleetRequestIds=[spot_fleet_id] SpotFleetRequestIds=[spot_fleet_id]
)["SpotFleetRequestConfigs"] )["SpotFleetRequestConfigs"]
len(spot_fleet_requests).should.equal(1) assert len(spot_fleet_requests) == 1
spot_fleet_request = spot_fleet_requests[0] spot_fleet_request = spot_fleet_requests[0]
spot_fleet_request["SpotFleetRequestState"].should.equal("active") assert spot_fleet_request["SpotFleetRequestState"] == "active"
spot_fleet_config = spot_fleet_request["SpotFleetRequestConfig"] spot_fleet_config = spot_fleet_request["SpotFleetRequestConfig"]
assert "SpotPrice" not in spot_fleet_config assert "SpotPrice" not in spot_fleet_config
len(spot_fleet_config["LaunchSpecifications"]).should.equal(2) assert len(spot_fleet_config["LaunchSpecifications"]) == 2
launch_spec1 = spot_fleet_config["LaunchSpecifications"][0] launch_spec1 = spot_fleet_config["LaunchSpecifications"][0]
launch_spec2 = spot_fleet_config["LaunchSpecifications"][1] launch_spec2 = spot_fleet_config["LaunchSpecifications"][1]
@ -575,7 +565,6 @@ def test_stack_spot_fleet_should_figure_out_default_price():
@mock_elbv2 @mock_elbv2
@mock_cloudformation @mock_cloudformation
def test_invalid_action_type_listener_rule(): def test_invalid_action_type_listener_rule():
invalid_listener_template = { invalid_listener_template = {
"AWSTemplateFormatVersion": "2010-09-09", "AWSTemplateFormatVersion": "2010-09-09",
"Resources": { "Resources": {
@ -634,9 +623,10 @@ def test_invalid_action_type_listener_rule():
listener_template_json = json.dumps(invalid_listener_template) listener_template_json = json.dumps(invalid_listener_template)
cfn_conn = boto3.client("cloudformation", "us-west-1") cfn_conn = boto3.client("cloudformation", "us-west-1")
cfn_conn.create_stack.when.called_with( with pytest.raises(ClientError) as exc:
StackName="listener_stack", TemplateBody=listener_template_json cfn_conn.create_stack(StackName="s", TemplateBody=listener_template_json)
).should.throw(ClientError) err = exc.value.response["Error"]
assert err["Code"] == "ValidationError"
@mock_ec2 @mock_ec2
@ -644,7 +634,6 @@ def test_invalid_action_type_listener_rule():
@mock_cloudformation @mock_cloudformation
@mock_events @mock_events
def test_update_stack_listener_and_rule(): def test_update_stack_listener_and_rule():
initial_template = { initial_template = {
"AWSTemplateFormatVersion": "2010-09-09", "AWSTemplateFormatVersion": "2010-09-09",
"Resources": { "Resources": {
@ -731,15 +720,11 @@ def test_update_stack_listener_and_rule():
listeners = elbv2_conn.describe_listeners( listeners = elbv2_conn.describe_listeners(
LoadBalancerArn=load_balancers[0]["LoadBalancerArn"] LoadBalancerArn=load_balancers[0]["LoadBalancerArn"]
)["Listeners"] )["Listeners"]
listeners[0]["Port"].should.equal(90) assert listeners[0]["Port"] == 90
listener_rule = elbv2_conn.describe_rules(ListenerArn=listeners[0]["ListenerArn"])[ l_rule = elbv2_conn.describe_rules(ListenerArn=listeners[0]["ListenerArn"])["Rules"]
"Rules"
]
listener_rule[0]["Conditions"].should.equal( assert l_rule[0]["Conditions"] == [{"Field": "host-header", "Values": ["*"]}]
[{"Field": "host-header", "Values": ["*"]}]
)
@mock_ec2 @mock_ec2
@ -887,96 +872,88 @@ def test_stack_elbv2_resources_integration():
elbv2_conn = boto3.client("elbv2", "us-west-1") elbv2_conn = boto3.client("elbv2", "us-west-1")
load_balancers = elbv2_conn.describe_load_balancers()["LoadBalancers"] lbs = elbv2_conn.describe_load_balancers()["LoadBalancers"]
len(load_balancers).should.equal(1) assert len(lbs) == 1
load_balancers[0]["LoadBalancerName"].should.equal("myelbv2") assert lbs[0]["LoadBalancerName"] == "myelbv2"
load_balancers[0]["Scheme"].should.equal("internet-facing") assert lbs[0]["Scheme"] == "internet-facing"
load_balancers[0]["Type"].should.equal("application") assert lbs[0]["Type"] == "application"
load_balancers[0]["IpAddressType"].should.equal("ipv4") assert lbs[0]["IpAddressType"] == "ipv4"
target_groups = sorted( target_groups = elbv2_conn.describe_target_groups()["TargetGroups"]
elbv2_conn.describe_target_groups()["TargetGroups"], # sort to do comparison with indexes
key=lambda tg: tg["TargetGroupName"], target_groups = sorted(target_groups, key=lambda tg: tg["TargetGroupName"])
) # sort to do comparison with indexes assert len(target_groups) == 2
len(target_groups).should.equal(2) assert target_groups[0]["HealthCheckIntervalSeconds"] == 30
target_groups[0]["HealthCheckIntervalSeconds"].should.equal(30) assert target_groups[0]["HealthCheckPath"] == "/status"
target_groups[0]["HealthCheckPath"].should.equal("/status") assert target_groups[0]["HealthCheckPort"] == "80"
target_groups[0]["HealthCheckPort"].should.equal("80") assert target_groups[0]["HealthCheckProtocol"] == "HTTP"
target_groups[0]["HealthCheckProtocol"].should.equal("HTTP") assert target_groups[0]["HealthCheckTimeoutSeconds"] == 5
target_groups[0]["HealthCheckTimeoutSeconds"].should.equal(5) assert target_groups[0]["HealthyThresholdCount"] == 30
target_groups[0]["HealthyThresholdCount"].should.equal(30) assert target_groups[0]["UnhealthyThresholdCount"] == 5
target_groups[0]["UnhealthyThresholdCount"].should.equal(5) assert target_groups[0]["Matcher"] == {"HttpCode": "200,201"}
target_groups[0]["Matcher"].should.equal({"HttpCode": "200,201"}) assert target_groups[0]["TargetGroupName"] == "mytargetgroup1"
target_groups[0]["TargetGroupName"].should.equal("mytargetgroup1") assert target_groups[0]["Port"] == 80
target_groups[0]["Port"].should.equal(80) assert target_groups[0]["Protocol"] == "HTTP"
target_groups[0]["Protocol"].should.equal("HTTP") assert target_groups[0]["TargetType"] == "instance"
target_groups[0]["TargetType"].should.equal("instance")
target_groups[1]["HealthCheckIntervalSeconds"].should.equal(30) assert target_groups[1]["HealthCheckIntervalSeconds"] == 30
target_groups[1]["HealthCheckPath"].should.equal("/status") assert target_groups[1]["HealthCheckPath"] == "/status"
target_groups[1]["HealthCheckPort"].should.equal("8080") assert target_groups[1]["HealthCheckPort"] == "8080"
target_groups[1]["HealthCheckProtocol"].should.equal("HTTP") assert target_groups[1]["HealthCheckProtocol"] == "HTTP"
target_groups[1]["HealthCheckTimeoutSeconds"].should.equal(5) assert target_groups[1]["HealthCheckTimeoutSeconds"] == 5
target_groups[1]["HealthyThresholdCount"].should.equal(30) assert target_groups[1]["HealthyThresholdCount"] == 30
target_groups[1]["UnhealthyThresholdCount"].should.equal(5) assert target_groups[1]["UnhealthyThresholdCount"] == 5
target_groups[1]["Matcher"].should.equal({"HttpCode": "200"}) assert target_groups[1]["Matcher"] == {"HttpCode": "200"}
target_groups[1]["TargetGroupName"].should.equal("mytargetgroup2") assert target_groups[1]["TargetGroupName"] == "mytargetgroup2"
target_groups[1]["Port"].should.equal(8080) assert target_groups[1]["Port"] == 8080
target_groups[1]["Protocol"].should.equal("HTTP") assert target_groups[1]["Protocol"] == "HTTP"
target_groups[1]["TargetType"].should.equal("instance") assert target_groups[1]["TargetType"] == "instance"
listeners = elbv2_conn.describe_listeners( lstnrs = elbv2_conn.describe_listeners(LoadBalancerArn=lbs[0]["LoadBalancerArn"])[
LoadBalancerArn=load_balancers[0]["LoadBalancerArn"] "Listeners"
)["Listeners"] ]
len(listeners).should.equal(1) assert len(lstnrs) == 1
listeners[0]["LoadBalancerArn"].should.equal(load_balancers[0]["LoadBalancerArn"]) assert lstnrs[0]["LoadBalancerArn"] == lbs[0]["LoadBalancerArn"]
listeners[0]["Port"].should.equal(80) assert lstnrs[0]["Port"] == 80
listeners[0]["Protocol"].should.equal("HTTP") assert lstnrs[0]["Protocol"] == "HTTP"
listeners[0]["DefaultActions"].should.equal( assert lstnrs[0]["DefaultActions"] == [
[{"Type": "forward", "TargetGroupArn": target_groups[0]["TargetGroupArn"]}] {"Type": "forward", "TargetGroupArn": target_groups[0]["TargetGroupArn"]}
)
listener_rule = elbv2_conn.describe_rules(ListenerArn=listeners[0]["ListenerArn"])[
"Rules"
] ]
len(listener_rule).should.equal(3)
listener_rule[0]["Priority"].should.equal("2")
listener_rule[0]["Actions"].should.equal(
[
{
"Type": "forward",
"ForwardConfig": {
"TargetGroups": [
{
"TargetGroupArn": target_groups[1]["TargetGroupArn"],
"Weight": 1,
},
{
"TargetGroupArn": target_groups[0]["TargetGroupArn"],
"Weight": 2,
},
],
"TargetGroupStickinessConfig": {"Enabled": False},
},
}
],
[{"Type": "forward", "TargetGroupArn": target_groups[1]["TargetGroupArn"]}],
)
listener_rule[0]["Conditions"].should.equal(
[{"Field": "path-pattern", "Values": ["/*"]}]
)
listener_rule[1]["Priority"].should.equal("30") rule = elbv2_conn.describe_rules(ListenerArn=lstnrs[0]["ListenerArn"])["Rules"]
listener_rule[1]["Actions"].should.equal( assert len(rule) == 3
[{"Type": "forward", "TargetGroupArn": target_groups[1]["TargetGroupArn"]}] assert rule[0]["Priority"] == "2"
) assert rule[0]["Actions"] == [
listener_rule[1]["Conditions"].should.equal( {
[{"Field": "host-header", "Values": ["example.com"]}] "Type": "forward",
) "ForwardConfig": {
"TargetGroups": [
{
"TargetGroupArn": target_groups[1]["TargetGroupArn"],
"Weight": 1,
},
{
"TargetGroupArn": target_groups[0]["TargetGroupArn"],
"Weight": 2,
},
],
"TargetGroupStickinessConfig": {"Enabled": False},
},
}
]
assert rule[0]["Conditions"] == [{"Field": "path-pattern", "Values": ["/*"]}]
assert rule[1]["Priority"] == "30"
assert rule[1]["Actions"] == [
{"Type": "forward", "TargetGroupArn": target_groups[1]["TargetGroupArn"]}
]
assert rule[1]["Conditions"] == [
{"Field": "host-header", "Values": ["example.com"]}
]
# test outputs # test outputs
stacks = cfn_conn.describe_stacks(StackName="elb_stack")["Stacks"] stacks = cfn_conn.describe_stacks(StackName="elb_stack")["Stacks"]
len(stacks).should.equal(1) assert len(stacks) == 1
dns = list( dns = list(
filter(lambda item: item["OutputKey"] == "albdns", stacks[0]["Outputs"]) filter(lambda item: item["OutputKey"] == "albdns", stacks[0]["Outputs"])
@ -985,8 +962,8 @@ def test_stack_elbv2_resources_integration():
filter(lambda item: item["OutputKey"] == "albname", stacks[0]["Outputs"]) filter(lambda item: item["OutputKey"] == "albname", stacks[0]["Outputs"])
)[0] )[0]
dns["OutputValue"].should.equal(load_balancers[0]["DNSName"]) assert dns["OutputValue"] == lbs[0]["DNSName"]
name["OutputValue"].should.equal(load_balancers[0]["LoadBalancerName"]) assert name["OutputValue"] == lbs[0]["LoadBalancerName"]
@mock_dynamodb @mock_dynamodb
@ -1073,13 +1050,14 @@ def test_stack_dynamodb_resources_integration():
dynamodb_client = boto3.client("dynamodb", region_name="us-east-1") dynamodb_client = boto3.client("dynamodb", region_name="us-east-1")
table_desc = dynamodb_client.describe_table(TableName="myTableName")["Table"] table_desc = dynamodb_client.describe_table(TableName="myTableName")["Table"]
table_desc["StreamSpecification"].should.equal( assert table_desc["StreamSpecification"] == {
{"StreamEnabled": True, "StreamViewType": "KEYS_ONLY"} "StreamEnabled": True,
) "StreamViewType": "KEYS_ONLY",
}
dynamodb_conn = boto3.resource("dynamodb", region_name="us-east-1") dynamodb_conn = boto3.resource("dynamodb", region_name="us-east-1")
table = dynamodb_conn.Table("myTableName") table = dynamodb_conn.Table("myTableName")
table.name.should.equal("myTableName") assert table.name == "myTableName"
table.put_item( table.put_item(
Item={"Album": "myAlbum", "Artist": "myArtist", "Sales": 10, "NumberOfSongs": 5} Item={"Album": "myAlbum", "Artist": "myArtist", "Sales": 10, "NumberOfSongs": 5}
@ -1087,10 +1065,10 @@ def test_stack_dynamodb_resources_integration():
response = table.get_item(Key={"Album": "myAlbum", "Artist": "myArtist"}) response = table.get_item(Key={"Album": "myAlbum", "Artist": "myArtist"})
response["Item"]["Album"].should.equal("myAlbum") assert response["Item"]["Album"] == "myAlbum"
response["Item"]["Sales"].should.equal(Decimal("10")) assert response["Item"]["Sales"] == Decimal("10")
response["Item"]["NumberOfSongs"].should.equal(Decimal("5")) assert response["Item"]["NumberOfSongs"] == Decimal("5")
response["Item"]["Album"].should.equal("myAlbum") assert response["Item"]["Album"] == "myAlbum"
@mock_cloudformation @mock_cloudformation
@ -1133,7 +1111,7 @@ def test_create_log_group_using_fntransform():
logs_conn = boto3.client("logs", region_name="us-west-2") logs_conn = boto3.client("logs", region_name="us-west-2")
log_group = logs_conn.describe_log_groups()["logGroups"][0] log_group = logs_conn.describe_log_groups()["logGroups"][0]
log_group["logGroupName"].should.equal("some-log-group") assert log_group["logGroupName"] == "some-log-group"
@mock_cloudformation @mock_cloudformation
@ -1188,27 +1166,30 @@ def test_create_cloudwatch_logs_resource_policy():
logs_conn = boto3.client("logs", region_name="us-east-1") logs_conn = boto3.client("logs", region_name="us-east-1")
policies = logs_conn.describe_resource_policies()["resourcePolicies"] policies = logs_conn.describe_resource_policies()["resourcePolicies"]
policies.should.have.length_of(1) assert len(policies) == 1
policies.should.be.containing_item_with_attributes(
policyName="TestPolicyA", policyDocument=policy_document assert policies[0]["policyName"] == "TestPolicyA"
) assert policies[0]["policyDocument"] == policy_document
cf_conn.update_stack(StackName="test_stack", TemplateBody=json.dumps(template2)) cf_conn.update_stack(StackName="test_stack", TemplateBody=json.dumps(template2))
policies = logs_conn.describe_resource_policies()["resourcePolicies"] policies = logs_conn.describe_resource_policies()["resourcePolicies"]
policies.should.have.length_of(2) assert len(policies) == 2
policies.should.be.containing_item_with_attributes(
policyName="TestPolicyB", policyDocument=policy_document policy_b = [pol for pol in policies if pol["policyName"] == "TestPolicyB"][0][
) "policyDocument"
policies.should.be.containing_item_with_attributes( ]
policyName="TestPolicyC", policyDocument=policy_document assert policy_b == policy_document
)
policy_c = [pol for pol in policies if pol["policyName"] == "TestPolicyC"][0][
"policyDocument"
]
assert policy_c == policy_document
cf_conn.update_stack(StackName="test_stack", TemplateBody=json.dumps(template1)) cf_conn.update_stack(StackName="test_stack", TemplateBody=json.dumps(template1))
policies = logs_conn.describe_resource_policies()["resourcePolicies"] policies = logs_conn.describe_resource_policies()["resourcePolicies"]
policies.should.have.length_of(1) assert len(policies) == 1
policies.should.be.containing_item_with_attributes( assert policies[0]["policyName"] == "TestPolicyA"
policyName="TestPolicyA", policyDocument=policy_document assert policies[0]["policyDocument"] == policy_document
)
@mock_cloudformation @mock_cloudformation
@ -1232,11 +1213,11 @@ def test_delete_stack_containing_cloudwatch_logs_resource_policy():
logs_conn = boto3.client("logs", region_name="us-east-1") logs_conn = boto3.client("logs", region_name="us-east-1")
policies = logs_conn.describe_resource_policies()["resourcePolicies"] policies = logs_conn.describe_resource_policies()["resourcePolicies"]
policies.should.have.length_of(1) assert len(policies) == 1
cf_conn.delete_stack(StackName="test_stack") cf_conn.delete_stack(StackName="test_stack")
policies = logs_conn.describe_resource_policies()["resourcePolicies"] policies = logs_conn.describe_resource_policies()["resourcePolicies"]
policies.should.have.length_of(0) assert len(policies) == 0
@mock_cloudformation @mock_cloudformation
@ -1261,12 +1242,10 @@ def test_delete_stack_with_deletion_policy_boto3():
TemplateBody=sqs_template_json, TemplateBody=sqs_template_json,
) )
sqs = boto3.client("sqs", region_name="us-west-1") sqs = boto3.client("sqs", region_name="us-west-1")
sqs.list_queues()["QueueUrls"].should.have.length_of(1) assert len(sqs.list_queues()["QueueUrls"]) == 1
cf.delete_stack( cf.delete_stack(StackName="test_stack")
StackName="test_stack", assert len(sqs.list_queues()["QueueUrls"]) == 1
)
sqs.list_queues()["QueueUrls"].should.have.length_of(1)
@mock_cloudformation @mock_cloudformation
@ -1291,10 +1270,10 @@ def test_stack_events_create_rule_integration():
) )
rules = boto3.client("events", "us-west-2").list_rules() rules = boto3.client("events", "us-west-2").list_rules()
rules["Rules"].should.have.length_of(1) assert len(rules["Rules"]) == 1
rules["Rules"][0]["Name"].should.equal("quick-fox") assert rules["Rules"][0]["Name"] == "quick-fox"
rules["Rules"][0]["State"].should.equal("ENABLED") assert rules["Rules"][0]["State"] == "ENABLED"
rules["Rules"][0]["ScheduleExpression"].should.equal("rate(5 minutes)") assert rules["Rules"][0]["ScheduleExpression"] == "rate(5 minutes)"
@mock_cloudformation @mock_cloudformation
@ -1319,12 +1298,12 @@ def test_stack_events_delete_rule_integration():
) )
rules = boto3.client("events", "us-west-2").list_rules() rules = boto3.client("events", "us-west-2").list_rules()
rules["Rules"].should.have.length_of(1) assert len(rules["Rules"]) == 1
cf_conn.delete_stack(StackName="test_stack") cf_conn.delete_stack(StackName="test_stack")
rules = boto3.client("events", "us-west-2").list_rules() rules = boto3.client("events", "us-west-2").list_rules()
rules["Rules"].should.have.length_of(0) assert len(rules["Rules"]) == 0
@mock_cloudformation @mock_cloudformation
@ -1348,7 +1327,7 @@ def test_stack_events_create_rule_without_name_integration():
) )
rules = boto3.client("events", "us-west-2").list_rules() rules = boto3.client("events", "us-west-2").list_rules()
rules["Rules"][0]["Name"].should.contain("test_stack-Event-") assert "test_stack-Event-" in rules["Rules"][0]["Name"]
@mock_cloudformation @mock_cloudformation
@ -1382,10 +1361,10 @@ def test_stack_events_create_rule_as_target():
rules = boto3.client("events", "us-west-2").list_rules() rules = boto3.client("events", "us-west-2").list_rules()
log_groups = boto3.client("logs", "us-west-2").describe_log_groups() log_groups = boto3.client("logs", "us-west-2").describe_log_groups()
rules["Rules"][0]["Name"].should.contain("test_stack-Event-") assert "test_stack-Event-" in rules["Rules"][0]["Name"]
log_groups["logGroups"][0]["logGroupName"].should.equal(rules["Rules"][0]["Arn"]) assert log_groups["logGroups"][0]["logGroupName"] == rules["Rules"][0]["Arn"]
log_groups["logGroups"][0]["retentionInDays"].should.equal(3) assert log_groups["logGroups"][0]["retentionInDays"] == 3
@mock_cloudformation @mock_cloudformation
@ -1413,18 +1392,18 @@ def test_stack_events_update_rule_integration():
cf_conn.create_stack(StackName="test_stack", TemplateBody=original_template) cf_conn.create_stack(StackName="test_stack", TemplateBody=original_template)
rules = boto3.client("events", "us-west-2").list_rules() rules = boto3.client("events", "us-west-2").list_rules()
rules["Rules"].should.have.length_of(1) assert len(rules["Rules"]) == 1
rules["Rules"][0]["Name"].should.equal("Foo") assert rules["Rules"][0]["Name"] == "Foo"
rules["Rules"][0]["State"].should.equal("ENABLED") assert rules["Rules"][0]["State"] == "ENABLED"
update_template = events_template.substitute(Name="Bar", State="DISABLED") update_template = events_template.substitute(Name="Bar", State="DISABLED")
cf_conn.update_stack(StackName="test_stack", TemplateBody=update_template) cf_conn.update_stack(StackName="test_stack", TemplateBody=update_template)
rules = boto3.client("events", "us-west-2").list_rules() rules = boto3.client("events", "us-west-2").list_rules()
rules["Rules"].should.have.length_of(1) assert len(rules["Rules"]) == 1
rules["Rules"][0]["Name"].should.equal("Bar") assert rules["Rules"][0]["Name"] == "Bar"
rules["Rules"][0]["State"].should.equal("DISABLED") assert rules["Rules"][0]["State"] == "DISABLED"
@mock_cloudformation @mock_cloudformation
@ -1519,8 +1498,8 @@ def test_stack_eventbus_create_from_cfn_integration():
NamePrefix="MyCustom" NamePrefix="MyCustom"
) )
event_buses["EventBuses"].should.have.length_of(1) assert len(event_buses["EventBuses"]) == 1
event_buses["EventBuses"][0]["Name"].should.equal("MyCustomEventBus") assert event_buses["EventBuses"][0]["Name"] == "MyCustomEventBus"
@mock_cloudformation @mock_cloudformation
@ -1543,14 +1522,14 @@ def test_stack_events_delete_eventbus_integration():
event_buses = boto3.client("events", "us-west-2").list_event_buses( event_buses = boto3.client("events", "us-west-2").list_event_buses(
NamePrefix="MyCustom" NamePrefix="MyCustom"
) )
event_buses["EventBuses"].should.have.length_of(1) assert len(event_buses["EventBuses"]) == 1
cf_conn.delete_stack(StackName="test_stack") cf_conn.delete_stack(StackName="test_stack")
event_buses = boto3.client("events", "us-west-2").list_event_buses( event_buses = boto3.client("events", "us-west-2").list_event_buses(
NamePrefix="MyCustom" NamePrefix="MyCustom"
) )
event_buses["EventBuses"].should.have.length_of(0) assert len(event_buses["EventBuses"]) == 0
@mock_cloudformation @mock_cloudformation
@ -1580,7 +1559,7 @@ def test_stack_events_delete_from_cfn_integration():
original_event_buses = boto3.client("events", "us-west-2").list_event_buses( original_event_buses = boto3.client("events", "us-west-2").list_event_buses(
NamePrefix="MyCustom" NamePrefix="MyCustom"
) )
original_event_buses["EventBuses"].should.have.length_of(1) assert len(original_event_buses["EventBuses"]) == 1
original_eventbus = original_event_buses["EventBuses"][0] original_eventbus = original_event_buses["EventBuses"][0]
@ -1592,8 +1571,8 @@ def test_stack_events_delete_from_cfn_integration():
update_event_buses = boto3.client("events", "us-west-2").list_event_buses( update_event_buses = boto3.client("events", "us-west-2").list_event_buses(
NamePrefix="AnotherEventBus" NamePrefix="AnotherEventBus"
) )
update_event_buses["EventBuses"].should.have.length_of(1) assert len(update_event_buses["EventBuses"]) == 1
update_event_buses["EventBuses"][0]["Arn"].shouldnt.equal(original_eventbus["Arn"]) assert update_event_buses["EventBuses"][0]["Arn"] != original_eventbus["Arn"]
@mock_cloudformation @mock_cloudformation
@ -1621,7 +1600,7 @@ def test_stack_events_update_from_cfn_integration():
original_event_buses = boto3.client("events", "us-west-2").list_event_buses( original_event_buses = boto3.client("events", "us-west-2").list_event_buses(
NamePrefix="MyCustom" NamePrefix="MyCustom"
) )
original_event_buses["EventBuses"].should.have.length_of(1) assert len(original_event_buses["EventBuses"]) == 1
original_eventbus = original_event_buses["EventBuses"][0] original_eventbus = original_event_buses["EventBuses"][0]
@ -1631,9 +1610,9 @@ def test_stack_events_update_from_cfn_integration():
update_event_buses = boto3.client("events", "us-west-2").list_event_buses( update_event_buses = boto3.client("events", "us-west-2").list_event_buses(
NamePrefix="NewEventBus" NamePrefix="NewEventBus"
) )
update_event_buses["EventBuses"].should.have.length_of(1) assert len(update_event_buses["EventBuses"]) == 1
update_event_buses["EventBuses"][0]["Name"].should.equal("NewEventBus") assert update_event_buses["EventBuses"][0]["Name"] == "NewEventBus"
update_event_buses["EventBuses"][0]["Arn"].shouldnt.equal(original_eventbus["Arn"]) assert update_event_buses["EventBuses"][0]["Arn"] != original_eventbus["Arn"]
@mock_cloudformation @mock_cloudformation
@ -1668,8 +1647,8 @@ def test_stack_events_get_attribute_integration():
event_bus = events.list_event_buses(NamePrefix="MyEventBus")["EventBuses"][0] event_bus = events.list_event_buses(NamePrefix="MyEventBus")["EventBuses"][0]
output_arn["OutputValue"].should.equal(event_bus["Arn"]) assert output_arn["OutputValue"] == event_bus["Arn"]
output_name["OutputValue"].should.equal(event_bus["Name"]) assert output_name["OutputValue"] == event_bus["Name"]
@mock_cloudformation @mock_cloudformation
@ -1699,13 +1678,13 @@ def test_dynamodb_table_creation():
# Verify the TableName is part of the outputs # Verify the TableName is part of the outputs
stack = cfn.describe_stacks(StackName=stack_name)["Stacks"][0] stack = cfn.describe_stacks(StackName=stack_name)["Stacks"][0]
outputs = stack["Outputs"] outputs = stack["Outputs"]
outputs.should.have.length_of(1) assert len(outputs) == 1
outputs[0]["OutputKey"].should.equal("MyTableName") assert outputs[0]["OutputKey"] == "MyTableName"
outputs[0]["OutputValue"].should.contain("foobar") assert "foobar" in outputs[0]["OutputValue"]
# Assert the table is created # Assert the table is created
ddb = boto3.client("dynamodb", "us-west-2") ddb = boto3.client("dynamodb", "us-west-2")
table_names = ddb.list_tables()["TableNames"] table_names = ddb.list_tables()["TableNames"]
table_names.should.equal([outputs[0]["OutputValue"]]) assert table_names == [outputs[0]["OutputValue"]]
@mock_cloudformation @mock_cloudformation
@ -1735,15 +1714,15 @@ def test_ssm_parameter():
stack_resources = cfn.list_stack_resources(StackName=stack_name) stack_resources = cfn.list_stack_resources(StackName=stack_name)
ssm_resource = stack_resources.get("StackResourceSummaries")[0] ssm_resource = stack_resources.get("StackResourceSummaries")[0]
ssm_resource.get("PhysicalResourceId").should.equal("test_ssm") assert ssm_resource.get("PhysicalResourceId") == "test_ssm"
ssm_client = boto3.client("ssm", region_name="us-west-2") ssm_client = boto3.client("ssm", region_name="us-west-2")
parameters = ssm_client.get_parameters(Names=["test_ssm"], WithDecryption=False)[ parameters = ssm_client.get_parameters(Names=["test_ssm"], WithDecryption=False)[
"Parameters" "Parameters"
] ]
parameters.should.have.length_of(1) assert len(parameters) == 1
parameters[0]["Name"].should.equal("test_ssm") assert parameters[0]["Name"] == "test_ssm"
parameters[0]["Value"].should.equal("Test SSM Parameter") assert parameters[0]["Value"] == "Test SSM Parameter"
@mock_cloudformation @mock_cloudformation
@ -1775,9 +1754,9 @@ def test_ssm_parameter_update_stack():
parameters = ssm_client.get_parameters(Names=["test_ssm"], WithDecryption=False)[ parameters = ssm_client.get_parameters(Names=["test_ssm"], WithDecryption=False)[
"Parameters" "Parameters"
] ]
parameters.should.have.length_of(1) assert len(parameters) == 1
parameters[0]["Name"].should.equal("test_ssm") assert parameters[0]["Name"] == "test_ssm"
parameters[0]["Value"].should.equal("Test SSM Parameter") assert parameters[0]["Value"] == "Test SSM Parameter"
parameter_template["Resources"]["BasicParameter"]["Properties"][ parameter_template["Resources"]["BasicParameter"]["Properties"][
"Value" "Value"
@ -1788,9 +1767,9 @@ def test_ssm_parameter_update_stack():
parameters = ssm_client.get_parameters(Names=["test_ssm"], WithDecryption=False)[ parameters = ssm_client.get_parameters(Names=["test_ssm"], WithDecryption=False)[
"Parameters" "Parameters"
] ]
parameters.should.have.length_of(1) assert len(parameters) == 1
parameters[0]["Name"].should.equal("test_ssm") assert parameters[0]["Name"] == "test_ssm"
parameters[0]["Value"].should.equal("Test SSM Parameter Updated") assert parameters[0]["Value"] == "Test SSM Parameter Updated"
@mock_cloudformation @mock_cloudformation
@ -1822,9 +1801,9 @@ def test_ssm_parameter_update_stack_and_remove_resource():
parameters = ssm_client.get_parameters(Names=["test_ssm"], WithDecryption=False)[ parameters = ssm_client.get_parameters(Names=["test_ssm"], WithDecryption=False)[
"Parameters" "Parameters"
] ]
parameters.should.have.length_of(1) assert len(parameters) == 1
parameters[0]["Name"].should.equal("test_ssm") assert parameters[0]["Name"] == "test_ssm"
parameters[0]["Value"].should.equal("Test SSM Parameter") assert parameters[0]["Value"] == "Test SSM Parameter"
parameter_template["Resources"].pop("BasicParameter") parameter_template["Resources"].pop("BasicParameter")
cfn.update_stack(StackName=stack_name, TemplateBody=json.dumps(parameter_template)) cfn.update_stack(StackName=stack_name, TemplateBody=json.dumps(parameter_template))
@ -1833,7 +1812,7 @@ def test_ssm_parameter_update_stack_and_remove_resource():
parameters = ssm_client.get_parameters(Names=["test_ssm"], WithDecryption=False)[ parameters = ssm_client.get_parameters(Names=["test_ssm"], WithDecryption=False)[
"Parameters" "Parameters"
] ]
parameters.should.have.length_of(0) assert len(parameters) == 0
@mock_cloudformation @mock_cloudformation
@ -1851,7 +1830,7 @@ def test_ssm_parameter_update_stack_and_add_resource():
parameters = ssm_client.get_parameters(Names=["test_ssm"], WithDecryption=False)[ parameters = ssm_client.get_parameters(Names=["test_ssm"], WithDecryption=False)[
"Parameters" "Parameters"
] ]
parameters.should.have.length_of(0) assert len(parameters) == 0
parameter_template = { parameter_template = {
"AWSTemplateFormatVersion": "2010-09-09", "AWSTemplateFormatVersion": "2010-09-09",
@ -1874,6 +1853,6 @@ def test_ssm_parameter_update_stack_and_add_resource():
parameters = ssm_client.get_parameters(Names=["test_ssm"], WithDecryption=False)[ parameters = ssm_client.get_parameters(Names=["test_ssm"], WithDecryption=False)[
"Parameters" "Parameters"
] ]
parameters.should.have.length_of(1) assert len(parameters) == 1
parameters[0]["Name"].should.equal("test_ssm") assert parameters[0]["Name"] == "test_ssm"
parameters[0]["Value"].should.equal("Test SSM Parameter") assert parameters[0]["Value"] == "Test SSM Parameter"

34
tests/test_cloudformation/test_cloudformation_stack_policies.py
View File

@ -15,9 +15,9 @@ def test_set_stack_policy_on_nonexisting_stack():
with pytest.raises(ClientError) as exc: with pytest.raises(ClientError) as exc:
cf_conn.set_stack_policy(StackName="unknown", StackPolicyBody="{}") cf_conn.set_stack_policy(StackName="unknown", StackPolicyBody="{}")
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("ValidationError") assert err["Code"] == "ValidationError"
err["Message"].should.equal("Stack: unknown does not exist") assert err["Message"] == "Stack: unknown does not exist"
err["Type"].should.equal("Sender") assert err["Type"] == "Sender"
@mock_cloudformation @mock_cloudformation
@ -27,9 +27,9 @@ def test_get_stack_policy_on_nonexisting_stack():
with pytest.raises(ClientError) as exc: with pytest.raises(ClientError) as exc:
cf_conn.get_stack_policy(StackName="unknown") cf_conn.get_stack_policy(StackName="unknown")
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("ValidationError") assert err["Code"] == "ValidationError"
err["Message"].should.equal("Stack: unknown does not exist") assert err["Message"] == "Stack: unknown does not exist"
err["Type"].should.equal("Sender") assert err["Type"] == "Sender"
@mock_cloudformation @mock_cloudformation
@ -38,7 +38,7 @@ def test_get_stack_policy_on_stack_without_policy():
cf_conn.create_stack(StackName="test_stack", TemplateBody=dummy_template_json) cf_conn.create_stack(StackName="test_stack", TemplateBody=dummy_template_json)
resp = cf_conn.get_stack_policy(StackName="test_stack") resp = cf_conn.get_stack_policy(StackName="test_stack")
resp.shouldnt.have.key("StackPolicyBody") assert "StackPolicyBody" not in resp
@mock_cloudformation @mock_cloudformation
@ -51,11 +51,11 @@ def test_set_stack_policy_with_both_body_and_url():
StackName="test_stack", StackPolicyBody="{}", StackPolicyURL="..." StackName="test_stack", StackPolicyBody="{}", StackPolicyURL="..."
) )
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("ValidationError") assert err["Code"] == "ValidationError"
err["Message"].should.equal( assert (
"You cannot specify both StackPolicyURL and StackPolicyBody" err["Message"] == "You cannot specify both StackPolicyURL and StackPolicyBody"
) )
err["Type"].should.equal("Sender") assert err["Type"] == "Sender"
@mock_cloudformation @mock_cloudformation
@ -68,7 +68,7 @@ def test_set_stack_policy_with_body():
cf_conn.set_stack_policy(StackName="test_stack", StackPolicyBody=policy) cf_conn.set_stack_policy(StackName="test_stack", StackPolicyBody=policy)
resp = cf_conn.get_stack_policy(StackName="test_stack") resp = cf_conn.get_stack_policy(StackName="test_stack")
resp.should.have.key("StackPolicyBody").equals(policy) assert resp["StackPolicyBody"] == policy
@mock_cloudformation @mock_cloudformation
@ -81,7 +81,7 @@ def test_set_stack_policy_on_create():
) )
resp = cf_conn.get_stack_policy(StackName="test_stack") resp = cf_conn.get_stack_policy(StackName="test_stack")
resp.should.have.key("StackPolicyBody").equals("stack_policy_body") assert resp["StackPolicyBody"] == "stack_policy_body"
@mock_cloudformation @mock_cloudformation
@ -101,7 +101,7 @@ def test_set_stack_policy_with_url():
cf_conn.set_stack_policy(StackName="test_stack", StackPolicyURL=key_url) cf_conn.set_stack_policy(StackName="test_stack", StackPolicyURL=key_url)
resp = cf_conn.get_stack_policy(StackName="test_stack") resp = cf_conn.get_stack_policy(StackName="test_stack")
resp.should.have.key("StackPolicyBody").equals(policy) assert resp["StackPolicyBody"] == policy
@mock_cloudformation @mock_cloudformation
@ -113,6 +113,6 @@ def test_set_stack_policy_with_url_pointing_to_unknown_key():
with pytest.raises(ClientError) as exc: with pytest.raises(ClientError) as exc:
cf_conn.set_stack_policy(StackName="test_stack", StackPolicyURL="...") cf_conn.set_stack_policy(StackName="test_stack", StackPolicyURL="...")
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("ValidationError") assert err["Code"] == "ValidationError"
err["Message"].should.contain("S3 error: Access Denied") assert "S3 error: Access Denied" in err["Message"]
err["Type"].should.equal("Sender") assert err["Type"] == "Sender"

13
tests/test_cloudformation/test_server.py
View File

@ -1,6 +1,5 @@
import json import json
import re import re
import sure # noqa # pylint: disable=unused-import
import moto.server as server import moto.server as server
@ -17,17 +16,15 @@ def test_cloudformation_server_get():
create_stack_resp = test_client.action_data( create_stack_resp = test_client.action_data(
"CreateStack", StackName=stack_name, TemplateBody=json.dumps(template_body) "CreateStack", StackName=stack_name, TemplateBody=json.dumps(template_body)
) )
create_stack_resp.should.match( assert "<CreateStackResponse>" in create_stack_resp
r"<CreateStackResponse>.*<CreateStackResult>.*<StackId>.*</StackId>.*</CreateStackResult>.*</CreateStackResponse>", assert "<StackId>" in create_stack_resp
re.DOTALL, stack_id_from_create = re.search(
)
stack_id_from_create_response = re.search(
"<StackId>(.*)</StackId>", create_stack_resp "<StackId>(.*)</StackId>", create_stack_resp
).groups()[0] ).groups()[0]
list_stacks_resp = test_client.action_data("ListStacks") list_stacks_resp = test_client.action_data("ListStacks")
stack_id_from_list_response = re.search( stack_id_from_list = re.search(
"<StackId>(.*)</StackId>", list_stacks_resp "<StackId>(.*)</StackId>", list_stacks_resp
).groups()[0] ).groups()[0]
stack_id_from_create_response.should.equal(stack_id_from_list_response) assert stack_id_from_create == stack_id_from_list

274
tests/test_cloudformation/test_stack_parsing.py
View File

@ -3,7 +3,6 @@ import json
import yaml import yaml
import pytest import pytest
import sure # noqa # pylint: disable=unused-import
from botocore.exceptions import ClientError from botocore.exceptions import ClientError
from unittest.mock import patch from unittest.mock import patch
@ -259,15 +258,15 @@ def test_parse_stack_resources():
region_name="us-west-1", region_name="us-west-1",
) )
stack.resource_map.should.have.length_of(2) assert len(stack.resource_map) == 2
queue = stack.resource_map["Queue"] queue = stack.resource_map["Queue"]
queue.should.be.a(Queue) assert isinstance(queue, Queue)
queue.name.should.equal("my-queue") assert queue.name == "my-queue"
bucket = stack.resource_map["S3Bucket"] bucket = stack.resource_map["S3Bucket"]
bucket.should.be.a(FakeBucket) assert isinstance(bucket, FakeBucket)
bucket.physical_resource_id.should.equal(bucket.name) assert bucket.physical_resource_id == bucket.name
@patch("moto.cloudformation.parsing.logger") @patch("moto.cloudformation.parsing.logger")
@ -286,10 +285,10 @@ def test_parse_stack_with_name_type_resource():
region_name="us-west-1", region_name="us-west-1",
) )
stack.resource_map.should.have.length_of(1) assert len(stack.resource_map) == 1
list(stack.resource_map.keys())[0].should.equal("Queue") assert list(stack.resource_map.keys())[0] == "Queue"
queue = list(stack.resource_map.values())[0] queue = list(stack.resource_map.values())[0]
queue.should.be.a(Queue) assert isinstance(queue, Queue)
def test_parse_stack_with_tabbed_json_template(): def test_parse_stack_with_tabbed_json_template():
@ -302,10 +301,10 @@ def test_parse_stack_with_tabbed_json_template():
region_name="us-west-1", region_name="us-west-1",
) )
stack.resource_map.should.have.length_of(1) assert len(stack.resource_map) == 1
list(stack.resource_map.keys())[0].should.equal("Queue") assert list(stack.resource_map.keys())[0] == "Queue"
queue = list(stack.resource_map.values())[0] queue = list(stack.resource_map.values())[0]
queue.should.be.a(Queue) assert isinstance(queue, Queue)
def test_parse_stack_with_yaml_template(): def test_parse_stack_with_yaml_template():
@ -318,10 +317,10 @@ def test_parse_stack_with_yaml_template():
region_name="us-west-1", region_name="us-west-1",
) )
stack.resource_map.should.have.length_of(1) assert len(stack.resource_map) == 1
list(stack.resource_map.keys())[0].should.equal("Queue") assert list(stack.resource_map.keys())[0] == "Queue"
queue = list(stack.resource_map.values())[0] queue = list(stack.resource_map.values())[0]
queue.should.be.a(Queue) assert isinstance(queue, Queue)
def test_parse_stack_with_outputs(): def test_parse_stack_with_outputs():
@ -334,11 +333,11 @@ def test_parse_stack_with_outputs():
region_name="us-west-1", region_name="us-west-1",
) )
stack.output_map.should.have.length_of(1) assert len(stack.output_map) == 1
list(stack.output_map.keys())[0].should.equal("Output1") assert list(stack.output_map.keys())[0] == "Output1"
output = list(stack.output_map.values())[0] output = list(stack.output_map.values())[0]
output.should.be.a(Output) assert isinstance(output, Output)
output.description.should.equal("This is a description.") assert output.description == "This is a description."
def test_parse_stack_with_get_attribute_outputs(): def test_parse_stack_with_get_attribute_outputs():
@ -351,11 +350,11 @@ def test_parse_stack_with_get_attribute_outputs():
region_name="us-west-1", region_name="us-west-1",
) )
stack.output_map.should.have.length_of(1) assert len(stack.output_map) == 1
list(stack.output_map.keys())[0].should.equal("Output1") assert list(stack.output_map.keys())[0] == "Output1"
output = list(stack.output_map.values())[0] output = list(stack.output_map.values())[0]
output.should.be.a(Output) assert isinstance(output, Output)
output.value.should.equal("my-queue") assert output.value == "my-queue"
def test_parse_stack_with_get_attribute_kms(): def test_parse_stack_with_get_attribute_kms():
@ -371,10 +370,10 @@ def test_parse_stack_with_get_attribute_kms():
region_name="us-west-1", region_name="us-west-1",
) )
stack.output_map.should.have.length_of(1) assert len(stack.output_map) == 1
list(stack.output_map.keys())[0].should.equal("KeyArn") assert list(stack.output_map.keys())[0] == "KeyArn"
output = list(stack.output_map.values())[0] output = list(stack.output_map.values())[0]
output.should.be.a(Output) assert isinstance(output, Output)
def test_parse_stack_with_get_availability_zones(): def test_parse_stack_with_get_availability_zones():
@ -387,11 +386,11 @@ def test_parse_stack_with_get_availability_zones():
region_name="us-east-1", region_name="us-east-1",
) )
stack.output_map.should.have.length_of(1) assert len(stack.output_map) == 1
list(stack.output_map.keys())[0].should.equal("Output1") assert list(stack.output_map.keys())[0] == "Output1"
output = list(stack.output_map.values())[0] output = list(stack.output_map.values())[0]
output.should.be.a(Output) assert isinstance(output, Output)
output.value.should.equal(["us-east-1a", "us-east-1b", "us-east-1c", "us-east-1d"]) assert output.value == ["us-east-1a", "us-east-1b", "us-east-1c", "us-east-1d"]
@mock_sqs @mock_sqs
@ -401,9 +400,10 @@ def test_parse_stack_with_bad_get_attribute_outputs_using_boto3():
with pytest.raises(ClientError) as exc: with pytest.raises(ClientError) as exc:
conn.create_stack(StackName="teststack", TemplateBody=bad_output_template_json) conn.create_stack(StackName="teststack", TemplateBody=bad_output_template_json)
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("ValidationError") assert err["Code"] == "ValidationError"
err["Message"].should.equal( assert (
"Template error: resource Queue does not support attribute type InvalidAttribute in Fn::GetAtt" err["Message"]
== "Template error: resource Queue does not support attribute type InvalidAttribute in Fn::GetAtt"
) )
@ -417,8 +417,7 @@ def test_parse_stack_with_null_outputs_section():
account_id=ACCOUNT_ID, account_id=ACCOUNT_ID,
region_name="us-west-1", region_name="us-west-1",
) )
err = str(exc.value) assert "[/Outputs] 'null' values are not allowed in templates" in str(exc.value)
err.should.contain("[/Outputs] 'null' values are not allowed in templates")
def test_parse_stack_with_parameters(): def test_parse_stack_with_parameters():
@ -436,98 +435,123 @@ def test_parse_stack_with_parameters():
region_name="us-west-1", region_name="us-west-1",
) )
stack.resource_map.no_echo_parameter_keys.should.have("NoEchoParam") assert "NoEchoParam" in stack.resource_map.no_echo_parameter_keys
stack.resource_map.no_echo_parameter_keys.should_not.have("Param") assert "Param" not in stack.resource_map.no_echo_parameter_keys
stack.resource_map.no_echo_parameter_keys.should_not.have("NumberParam") assert "NumberParam" not in stack.resource_map.no_echo_parameter_keys
stack.resource_map.no_echo_parameter_keys.should_not.have("NumberListParam") assert "NumberListParam" not in stack.resource_map.no_echo_parameter_keys
stack.resource_map.resolved_parameters["NumberParam"].should.equal(42) assert stack.resource_map.resolved_parameters["NumberParam"] == 42
stack.resource_map.resolved_parameters["NumberListParam"].should.equal( assert stack.resource_map.resolved_parameters["NumberListParam"] == [42, 3.14159]
[42, 3.14159]
)
def test_parse_equals_condition(): def test_parse_equals_condition():
parse_condition( assert (
condition={"Fn::Equals": [{"Ref": "EnvType"}, "prod"]}, parse_condition(
resources_map={"EnvType": "prod"}, condition={"Fn::Equals": [{"Ref": "EnvType"}, "prod"]},
condition_map={}, resources_map={"EnvType": "prod"},
).should.equal(True) condition_map={},
)
is True
)
parse_condition( assert (
condition={"Fn::Equals": [{"Ref": "EnvType"}, "prod"]}, parse_condition(
resources_map={"EnvType": "staging"}, condition={"Fn::Equals": [{"Ref": "EnvType"}, "prod"]},
condition_map={}, resources_map={"EnvType": "staging"},
).should.equal(False) condition_map={},
)
is False
)
def test_parse_not_condition(): def test_parse_not_condition():
parse_condition( assert (
condition={"Fn::Not": [{"Fn::Equals": [{"Ref": "EnvType"}, "prod"]}]}, parse_condition(
resources_map={"EnvType": "prod"}, condition={"Fn::Not": [{"Fn::Equals": [{"Ref": "EnvType"}, "prod"]}]},
condition_map={}, resources_map={"EnvType": "prod"},
).should.equal(False) condition_map={},
)
is False
)
parse_condition( assert (
condition={"Fn::Not": [{"Fn::Equals": [{"Ref": "EnvType"}, "prod"]}]}, parse_condition(
resources_map={"EnvType": "staging"}, condition={"Fn::Not": [{"Fn::Equals": [{"Ref": "EnvType"}, "prod"]}]},
condition_map={}, resources_map={"EnvType": "staging"},
).should.equal(True) condition_map={},
)
is True
)
def test_parse_and_condition(): def test_parse_and_condition():
parse_condition( assert (
condition={ parse_condition(
"Fn::And": [ condition={
{"Fn::Equals": [{"Ref": "EnvType"}, "prod"]}, "Fn::And": [
{"Fn::Equals": [{"Ref": "EnvType"}, "staging"]}, {"Fn::Equals": [{"Ref": "EnvType"}, "prod"]},
] {"Fn::Equals": [{"Ref": "EnvType"}, "staging"]},
}, ]
resources_map={"EnvType": "prod"}, },
condition_map={}, resources_map={"EnvType": "prod"},
).should.equal(False) condition_map={},
)
is False
)
parse_condition( assert (
condition={ parse_condition(
"Fn::And": [ condition={
{"Fn::Equals": [{"Ref": "EnvType"}, "prod"]}, "Fn::And": [
{"Fn::Equals": [{"Ref": "EnvType"}, "prod"]}, {"Fn::Equals": [{"Ref": "EnvType"}, "prod"]},
] {"Fn::Equals": [{"Ref": "EnvType"}, "prod"]},
}, ]
resources_map={"EnvType": "prod"}, },
condition_map={}, resources_map={"EnvType": "prod"},
).should.equal(True) condition_map={},
)
is True
)
def test_parse_or_condition(): def test_parse_or_condition():
parse_condition( assert (
condition={ parse_condition(
"Fn::Or": [ condition={
{"Fn::Equals": [{"Ref": "EnvType"}, "prod"]}, "Fn::Or": [
{"Fn::Equals": [{"Ref": "EnvType"}, "staging"]}, {"Fn::Equals": [{"Ref": "EnvType"}, "prod"]},
] {"Fn::Equals": [{"Ref": "EnvType"}, "staging"]},
}, ]
resources_map={"EnvType": "prod"}, },
condition_map={}, resources_map={"EnvType": "prod"},
).should.equal(True) condition_map={},
)
is True
)
parse_condition( assert (
condition={ parse_condition(
"Fn::Or": [ condition={
{"Fn::Equals": [{"Ref": "EnvType"}, "staging"]}, "Fn::Or": [
{"Fn::Equals": [{"Ref": "EnvType"}, "staging"]}, {"Fn::Equals": [{"Ref": "EnvType"}, "staging"]},
] {"Fn::Equals": [{"Ref": "EnvType"}, "staging"]},
}, ]
resources_map={"EnvType": "prod"}, },
condition_map={}, resources_map={"EnvType": "prod"},
).should.equal(False) condition_map={},
)
is False
)
def test_reference_other_conditions(): def test_reference_other_conditions():
parse_condition( assert (
condition={"Fn::Not": [{"Condition": "OtherCondition"}]}, parse_condition(
resources_map={}, condition={"Fn::Not": [{"Condition": "OtherCondition"}]},
condition_map={"OtherCondition": True}, resources_map={},
).should.equal(False) condition_map={"OtherCondition": True},
)
is False
)
def test_parse_split_and_select(): def test_parse_split_and_select():
@ -540,9 +564,9 @@ def test_parse_split_and_select():
region_name="us-west-1", region_name="us-west-1",
) )
stack.resource_map.should.have.length_of(1) assert len(stack.resource_map) == 1
queue = stack.resource_map["Queue"] queue = stack.resource_map["Queue"]
queue.name.should.equal("myqueue") assert queue.name == "myqueue"
def test_sub(): def test_sub():
@ -557,7 +581,7 @@ def test_sub():
queue1 = stack.resource_map["Queue1"] queue1 = stack.resource_map["Queue1"]
queue2 = stack.resource_map["Queue2"] queue2 = stack.resource_map["Queue2"]
queue2.name.should.equal(queue1.name) assert queue2.name == queue1.name
def test_sub_num(): def test_sub_num():
@ -572,7 +596,7 @@ def test_sub_num():
# Errors on moto<=4.0.10 because int(42) is used with str.replace # Errors on moto<=4.0.10 because int(42) is used with str.replace
queue = stack.resource_map["Queue"] queue = stack.resource_map["Queue"]
queue.name.should.equal("test_stack-queue-42") assert queue.name == "test_stack-queue-42"
def test_sub_mapping(): def test_sub_mapping():
@ -585,7 +609,7 @@ def test_sub_mapping():
region_name="us-west-1", region_name="us-west-1",
) )
queue = stack.resource_map["Queue"] queue = stack.resource_map["Queue"]
queue.name.should.equal("test_stack-queue-apple-yes") assert queue.name == "test_stack-queue-apple-yes"
stack = FakeStack( stack = FakeStack(
stack_id="test_id", stack_id="test_id",
@ -596,7 +620,7 @@ def test_sub_mapping():
region_name="us-west-1", region_name="us-west-1",
) )
queue = stack.resource_map["Queue"] queue = stack.resource_map["Queue"]
queue.name.should.equal("test_stack-queue-banana-no") assert queue.name == "test_stack-queue-banana-no"
def test_import(): def test_import():
@ -619,7 +643,7 @@ def test_import():
) )
queue = import_stack.resource_map["Queue"] queue = import_stack.resource_map["Queue"]
queue.name.should.equal("value") assert queue.name == "value"
def test_to_json_string(): def test_to_json_string():
@ -632,8 +656,8 @@ def test_to_json_string():
region_name="us-west-1", region_name="us-west-1",
) )
queue = stack.resource_map["Queue"] queue = stack.resource_map["Queue"]
queue.name.should.equal("test") assert queue.name == "test"
queue.dead_letter_queue.name.should.equal("deadletter") assert queue.dead_letter_queue.name == "deadletter"
def test_short_form_func_in_yaml_teamplate(): def test_short_form_func_in_yaml_teamplate():
@ -676,7 +700,7 @@ def test_short_form_func_in_yaml_teamplate():
["KeySub", {"Fn::Sub": "A"}], ["KeySub", {"Fn::Sub": "A"}],
] ]
for k, v in key_and_expects: for k, v in key_and_expects:
template_dict.should.have.key(k).which.should.be.equal(v) assert template_dict[k] == v
@mock_ssm @mock_ssm
@ -700,10 +724,9 @@ def test_ssm_parameter_parsing():
region_name="us-west-1", region_name="us-west-1",
) )
stack.resource_map.resolved_parameters["SingleParamCfn"].should.equal("string") params = stack.resource_map.resolved_parameters
stack.resource_map.resolved_parameters["ListParamCfn"].should.equal( assert params["SingleParamCfn"] == "string"
["comma", "separated", "string"] assert params["ListParamCfn"] == ["comma", "separated", "string"]
)
# Not passing in a value for ListParamCfn to test Default value # Not passing in a value for ListParamCfn to test Default value
if not settings.TEST_SERVER_MODE: if not settings.TEST_SERVER_MODE:
@ -716,7 +739,6 @@ def test_ssm_parameter_parsing():
region_name="us-west-1", region_name="us-west-1",
) )
stack.resource_map.resolved_parameters["SingleParamCfn"].should.equal("string") params = stack.resource_map.resolved_parameters
stack.resource_map.resolved_parameters["ListParamCfn"].should.equal( assert params["SingleParamCfn"] == "string"
["comma", "separated", "string"] assert params["ListParamCfn"] == ["comma", "separated", "string"]
)

30
tests/test_cloudformation/test_validate.py
View File

@ -1,7 +1,7 @@
import json import json
import boto3 import boto3
import botocore import botocore
import sure # noqa # pylint: disable=unused-import import pytest
from moto import mock_cloudformation, mock_s3 from moto import mock_cloudformation, mock_s3
from tests import EXAMPLE_AMI_ID from tests import EXAMPLE_AMI_ID
@ -63,15 +63,13 @@ def test_boto3_json_with_tabs_validate_successful():
@mock_cloudformation @mock_cloudformation
def test_boto3_json_invalid_missing_resource(): def test_boto3_json_invalid_missing_resource():
cf_conn = boto3.client("cloudformation", region_name="us-east-1") cf_conn = boto3.client("cloudformation", region_name="us-east-1")
try: with pytest.raises(botocore.exceptions.ClientError) as exc:
cf_conn.validate_template(TemplateBody=dummy_bad_template_json) cf_conn.validate_template(TemplateBody=dummy_bad_template_json)
assert False err = exc.value.response["Error"]
except botocore.exceptions.ClientError as e: assert (
str(e).should.contain( err["Message"]
"An error occurred (ValidationError) when calling the ValidateTemplate operation: Stack" == "Stack with id Missing top level template section Resources does not exist"
" with id Missing top level" )
)
assert True
yaml_template = """ yaml_template = """
@ -122,12 +120,10 @@ def test_boto3_yaml_validate_template_url_successful():
@mock_cloudformation @mock_cloudformation
def test_boto3_yaml_invalid_missing_resource(): def test_boto3_yaml_invalid_missing_resource():
cf_conn = boto3.client("cloudformation", region_name="us-east-1") cf_conn = boto3.client("cloudformation", region_name="us-east-1")
try: with pytest.raises(botocore.exceptions.ClientError) as exc:
cf_conn.validate_template(TemplateBody=yaml_bad_template) cf_conn.validate_template(TemplateBody=yaml_bad_template)
assert False err = exc.value.response["Error"]
except botocore.exceptions.ClientError as e: assert (
str(e).should.contain( err["Message"]
"An error occurred (ValidationError) when calling the ValidateTemplate operation: Stack" == "Stack with id Missing top level template section Resources does not exist"
" with id Missing top level" )
)
assert True