1241 lines
43 KiB
Python
1241 lines
43 KiB
Python
import json
|
|
import random
|
|
from unittest import SkipTest
|
|
from uuid import uuid4
|
|
|
|
import boto3
|
|
import pytest
|
|
from botocore.exceptions import ClientError
|
|
|
|
from moto import mock_aws, settings
|
|
|
|
from .test_tags import retrieve_all_tagged
|
|
|
|
SAMPLE_DOMAIN_NAME = "example.com"
|
|
SAMPLE_NAME_SERVERS = ["10.0.0.6", "10.0.0.7"]
|
|
|
|
|
|
@mock_aws
|
|
def test_creating_a_vpc_in_empty_region_does_not_make_this_vpc_the_default():
|
|
if settings.TEST_SERVER_MODE:
|
|
raise SkipTest("Lets not start deleting VPC's while other tests are using it")
|
|
# Delete VPC that's created by default
|
|
client = boto3.client("ec2", region_name="eu-north-1")
|
|
all_vpcs = retrieve_all_vpcs(client)
|
|
for vpc in all_vpcs:
|
|
client.delete_vpc(VpcId=vpc["VpcId"])
|
|
# create vpc
|
|
client.create_vpc(CidrBlock="10.0.0.0/16")
|
|
# verify this is not the default
|
|
all_vpcs = retrieve_all_vpcs(client)
|
|
assert len(all_vpcs) == 1
|
|
assert all_vpcs[0]["IsDefault"] is False
|
|
|
|
|
|
@mock_aws
|
|
def test_create_default_vpc():
|
|
if settings.TEST_SERVER_MODE:
|
|
raise SkipTest("Lets not start deleting VPC's while other tests are using it")
|
|
# Delete VPC that's created by default
|
|
client = boto3.client("ec2", region_name="eu-north-1")
|
|
all_vpcs = retrieve_all_vpcs(client)
|
|
for vpc in all_vpcs:
|
|
client.delete_vpc(VpcId=vpc["VpcId"])
|
|
# create default vpc
|
|
client.create_default_vpc()
|
|
# verify this is the default
|
|
all_vpcs = retrieve_all_vpcs(client)
|
|
assert len(all_vpcs) == 1
|
|
assert all_vpcs[0]["IsDefault"] is True
|
|
|
|
|
|
@mock_aws
|
|
def test_create_multiple_default_vpcs():
|
|
client = boto3.client("ec2", region_name="eu-north-1")
|
|
with pytest.raises(ClientError) as exc:
|
|
client.create_default_vpc()
|
|
err = exc.value.response["Error"]
|
|
assert err["Code"] == "DefaultVpcAlreadyExists"
|
|
assert (
|
|
err["Message"]
|
|
== "A Default VPC already exists for this account in this region."
|
|
)
|
|
|
|
|
|
@mock_aws
|
|
def test_create_and_delete_vpc():
|
|
ec2 = boto3.resource("ec2", region_name="eu-north-1")
|
|
client = boto3.client("ec2", region_name="eu-north-1")
|
|
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
assert vpc.cidr_block == "10.0.0.0/16"
|
|
|
|
all_vpcs = retrieve_all_vpcs(client)
|
|
assert vpc.id in [v["VpcId"] for v in all_vpcs]
|
|
|
|
vpc.delete()
|
|
|
|
all_vpcs = retrieve_all_vpcs(client)
|
|
assert vpc.id not in [v["VpcId"] for v in all_vpcs]
|
|
|
|
with pytest.raises(ClientError) as ex:
|
|
client.delete_vpc(VpcId="vpc-1234abcd")
|
|
assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
|
|
assert "RequestId" in ex.value.response["ResponseMetadata"]
|
|
assert ex.value.response["Error"]["Code"] == "InvalidVpcID.NotFound"
|
|
|
|
|
|
@mock_aws
|
|
def test_vpc_defaults():
|
|
ec2 = boto3.resource("ec2", region_name="eu-north-1")
|
|
client = boto3.client("ec2", region_name="eu-north-1")
|
|
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
|
|
filters = [{"Name": "vpc-id", "Values": [vpc.id]}]
|
|
|
|
assert len(client.describe_route_tables(Filters=filters)["RouteTables"]) == 1
|
|
assert len(client.describe_security_groups(Filters=filters)["SecurityGroups"]) == 1
|
|
|
|
vpc.delete()
|
|
|
|
assert len(client.describe_route_tables(Filters=filters)["RouteTables"]) == 0
|
|
assert len(client.describe_security_groups(Filters=filters)["SecurityGroups"]) == 0
|
|
|
|
|
|
@mock_aws
|
|
def test_vpc_isdefault_filter():
|
|
ec2 = boto3.resource("ec2", region_name="eu-west-1")
|
|
client = boto3.client("ec2", region_name="eu-west-1")
|
|
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
default_vpcs = client.describe_vpcs(
|
|
Filters=[{"Name": "isDefault", "Values": ["true"]}]
|
|
)["Vpcs"]
|
|
assert len(default_vpcs) == 1
|
|
|
|
vpc.delete()
|
|
|
|
default_vpcs = client.describe_vpcs(
|
|
Filters=[{"Name": "isDefault", "Values": ["true"]}]
|
|
)["Vpcs"]
|
|
assert len(default_vpcs) == 1
|
|
|
|
|
|
@mock_aws
|
|
def test_multiple_vpcs_default_filter():
|
|
ec2 = boto3.resource("ec2", region_name="eu-west-1")
|
|
client = boto3.client("ec2", region_name="eu-west-1")
|
|
ec2.create_vpc(CidrBlock="10.8.0.0/16")
|
|
ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
ec2.create_vpc(CidrBlock="192.168.0.0/16")
|
|
default_vpcs = retrieve_all_vpcs(
|
|
client, [{"Name": "isDefault", "Values": ["true"]}]
|
|
)
|
|
assert "172.31.0.0/16" in [v["CidrBlock"] for v in default_vpcs]
|
|
|
|
|
|
@mock_aws
|
|
def test_vpc_state_available_filter():
|
|
ec2 = boto3.resource("ec2", region_name="eu-west-1")
|
|
client = boto3.client("ec2", region_name="eu-west-1")
|
|
vpc1 = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
vpc2 = ec2.create_vpc(CidrBlock="10.1.0.0/16")
|
|
|
|
available = retrieve_all_vpcs(client, [{"Name": "state", "Values": ["available"]}])
|
|
assert vpc1.id in [v["VpcId"] for v in available]
|
|
assert vpc2.id in [v["VpcId"] for v in available]
|
|
|
|
vpc1.delete()
|
|
|
|
available = retrieve_all_vpcs(client, [{"Name": "state", "Values": ["available"]}])
|
|
assert vpc1.id not in [v["VpcId"] for v in available]
|
|
assert vpc2.id in [v["VpcId"] for v in available]
|
|
|
|
|
|
def retrieve_all_vpcs(client, filters=[]): # pylint: disable=W0102
|
|
resp = client.describe_vpcs(Filters=filters)
|
|
all_vpcs = resp["Vpcs"]
|
|
token = resp.get("NextToken")
|
|
while token:
|
|
resp = client.describe_vpcs(Filters=filters, NextToken=token)
|
|
all_vpcs.extend(resp["Vpcs"])
|
|
token = resp.get("NextToken")
|
|
return all_vpcs
|
|
|
|
|
|
@mock_aws
|
|
def test_vpc_tagging():
|
|
ec2 = boto3.resource("ec2", region_name="eu-west-1")
|
|
client = boto3.client("ec2", region_name="eu-west-1")
|
|
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
|
|
vpc.create_tags(Tags=[{"Key": "a key", "Value": "some value"}])
|
|
|
|
all_tags = retrieve_all_tagged(client)
|
|
ours = [t for t in all_tags if t["ResourceId"] == vpc.id][0]
|
|
assert ours["Key"] == "a key"
|
|
assert ours["Value"] == "some value"
|
|
|
|
# Refresh the vpc
|
|
vpc = client.describe_vpcs(VpcIds=[vpc.id])["Vpcs"][0]
|
|
assert vpc["Tags"] == [{"Key": "a key", "Value": "some value"}]
|
|
|
|
|
|
@mock_aws
|
|
def test_vpc_get_by_id():
|
|
ec2 = boto3.resource("ec2", region_name="eu-west-1")
|
|
client = boto3.client("ec2", region_name="eu-west-1")
|
|
vpc1 = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
vpc2 = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
|
|
vpcs = client.describe_vpcs(VpcIds=[vpc1.id, vpc2.id])["Vpcs"]
|
|
assert len(vpcs) == 2
|
|
vpc_ids = tuple(map(lambda v: v["VpcId"], vpcs))
|
|
assert vpc1.id in vpc_ids
|
|
assert vpc2.id in vpc_ids
|
|
|
|
with pytest.raises(ClientError) as ex:
|
|
client.describe_vpcs(VpcIds=["vpc-does_not_exist"])
|
|
assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
|
|
assert "RequestId" in ex.value.response["ResponseMetadata"]
|
|
assert ex.value.response["Error"]["Code"] == "InvalidVpcID.NotFound"
|
|
|
|
|
|
@mock_aws
|
|
def test_vpc_get_by_cidr_block():
|
|
ec2 = boto3.resource("ec2", region_name="eu-west-1")
|
|
client = boto3.client("ec2", region_name="eu-west-1")
|
|
random_ip = ".".join(map(str, (random.randint(0, 99) for _ in range(4))))
|
|
random_cidr = f"{random_ip}/16"
|
|
vpc1 = ec2.create_vpc(CidrBlock=random_cidr)
|
|
vpc2 = ec2.create_vpc(CidrBlock=random_cidr)
|
|
ec2.create_vpc(CidrBlock="10.0.0.0/24")
|
|
|
|
vpcs = client.describe_vpcs(Filters=[{"Name": "cidr", "Values": [random_cidr]}])[
|
|
"Vpcs"
|
|
]
|
|
assert set([vpc["VpcId"] for vpc in vpcs]) == {vpc1.id, vpc2.id}
|
|
|
|
|
|
@mock_aws
|
|
def test_vpc_get_by_dhcp_options_id():
|
|
ec2 = boto3.resource("ec2", region_name="us-east-1")
|
|
client = boto3.client("ec2", region_name="us-east-1")
|
|
dhcp_options = ec2.create_dhcp_options(
|
|
DhcpConfigurations=[
|
|
{"Key": "domain-name", "Values": [SAMPLE_DOMAIN_NAME]},
|
|
{"Key": "domain-name-servers", "Values": SAMPLE_NAME_SERVERS},
|
|
]
|
|
)
|
|
vpc1 = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
vpc2 = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
ec2.create_vpc(CidrBlock="10.0.0.0/24")
|
|
|
|
client.associate_dhcp_options(DhcpOptionsId=dhcp_options.id, VpcId=vpc1.id)
|
|
client.associate_dhcp_options(DhcpOptionsId=dhcp_options.id, VpcId=vpc2.id)
|
|
|
|
vpcs = client.describe_vpcs(
|
|
Filters=[{"Name": "dhcp-options-id", "Values": [dhcp_options.id]}]
|
|
)["Vpcs"]
|
|
assert len(vpcs) == 2
|
|
vpc_ids = tuple(map(lambda v: v["VpcId"], vpcs))
|
|
assert vpc1.id in vpc_ids
|
|
assert vpc2.id in vpc_ids
|
|
|
|
|
|
@mock_aws
|
|
def test_vpc_get_by_tag():
|
|
ec2 = boto3.resource("ec2", region_name="us-east-1")
|
|
client = boto3.client("ec2", region_name="us-east-1")
|
|
vpc1 = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
vpc2 = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
vpc3 = ec2.create_vpc(CidrBlock="10.0.0.0/24")
|
|
|
|
value1 = str(uuid4())
|
|
vpc1.create_tags(Tags=[{"Key": "Name", "Value": value1}])
|
|
vpc2.create_tags(Tags=[{"Key": "Name", "Value": value1}])
|
|
vpc3.create_tags(Tags=[{"Key": "Name", "Value": "TestVPC2"}])
|
|
|
|
vpcs = client.describe_vpcs(Filters=[{"Name": "tag:Name", "Values": [value1]}])[
|
|
"Vpcs"
|
|
]
|
|
assert len(vpcs) == 2
|
|
assert set([vpc["VpcId"] for vpc in vpcs]) == {vpc1.id, vpc2.id}
|
|
|
|
|
|
@mock_aws
|
|
def test_vpc_get_by_tag_key_superset():
|
|
ec2 = boto3.resource("ec2", region_name="us-east-1")
|
|
client = boto3.client("ec2", region_name="us-east-1")
|
|
vpc1 = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
vpc2 = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
vpc3 = ec2.create_vpc(CidrBlock="10.0.0.0/24")
|
|
|
|
tag_key = str(uuid4())[0:6]
|
|
vpc1.create_tags(Tags=[{"Key": tag_key, "Value": "TestVPC"}])
|
|
vpc1.create_tags(Tags=[{"Key": "Key", "Value": "TestVPC2"}])
|
|
vpc2.create_tags(Tags=[{"Key": tag_key, "Value": "TestVPC"}])
|
|
vpc2.create_tags(Tags=[{"Key": "Key", "Value": "TestVPC2"}])
|
|
vpc3.create_tags(Tags=[{"Key": "Key", "Value": "TestVPC2"}])
|
|
|
|
vpcs = client.describe_vpcs(Filters=[{"Name": "tag-key", "Values": [tag_key]}])[
|
|
"Vpcs"
|
|
]
|
|
assert len(vpcs) == 2
|
|
assert set([vpc["VpcId"] for vpc in vpcs]) == {vpc1.id, vpc2.id}
|
|
|
|
|
|
@mock_aws
|
|
def test_vpc_get_by_tag_key_subset():
|
|
ec2 = boto3.resource("ec2", region_name="us-east-1")
|
|
client = boto3.client("ec2", region_name="us-east-1")
|
|
vpc1 = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
vpc2 = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
vpc3 = ec2.create_vpc(CidrBlock="10.0.0.0/24")
|
|
|
|
tag_key1 = str(uuid4())[0:6]
|
|
tag_key2 = str(uuid4())[0:6]
|
|
vpc1.create_tags(Tags=[{"Key": tag_key1, "Value": "TestVPC"}])
|
|
vpc1.create_tags(Tags=[{"Key": tag_key2, "Value": "TestVPC2"}])
|
|
vpc2.create_tags(Tags=[{"Key": tag_key1, "Value": "TestVPC"}])
|
|
vpc2.create_tags(Tags=[{"Key": tag_key2, "Value": "TestVPC2"}])
|
|
vpc3.create_tags(Tags=[{"Key": "Test", "Value": "TestVPC2"}])
|
|
|
|
vpcs = client.describe_vpcs(
|
|
Filters=[{"Name": "tag-key", "Values": [tag_key1, tag_key2]}]
|
|
)["Vpcs"]
|
|
assert len(vpcs) == 2
|
|
assert set([vpc["VpcId"] for vpc in vpcs]) == {vpc1.id, vpc2.id}
|
|
|
|
|
|
@mock_aws
|
|
def test_vpc_get_by_tag_value_superset():
|
|
ec2 = boto3.resource("ec2", region_name="us-east-1")
|
|
client = boto3.client("ec2", region_name="us-east-1")
|
|
vpc1 = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
vpc2 = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
vpc3 = ec2.create_vpc(CidrBlock="10.0.0.0/24")
|
|
|
|
tag_value = str(uuid4())
|
|
vpc1.create_tags(Tags=[{"Key": "Name", "Value": tag_value}])
|
|
vpc1.create_tags(Tags=[{"Key": "Key", "Value": "TestVPC2"}])
|
|
vpc2.create_tags(Tags=[{"Key": "Name", "Value": tag_value}])
|
|
vpc2.create_tags(Tags=[{"Key": "Key", "Value": "TestVPC2"}])
|
|
vpc3.create_tags(Tags=[{"Key": "Key", "Value": "TestVPC2"}])
|
|
|
|
vpcs = client.describe_vpcs(Filters=[{"Name": "tag-value", "Values": [tag_value]}])[
|
|
"Vpcs"
|
|
]
|
|
assert len(vpcs) == 2
|
|
assert set([vpc["VpcId"] for vpc in vpcs]) == {vpc1.id, vpc2.id}
|
|
|
|
|
|
@mock_aws
|
|
def test_vpc_get_by_tag_value_subset():
|
|
ec2 = boto3.resource("ec2", region_name="us-east-1")
|
|
client = boto3.client("ec2", region_name="us-east-1")
|
|
vpc1 = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
vpc2 = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
ec2.create_vpc(CidrBlock="10.0.0.0/24")
|
|
|
|
value1 = str(uuid4())[0:6]
|
|
value2 = str(uuid4())[0:6]
|
|
vpc1.create_tags(Tags=[{"Key": "Name", "Value": value1}])
|
|
vpc1.create_tags(Tags=[{"Key": "Key", "Value": value2}])
|
|
vpc2.create_tags(Tags=[{"Key": "Name", "Value": value1}])
|
|
vpc2.create_tags(Tags=[{"Key": "Key", "Value": value2}])
|
|
|
|
vpcs = client.describe_vpcs(
|
|
Filters=[{"Name": "tag-value", "Values": [value1, value2]}]
|
|
)["Vpcs"]
|
|
assert len(vpcs) == 2
|
|
vpc_ids = tuple(map(lambda v: v["VpcId"], vpcs))
|
|
assert vpc1.id in vpc_ids
|
|
assert vpc2.id in vpc_ids
|
|
|
|
|
|
@mock_aws
|
|
def test_default_vpc():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
# Create the default VPC
|
|
default_vpc = list(ec2.vpcs.all())[0]
|
|
assert default_vpc.cidr_block == "172.31.0.0/16"
|
|
assert default_vpc.instance_tenancy == "default"
|
|
default_vpc.reload()
|
|
assert default_vpc.is_default is True
|
|
|
|
# Test default values for VPC attributes
|
|
response = default_vpc.describe_attribute(Attribute="enableDnsSupport")
|
|
attr = response.get("EnableDnsSupport")
|
|
assert attr.get("Value") is True
|
|
|
|
response = default_vpc.describe_attribute(Attribute="enableDnsHostnames")
|
|
attr = response.get("EnableDnsHostnames")
|
|
assert attr.get("Value") is True
|
|
|
|
response = default_vpc.describe_attribute(
|
|
Attribute="enableNetworkAddressUsageMetrics"
|
|
)
|
|
attr = response.get("EnableNetworkAddressUsageMetrics")
|
|
assert attr.get("Value") is False
|
|
|
|
|
|
@mock_aws
|
|
def test_non_default_vpc():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
# Create the default VPC - this already exists when backend instantiated!
|
|
# ec2.create_vpc(CidrBlock='172.31.0.0/16')
|
|
|
|
# Create the non default VPC
|
|
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
vpc.reload()
|
|
assert vpc.is_default is False
|
|
|
|
# Test default instance_tenancy
|
|
assert vpc.instance_tenancy == "default"
|
|
|
|
# Test default values for VPC attributes
|
|
response = vpc.describe_attribute(Attribute="enableDnsSupport")
|
|
attr = response.get("EnableDnsSupport")
|
|
assert attr.get("Value") is True
|
|
|
|
response = vpc.describe_attribute(Attribute="enableDnsHostnames")
|
|
attr = response.get("EnableDnsHostnames")
|
|
assert attr.get("Value") is False
|
|
|
|
response = vpc.describe_attribute(Attribute="enableNetworkAddressUsageMetrics")
|
|
attr = response.get("EnableNetworkAddressUsageMetrics")
|
|
assert attr.get("Value") is False
|
|
|
|
# Check Primary CIDR Block Associations
|
|
cidr_block_association_set = next(iter(vpc.cidr_block_association_set), None)
|
|
assert cidr_block_association_set["CidrBlockState"]["State"] == "associated"
|
|
assert cidr_block_association_set["CidrBlock"] == vpc.cidr_block
|
|
assert "vpc-cidr-assoc" in cidr_block_association_set["AssociationId"]
|
|
|
|
|
|
@mock_aws
|
|
def test_vpc_dedicated_tenancy():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
# Create the default VPC
|
|
ec2.create_vpc(CidrBlock="172.31.0.0/16")
|
|
|
|
# Create the non default VPC
|
|
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16", InstanceTenancy="dedicated")
|
|
vpc.reload()
|
|
assert vpc.is_default is False
|
|
|
|
assert vpc.instance_tenancy == "dedicated"
|
|
|
|
|
|
@mock_aws
|
|
def test_vpc_modify_tenancy_unknown():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
ec2_client = boto3.client("ec2", region_name="us-west-1")
|
|
|
|
# Create the default VPC
|
|
ec2.create_vpc(CidrBlock="172.31.0.0/16")
|
|
|
|
# Create the non default VPC
|
|
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16", InstanceTenancy="dedicated")
|
|
assert vpc.instance_tenancy == "dedicated"
|
|
|
|
with pytest.raises(ClientError) as ex:
|
|
ec2_client.modify_vpc_tenancy(VpcId=vpc.id, InstanceTenancy="unknown")
|
|
err = ex.value.response["Error"]
|
|
assert err["Message"] == "The tenancy value unknown is not supported."
|
|
assert err["Code"] == "UnsupportedTenancy"
|
|
|
|
ec2_client.modify_vpc_tenancy(VpcId=vpc.id, InstanceTenancy="default")
|
|
|
|
vpc.reload()
|
|
|
|
assert vpc.instance_tenancy == "default"
|
|
|
|
|
|
@mock_aws
|
|
def test_vpc_modify_enable_dns_support():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
# Create the default VPC
|
|
ec2.create_vpc(CidrBlock="172.31.0.0/16")
|
|
|
|
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
|
|
# Test default values for VPC attributes
|
|
response = vpc.describe_attribute(Attribute="enableDnsSupport")
|
|
attr = response.get("EnableDnsSupport")
|
|
assert attr.get("Value") is not None
|
|
|
|
vpc.modify_attribute(EnableDnsSupport={"Value": False})
|
|
|
|
response = vpc.describe_attribute(Attribute="enableDnsSupport")
|
|
attr = response.get("EnableDnsSupport")
|
|
assert attr.get("Value") is False
|
|
|
|
|
|
@mock_aws
|
|
def test_vpc_modify_enable_dns_hostnames():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
# Create the default VPC
|
|
ec2.create_vpc(CidrBlock="172.31.0.0/16")
|
|
|
|
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
# Test default values for VPC attributes
|
|
response = vpc.describe_attribute(Attribute="enableDnsHostnames")
|
|
attr = response.get("EnableDnsHostnames")
|
|
assert attr.get("Value") is False
|
|
|
|
vpc.modify_attribute(EnableDnsHostnames={"Value": True})
|
|
|
|
response = vpc.describe_attribute(Attribute="enableDnsHostnames")
|
|
attr = response.get("EnableDnsHostnames")
|
|
assert attr.get("Value") is not None
|
|
|
|
|
|
@mock_aws
|
|
def test_vpc_modify_enable_network_address_usage_metrics():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
# Create the default VPC
|
|
ec2.create_vpc(CidrBlock="172.31.0.0/16")
|
|
|
|
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
# Test default values for VPC attributes
|
|
response = vpc.describe_attribute(Attribute="enableNetworkAddressUsageMetrics")
|
|
attr = response.get("EnableNetworkAddressUsageMetrics")
|
|
assert attr.get("Value") is False
|
|
|
|
vpc.modify_attribute(EnableNetworkAddressUsageMetrics={"Value": True})
|
|
|
|
response = vpc.describe_attribute(Attribute="enableNetworkAddressUsageMetrics")
|
|
attr = response.get("EnableNetworkAddressUsageMetrics")
|
|
assert attr.get("Value") is True
|
|
|
|
|
|
@mock_aws
|
|
def test_vpc_associate_dhcp_options():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
client = boto3.client("ec2", region_name="us-west-1")
|
|
dhcp_options = ec2.create_dhcp_options(
|
|
DhcpConfigurations=[
|
|
{"Key": "domain-name", "Values": [SAMPLE_DOMAIN_NAME]},
|
|
{"Key": "domain-name-servers", "Values": SAMPLE_NAME_SERVERS},
|
|
]
|
|
)
|
|
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
|
|
client.associate_dhcp_options(DhcpOptionsId=dhcp_options.id, VpcId=vpc.id)
|
|
|
|
vpc.reload()
|
|
assert dhcp_options.id == vpc.dhcp_options_id
|
|
|
|
|
|
@mock_aws
|
|
def test_associate_vpc_ipv4_cidr_block():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
vpc = ec2.create_vpc(CidrBlock="10.10.42.0/24")
|
|
|
|
# Associate/Extend vpc CIDR range up to 5 ciders
|
|
for i in range(43, 47):
|
|
response = ec2.meta.client.associate_vpc_cidr_block(
|
|
VpcId=vpc.id, CidrBlock=f"10.10.{i}.0/24"
|
|
)
|
|
assert (
|
|
response["CidrBlockAssociation"]["CidrBlockState"]["State"] == "associating"
|
|
)
|
|
assert response["CidrBlockAssociation"]["CidrBlock"] == f"10.10.{i}.0/24"
|
|
assert "vpc-cidr-assoc" in response["CidrBlockAssociation"]["AssociationId"]
|
|
|
|
# Check all associations exist
|
|
vpc = ec2.Vpc(vpc.id)
|
|
assert len(vpc.cidr_block_association_set) == 5
|
|
assert vpc.cidr_block_association_set[2]["CidrBlockState"]["State"] == "associated"
|
|
assert vpc.cidr_block_association_set[4]["CidrBlockState"]["State"] == "associated"
|
|
|
|
# Check error on adding 6th association.
|
|
with pytest.raises(ClientError) as ex:
|
|
response = ec2.meta.client.associate_vpc_cidr_block(
|
|
VpcId=vpc.id, CidrBlock="10.10.50.0/22"
|
|
)
|
|
assert (
|
|
str(ex.value)
|
|
== f"An error occurred (CidrLimitExceeded) when calling the AssociateVpcCidrBlock operation: This network '{vpc.id}' has met its maximum number of allowed CIDRs: 5"
|
|
)
|
|
|
|
|
|
@mock_aws
|
|
def test_disassociate_vpc_ipv4_cidr_block():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
vpc = ec2.create_vpc(CidrBlock="10.10.42.0/24")
|
|
ec2.meta.client.associate_vpc_cidr_block(VpcId=vpc.id, CidrBlock="10.10.43.0/24")
|
|
|
|
# Remove an extended cidr block
|
|
vpc = ec2.Vpc(vpc.id)
|
|
non_default_assoc_cidr_block = next(
|
|
iter(
|
|
[
|
|
x
|
|
for x in vpc.cidr_block_association_set
|
|
if vpc.cidr_block != x["CidrBlock"]
|
|
]
|
|
),
|
|
None,
|
|
)
|
|
response = ec2.meta.client.disassociate_vpc_cidr_block(
|
|
AssociationId=non_default_assoc_cidr_block["AssociationId"]
|
|
)
|
|
assert (
|
|
response["CidrBlockAssociation"]["CidrBlockState"]["State"] == "disassociating"
|
|
)
|
|
assert (
|
|
response["CidrBlockAssociation"]["CidrBlock"]
|
|
== non_default_assoc_cidr_block["CidrBlock"]
|
|
)
|
|
assert (
|
|
response["CidrBlockAssociation"]["AssociationId"]
|
|
== non_default_assoc_cidr_block["AssociationId"]
|
|
)
|
|
|
|
# Error attempting to delete a non-existent CIDR_BLOCK association
|
|
with pytest.raises(ClientError) as ex:
|
|
response = ec2.meta.client.disassociate_vpc_cidr_block(
|
|
AssociationId="vpc-cidr-assoc-BORING123"
|
|
)
|
|
assert (
|
|
str(ex.value)
|
|
== "An error occurred (InvalidVpcCidrBlockAssociationIdError.NotFound) when calling the DisassociateVpcCidrBlock operation: The vpc CIDR block association ID 'vpc-cidr-assoc-BORING123' does not exist"
|
|
)
|
|
|
|
# Error attempting to delete Primary CIDR BLOCK association
|
|
vpc_base_cidr_assoc_id = next(
|
|
iter(
|
|
[
|
|
x
|
|
for x in vpc.cidr_block_association_set
|
|
if vpc.cidr_block == x["CidrBlock"]
|
|
]
|
|
),
|
|
{},
|
|
)["AssociationId"]
|
|
|
|
with pytest.raises(ClientError) as ex:
|
|
ec2.meta.client.disassociate_vpc_cidr_block(
|
|
AssociationId=vpc_base_cidr_assoc_id
|
|
)
|
|
assert (
|
|
str(ex.value)
|
|
== f"An error occurred (OperationNotPermitted) when calling the DisassociateVpcCidrBlock operation: The vpc CIDR block with association ID {vpc_base_cidr_assoc_id} may not be disassociated. It is the primary IPv4 CIDR block of the VPC"
|
|
)
|
|
|
|
|
|
@mock_aws
|
|
def test_cidr_block_association_filters():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
vpc1 = ec2.create_vpc(CidrBlock="10.90.0.0/16")
|
|
vpc2 = ec2.create_vpc(CidrBlock="10.91.0.0/16")
|
|
ec2.meta.client.associate_vpc_cidr_block(VpcId=vpc2.id, CidrBlock="10.10.0.0/19")
|
|
vpc3 = ec2.create_vpc(CidrBlock="10.92.0.0/24")
|
|
ec2.meta.client.associate_vpc_cidr_block(VpcId=vpc3.id, CidrBlock="10.92.1.0/24")
|
|
ec2.meta.client.associate_vpc_cidr_block(VpcId=vpc3.id, CidrBlock="10.92.2.0/24")
|
|
vpc3_assoc_response = ec2.meta.client.associate_vpc_cidr_block(
|
|
VpcId=vpc3.id, CidrBlock="10.92.3.0/24"
|
|
)
|
|
|
|
# Test filters for a cidr-block in all VPCs cidr-block-associations
|
|
filtered_vpcs = list(
|
|
ec2.vpcs.filter(
|
|
Filters=[
|
|
{
|
|
"Name": "cidr-block-association.cidr-block",
|
|
"Values": ["10.10.0.0/19"],
|
|
}
|
|
]
|
|
)
|
|
)
|
|
assert vpc1.id not in [vpc.id for vpc in filtered_vpcs]
|
|
assert vpc2.id in [vpc.id for vpc in filtered_vpcs]
|
|
assert vpc3.id not in [vpc.id for vpc in filtered_vpcs]
|
|
|
|
# Test filter for association id in VPCs
|
|
association_id = vpc3_assoc_response["CidrBlockAssociation"]["AssociationId"]
|
|
filtered_vpcs = list(
|
|
ec2.vpcs.filter(
|
|
Filters=[
|
|
{
|
|
"Name": "cidr-block-association.association-id",
|
|
"Values": [association_id],
|
|
}
|
|
]
|
|
)
|
|
)
|
|
assert len(filtered_vpcs) == 1
|
|
assert filtered_vpcs[0].id == vpc3.id
|
|
|
|
# Test filter for association state in VPC - this will never show anything in this test
|
|
filtered_vpcs = list(
|
|
ec2.vpcs.filter(
|
|
Filters=[
|
|
{"Name": "cidr-block-association.association-id", "Values": ["failing"]}
|
|
]
|
|
)
|
|
)
|
|
assert len(filtered_vpcs) == 0
|
|
|
|
|
|
@mock_aws
|
|
def test_vpc_associate_ipv6_cidr_block():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
# Test create VPC with IPV6 cidr range
|
|
vpc = ec2.create_vpc(CidrBlock="10.10.42.0/24", AmazonProvidedIpv6CidrBlock=True)
|
|
assoc_set = next(iter(vpc.ipv6_cidr_block_association_set), None)
|
|
assert assoc_set["Ipv6CidrBlockState"]["State"] == "associated"
|
|
assert "::/56" in assoc_set["Ipv6CidrBlock"]
|
|
assert "vpc-cidr-assoc" in assoc_set["AssociationId"]
|
|
|
|
# Test Fail on adding 2nd IPV6 association - AWS only allows 1 at this time!
|
|
with pytest.raises(ClientError) as ex:
|
|
ec2.meta.client.associate_vpc_cidr_block(
|
|
VpcId=vpc.id, AmazonProvidedIpv6CidrBlock=True
|
|
)
|
|
assert (
|
|
str(ex.value)
|
|
== f"An error occurred (CidrLimitExceeded) when calling the AssociateVpcCidrBlock operation: This network '{vpc.id}' has met its maximum number of allowed CIDRs: 1"
|
|
)
|
|
|
|
# Test associate ipv6 cidr block after vpc created
|
|
vpc = ec2.create_vpc(CidrBlock="10.10.50.0/24")
|
|
cidr_block = ec2.meta.client.associate_vpc_cidr_block(
|
|
VpcId=vpc.id, AmazonProvidedIpv6CidrBlock=True
|
|
)["Ipv6CidrBlockAssociation"]
|
|
assert cidr_block["Ipv6CidrBlockState"]["State"] == "associating"
|
|
assert "::/56" in cidr_block["Ipv6CidrBlock"]
|
|
assert "vpc-cidr-assoc-" in cidr_block["AssociationId"]
|
|
|
|
# Check on describe vpc that has ipv6 cidr block association
|
|
vpc = ec2.Vpc(vpc.id)
|
|
assert len(vpc.ipv6_cidr_block_association_set) == 1
|
|
|
|
|
|
@mock_aws
|
|
def test_vpc_disassociate_ipv6_cidr_block():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
# Test create VPC with IPV6 cidr range
|
|
vpc = ec2.create_vpc(CidrBlock="10.10.42.0/24", AmazonProvidedIpv6CidrBlock=True)
|
|
# Test disassociating the only IPV6
|
|
assoc_id = vpc.ipv6_cidr_block_association_set[0]["AssociationId"]
|
|
cidr = ec2.meta.client.disassociate_vpc_cidr_block(AssociationId=assoc_id)[
|
|
"Ipv6CidrBlockAssociation"
|
|
]
|
|
assert cidr["Ipv6CidrBlockState"]["State"] == "disassociating"
|
|
assert "::/56" in cidr["Ipv6CidrBlock"]
|
|
assert cidr["AssociationId"] == assoc_id
|
|
|
|
|
|
@mock_aws
|
|
def test_ipv6_cidr_block_association_filters():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
vpc1 = ec2.create_vpc(CidrBlock="10.90.0.0/16")
|
|
|
|
vpc2 = ec2.create_vpc(CidrBlock="10.91.0.0/16", AmazonProvidedIpv6CidrBlock=True)
|
|
vpc2_assoc_ipv6_assoc_id = vpc2.ipv6_cidr_block_association_set[0]["AssociationId"]
|
|
ec2.meta.client.associate_vpc_cidr_block(VpcId=vpc2.id, CidrBlock="10.10.0.0/19")
|
|
|
|
vpc3 = ec2.create_vpc(CidrBlock="10.92.0.0/24")
|
|
ec2.meta.client.associate_vpc_cidr_block(VpcId=vpc3.id, CidrBlock="10.92.1.0/24")
|
|
ec2.meta.client.associate_vpc_cidr_block(VpcId=vpc3.id, CidrBlock="10.92.2.0/24")
|
|
response = ec2.meta.client.associate_vpc_cidr_block(
|
|
VpcId=vpc3.id, AmazonProvidedIpv6CidrBlock=True
|
|
)
|
|
vpc3_ipv6_cidr_block = response["Ipv6CidrBlockAssociation"]["Ipv6CidrBlock"]
|
|
|
|
vpc4 = ec2.create_vpc(CidrBlock="10.95.0.0/16") # Here for its looks
|
|
|
|
# Test filters for an ipv6 cidr-block in all VPCs cidr-block-associations
|
|
filtered_vpcs = list(
|
|
ec2.vpcs.filter(
|
|
Filters=[
|
|
{
|
|
"Name": "ipv6-cidr-block-association.ipv6-cidr-block",
|
|
"Values": [vpc3_ipv6_cidr_block],
|
|
}
|
|
]
|
|
)
|
|
)
|
|
assert len(filtered_vpcs) == 1
|
|
assert filtered_vpcs[0].id == vpc3.id
|
|
|
|
# Test filter for association id in VPCs
|
|
filtered_vpcs = list(
|
|
ec2.vpcs.filter(
|
|
Filters=[
|
|
{
|
|
"Name": "ipv6-cidr-block-association.association-id",
|
|
"Values": [vpc2_assoc_ipv6_assoc_id],
|
|
}
|
|
]
|
|
)
|
|
)
|
|
assert len(filtered_vpcs) == 1
|
|
assert filtered_vpcs[0].id == vpc2.id
|
|
|
|
# Test filter for association state in VPC - this will never show anything in this test
|
|
assoc_vpcs = [
|
|
vpc.id
|
|
for vpc in ec2.vpcs.filter(
|
|
Filters=[
|
|
{"Name": "ipv6-cidr-block-association.state", "Values": ["associated"]}
|
|
]
|
|
)
|
|
]
|
|
assert vpc1.id not in assoc_vpcs
|
|
assert vpc2.id in assoc_vpcs
|
|
assert vpc3.id in assoc_vpcs
|
|
assert vpc4.id not in assoc_vpcs
|
|
|
|
|
|
@mock_aws
|
|
def test_create_vpc_with_invalid_cidr_block_parameter():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
vpc_cidr_block = "1000.1.0.0/20"
|
|
with pytest.raises(ClientError) as ex:
|
|
ec2.create_vpc(CidrBlock=vpc_cidr_block)
|
|
assert (
|
|
str(ex.value)
|
|
== f"An error occurred (InvalidParameterValue) when calling the CreateVpc operation: Value ({vpc_cidr_block}) for parameter cidrBlock is invalid. This is not a valid CIDR block."
|
|
)
|
|
|
|
|
|
@mock_aws
|
|
def test_create_vpc_with_invalid_cidr_range():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
vpc_cidr_block = "10.1.0.0/29"
|
|
with pytest.raises(ClientError) as ex:
|
|
ec2.create_vpc(CidrBlock=vpc_cidr_block)
|
|
assert (
|
|
str(ex.value)
|
|
== f"An error occurred (InvalidVpc.Range) when calling the CreateVpc operation: The CIDR '{vpc_cidr_block}' is invalid."
|
|
)
|
|
|
|
|
|
@mock_aws
|
|
def test_create_vpc_with_tags():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
# Create VPC
|
|
vpc = ec2.create_vpc(
|
|
CidrBlock="10.0.0.0/16",
|
|
TagSpecifications=[
|
|
{"ResourceType": "vpc", "Tags": [{"Key": "name", "Value": "some-vpc"}]}
|
|
],
|
|
)
|
|
assert vpc.tags == [{"Key": "name", "Value": "some-vpc"}]
|
|
|
|
|
|
@mock_aws
|
|
def test_enable_vpc_classic_link():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
# Create VPC
|
|
vpc = ec2.create_vpc(CidrBlock="10.1.0.0/16")
|
|
|
|
response = ec2.meta.client.enable_vpc_classic_link(VpcId=vpc.id)
|
|
assert response.get("Return") is True
|
|
|
|
|
|
@mock_aws
|
|
def test_enable_vpc_classic_link_failure():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
# Create VPC
|
|
vpc = ec2.create_vpc(CidrBlock="10.90.0.0/16")
|
|
|
|
response = ec2.meta.client.enable_vpc_classic_link(VpcId=vpc.id)
|
|
assert response.get("Return") is False
|
|
|
|
|
|
@mock_aws
|
|
def test_disable_vpc_classic_link():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
# Create VPC
|
|
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
|
|
ec2.meta.client.enable_vpc_classic_link(VpcId=vpc.id)
|
|
response = ec2.meta.client.disable_vpc_classic_link(VpcId=vpc.id)
|
|
assert response.get("Return") is False
|
|
|
|
|
|
@mock_aws
|
|
def test_describe_classic_link_enabled():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
# Create VPC
|
|
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
|
|
ec2.meta.client.enable_vpc_classic_link(VpcId=vpc.id)
|
|
response = ec2.meta.client.describe_vpc_classic_link(VpcIds=[vpc.id])
|
|
assert response.get("Vpcs")[0].get("ClassicLinkEnabled") is True
|
|
|
|
|
|
@mock_aws
|
|
def test_describe_classic_link_disabled():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
# Create VPC
|
|
vpc = ec2.create_vpc(CidrBlock="10.90.0.0/16")
|
|
|
|
response = ec2.meta.client.describe_vpc_classic_link(VpcIds=[vpc.id])
|
|
assert response.get("Vpcs")[0].get("ClassicLinkEnabled") is False
|
|
|
|
|
|
@mock_aws
|
|
def test_describe_classic_link_multiple():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
# Create VPC
|
|
vpc1 = ec2.create_vpc(CidrBlock="10.90.0.0/16")
|
|
vpc2 = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
|
|
ec2.meta.client.enable_vpc_classic_link(VpcId=vpc2.id)
|
|
response = ec2.meta.client.describe_vpc_classic_link(VpcIds=[vpc1.id, vpc2.id])
|
|
expected = [
|
|
{"VpcId": vpc1.id, "ClassicLinkDnsSupported": False},
|
|
{"VpcId": vpc2.id, "ClassicLinkDnsSupported": True},
|
|
]
|
|
|
|
# Ensure response is sorted, because they can come in random order
|
|
assert response.get("Vpcs").sort(key=lambda x: x["VpcId"]) == expected.sort(
|
|
key=lambda x: x["VpcId"]
|
|
)
|
|
|
|
|
|
@mock_aws
|
|
def test_enable_vpc_classic_link_dns_support():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
# Create VPC
|
|
vpc = ec2.create_vpc(CidrBlock="10.1.0.0/16")
|
|
|
|
response = ec2.meta.client.enable_vpc_classic_link_dns_support(VpcId=vpc.id)
|
|
assert response.get("Return") is True
|
|
|
|
|
|
@mock_aws
|
|
def test_disable_vpc_classic_link_dns_support():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
# Create VPC
|
|
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
|
|
ec2.meta.client.enable_vpc_classic_link_dns_support(VpcId=vpc.id)
|
|
response = ec2.meta.client.disable_vpc_classic_link_dns_support(VpcId=vpc.id)
|
|
assert response.get("Return") is False
|
|
|
|
|
|
@mock_aws
|
|
def test_describe_classic_link_dns_support_enabled():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
# Create VPC
|
|
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
|
|
ec2.meta.client.enable_vpc_classic_link_dns_support(VpcId=vpc.id)
|
|
response = ec2.meta.client.describe_vpc_classic_link_dns_support(VpcIds=[vpc.id])
|
|
assert response.get("Vpcs")[0].get("ClassicLinkDnsSupported") is True
|
|
|
|
|
|
@mock_aws
|
|
def test_describe_classic_link_dns_support_disabled():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
# Create VPC
|
|
vpc = ec2.create_vpc(CidrBlock="10.90.0.0/16")
|
|
|
|
response = ec2.meta.client.describe_vpc_classic_link_dns_support(VpcIds=[vpc.id])
|
|
assert response.get("Vpcs")[0].get("ClassicLinkDnsSupported") is False
|
|
|
|
|
|
@mock_aws
|
|
def test_describe_classic_link_dns_support_multiple():
|
|
ec2 = boto3.resource("ec2", region_name="us-west-1")
|
|
|
|
# Create VPC
|
|
vpc1 = ec2.create_vpc(CidrBlock="10.90.0.0/16")
|
|
vpc2 = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
|
|
|
ec2.meta.client.enable_vpc_classic_link_dns_support(VpcId=vpc2.id)
|
|
response = ec2.meta.client.describe_vpc_classic_link_dns_support(
|
|
VpcIds=[vpc1.id, vpc2.id]
|
|
)
|
|
expected = [
|
|
{"VpcId": vpc1.id, "ClassicLinkDnsSupported": False},
|
|
{"VpcId": vpc2.id, "ClassicLinkDnsSupported": True},
|
|
]
|
|
|
|
# Ensure response is sorted, because they can come in random order
|
|
assert response.get("Vpcs").sort(key=lambda x: x["VpcId"]) == expected.sort(
|
|
key=lambda x: x["VpcId"]
|
|
)
|
|
|
|
|
|
@mock_aws
|
|
def test_create_vpc_endpoint__policy():
|
|
ec2 = boto3.client("ec2", region_name="us-west-1")
|
|
vpc_id = ec2.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]["VpcId"]
|
|
# create without policy --> verify the default policy is created
|
|
default_policy = {
|
|
"Version": "2008-10-17",
|
|
"Statement": [
|
|
{"Effect": "Allow", "Principal": "*", "Action": "*", "Resource": "*"}
|
|
],
|
|
}
|
|
vpc_end_point = ec2.create_vpc_endpoint(
|
|
VpcId=vpc_id,
|
|
ServiceName="com.amazonaws.us-east-1.s3",
|
|
VpcEndpointType="Gateway",
|
|
)["VpcEndpoint"]
|
|
|
|
assert "PolicyDocument" in vpc_end_point
|
|
assert json.loads(vpc_end_point["PolicyDocument"]) == default_policy
|
|
|
|
# create with policy --> verify the passed policy is returned
|
|
vpc_end_point = ec2.create_vpc_endpoint(
|
|
VpcId=vpc_id,
|
|
ServiceName="com.amazonaws.us-east-1.s3",
|
|
PolicyDocument="my policy document",
|
|
VpcEndpointType="Gateway",
|
|
)["VpcEndpoint"]
|
|
assert vpc_end_point["PolicyDocument"] == "my policy document"
|
|
|
|
|
|
@mock_aws
|
|
def test_describe_vpc_gateway_end_points():
|
|
ec2 = boto3.client("ec2", region_name="us-west-1")
|
|
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]
|
|
|
|
route_table = ec2.create_route_table(VpcId=vpc["VpcId"])["RouteTable"]
|
|
vpc_end_point = ec2.create_vpc_endpoint(
|
|
VpcId=vpc["VpcId"],
|
|
ServiceName="com.amazonaws.us-east-1.s3",
|
|
RouteTableIds=[route_table["RouteTableId"]],
|
|
VpcEndpointType="Gateway",
|
|
)["VpcEndpoint"]
|
|
our_id = vpc_end_point["VpcEndpointId"]
|
|
|
|
all_endpoints = retrieve_all_endpoints(ec2)
|
|
assert our_id in [e["VpcEndpointId"] for e in all_endpoints]
|
|
our_endpoint = [e for e in all_endpoints if e["VpcEndpointId"] == our_id][0]
|
|
assert vpc_end_point["PrivateDnsEnabled"] is True
|
|
assert our_endpoint["PrivateDnsEnabled"] is True
|
|
|
|
assert our_endpoint["VpcId"] == vpc["VpcId"]
|
|
assert our_endpoint["RouteTableIds"] == [route_table["RouteTableId"]]
|
|
|
|
assert our_endpoint["VpcEndpointType"] == "Gateway"
|
|
assert our_endpoint["ServiceName"] == "com.amazonaws.us-east-1.s3"
|
|
assert our_endpoint["State"] == "available"
|
|
|
|
endpoint_by_id = ec2.describe_vpc_endpoints(VpcEndpointIds=[our_id])[
|
|
"VpcEndpoints"
|
|
][0]
|
|
assert endpoint_by_id["VpcEndpointId"] == our_id
|
|
assert endpoint_by_id["VpcId"] == vpc["VpcId"]
|
|
assert endpoint_by_id["RouteTableIds"] == [route_table["RouteTableId"]]
|
|
assert endpoint_by_id["VpcEndpointType"] == "Gateway"
|
|
assert endpoint_by_id["ServiceName"] == "com.amazonaws.us-east-1.s3"
|
|
assert endpoint_by_id["State"] == "available"
|
|
|
|
gateway_endpoints = ec2.describe_vpc_endpoints(
|
|
Filters=[{"Name": "vpc-endpoint-type", "Values": ["Gateway"]}]
|
|
)["VpcEndpoints"]
|
|
assert our_id in [e["VpcEndpointId"] for e in gateway_endpoints]
|
|
|
|
with pytest.raises(ClientError) as ex:
|
|
ec2.describe_vpc_endpoints(VpcEndpointIds=[route_table["RouteTableId"]])
|
|
err = ex.value.response["Error"]
|
|
assert err["Code"] == "InvalidVpcEndpointId.NotFound"
|
|
|
|
|
|
@mock_aws
|
|
def test_describe_vpc_interface_end_points():
|
|
ec2 = boto3.client("ec2", region_name="us-west-1")
|
|
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]
|
|
subnet = ec2.create_subnet(VpcId=vpc["VpcId"], CidrBlock="10.0.1.0/24")["Subnet"]
|
|
|
|
route_table = ec2.create_route_table(VpcId=vpc["VpcId"])["RouteTable"]
|
|
vpc_end_point = ec2.create_vpc_endpoint(
|
|
VpcId=vpc["VpcId"],
|
|
ServiceName="com.tester.my-test-endpoint",
|
|
VpcEndpointType="interface",
|
|
SubnetIds=[subnet["SubnetId"]],
|
|
)["VpcEndpoint"]
|
|
our_id = vpc_end_point["VpcEndpointId"]
|
|
|
|
assert len(vpc_end_point["DnsEntries"]) == 1
|
|
assert "com.tester.my-test-endpoint" in vpc_end_point["DnsEntries"][0]["DnsName"]
|
|
assert "HostedZoneId" in vpc_end_point["DnsEntries"][0]
|
|
|
|
all_endpoints = retrieve_all_endpoints(ec2)
|
|
assert our_id in [e["VpcEndpointId"] for e in all_endpoints]
|
|
our_endpoint = [e for e in all_endpoints if e["VpcEndpointId"] == our_id][0]
|
|
assert vpc_end_point["PrivateDnsEnabled"] is True
|
|
assert our_endpoint["PrivateDnsEnabled"] is True
|
|
|
|
assert our_endpoint["VpcId"] == vpc["VpcId"]
|
|
assert "RouteTableIds" not in our_endpoint
|
|
|
|
assert our_endpoint["DnsEntries"] == vpc_end_point["DnsEntries"]
|
|
|
|
assert our_endpoint["VpcEndpointType"] == "interface"
|
|
assert our_endpoint["ServiceName"] == "com.tester.my-test-endpoint"
|
|
assert our_endpoint["State"] == "available"
|
|
|
|
endpoint_by_id = ec2.describe_vpc_endpoints(VpcEndpointIds=[our_id])[
|
|
"VpcEndpoints"
|
|
][0]
|
|
assert endpoint_by_id["VpcEndpointId"] == our_id
|
|
assert endpoint_by_id["VpcId"] == vpc["VpcId"]
|
|
assert "RouteTableIds" not in endpoint_by_id
|
|
assert endpoint_by_id["VpcEndpointType"] == "interface"
|
|
assert endpoint_by_id["ServiceName"] == "com.tester.my-test-endpoint"
|
|
assert endpoint_by_id["State"] == "available"
|
|
assert endpoint_by_id["DnsEntries"] == vpc_end_point["DnsEntries"]
|
|
|
|
with pytest.raises(ClientError) as ex:
|
|
ec2.describe_vpc_endpoints(VpcEndpointIds=[route_table["RouteTableId"]])
|
|
err = ex.value.response["Error"]
|
|
assert err["Code"] == "InvalidVpcEndpointId.NotFound"
|
|
|
|
|
|
def retrieve_all_endpoints(ec2):
|
|
resp = ec2.describe_vpc_endpoints()
|
|
all_endpoints = resp["VpcEndpoints"]
|
|
next_token = resp.get("NextToken")
|
|
while next_token:
|
|
resp = ec2.describe_vpc_endpoints(NextToken=next_token)
|
|
all_endpoints.extend(resp["VpcEndpoints"])
|
|
next_token = resp.get("NextToken")
|
|
return all_endpoints
|
|
|
|
|
|
@mock_aws
|
|
def test_modify_vpc_endpoint():
|
|
ec2 = boto3.client("ec2", region_name="us-west-1")
|
|
vpc_id = ec2.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]["VpcId"]
|
|
subnet_id1 = ec2.create_subnet(VpcId=vpc_id, CidrBlock="10.0.1.0/24")["Subnet"][
|
|
"SubnetId"
|
|
]
|
|
subnet_id2 = ec2.create_subnet(VpcId=vpc_id, CidrBlock="10.0.2.0/24")["Subnet"][
|
|
"SubnetId"
|
|
]
|
|
|
|
rt_id = ec2.create_route_table(VpcId=vpc_id)["RouteTable"]["RouteTableId"]
|
|
endpoint = ec2.create_vpc_endpoint(
|
|
VpcId=vpc_id,
|
|
ServiceName="com.tester.my-test-endpoint",
|
|
VpcEndpointType="interface",
|
|
SubnetIds=[subnet_id1],
|
|
)["VpcEndpoint"]
|
|
vpc_id = endpoint["VpcEndpointId"]
|
|
|
|
ec2.modify_vpc_endpoint(
|
|
VpcEndpointId=vpc_id,
|
|
AddSubnetIds=[subnet_id2],
|
|
)
|
|
|
|
endpoint = ec2.describe_vpc_endpoints(VpcEndpointIds=[vpc_id])["VpcEndpoints"][0]
|
|
assert endpoint["SubnetIds"] == [subnet_id1, subnet_id2]
|
|
|
|
ec2.modify_vpc_endpoint(VpcEndpointId=vpc_id, AddRouteTableIds=[rt_id])
|
|
endpoint = ec2.describe_vpc_endpoints(VpcEndpointIds=[vpc_id])["VpcEndpoints"][0]
|
|
assert endpoint["RouteTableIds"] == [rt_id]
|
|
|
|
ec2.modify_vpc_endpoint(VpcEndpointId=vpc_id, RemoveRouteTableIds=[rt_id])
|
|
endpoint = ec2.describe_vpc_endpoints(VpcEndpointIds=[vpc_id])["VpcEndpoints"][0]
|
|
assert "RouteTableIds" not in endpoint
|
|
|
|
ec2.modify_vpc_endpoint(
|
|
VpcEndpointId=vpc_id,
|
|
PolicyDocument="doc",
|
|
)
|
|
endpoint = ec2.describe_vpc_endpoints(VpcEndpointIds=[vpc_id])["VpcEndpoints"][0]
|
|
assert endpoint["PolicyDocument"] == "doc"
|
|
|
|
|
|
@mock_aws
|
|
def test_delete_vpc_end_points():
|
|
ec2 = boto3.client("ec2", region_name="us-west-1")
|
|
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]
|
|
|
|
route_table = ec2.create_route_table(VpcId=vpc["VpcId"])["RouteTable"]
|
|
vpc_end_point1 = ec2.create_vpc_endpoint(
|
|
VpcId=vpc["VpcId"],
|
|
ServiceName="com.amazonaws.us-west-1.s3",
|
|
RouteTableIds=[route_table["RouteTableId"]],
|
|
VpcEndpointType="gateway",
|
|
)["VpcEndpoint"]
|
|
vpc_end_point2 = ec2.create_vpc_endpoint(
|
|
VpcId=vpc["VpcId"],
|
|
ServiceName="com.amazonaws.us-west-1.s3",
|
|
RouteTableIds=[route_table["RouteTableId"]],
|
|
VpcEndpointType="gateway",
|
|
)["VpcEndpoint"]
|
|
|
|
vpc_endpoints = retrieve_all_endpoints(ec2)
|
|
all_ids = [e["VpcEndpointId"] for e in vpc_endpoints]
|
|
assert vpc_end_point1["VpcEndpointId"] in all_ids
|
|
assert vpc_end_point2["VpcEndpointId"] in all_ids
|
|
|
|
ec2.delete_vpc_endpoints(VpcEndpointIds=[vpc_end_point1["VpcEndpointId"]])
|
|
|
|
vpc_endpoints = retrieve_all_endpoints(ec2)
|
|
all_ids = [e["VpcEndpointId"] for e in vpc_endpoints]
|
|
assert vpc_end_point1["VpcEndpointId"] in all_ids
|
|
assert vpc_end_point2["VpcEndpointId"] in all_ids
|
|
|
|
ep1 = ec2.describe_vpc_endpoints(VpcEndpointIds=[vpc_end_point1["VpcEndpointId"]])[
|
|
"VpcEndpoints"
|
|
][0]
|
|
assert ep1["State"] == "deleted"
|
|
|
|
ep2 = ec2.describe_vpc_endpoints(VpcEndpointIds=[vpc_end_point2["VpcEndpointId"]])[
|
|
"VpcEndpoints"
|
|
][0]
|
|
assert ep2["State"] == "available"
|
|
|
|
|
|
@mock_aws
|
|
def test_describe_vpcs_dryrun():
|
|
client = boto3.client("ec2", region_name="us-east-1")
|
|
|
|
with pytest.raises(ClientError) as ex:
|
|
client.describe_vpcs(DryRun=True)
|
|
assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 412
|
|
assert ex.value.response["Error"]["Code"] == "DryRunOperation"
|
|
assert (
|
|
ex.value.response["Error"]["Message"]
|
|
== "An error occurred (DryRunOperation) when calling the DescribeVpcs operation: Request would have succeeded, but DryRun flag is set"
|
|
)
|
|
|
|
|
|
@mock_aws
|
|
def test_describe_prefix_lists():
|
|
client = boto3.client("ec2", region_name="us-east-1")
|
|
result_unfiltered = client.describe_prefix_lists()
|
|
assert len(result_unfiltered["PrefixLists"]) > 1
|
|
result_filtered = client.describe_prefix_lists(
|
|
Filters=[
|
|
{"Name": "prefix-list-name", "Values": ["com.amazonaws.us-east-1.s3"]},
|
|
]
|
|
)
|
|
assert len(result_filtered["PrefixLists"]) == 1
|