diff --git a/tests/test_ec2/test_carrier_gateways.py b/tests/test_ec2/test_carrier_gateways.py new file mode 100644 index 000000000..c01cafa72 --- /dev/null +++ b/tests/test_ec2/test_carrier_gateways.py @@ -0,0 +1,89 @@ +import boto3 +import sure # noqa +import pytest +from botocore.exceptions import ClientError +from moto import mock_ec2 +from moto.core import ACCOUNT_ID + + +@mock_ec2 +def test_describe_carrier_gateways_none(): + ec2 = boto3.client("ec2", region_name="us-east-1") + ec2.describe_carrier_gateways()["CarrierGateways"].should.equal([]) + + +@mock_ec2 +def test_describe_carrier_gateways_multiple(): + client = boto3.client("ec2", region_name="us-east-1") + ec2 = boto3.resource("ec2", region_name="us-east-1") + vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") + cg1 = client.create_carrier_gateway(VpcId=vpc.id)["CarrierGateway"] + cg2 = client.create_carrier_gateway(VpcId=vpc.id)["CarrierGateway"] + + client.describe_carrier_gateways()["CarrierGateways"].should.have.length_of(2) + + find_one = client.describe_carrier_gateways( + CarrierGatewayIds=[cg1["CarrierGatewayId"]] + )["CarrierGateways"] + find_one.should.have.length_of(1) + find_one[0]["CarrierGatewayId"].should.equal(cg1["CarrierGatewayId"]) + + find_one = client.describe_carrier_gateways( + CarrierGatewayIds=[cg2["CarrierGatewayId"], "non-existant"] + )["CarrierGateways"] + find_one.should.have.length_of(1) + find_one[0]["CarrierGatewayId"].should.equal(cg2["CarrierGatewayId"]) + + +@mock_ec2 +def test_create_carrier_gateways_without_tags(): + client = boto3.client("ec2", region_name="us-east-1") + ec2 = boto3.resource("ec2", region_name="us-east-1") + vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") + cg = client.create_carrier_gateway(VpcId=vpc.id)["CarrierGateway"] + + cg.should.have.key("CarrierGatewayId").match("cagw-[a-z0-9]+") + cg.should.have.key("VpcId").equal(vpc.id) + cg.should.have.key("State").equal("available") + cg.should.have.key("OwnerId").equal(ACCOUNT_ID) + cg.should.have.key("Tags").equal([]) + + +@mock_ec2 +def test_create_carrier_gateways_with_tags(): + client = boto3.client("ec2", region_name="us-east-1") + ec2 = boto3.resource("ec2", region_name="us-east-1") + vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") + cg = client.create_carrier_gateway( + VpcId=vpc.id, + TagSpecifications=[ + {"ResourceType": "CarrierGateway", "Tags": [{"Key": "tk", "Value": "tv"}]} + ], + )["CarrierGateway"] + + cg.should.have.key("CarrierGatewayId").match("cagw-[a-z0-9]+") + cg.should.have.key("VpcId").equal(vpc.id) + cg.should.have.key("State").equal("available") + cg.should.have.key("OwnerId").equal(ACCOUNT_ID) + cg.should.have.key("Tags").should.equal([{"Key": "tk", "Value": "tv"}]) + + +@mock_ec2 +def test_create_carrier_gateways_invalid_vpc(): + ec2 = boto3.client("ec2", region_name="us-east-1") + with pytest.raises(ClientError) as exc: + ec2.create_carrier_gateway(VpcId="vpc-asdf") + err = exc.value.response["Error"] + err["Code"].should.equal("InvalidVpcID.NotFound") + err["Message"].should.equal("VpcID vpc-asdf does not exist.") + + +@mock_ec2 +def test_delete_carrier_gateways(): + client = boto3.client("ec2", region_name="us-east-1") + ec2 = boto3.resource("ec2", region_name="us-east-1") + vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") + cg = client.create_carrier_gateway(VpcId=vpc.id)["CarrierGateway"] + client.delete_carrier_gateway(CarrierGatewayId=cg["CarrierGatewayId"]) + + client.describe_carrier_gateways()["CarrierGateways"].should.equal([]) diff --git a/tests/test_ec2/test_security_groups.py b/tests/test_ec2/test_security_groups.py index 2542051c9..432d4441a 100644 --- a/tests/test_ec2/test_security_groups.py +++ b/tests/test_ec2/test_security_groups.py @@ -1071,6 +1071,87 @@ def test_revoke_security_group_egress(): sg.ip_permissions_egress.should.have.length_of(0) +@mock_ec2 +def test_update_security_group_rule_descriptions_egress(): + ec2 = boto3.resource("ec2", "us-east-1") + client = boto3.client("ec2", "us-east-1") + vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") + sg = ec2.create_security_group( + Description="Test SG", GroupName="test-sg", VpcId=vpc.id + ) + sg_id = sg.id + + ip_ranges = client.describe_security_groups(GroupIds=[sg_id])["SecurityGroups"][0][ + "IpPermissionsEgress" + ][0]["IpRanges"] + ip_ranges.should.have.length_of(1) + ip_ranges[0].should.equal({"CidrIp": "0.0.0.0/0"}) + + client.update_security_group_rule_descriptions_egress( + GroupName="test-sg", + IpPermissions=[ + { + "IpProtocol": "-1", + "IpRanges": [{"CidrIp": "0.0.0.0/0", "Description": "my d3scription"}], + "UserIdGroupPairs": [], + "Ipv6Ranges": [], + "PrefixListIds": [], + } + ], + ) + + ip_ranges = client.describe_security_groups(GroupIds=[sg_id])["SecurityGroups"][0][ + "IpPermissionsEgress" + ][0]["IpRanges"] + ip_ranges.should.have.length_of(1) + ip_ranges[0].should.equal({"CidrIp": "0.0.0.0/0", "Description": "my d3scription"}) + + +@mock_ec2 +def test_update_security_group_rule_descriptions_ingress(): + ec2 = boto3.resource("ec2", "us-east-1") + client = boto3.client("ec2", "us-east-1") + vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") + sg = ec2.create_security_group( + Description="Test SG", GroupName="test-sg", VpcId=vpc.id + ) + sg_id = sg.id + + ip_permissions = [ + { + "IpProtocol": "tcp", + "FromPort": 27017, + "ToPort": 27017, + "IpRanges": [{"CidrIp": "1.2.3.4/32", "Description": "first desc"}], + } + ] + client.authorize_security_group_ingress(GroupId=sg_id, IpPermissions=ip_permissions) + + ip_ranges = client.describe_security_groups(GroupIds=[sg_id])["SecurityGroups"][0][ + "IpPermissions" + ][0]["IpRanges"] + ip_ranges.should.have.length_of(1) + ip_ranges[0].should.equal({"CidrIp": "1.2.3.4/32", "Description": "first desc"}) + + client.update_security_group_rule_descriptions_ingress( + GroupName="test-sg", + IpPermissions=[ + { + "IpProtocol": "tcp", + "FromPort": 27017, + "ToPort": 27017, + "IpRanges": [{"CidrIp": "1.2.3.4/32", "Description": "second desc"}], + } + ], + ) + + ip_ranges = client.describe_security_groups(GroupIds=[sg_id])["SecurityGroups"][0][ + "IpPermissions" + ][0]["IpRanges"] + ip_ranges.should.have.length_of(1) + ip_ranges[0].should.equal({"CidrIp": "1.2.3.4/32", "Description": "second desc"}) + + @mock_ec2 def test_non_existent_security_group_raises_error_on_authorize(): client = boto3.client("ec2", "us-east-1") diff --git a/tests/test_ec2/test_subnets.py b/tests/test_ec2/test_subnets.py index b373de308..3f439a5c9 100644 --- a/tests/test_ec2/test_subnets.py +++ b/tests/test_ec2/test_subnets.py @@ -731,3 +731,64 @@ def test_describe_subnets_by_state(): ).get("Subnets", []) for subnet in subnets: subnet["State"].should.equal("available") + + +@mock_ec2 +def test_associate_subnet_cidr_block(): + ec2 = boto3.resource("ec2", region_name="us-west-1") + client = boto3.client("ec2", region_name="us-west-1") + + vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") + subnet_object = ec2.create_subnet( + VpcId=vpc.id, CidrBlock="10.0.0.0/24", AvailabilityZone="us-west-1a" + ) + + subnets = client.describe_subnets(SubnetIds=[subnet_object.id])["Subnets"] + association_set = subnets[0]["Ipv6CidrBlockAssociationSet"] + association_set.should.equal([]) + + res = client.associate_subnet_cidr_block( + Ipv6CidrBlock="1080::1:200C:417A/112", SubnetId=subnet_object.id + ) + res.should.have.key("Ipv6CidrBlockAssociation") + association = res["Ipv6CidrBlockAssociation"] + association.should.have.key("AssociationId").match("subnet-cidr-assoc-[a-z0-9]+") + association.should.have.key("Ipv6CidrBlock").equals("1080::1:200C:417A/112") + association.should.have.key("Ipv6CidrBlockState").equals({"State": "associated"}) + + subnets = client.describe_subnets(SubnetIds=[subnet_object.id])["Subnets"] + association_set = subnets[0]["Ipv6CidrBlockAssociationSet"] + association_set.should.have.length_of(1) + association_set[0].should.have.key("AssociationId").equal( + association["AssociationId"] + ) + association_set[0].should.have.key("Ipv6CidrBlock").equals("1080::1:200C:417A/112") + + +@mock_ec2 +def test_disassociate_subnet_cidr_block(): + ec2 = boto3.resource("ec2", region_name="us-west-1") + client = boto3.client("ec2", region_name="us-west-1") + + vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") + subnet_object = ec2.create_subnet( + VpcId=vpc.id, CidrBlock="10.0.0.0/24", AvailabilityZone="us-west-1a" + ) + + client.associate_subnet_cidr_block( + Ipv6CidrBlock="1080::1:200C:417A/111", SubnetId=subnet_object.id + ) + association_id = client.associate_subnet_cidr_block( + Ipv6CidrBlock="1080::1:200C:417A/999", SubnetId=subnet_object.id + )["Ipv6CidrBlockAssociation"]["AssociationId"] + + subnets = client.describe_subnets(SubnetIds=[subnet_object.id])["Subnets"] + association_set = subnets[0]["Ipv6CidrBlockAssociationSet"] + association_set.should.have.length_of(2) + + client.disassociate_subnet_cidr_block(AssociationId=association_id) + + subnets = client.describe_subnets(SubnetIds=[subnet_object.id])["Subnets"] + association_set = subnets[0]["Ipv6CidrBlockAssociationSet"] + association_set.should.have.length_of(1) + association_set[0]["Ipv6CidrBlock"].should.equal("1080::1:200C:417A/111")