from __future__ import unicode_literals import json import boto import boto.cloudformation import boto.ec2 import boto.ec2.autoscale import boto.ec2.elb from boto.exception import BotoServerError import boto.iam import boto.redshift import boto.sns import boto.sqs import boto.vpc import sure # noqa from moto import ( mock_autoscaling, mock_cloudformation, mock_ec2, mock_elb, mock_iam, mock_rds, mock_redshift, mock_route53, mock_sns, mock_sqs, ) from .fixtures import ( ec2_classic_eip, fn_join, rds_mysql_with_read_replica, redshift, route53_ec2_instance_with_public_ip, route53_health_check, route53_roundrobin, single_instance_with_ebs_volume, vpc_eip, vpc_single_instance_in_subnet, ) @mock_cloudformation() def test_stack_sqs_integration(): sqs_template = { "AWSTemplateFormatVersion": "2010-09-09", "Resources": { "QueueGroup": { "Type": "AWS::SQS::Queue", "Properties": { "QueueName": "my-queue", "VisibilityTimeout": 60, } }, }, } sqs_template_json = json.dumps(sqs_template) conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack( "test_stack", template_body=sqs_template_json, ) stack = conn.describe_stacks()[0] queue = stack.describe_resources()[0] queue.resource_type.should.equal('AWS::SQS::Queue') queue.logical_resource_id.should.equal("QueueGroup") queue.physical_resource_id.should.equal("my-queue") @mock_cloudformation() def test_stack_list_resources(): sqs_template = { "AWSTemplateFormatVersion": "2010-09-09", "Resources": { "QueueGroup": { "Type": "AWS::SQS::Queue", "Properties": { "QueueName": "my-queue", "VisibilityTimeout": 60, } }, }, } sqs_template_json = json.dumps(sqs_template) conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack( "test_stack", template_body=sqs_template_json, ) resources = conn.list_stack_resources("test_stack") assert len(resources) == 1 queue = resources[0] queue.resource_type.should.equal('AWS::SQS::Queue') queue.logical_resource_id.should.equal("QueueGroup") queue.physical_resource_id.should.equal("my-queue") @mock_cloudformation() @mock_sqs() def test_update_stack(): sqs_template = { "AWSTemplateFormatVersion": "2010-09-09", "Resources": { "QueueGroup": { "Type": "AWS::SQS::Queue", "Properties": { "QueueName": "my-queue", "VisibilityTimeout": 60, } }, }, } sqs_template_json = json.dumps(sqs_template) conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack( "test_stack", template_body=sqs_template_json, ) sqs_conn = boto.sqs.connect_to_region("us-west-1") queues = sqs_conn.get_all_queues() queues.should.have.length_of(1) queues[0].get_attributes('VisibilityTimeout')['VisibilityTimeout'].should.equal('60') sqs_template['Resources']['QueueGroup']['Properties']['VisibilityTimeout'] = 100 sqs_template_json = json.dumps(sqs_template) conn.update_stack("test_stack", sqs_template_json) queues = sqs_conn.get_all_queues() queues.should.have.length_of(1) queues[0].get_attributes('VisibilityTimeout')['VisibilityTimeout'].should.equal('100') @mock_cloudformation() @mock_sqs() def test_update_stack_and_remove_resource(): sqs_template = { "AWSTemplateFormatVersion": "2010-09-09", "Resources": { "QueueGroup": { "Type": "AWS::SQS::Queue", "Properties": { "QueueName": "my-queue", "VisibilityTimeout": 60, } }, }, } sqs_template_json = json.dumps(sqs_template) conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack( "test_stack", template_body=sqs_template_json, ) sqs_conn = boto.sqs.connect_to_region("us-west-1") queues = sqs_conn.get_all_queues() queues.should.have.length_of(1) sqs_template['Resources'].pop('QueueGroup') sqs_template_json = json.dumps(sqs_template) conn.update_stack("test_stack", sqs_template_json) queues = sqs_conn.get_all_queues() queues.should.have.length_of(0) @mock_cloudformation() @mock_sqs() def test_update_stack_and_add_resource(): sqs_template = { "AWSTemplateFormatVersion": "2010-09-09", "Resources": {}, } sqs_template_json = json.dumps(sqs_template) conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack( "test_stack", template_body=sqs_template_json, ) sqs_conn = boto.sqs.connect_to_region("us-west-1") queues = sqs_conn.get_all_queues() queues.should.have.length_of(0) sqs_template = { "AWSTemplateFormatVersion": "2010-09-09", "Resources": { "QueueGroup": { "Type": "AWS::SQS::Queue", "Properties": { "QueueName": "my-queue", "VisibilityTimeout": 60, } }, }, } sqs_template_json = json.dumps(sqs_template) conn.update_stack("test_stack", sqs_template_json) queues = sqs_conn.get_all_queues() queues.should.have.length_of(1) @mock_ec2() @mock_cloudformation() def test_stack_ec2_integration(): ec2_template = { "AWSTemplateFormatVersion": "2010-09-09", "Resources": { "WebServerGroup": { "Type": "AWS::EC2::Instance", "Properties": { "ImageId": "ami-1234abcd", "UserData": "some user data", } }, }, } ec2_template_json = json.dumps(ec2_template) conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack( "ec2_stack", template_body=ec2_template_json, ) ec2_conn = boto.ec2.connect_to_region("us-west-1") reservation = ec2_conn.get_all_instances()[0] ec2_instance = reservation.instances[0] stack = conn.describe_stacks()[0] instance = stack.describe_resources()[0] instance.resource_type.should.equal('AWS::EC2::Instance') instance.logical_resource_id.should.contain("WebServerGroup") instance.physical_resource_id.should.equal(ec2_instance.id) @mock_ec2() @mock_elb() @mock_cloudformation() def test_stack_elb_integration_with_attached_ec2_instances(): elb_template = { "AWSTemplateFormatVersion": "2010-09-09", "Resources": { "MyELB": { "Type": "AWS::ElasticLoadBalancing::LoadBalancer", "Instances": [{"Ref": "Ec2Instance1"}], "Properties": { "LoadBalancerName": "test-elb", "AvailabilityZones": ['us-east1'], } }, "Ec2Instance1": { "Type": "AWS::EC2::Instance", "Properties": { "ImageId": "ami-1234abcd", "UserData": "some user data", } }, }, } elb_template_json = json.dumps(elb_template) conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack( "elb_stack", template_body=elb_template_json, ) elb_conn = boto.ec2.elb.connect_to_region("us-west-1") load_balancer = elb_conn.get_all_load_balancers()[0] ec2_conn = boto.ec2.connect_to_region("us-west-1") reservation = ec2_conn.get_all_instances()[0] ec2_instance = reservation.instances[0] instance_id = ec2_instance.id load_balancer.instances[0].id.should.equal(ec2_instance.id) list(load_balancer.availability_zones).should.equal(['us-east1']) @mock_ec2() @mock_redshift() @mock_cloudformation() def test_redshift_stack(): redshift_template_json = json.dumps(redshift.template) vpc_conn = boto.vpc.connect_to_region("us-west-2") conn = boto.cloudformation.connect_to_region("us-west-2") conn.create_stack( "redshift_stack", template_body=redshift_template_json, parameters=[ ("DatabaseName", "mydb"), ("ClusterType", "multi-node"), ("NumberOfNodes", 2), ("NodeType", "dw1.xlarge"), ("MasterUsername", "myuser"), ("MasterUserPassword", "mypass"), ("InboundTraffic", "10.0.0.1/16"), ("PortNumber", 5439), ] ) redshift_conn = boto.redshift.connect_to_region("us-west-2") cluster_res = redshift_conn.describe_clusters() clusters = cluster_res['DescribeClustersResponse']['DescribeClustersResult']['Clusters'] clusters.should.have.length_of(1) cluster = clusters[0] cluster['DBName'].should.equal("mydb") cluster['NumberOfNodes'].should.equal(2) cluster['NodeType'].should.equal("dw1.xlarge") cluster['MasterUsername'].should.equal("myuser") cluster['Port'].should.equal(5439) cluster['VpcSecurityGroups'].should.have.length_of(1) security_group_id = cluster['VpcSecurityGroups'][0]['VpcSecurityGroupId'] groups = vpc_conn.get_all_security_groups(group_ids=[security_group_id]) groups.should.have.length_of(1) group = groups[0] group.rules.should.have.length_of(1) group.rules[0].grants[0].cidr_ip.should.equal("10.0.0.1/16") @mock_ec2() @mock_cloudformation() def test_stack_security_groups(): security_group_template = { "AWSTemplateFormatVersion": "2010-09-09", "Resources": { "my-security-group": { "Type": "AWS::EC2::SecurityGroup", "Properties": { "GroupDescription": "My other group", }, }, "Ec2Instance2": { "Type": "AWS::EC2::Instance", "Properties": { "SecurityGroups": [{"Ref": "InstanceSecurityGroup"}], "ImageId": "ami-1234abcd", } }, "InstanceSecurityGroup": { "Type": "AWS::EC2::SecurityGroup", "Properties": { "GroupDescription": "My security group", "SecurityGroupIngress": [{ "IpProtocol": "tcp", "FromPort": "22", "ToPort": "22", "CidrIp": "123.123.123.123/32", }, { "IpProtocol": "tcp", "FromPort": "80", "ToPort": "8000", "SourceSecurityGroupId": {"Ref": "my-security-group"}, }] } } }, } security_group_template_json = json.dumps(security_group_template) conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack( "security_group_stack", template_body=security_group_template_json, ) ec2_conn = boto.ec2.connect_to_region("us-west-1") instance_group = ec2_conn.get_all_security_groups(filters={'description': ['My security group']})[0] other_group = ec2_conn.get_all_security_groups(filters={'description': ['My other group']})[0] reservation = ec2_conn.get_all_instances()[0] ec2_instance = reservation.instances[0] ec2_instance.groups[0].id.should.equal(instance_group.id) instance_group.description.should.equal("My security group") rule1, rule2 = instance_group.rules int(rule1.to_port).should.equal(22) int(rule1.from_port).should.equal(22) rule1.grants[0].cidr_ip.should.equal("123.123.123.123/32") rule1.ip_protocol.should.equal('tcp') int(rule2.to_port).should.equal(8000) int(rule2.from_port).should.equal(80) rule2.ip_protocol.should.equal('tcp') rule2.grants[0].group_id.should.equal(other_group.id) @mock_autoscaling() @mock_elb() @mock_cloudformation() def test_autoscaling_group_with_elb(): web_setup_template = { "AWSTemplateFormatVersion": "2010-09-09", "Resources": { "my-as-group": { "Type": "AWS::AutoScaling::AutoScalingGroup", "Properties": { "AvailabilityZones": ['us-east1'], "LaunchConfigurationName": {"Ref": "my-launch-config"}, "MinSize": "2", "MaxSize": "2", "LoadBalancerNames": [{"Ref": "my-elb"}] }, }, "my-launch-config": { "Type": "AWS::AutoScaling::LaunchConfiguration", "Properties": { "ImageId": "ami-1234abcd", "UserData": "some user data", } }, "my-elb": { "Type": "AWS::ElasticLoadBalancing::LoadBalancer", "Properties": { "AvailabilityZones": ['us-east1'], "Listeners": [{ "LoadBalancerPort": "80", "InstancePort": "80", "Protocol": "HTTP" }], "LoadBalancerName": "my-elb", "HealthCheck": { "Target": "80", "HealthyThreshold": "3", "UnhealthyThreshold": "5", "Interval": "30", "Timeout": "5", }, }, }, } } web_setup_template_json = json.dumps(web_setup_template) conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack( "web_stack", template_body=web_setup_template_json, ) autoscale_conn = boto.ec2.autoscale.connect_to_region("us-west-1") autoscale_group = autoscale_conn.get_all_groups()[0] autoscale_group.launch_config_name.should.contain("my-launch-config") autoscale_group.load_balancers[0].should.equal('my-elb') # Confirm the Launch config was actually created autoscale_conn.get_all_launch_configurations().should.have.length_of(1) # Confirm the ELB was actually created elb_conn = boto.ec2.elb.connect_to_region("us-west-1") elb_conn.get_all_load_balancers().should.have.length_of(1) stack = conn.describe_stacks()[0] resources = stack.describe_resources() as_group_resource = [resource for resource in resources if resource.resource_type == 'AWS::AutoScaling::AutoScalingGroup'][0] as_group_resource.physical_resource_id.should.contain("my-as-group") launch_config_resource = [resource for resource in resources if resource.resource_type == 'AWS::AutoScaling::LaunchConfiguration'][0] launch_config_resource.physical_resource_id.should.contain("my-launch-config") elb_resource = [resource for resource in resources if resource.resource_type == 'AWS::ElasticLoadBalancing::LoadBalancer'][0] elb_resource.physical_resource_id.should.contain("my-elb") @mock_ec2() @mock_cloudformation() def test_vpc_single_instance_in_subnet(): template_json = json.dumps(vpc_single_instance_in_subnet.template) conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack( "test_stack", template_body=template_json, parameters=[("KeyName", "my_key")], ) vpc_conn = boto.vpc.connect_to_region("us-west-1") vpc = vpc_conn.get_all_vpcs()[0] vpc.cidr_block.should.equal("10.0.0.0/16") # Add this once we implement the endpoint # vpc_conn.get_all_internet_gateways().should.have.length_of(1) subnet = vpc_conn.get_all_subnets()[0] subnet.vpc_id.should.equal(vpc.id) ec2_conn = boto.ec2.connect_to_region("us-west-1") reservation = ec2_conn.get_all_instances()[0] instance = reservation.instances[0] # Check that the EIP is attached the the EC2 instance eip = ec2_conn.get_all_addresses()[0] eip.domain.should.equal('vpc') eip.instance_id.should.equal(instance.id) security_group = ec2_conn.get_all_security_groups(filters={'vpc_id': [vpc.id]})[0] security_group.vpc_id.should.equal(vpc.id) stack = conn.describe_stacks()[0] resources = stack.describe_resources() vpc_resource = [resource for resource in resources if resource.resource_type == 'AWS::EC2::VPC'][0] vpc_resource.physical_resource_id.should.equal(vpc.id) subnet_resource = [resource for resource in resources if resource.resource_type == 'AWS::EC2::Subnet'][0] subnet_resource.physical_resource_id.should.equal(subnet.id) eip_resource = [resource for resource in resources if resource.resource_type == 'AWS::EC2::EIP'][0] eip_resource.physical_resource_id.should.equal(eip.allocation_id) @mock_cloudformation() @mock_ec2() @mock_rds() def test_rds_mysql_with_read_replica(): ec2_conn = boto.ec2.connect_to_region("us-west-1") ec2_conn.create_security_group('application', 'Our Application Group') template_json = json.dumps(rds_mysql_with_read_replica.template) conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack( "test_stack", template_body=template_json, parameters=[ ("DBInstanceIdentifier", "master_db"), ("DBName", "my_db"), ("DBUser", "my_user"), ("DBPassword", "my_password"), ("DBAllocatedStorage", "20"), ("DBInstanceClass", "db.m1.medium"), ("EC2SecurityGroup", "application"), ("MultiAZ", "true"), ], ) rds_conn = boto.rds.connect_to_region("us-west-1") primary = rds_conn.get_all_dbinstances("master_db")[0] primary.master_username.should.equal("my_user") primary.allocated_storage.should.equal(20) primary.instance_class.should.equal("db.m1.medium") primary.multi_az.should.equal(True) list(primary.read_replica_dbinstance_identifiers).should.have.length_of(1) replica_id = primary.read_replica_dbinstance_identifiers[0] replica = rds_conn.get_all_dbinstances(replica_id)[0] replica.instance_class.should.equal("db.m1.medium") security_group_name = primary.security_groups[0].name security_group = rds_conn.get_all_dbsecurity_groups(security_group_name)[0] security_group.ec2_groups[0].name.should.equal("application") @mock_cloudformation() @mock_ec2() @mock_rds() def test_rds_mysql_with_read_replica_in_vpc(): template_json = json.dumps(rds_mysql_with_read_replica.template) conn = boto.cloudformation.connect_to_region("eu-central-1") conn.create_stack( "test_stack", template_body=template_json, parameters=[ ("DBInstanceIdentifier", "master_db"), ("DBName", "my_db"), ("DBUser", "my_user"), ("DBPassword", "my_password"), ("DBAllocatedStorage", "20"), ("DBInstanceClass", "db.m1.medium"), ("MultiAZ", "true"), ], ) rds_conn = boto.rds.connect_to_region("eu-central-1") primary = rds_conn.get_all_dbinstances("master_db")[0] subnet_group_name = primary.subnet_group.name subnet_group = rds_conn.get_all_db_subnet_groups(subnet_group_name)[0] subnet_group.description.should.equal("my db subnet group") @mock_autoscaling() @mock_iam() @mock_cloudformation() def test_iam_roles(): iam_template = { "AWSTemplateFormatVersion": "2010-09-09", "Resources": { "my-launch-config": { "Properties": { "IamInstanceProfile": {"Ref": "my-instance-profile"}, "ImageId": "ami-1234abcd", }, "Type": "AWS::AutoScaling::LaunchConfiguration" }, "my-instance-profile": { "Properties": { "Path": "my-path", "Roles": [{"Ref": "my-role"}], }, "Type": "AWS::IAM::InstanceProfile" }, "my-role": { "Properties": { "AssumeRolePolicyDocument": { "Statement": [ { "Action": [ "sts:AssumeRole" ], "Effect": "Allow", "Principal": { "Service": [ "ec2.amazonaws.com" ] } } ] }, "Path": "my-path", "Policies": [ { "PolicyDocument": { "Statement": [ { "Action": [ "ec2:CreateTags", "ec2:DescribeInstances", "ec2:DescribeTags" ], "Effect": "Allow", "Resource": [ "*" ] } ], "Version": "2012-10-17" }, "PolicyName": "EC2_Tags" }, { "PolicyDocument": { "Statement": [ { "Action": [ "sqs:*" ], "Effect": "Allow", "Resource": [ "*" ] } ], "Version": "2012-10-17" }, "PolicyName": "SQS" }, ] }, "Type": "AWS::IAM::Role" } } } iam_template_json = json.dumps(iam_template) conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack( "test_stack", template_body=iam_template_json, ) iam_conn = boto.iam.connect_to_region("us-west-1") role_result = iam_conn.list_roles()['list_roles_response']['list_roles_result']['roles'][0] role = iam_conn.get_role(role_result.role_name) role.role_name.should.contain("my-role") role.path.should.equal("my-path") instance_profile_response = iam_conn.list_instance_profiles()['list_instance_profiles_response'] cfn_instance_profile = instance_profile_response['list_instance_profiles_result']['instance_profiles'][0] instance_profile = iam_conn.get_instance_profile(cfn_instance_profile.instance_profile_name) instance_profile.instance_profile_name.should.contain("my-instance-profile") instance_profile.path.should.equal("my-path") instance_profile.role_id.should.equal(role.role_id) autoscale_conn = boto.ec2.autoscale.connect_to_region("us-west-1") launch_config = autoscale_conn.get_all_launch_configurations()[0] launch_config.instance_profile_name.should.contain("my-instance-profile") stack = conn.describe_stacks()[0] resources = stack.describe_resources() instance_profile_resource = [resource for resource in resources if resource.resource_type == 'AWS::IAM::InstanceProfile'][0] instance_profile_resource.physical_resource_id.should.equal(instance_profile.instance_profile_name) role_resource = [resource for resource in resources if resource.resource_type == 'AWS::IAM::Role'][0] role_resource.physical_resource_id.should.equal(role.role_id) @mock_ec2() @mock_cloudformation() def test_single_instance_with_ebs_volume(): template_json = json.dumps(single_instance_with_ebs_volume.template) conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack( "test_stack", template_body=template_json, parameters=[("KeyName", "key_name")] ) ec2_conn = boto.ec2.connect_to_region("us-west-1") reservation = ec2_conn.get_all_instances()[0] ec2_instance = reservation.instances[0] volume = ec2_conn.get_all_volumes()[0] volume.volume_state().should.equal('in-use') volume.attach_data.instance_id.should.equal(ec2_instance.id) stack = conn.describe_stacks()[0] resources = stack.describe_resources() ebs_volume = [resource for resource in resources if resource.resource_type == 'AWS::EC2::Volume'][0] ebs_volume.physical_resource_id.should.equal(volume.id) @mock_cloudformation() def test_create_template_without_required_param(): template_json = json.dumps(single_instance_with_ebs_volume.template) conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack.when.called_with( "test_stack", template_body=template_json, ).should.throw(BotoServerError) @mock_ec2() @mock_cloudformation() def test_classic_eip(): template_json = json.dumps(ec2_classic_eip.template) conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack("test_stack", template_body=template_json) ec2_conn = boto.ec2.connect_to_region("us-west-1") eip = ec2_conn.get_all_addresses()[0] stack = conn.describe_stacks()[0] resources = stack.describe_resources() cfn_eip = [resource for resource in resources if resource.resource_type == 'AWS::EC2::EIP'][0] cfn_eip.physical_resource_id.should.equal(eip.public_ip) @mock_ec2() @mock_cloudformation() def test_vpc_eip(): template_json = json.dumps(vpc_eip.template) conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack("test_stack", template_body=template_json) ec2_conn = boto.ec2.connect_to_region("us-west-1") eip = ec2_conn.get_all_addresses()[0] stack = conn.describe_stacks()[0] resources = stack.describe_resources() cfn_eip = [resource for resource in resources if resource.resource_type == 'AWS::EC2::EIP'][0] cfn_eip.physical_resource_id.should.equal(eip.allocation_id) @mock_ec2() @mock_cloudformation() def test_fn_join(): template_json = json.dumps(fn_join.template) conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack("test_stack", template_body=template_json) ec2_conn = boto.ec2.connect_to_region("us-west-1") eip = ec2_conn.get_all_addresses()[0] stack = conn.describe_stacks()[0] fn_join_output = stack.outputs[0] fn_join_output.value.should.equal('test eip:{0}'.format(eip.public_ip)) @mock_cloudformation() @mock_sqs() def test_conditional_resources(): sqs_template = { "AWSTemplateFormatVersion": "2010-09-09", "Parameters": { "EnvType": { "Description": "Environment type.", "Type": "String", } }, "Conditions": { "CreateQueue": {"Fn::Equals": [{"Ref": "EnvType"}, "prod"]} }, "Resources": { "QueueGroup": { "Condition": "CreateQueue", "Type": "AWS::SQS::Queue", "Properties": { "QueueName": "my-queue", "VisibilityTimeout": 60, } }, }, } sqs_template_json = json.dumps(sqs_template) conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack( "test_stack_without_queue", template_body=sqs_template_json, parameters=[("EnvType", "staging")], ) sqs_conn = boto.sqs.connect_to_region("us-west-1") list(sqs_conn.get_all_queues()).should.have.length_of(0) conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack( "test_stack_with_queue", template_body=sqs_template_json, parameters=[("EnvType", "prod")], ) sqs_conn = boto.sqs.connect_to_region("us-west-1") list(sqs_conn.get_all_queues()).should.have.length_of(1) @mock_cloudformation() @mock_ec2() def test_conditional_if_handling(): dummy_template = { "AWSTemplateFormatVersion": "2010-09-09", "Conditions": { "EnvEqualsPrd": { "Fn::Equals": [ { "Ref": "ENV" }, "prd" ] } }, "Parameters": { "ENV": { "Default": "dev", "Description": "Deployment environment for the stack (dev/prd)", "Type": "String" }, }, "Description": "Stack 1", "Resources": { "App1": { "Properties": { "ImageId": { "Fn::If": [ "EnvEqualsPrd", "ami-00000000", "ami-ffffffff" ] }, }, "Type": "AWS::EC2::Instance" }, } } dummy_template_json = json.dumps(dummy_template) conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack('test_stack1', template_body=dummy_template_json) ec2_conn = boto.ec2.connect_to_region("us-west-1") reservation = ec2_conn.get_all_instances()[0] ec2_instance = reservation.instances[0] ec2_instance.image_id.should.equal("ami-ffffffff") ec2_instance.terminate() conn = boto.cloudformation.connect_to_region("us-west-2") conn.create_stack('test_stack1', template_body=dummy_template_json, parameters=[("ENV", "prd")]) ec2_conn = boto.ec2.connect_to_region("us-west-2") reservation = ec2_conn.get_all_instances()[0] ec2_instance = reservation.instances[0] ec2_instance.image_id.should.equal("ami-00000000") @mock_cloudformation() @mock_ec2() def test_cloudformation_mapping(): dummy_template = { "AWSTemplateFormatVersion": "2010-09-09", "Mappings": { "RegionMap": { "us-east-1": {"32": "ami-6411e20d", "64": "ami-7a11e213"}, "us-west-1": {"32": "ami-c9c7978c", "64": "ami-cfc7978a"}, "eu-west-1": {"32": "ami-37c2f643", "64": "ami-31c2f645"}, "ap-southeast-1": {"32": "ami-66f28c34", "64": "ami-60f28c32"}, "ap-northeast-1": {"32": "ami-9c03a89d", "64": "ami-a003a8a1"} } }, "Resources": { "WebServer": { "Type": "AWS::EC2::Instance", "Properties": { "ImageId": { "Fn::FindInMap": ["RegionMap", {"Ref": "AWS::Region"}, "32"] }, "InstanceType": "m1.small" }, "Type": "AWS::EC2::Instance", }, }, } dummy_template_json = json.dumps(dummy_template) conn = boto.cloudformation.connect_to_region("us-east-1") conn.create_stack('test_stack1', template_body=dummy_template_json) ec2_conn = boto.ec2.connect_to_region("us-east-1") reservation = ec2_conn.get_all_instances()[0] ec2_instance = reservation.instances[0] ec2_instance.image_id.should.equal("ami-6411e20d") conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack('test_stack1', template_body=dummy_template_json) ec2_conn = boto.ec2.connect_to_region("us-west-1") reservation = ec2_conn.get_all_instances()[0] ec2_instance = reservation.instances[0] ec2_instance.image_id.should.equal("ami-c9c7978c") @mock_cloudformation() @mock_route53() def test_route53_roundrobin(): route53_conn = boto.connect_route53() template_json = json.dumps(route53_roundrobin.template) conn = boto.cloudformation.connect_to_region("us-west-1") stack = conn.create_stack( "test_stack", template_body=template_json, ) zones = route53_conn.get_all_hosted_zones()['ListHostedZonesResponse']['HostedZones'] list(zones).should.have.length_of(1) zone_id = zones[0]['Id'] rrsets = route53_conn.get_all_rrsets(zone_id) rrsets.hosted_zone_id.should.equal(zone_id) rrsets.should.have.length_of(2) record_set1 = rrsets[0] record_set1.name.should.equal('test_stack.us-west-1.my_zone.') record_set1.identifier.should.equal("test_stack AWS") record_set1.type.should.equal('CNAME') record_set1.ttl.should.equal('900') record_set1.weight.should.equal('3') record_set1.resource_records[0].should.equal("aws.amazon.com") record_set2 = rrsets[1] record_set2.name.should.equal('test_stack.us-west-1.my_zone.') record_set2.identifier.should.equal("test_stack Amazon") record_set2.type.should.equal('CNAME') record_set2.ttl.should.equal('900') record_set2.weight.should.equal('1') record_set2.resource_records[0].should.equal("www.amazon.com") stack = conn.describe_stacks()[0] output = stack.outputs[0] output.key.should.equal('DomainName') output.value.should.equal('arn:aws:route53:::hostedzone/{0}'.format(zone_id)) @mock_cloudformation() @mock_ec2() @mock_route53() def test_route53_ec2_instance_with_public_ip(): route53_conn = boto.connect_route53() ec2_conn = boto.ec2.connect_to_region("us-west-1") template_json = json.dumps(route53_ec2_instance_with_public_ip.template) conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack( "test_stack", template_body=template_json, ) instance_id = ec2_conn.get_all_reservations()[0].instances[0].id zones = route53_conn.get_all_hosted_zones()['ListHostedZonesResponse']['HostedZones'] list(zones).should.have.length_of(1) zone_id = zones[0]['Id'] rrsets = route53_conn.get_all_rrsets(zone_id) rrsets.should.have.length_of(1) record_set1 = rrsets[0] record_set1.name.should.equal('{0}.us-west-1.my_zone.'.format(instance_id)) record_set1.identifier.should.equal(None) record_set1.type.should.equal('A') record_set1.ttl.should.equal('900') record_set1.weight.should.equal(None) record_set1.resource_records[0].should.equal("10.0.0.25") @mock_cloudformation() @mock_route53() def test_route53_associate_health_check(): route53_conn = boto.connect_route53() template_json = json.dumps(route53_health_check.template) conn = boto.cloudformation.connect_to_region("us-west-1") conn.create_stack( "test_stack", template_body=template_json, ) checks = route53_conn.get_list_health_checks()['ListHealthChecksResponse']['HealthChecks'] list(checks).should.have.length_of(1) check = checks[0] health_check_id = check['Id'] config = check['HealthCheckConfig'] config["FailureThreshold"].should.equal("3") config["IPAddress"].should.equal("10.0.0.4") config["Port"].should.equal("80") config["RequestInterval"].should.equal("10") config["ResourcePath"].should.equal("/") config["Type"].should.equal("HTTP") zones = route53_conn.get_all_hosted_zones()['ListHostedZonesResponse']['HostedZones'] list(zones).should.have.length_of(1) zone_id = zones[0]['Id'] rrsets = route53_conn.get_all_rrsets(zone_id) rrsets.should.have.length_of(1) record_set = rrsets[0] record_set.health_check.should.equal(health_check_id) @mock_cloudformation() @mock_sns() def test_sns_topic(): dummy_template = { "AWSTemplateFormatVersion": "2010-09-09", "Resources": { "MySNSTopic": { "Type": "AWS::SNS::Topic", "Properties": { "Subscription": [ {"Endpoint": "https://example.com", "Protocol": "https"}, ], "TopicName": "my_topics", } } }, "Outputs": { "topic_name": { "Value": {"Fn::GetAtt": ["MySNSTopic", "TopicName"]} }, "topic_arn": { "Value": {"Ref": "MySNSTopic"} }, } } template_json = json.dumps(dummy_template) conn = boto.cloudformation.connect_to_region("us-west-1") stack = conn.create_stack( "test_stack", template_body=template_json, ) sns_conn = boto.sns.connect_to_region("us-west-1") topics = sns_conn.get_all_topics()["ListTopicsResponse"]["ListTopicsResult"]["Topics"] topics.should.have.length_of(1) topic_arn = topics[0]['TopicArn'] topic_arn.should.contain("my_topics") subscriptions = sns_conn.get_all_subscriptions()["ListSubscriptionsResponse"]["ListSubscriptionsResult"]["Subscriptions"] subscriptions.should.have.length_of(1) subscription = subscriptions[0] subscription["TopicArn"].should.equal(topic_arn) subscription["Protocol"].should.equal("https") subscription["SubscriptionArn"].should.contain(topic_arn) subscription["Endpoint"].should.equal("https://example.com") stack = conn.describe_stacks()[0] topic_name_output = [x for x in stack.outputs if x.key == 'topic_name'][0] topic_name_output.value.should.equal("my_topics") topic_arn_output = [x for x in stack.outputs if x.key == 'topic_arn'][0] topic_arn_output.value.should.equal(topic_arn) @mock_cloudformation @mock_ec2 def test_vpc_gateway_attachment_creation_should_attach_itself_to_vpc(): template = { "AWSTemplateFormatVersion": "2010-09-09", "Resources": { "internetgateway": { "Type": "AWS::EC2::InternetGateway" }, "testvpc": { "Type": "AWS::EC2::VPC", "Properties": { "CidrBlock": "10.0.0.0/16", "EnableDnsHostnames": "true", "EnableDnsSupport": "true", "InstanceTenancy": "default" }, }, "vpcgatewayattachment": { "Type": "AWS::EC2::VPCGatewayAttachment", "Properties": { "InternetGatewayId": { "Ref": "internetgateway" }, "VpcId": { "Ref": "testvpc" } }, }, } } template_json = json.dumps(template) cf_conn = boto.cloudformation.connect_to_region("us-west-1") cf_conn.create_stack( "test_stack", template_body=template_json, ) vpc_conn = boto.vpc.connect_to_region("us-west-1") vpc = vpc_conn.get_all_vpcs()[0] igws = vpc_conn.get_all_internet_gateways( filters={'attachment.vpc-id': vpc.id} ) igws.should.have.length_of(1) @mock_cloudformation @mock_ec2 def test_vpc_peering_creation(): vpc_conn = boto.vpc.connect_to_region("us-west-1") vpc_source = vpc_conn.create_vpc("10.0.0.0/16") peer_vpc = vpc_conn.create_vpc("10.1.0.0/16") template = { "AWSTemplateFormatVersion": "2010-09-09", "Resources": { "vpcpeeringconnection": { "Type": "AWS::EC2::VPCPeeringConnection", "Properties": { "PeerVpcId": peer_vpc.id, "VpcId": vpc_source.id, } }, } } template_json = json.dumps(template) cf_conn = boto.cloudformation.connect_to_region("us-west-1") cf_conn.create_stack( "test_stack", template_body=template_json, ) peering_connections = vpc_conn.get_all_vpc_peering_connections() peering_connections.should.have.length_of(1) @mock_cloudformation @mock_ec2 def test_multiple_security_group_ingress_separate_from_security_group_by_id(): template = { "AWSTemplateFormatVersion": "2010-09-09", "Resources": { "test-security-group1": { "Type": "AWS::EC2::SecurityGroup", "Properties": { "GroupDescription": "test security group", "Tags": [ { "Key": "sg-name", "Value": "sg1" } ] }, }, "test-security-group2": { "Type": "AWS::EC2::SecurityGroup", "Properties": { "GroupDescription": "test security group", "Tags": [ { "Key": "sg-name", "Value": "sg2" } ] }, }, "test-sg-ingress": { "Type": "AWS::EC2::SecurityGroupIngress", "Properties": { "GroupId": {"Ref": "test-security-group1"}, "IpProtocol": "tcp", "FromPort": "80", "ToPort": "8080", "SourceSecurityGroupId": {"Ref": "test-security-group2"}, } } } } template_json = json.dumps(template) cf_conn = boto.cloudformation.connect_to_region("us-west-1") cf_conn.create_stack( "test_stack", template_body=template_json, ) ec2_conn = boto.ec2.connect_to_region("us-west-1") security_group1 = ec2_conn.get_all_security_groups(filters={"tag:sg-name": "sg1"})[0] security_group2 = ec2_conn.get_all_security_groups(filters={"tag:sg-name": "sg2"})[0] security_group1.rules.should.have.length_of(1) security_group1.rules[0].grants.should.have.length_of(1) security_group1.rules[0].grants[0].group_id.should.equal(security_group2.id) security_group1.rules[0].ip_protocol.should.equal('tcp') security_group1.rules[0].from_port.should.equal('80') security_group1.rules[0].to_port.should.equal('8080') @mock_cloudformation @mock_ec2 def test_security_group_ingress_separate_from_security_group_by_id(): ec2_conn = boto.ec2.connect_to_region("us-west-1") ec2_conn.create_security_group("test-security-group1", "test security group") template = { "AWSTemplateFormatVersion": "2010-09-09", "Resources": { "test-security-group2": { "Type": "AWS::EC2::SecurityGroup", "Properties": { "GroupDescription": "test security group", "Tags": [ { "Key": "sg-name", "Value": "sg2" } ] }, }, "test-sg-ingress": { "Type": "AWS::EC2::SecurityGroupIngress", "Properties": { "GroupName": "test-security-group1", "IpProtocol": "tcp", "FromPort": "80", "ToPort": "8080", "SourceSecurityGroupId": {"Ref": "test-security-group2"}, } } } } template_json = json.dumps(template) cf_conn = boto.cloudformation.connect_to_region("us-west-1") cf_conn.create_stack( "test_stack", template_body=template_json, ) security_group1 = ec2_conn.get_all_security_groups(groupnames=["test-security-group1"])[0] security_group2 = ec2_conn.get_all_security_groups(filters={"tag:sg-name": "sg2"})[0] security_group1.rules.should.have.length_of(1) security_group1.rules[0].grants.should.have.length_of(1) security_group1.rules[0].grants[0].group_id.should.equal(security_group2.id) security_group1.rules[0].ip_protocol.should.equal('tcp') security_group1.rules[0].from_port.should.equal('80') security_group1.rules[0].to_port.should.equal('8080') @mock_cloudformation @mock_ec2 def test_security_group_ingress_separate_from_security_group_by_id_using_vpc(): vpc_conn = boto.vpc.connect_to_region("us-west-1") vpc = vpc_conn.create_vpc("10.0.0.0/16") template = { "AWSTemplateFormatVersion": "2010-09-09", "Resources": { "test-security-group1": { "Type": "AWS::EC2::SecurityGroup", "Properties": { "GroupDescription": "test security group", "VpcId": vpc.id, "Tags": [ { "Key": "sg-name", "Value": "sg1" } ] }, }, "test-security-group2": { "Type": "AWS::EC2::SecurityGroup", "Properties": { "GroupDescription": "test security group", "VpcId": vpc.id, "Tags": [ { "Key": "sg-name", "Value": "sg2" } ] }, }, "test-sg-ingress": { "Type": "AWS::EC2::SecurityGroupIngress", "Properties": { "GroupId": {"Ref": "test-security-group1"}, "VpcId": vpc.id, "IpProtocol": "tcp", "FromPort": "80", "ToPort": "8080", "SourceSecurityGroupId": {"Ref": "test-security-group2"}, } } } } template_json = json.dumps(template) cf_conn = boto.cloudformation.connect_to_region("us-west-1") cf_conn.create_stack( "test_stack", template_body=template_json, ) security_group1 = vpc_conn.get_all_security_groups(filters={"tag:sg-name": "sg1"})[0] security_group2 = vpc_conn.get_all_security_groups(filters={"tag:sg-name": "sg2"})[0] security_group1.rules.should.have.length_of(1) security_group1.rules[0].grants.should.have.length_of(1) security_group1.rules[0].grants[0].group_id.should.equal(security_group2.id) security_group1.rules[0].ip_protocol.should.equal('tcp') security_group1.rules[0].from_port.should.equal('80') security_group1.rules[0].to_port.should.equal('8080') @mock_cloudformation @mock_ec2 def test_subnets_should_be_created_with_availability_zone(): vpc_conn = boto.vpc.connect_to_region('us-west-1') vpc = vpc_conn.create_vpc("10.0.0.0/16") subnet_template = { "AWSTemplateFormatVersion": "2010-09-09", "Resources": { "testSubnet": { "Type": "AWS::EC2::Subnet", "Properties": { "VpcId": vpc.id, "CidrBlock": "10.0.0.0/24", "AvailabilityZone": "us-west-1b", } } } } cf_conn = boto.cloudformation.connect_to_region("us-west-1") template_json = json.dumps(subnet_template) cf_conn.create_stack( "test_stack", template_body=template_json, ) subnet = vpc_conn.get_all_subnets(filters={'cidrBlock': '10.0.0.0/24'})[0] subnet.availability_zone.should.equal('us-west-1b')