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

Enable Extended CIDR Associations on VPC (#1511)

Browse Source 
* Enable Extended CIDR Associations on VPC

* Ooops missed the utils, try to be more flakey?, remove unnecessary part in tests

* try to be even more flakey
This commit is contained in:
Rob Walker 2018-03-22 02:10:38 +10:00 committed by Jack Danger
parent b7ae704ad2
commit 4b3469292a
5 changed files with 455 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
moto/ec2/exceptions.py
View File

@ -280,6 +280,15 @@ class InvalidAssociationIdError(EC2ClientError):
.format(association_id))
class InvalidVpcCidrBlockAssociationIdError(EC2ClientError):
def __init__(self, association_id):
super(InvalidVpcCidrBlockAssociationIdError, self).__init__(
"InvalidVpcCidrBlockAssociationIdError.NotFound",
"The vpc CIDR block association ID '{0}' does not exist"
.format(association_id))
class InvalidVPCPeeringConnectionIdError(EC2ClientError):
def __init__(self, vpc_peering_connection_id):
@ -392,3 +401,22 @@ class FilterNotImplementedError(MotoNotImplementedError):
super(FilterNotImplementedError, self).__init__(
"The filter '{0}' for {1}".format(
filter_name, method_name))
class CidrLimitExceeded(EC2ClientError):
def __init__(self, vpc_id, max_cidr_limit):
super(CidrLimitExceeded, self).__init__(
"CidrLimitExceeded",
"This network '{0}' has met its maximum number of allowed CIDRs: {1}".format(vpc_id, max_cidr_limit)
)
class OperationNotPermitted(EC2ClientError):
def __init__(self, association_id):
super(OperationNotPermitted, self).__init__(
"OperationNotPermitted",
"The vpc CIDR block with association ID {} may not be disassociated. "
"It is the primary IPv4 CIDR block of the VPC".format(association_id)
)

130
moto/ec2/models.py
View File

@ -24,51 +24,54 @@ from moto.core import BaseBackend
from moto.core.models import Model, BaseModel
from moto.core.utils import iso_8601_datetime_with_milliseconds, camelcase_to_underscores
from .exceptions import (
EC2ClientError,
CidrLimitExceeded,
DependencyViolationError,
MissingParameterError,
EC2ClientError,
FilterNotImplementedError,
GatewayNotAttachedError,
InvalidAddressError,
InvalidAllocationIdError,
InvalidAMIIdError,
InvalidAMIAttributeItemValueError,
InvalidAssociationIdError,
InvalidCIDRSubnetError,
InvalidCustomerGatewayIdError,
InvalidDHCPOptionsIdError,
InvalidDomainError,
InvalidID,
InvalidInstanceIdError,
InvalidInternetGatewayIdError,
InvalidKeyPairDuplicateError,
InvalidKeyPairNameError,
InvalidNetworkAclIdError,
InvalidNetworkAttachmentIdError,
InvalidNetworkInterfaceIdError,
InvalidParameterValueError,
InvalidParameterValueErrorTagNull,
InvalidDHCPOptionsIdError,
MalformedDHCPOptionsIdError,
InvalidKeyPairNameError,
InvalidKeyPairDuplicateError,
InvalidInternetGatewayIdError,
GatewayNotAttachedError,
ResourceAlreadyAssociatedError,
InvalidVPCIdError,
InvalidSubnetIdError,
InvalidNetworkInterfaceIdError,
InvalidNetworkAttachmentIdError,
InvalidSecurityGroupDuplicateError,
InvalidSecurityGroupNotFoundError,
InvalidPermissionNotFoundError,
InvalidPermissionDuplicateError,
InvalidRouteTableIdError,
InvalidRouteError,
InvalidInstanceIdError,
InvalidAMIIdError,
InvalidAMIAttributeItemValueError,
InvalidSecurityGroupDuplicateError,
InvalidSecurityGroupNotFoundError,
InvalidSnapshotIdError,
InvalidSubnetIdError,
InvalidVolumeIdError,
InvalidVolumeAttachmentError,
InvalidDomainError,
InvalidAddressError,
InvalidAllocationIdError,
InvalidAssociationIdError,
InvalidVpcCidrBlockAssociationIdError,
InvalidVPCPeeringConnectionIdError,
InvalidVPCPeeringConnectionStateTransitionError,
TagLimitExceeded,
InvalidID,
InvalidCIDRSubnetError,
InvalidNetworkAclIdError,
InvalidVPCIdError,
InvalidVpnGatewayIdError,
InvalidVpnConnectionIdError,
InvalidCustomerGatewayIdError,
RulesPerSecurityGroupLimitExceededError,
MalformedAMIIdError,
MalformedDHCPOptionsIdError,
MissingParameterError,
MotoNotImplementedError,
FilterNotImplementedError,
MalformedAMIIdError)
OperationNotPermitted,
ResourceAlreadyAssociatedError,
RulesPerSecurityGroupLimitExceededError,
TagLimitExceeded)
from .utils import (
EC2_RESOURCE_TO_PREFIX,
EC2_PREFIX_TO_RESOURCE,
@ -81,6 +84,7 @@ from .utils import (
random_instance_id,
random_internet_gateway_id,
random_ip,
random_ipv6_cidr,
random_nat_gateway_id,
random_key_pair,
random_private_ip,
@ -97,6 +101,7 @@ from .utils import (
random_subnet_association_id,
random_volume_id,
random_vpc_id,
random_vpc_cidr_association_id,
random_vpc_peering_connection_id,
generic_filter,
is_valid_resource_id,
@ -2005,10 +2010,13 @@ class EBSBackend(object):
class VPC(TaggedEC2Resource):
def __init__(self, ec2_backend, vpc_id, cidr_block, is_default, instance_tenancy='default'):
def __init__(self, ec2_backend, vpc_id, cidr_block, is_default, instance_tenancy='default',
amazon_provided_ipv6_cidr_block=False):
self.ec2_backend = ec2_backend
self.id = vpc_id
self.cidr_block = cidr_block
self.cidr_block_association_set = {}
self.dhcp_options = None
self.state = 'available'
self.instance_tenancy = instance_tenancy
@ -2018,6 +2026,10 @@ class VPC(TaggedEC2Resource):
# or VPCs created using the wizard of the VPC console
self.enable_dns_hostnames = 'true' if is_default else 'false'
self.associate_vpc_cidr_block(cidr_block)
if amazon_provided_ipv6_cidr_block:
self.associate_vpc_cidr_block(cidr_block, amazon_provided_ipv6_cidr_block=amazon_provided_ipv6_cidr_block)
@classmethod
def create_from_cloudformation_json(cls, resource_name, cloudformation_json, region_name):
properties = cloudformation_json['Properties']
@ -2043,6 +2055,12 @@ class VPC(TaggedEC2Resource):
return self.id
elif filter_name in ('cidr', 'cidr-block', 'cidrBlock'):
return self.cidr_block
elif filter_name in ('cidr-block-association.cidr-block', 'ipv6-cidr-block-association.ipv6-cidr-block'):
return [c['cidr_block'] for c in self.get_cidr_block_association_set(ipv6='ipv6' in filter_name)]
elif filter_name in ('cidr-block-association.association-id', 'ipv6-cidr-block-association.association-id'):
return self.cidr_block_association_set.keys()
elif filter_name in ('cidr-block-association.state', 'ipv6-cidr-block-association.state'):
return [c['cidr_block_state']['state'] for c in self.get_cidr_block_association_set(ipv6='ipv6' in filter_name)]
elif filter_name in ('instance_tenancy', 'InstanceTenancy'):
return self.instance_tenancy
elif filter_name in ('is-default', 'isDefault'):
@ -2054,8 +2072,37 @@ class VPC(TaggedEC2Resource):
return None
return self.dhcp_options.id
else:
return super(VPC, self).get_filter_value(
filter_name, 'DescribeVpcs')
return super(VPC, self).get_filter_value(filter_name, 'DescribeVpcs')
def associate_vpc_cidr_block(self, cidr_block, amazon_provided_ipv6_cidr_block=False):
max_associations = 5 if not amazon_provided_ipv6_cidr_block else 1
if len(self.get_cidr_block_association_set(amazon_provided_ipv6_cidr_block)) >= max_associations:
raise CidrLimitExceeded(self.id, max_associations)
association_id = random_vpc_cidr_association_id()
association_set = {
'association_id': association_id,
'cidr_block_state': {'state': 'associated', 'StatusMessage': ''}
}
association_set['cidr_block'] = random_ipv6_cidr() if amazon_provided_ipv6_cidr_block else cidr_block
self.cidr_block_association_set[association_id] = association_set
return association_set
def disassociate_vpc_cidr_block(self, association_id):
if self.cidr_block == self.cidr_block_association_set.get(association_id, {}).get('cidr_block'):
raise OperationNotPermitted(association_id)
response = self.cidr_block_association_set.pop(association_id, {})
if response:
response['vpc_id'] = self.id
response['cidr_block_state']['state'] = 'disassociating'
return response
def get_cidr_block_association_set(self, ipv6=False):
return [c for c in self.cidr_block_association_set.values() if ('::/' if ipv6 else '.') in c.get('cidr_block')]
class VPCBackend(object):
@ -2063,10 +2110,9 @@ class VPCBackend(object):
self.vpcs = {}
super(VPCBackend, self).__init__()
def create_vpc(self, cidr_block, instance_tenancy='default'):
def create_vpc(self, cidr_block, instance_tenancy='default', amazon_provided_ipv6_cidr_block=False):
vpc_id = random_vpc_id()
vpc = VPC(self, vpc_id, cidr_block, len(
self.vpcs) == 0, instance_tenancy)
vpc = VPC(self, vpc_id, cidr_block, len(self.vpcs) == 0, instance_tenancy, amazon_provided_ipv6_cidr_block)
self.vpcs[vpc_id] = vpc
# AWS creates a default main route table and security group.
@ -2139,6 +2185,18 @@ class VPCBackend(object):
else:
raise InvalidParameterValueError(attr_name)
def disassociate_vpc_cidr_block(self, association_id):
for vpc in self.vpcs.values():
response = vpc.disassociate_vpc_cidr_block(association_id)
if response:
return response
else:
raise InvalidVpcCidrBlockAssociationIdError(association_id)
def associate_vpc_cidr_block(self, vpc_id, cidr_block, amazon_provided_ipv6_cidr_block):
vpc = self.get_vpc(vpc_id)
return vpc.associate_vpc_cidr_block(cidr_block, amazon_provided_ipv6_cidr_block)
class VPCPeeringConnectionStatus(object):
def __init__(self, code='initiating-request', message=''):

147
moto/ec2/responses/vpcs.py
View File

@ -9,9 +9,12 @@ class VPCs(BaseResponse):
def create_vpc(self):
cidr_block = self._get_param('CidrBlock')
instance_tenancy = self._get_param('InstanceTenancy', if_none='default')
vpc = self.ec2_backend.create_vpc(cidr_block, instance_tenancy)
amazon_provided_ipv6_cidr_blocks = self._get_param('AmazonProvidedIpv6CidrBlock')
vpc = self.ec2_backend.create_vpc(cidr_block, instance_tenancy,
amazon_provided_ipv6_cidr_block=amazon_provided_ipv6_cidr_blocks)
doc_date = '2013-10-15' if 'Boto/' in self.headers.get('user-agent', '') else '2016-11-15'
template = self.response_template(CREATE_VPC_RESPONSE)
return template.render(vpc=vpc)
return template.render(vpc=vpc, doc_date=doc_date)
def delete_vpc(self):
vpc_id = self._get_param('VpcId')
@ -23,8 +26,9 @@ class VPCs(BaseResponse):
vpc_ids = self._get_multi_param('VpcId')
filters = filters_from_querystring(self.querystring)
vpcs = self.ec2_backend.get_all_vpcs(vpc_ids=vpc_ids, filters=filters)
doc_date = '2013-10-15' if 'Boto/' in self.headers.get('user-agent', '') else '2016-11-15'
template = self.response_template(DESCRIBE_VPCS_RESPONSE)
return template.render(vpcs=vpcs)
return template.render(vpcs=vpcs, doc_date=doc_date)
def describe_vpc_attribute(self):
vpc_id = self._get_param('VpcId')
@ -45,14 +49,63 @@ class VPCs(BaseResponse):
vpc_id, attr_name, attr_value)
return MODIFY_VPC_ATTRIBUTE_RESPONSE
def associate_vpc_cidr_block(self):
vpc_id = self._get_param('VpcId')
amazon_provided_ipv6_cidr_blocks = self._get_param('AmazonProvidedIpv6CidrBlock')
# todo test on AWS if can create an association for IPV4 and IPV6 in the same call?
cidr_block = self._get_param('CidrBlock') if not amazon_provided_ipv6_cidr_blocks else None
value = self.ec2_backend.associate_vpc_cidr_block(vpc_id, cidr_block, amazon_provided_ipv6_cidr_blocks)
if not amazon_provided_ipv6_cidr_blocks:
render_template = ASSOCIATE_VPC_CIDR_BLOCK_RESPONSE
else:
render_template = IPV6_ASSOCIATE_VPC_CIDR_BLOCK_RESPONSE
template = self.response_template(render_template)
return template.render(vpc_id=vpc_id, value=value, cidr_block=value['cidr_block'],
association_id=value['association_id'], cidr_block_state='associating')
def disassociate_vpc_cidr_block(self):
association_id = self._get_param('AssociationId')
value = self.ec2_backend.disassociate_vpc_cidr_block(association_id)
if "::" in value.get('cidr_block', ''):
render_template = IPV6_DISASSOCIATE_VPC_CIDR_BLOCK_RESPONSE
else:
render_template = DISASSOCIATE_VPC_CIDR_BLOCK_RESPONSE
template = self.response_template(render_template)
return template.render(vpc_id=value['vpc_id'], cidr_block=value['cidr_block'],
association_id=value['association_id'], cidr_block_state='disassociating')
CREATE_VPC_RESPONSE = """
<CreateVpcResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
<CreateVpcResponse xmlns="http://ec2.amazonaws.com/doc/{{doc_date}}/">
<requestId>7a62c49f-347e-4fc4-9331-6e8eEXAMPLE</requestId>
<vpc>
<vpcId>{{ vpc.id }}</vpcId>
<state>pending</state>
<cidrBlock>{{ vpc.cidr_block }}</cidrBlock>
{% if doc_date == "2016-11-15" %}
<cidrBlockAssociationSet>
{% for assoc in vpc.get_cidr_block_association_set() %}
<item>
<cidrBlock>{{assoc.cidr_block}}</cidrBlock>
<associationId>{{ assoc.association_id }}</associationId>
<cidrBlockState>
<state>{{assoc.cidr_block_state.state}}</state>
</cidrBlockState>
</item>
{% endfor %}
</cidrBlockAssociationSet>
<ipv6CidrBlockAssociationSet>
{% for assoc in vpc.get_cidr_block_association_set(ipv6=True) %}
<item>
<ipv6CidrBlock>{{assoc.cidr_block}}</ipv6CidrBlock>
<associationId>{{ assoc.association_id }}</associationId>
<ipv6CidrBlockState>
<state>{{assoc.cidr_block_state.state}}</state>
</ipv6CidrBlockState>
</item>
{% endfor %}
</ipv6CidrBlockAssociationSet>
{% endif %}
<dhcpOptionsId>{% if vpc.dhcp_options %}{{ vpc.dhcp_options.id }}{% else %}dopt-1a2b3c4d2{% endif %}</dhcpOptionsId>
<instanceTenancy>{{ vpc.instance_tenancy }}</instanceTenancy>
<tagSet>
@ -69,14 +122,38 @@ CREATE_VPC_RESPONSE = """
</CreateVpcResponse>"""
DESCRIBE_VPCS_RESPONSE = """
<DescribeVpcsResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
<requestId>7a62c49f-347e-4fc4-9331-6e8eEXAMPLE</requestId>
<DescribeVpcsResponse xmlns="http://ec2.amazonaws.com/doc/{{doc_date}}/">
<requestId>7a62c442-3484-4f42-9342-6942EXAMPLE</requestId>
<vpcSet>
{% for vpc in vpcs %}
<item>
<vpcId>{{ vpc.id }}</vpcId>
<state>{{ vpc.state }}</state>
<cidrBlock>{{ vpc.cidr_block }}</cidrBlock>
{% if doc_date == "2016-11-15" %}
<cidrBlockAssociationSet>
{% for assoc in vpc.get_cidr_block_association_set() %}
<item>
<cidrBlock>{{assoc.cidr_block}}</cidrBlock>
<associationId>{{ assoc.association_id }}</associationId>
<cidrBlockState>
<state>{{assoc.cidr_block_state.state}}</state>
</cidrBlockState>
</item>
{% endfor %}
</cidrBlockAssociationSet>
<ipv6CidrBlockAssociationSet>
{% for assoc in vpc.get_cidr_block_association_set(ipv6=True) %}
<item>
<ipv6CidrBlock>{{assoc.cidr_block}}</ipv6CidrBlock>
<associationId>{{ assoc.association_id }}</associationId>
<ipv6CidrBlockState>
<state>{{assoc.cidr_block_state.state}}</state>
</ipv6CidrBlockState>
</item>
{% endfor %}
</ipv6CidrBlockAssociationSet>
{% endif %}
<dhcpOptionsId>{% if vpc.dhcp_options %}{{ vpc.dhcp_options.id }}{% else %}dopt-7a8b9c2d{% endif %}</dhcpOptionsId>
<instanceTenancy>{{ vpc.instance_tenancy }}</instanceTenancy>
<isDefault>{{ vpc.is_default }}</isDefault>
@ -96,14 +173,14 @@ DESCRIBE_VPCS_RESPONSE = """
</DescribeVpcsResponse>"""
DELETE_VPC_RESPONSE = """
<DeleteVpcResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
<DeleteVpcResponse xmlns="http://ec2.amazonaws.com/doc/2016-11-15/">
<requestId>7a62c49f-347e-4fc4-9331-6e8eEXAMPLE</requestId>
<return>true</return>
</DeleteVpcResponse>
"""
DESCRIBE_VPC_ATTRIBUTE_RESPONSE = """
<DescribeVpcAttributeResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
<DescribeVpcAttributeResponse xmlns="http://ec2.amazonaws.com/doc/2016-11-15/">
<requestId>7a62c49f-347e-4fc4-9331-6e8eEXAMPLE</requestId>
<vpcId>{{ vpc_id }}</vpcId>
<{{ attribute }}>
@ -112,7 +189,59 @@ DESCRIBE_VPC_ATTRIBUTE_RESPONSE = """
</DescribeVpcAttributeResponse>"""
MODIFY_VPC_ATTRIBUTE_RESPONSE = """
<ModifyVpcAttributeResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
<ModifyVpcAttributeResponse xmlns="http://ec2.amazonaws.com/doc/2016-11-15/">
<requestId>7a62c49f-347e-4fc4-9331-6e8eEXAMPLE</requestId>
<return>true</return>
</ModifyVpcAttributeResponse>"""
ASSOCIATE_VPC_CIDR_BLOCK_RESPONSE = """
<AssociateVpcCidrBlockResponse xmlns="http://ec2.amazonaws.com/doc/2016-11-15/">
<requestId>7a62c49f-347e-4fc4-9331-6e8eEXAMPLE</requestId>
<vpcId>{{vpc_id}}</vpcId>
<cidrBlockAssociation>
<associationId>{{association_id}}</associationId>
<cidrBlock>{{cidr_block}}</cidrBlock>
<cidrBlockState>
<state>{{cidr_block_state}}</state>
</cidrBlockState>
</cidrBlockAssociation>
</AssociateVpcCidrBlockResponse>"""
DISASSOCIATE_VPC_CIDR_BLOCK_RESPONSE = """
<DisassociateVpcCidrBlockResponse xmlns="http://ec2.amazonaws.com/doc/2016-11-15/">
<requestId>7a62c49f-347e-4fc4-9331-6e8eEXAMPLE</requestId>
<vpcId>{{vpc_id}}</vpcId>
<cidrBlockAssociation>
<associationId>{{association_id}}</associationId>
<cidrBlock>{{cidr_block}}</cidrBlock>
<cidrBlockState>
<state>{{cidr_block_state}}</state>
</cidrBlockState>
</cidrBlockAssociation>
</DisassociateVpcCidrBlockResponse>"""
IPV6_ASSOCIATE_VPC_CIDR_BLOCK_RESPONSE = """
<AssociateVpcCidrBlockResponse xmlns="http://ec2.amazonaws.com/doc/2016-11-15/">
<requestId>33af6c54-1139-4d50-b4f7-15a8example</requestId>
<vpcId>{{vpc_id}}</vpcId>
<ipv6CidrBlockAssociation>
<associationId>{{association_id}}</associationId>
<ipv6CidrBlock>{{cidr_block}}</ipv6CidrBlock>
<ipv6CidrBlockState>
<state>{{cidr_block_state}}</state>
</ipv6CidrBlockState>
</ipv6CidrBlockAssociation>
</AssociateVpcCidrBlockResponse>"""
IPV6_DISASSOCIATE_VPC_CIDR_BLOCK_RESPONSE = """
<DisassociateVpcCidrBlockResponse xmlns="http://ec2.amazonaws.com/doc/2016-11-15/">
<requestId>33af6c54-1139-4d50-b4f7-15a8example</requestId>
<vpcId>{{vpc_id}}</vpcId>
<ipv6CidrBlockAssociation>
<associationId>{{association_id}}</associationId>
<ipv6CidrBlock>{{cidr_block}}</ipv6CidrBlock>
<ipv6CidrBlockState>
<state>{{cidr_block_state}}</state>
</ipv6CidrBlockState>
</ipv6CidrBlockAssociation>
</DisassociateVpcCidrBlockResponse>"""

24
moto/ec2/utils.py
View File

@ -27,6 +27,7 @@ EC2_RESOURCE_TO_PREFIX = {
'reservation': 'r',
'volume': 'vol',
'vpc': 'vpc',
'vpc-cidr-association-id': 'vpc-cidr-assoc',
'vpc-elastic-ip': 'eipalloc',
'vpc-elastic-ip-association': 'eipassoc',
'vpc-peering-connection': 'pcx',
@ -34,16 +35,17 @@ EC2_RESOURCE_TO_PREFIX = {
'vpn-gateway': 'vgw'}
EC2_PREFIX_TO_RESOURCE = dict((v, k)
for (k, v) in EC2_RESOURCE_TO_PREFIX.items())
EC2_PREFIX_TO_RESOURCE = dict((v, k) for (k, v) in EC2_RESOURCE_TO_PREFIX.items())
def random_resource_id(size=8):
chars = list(range(10)) + ['a', 'b', 'c', 'd', 'e', 'f']
resource_id = ''.join(six.text_type(random.choice(chars)) for x in range(size))
return resource_id
def random_id(prefix='', size=8):
chars = list(range(10)) + ['a', 'b', 'c', 'd', 'e', 'f']
resource_id = ''.join(six.text_type(random.choice(chars))
for x in range(size))
return '{0}-{1}'.format(prefix, resource_id)
return '{0}-{1}'.format(prefix, random_resource_id(size))
def random_ami_id():
@ -110,6 +112,10 @@ def random_vpc_id():
return random_id(prefix=EC2_RESOURCE_TO_PREFIX['vpc'])
def random_vpc_cidr_association_id():
return random_id(prefix=EC2_RESOURCE_TO_PREFIX['vpc-cidr-association-id'])
def random_vpc_peering_connection_id():
return random_id(prefix=EC2_RESOURCE_TO_PREFIX['vpc-peering-connection'])
@ -165,6 +171,10 @@ def random_ip():
)
def random_ipv6_cidr():
return "2400:6500:{}:{}::/56".format(random_resource_id(4), random_resource_id(4))
def generate_route_id(route_table_id, cidr_block):
return "%s~%s" % (route_table_id, cidr_block)

181
tests/test_ec2/test_vpcs.py
View File

@ -2,6 +2,8 @@ from __future__ import unicode_literals
# Ensure 'assert_raises' context manager support for Python 2.6
import tests.backport_assert_raises # flake8: noqa
from nose.tools import assert_raises
from moto.ec2.exceptions import EC2ClientError
from botocore.exceptions import ClientError
import boto3
import boto
@ -275,8 +277,8 @@ def test_default_vpc():
def test_non_default_vpc():
ec2 = boto3.resource('ec2', region_name='us-west-1')
# Create the default VPC
ec2.create_vpc(CidrBlock='172.31.0.0/16')
# 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')
@ -295,6 +297,12 @@ def test_non_default_vpc():
attr = response.get('EnableDnsHostnames')
attr.get('Value').shouldnt.be.ok
# Check Primary CIDR Block Associations
cidr_block_association_set = next(iter(vpc.cidr_block_association_set), None)
cidr_block_association_set['CidrBlockState']['State'].should.equal('associated')
cidr_block_association_set['CidrBlock'].should.equal(vpc.cidr_block)
cidr_block_association_set['AssociationId'].should.contain('vpc-cidr-assoc')
@mock_ec2
def test_vpc_dedicated_tenancy():
@ -340,7 +348,6 @@ def test_vpc_modify_enable_dns_hostnames():
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')
@ -364,3 +371,171 @@ def test_vpc_associate_dhcp_options():
vpc.update()
dhcp_options.id.should.equal(vpc.dhcp_options_id)
@mock_ec2
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='10.10.{}.0/24'.format(i))
response['CidrBlockAssociation']['CidrBlockState']['State'].should.equal('associating')
response['CidrBlockAssociation']['CidrBlock'].should.equal('10.10.{}.0/24'.format(i))
response['CidrBlockAssociation']['AssociationId'].should.contain('vpc-cidr-assoc')
# Check all associations exist
vpc = ec2.Vpc(vpc.id)
vpc.cidr_block_association_set.should.have.length_of(5)
vpc.cidr_block_association_set[2]['CidrBlockState']['State'].should.equal('associated')
vpc.cidr_block_association_set[4]['CidrBlockState']['State'].should.equal('associated')
# Check error on adding 6th association.
with assert_raises(ClientError) as ex:
response = ec2.meta.client.associate_vpc_cidr_block(VpcId=vpc.id, CidrBlock='10.10.50.0/22')
str(ex.exception).should.equal(
"An error occurred (CidrLimitExceeded) when calling the AssociateVpcCidrBlock "
"operation: This network '{}' has met its maximum number of allowed CIDRs: 5".format(vpc.id))
@mock_ec2
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'])
response['CidrBlockAssociation']['CidrBlockState']['State'].should.equal('disassociating')
response['CidrBlockAssociation']['CidrBlock'].should.equal(non_default_assoc_cidr_block['CidrBlock'])
response['CidrBlockAssociation']['AssociationId'].should.equal(non_default_assoc_cidr_block['AssociationId'])
# Error attempting to delete a non-existent CIDR_BLOCK association
with assert_raises(ClientError) as ex:
response = ec2.meta.client.disassociate_vpc_cidr_block(AssociationId='vpc-cidr-assoc-BORING123')
str(ex.exception).should.equal(
"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 assert_raises(ClientError) as ex:
response = ec2.meta.client.disassociate_vpc_cidr_block(AssociationId=vpc_base_cidr_assoc_id)
str(ex.exception).should.equal(
"An error occurred (OperationNotPermitted) when calling the DisassociateVpcCidrBlock operation: "
"The vpc CIDR block with association ID {} may not be disassociated. It is the primary "
"IPv4 CIDR block of the VPC".format(vpc_base_cidr_assoc_id))
@mock_ec2
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']}]))
filtered_vpcs.should.be.length_of(1)
filtered_vpcs[0].id.should.equal(vpc2.id)
# 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]}]))
filtered_vpcs.should.be.length_of(1)
filtered_vpcs[0].id.should.equal(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']}]))
filtered_vpcs.should.be.length_of(0)
@mock_ec2
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)
ipv6_cidr_block_association_set = next(iter(vpc.ipv6_cidr_block_association_set), None)
ipv6_cidr_block_association_set['Ipv6CidrBlockState']['State'].should.equal('associated')
ipv6_cidr_block_association_set['Ipv6CidrBlock'].should.contain('::/56')
ipv6_cidr_block_association_set['AssociationId'].should.contain('vpc-cidr-assoc')
# Test Fail on adding 2nd IPV6 association - AWS only allows 1 at this time!
with assert_raises(ClientError) as ex:
response = ec2.meta.client.associate_vpc_cidr_block(VpcId=vpc.id, AmazonProvidedIpv6CidrBlock=True)
str(ex.exception).should.equal(
"An error occurred (CidrLimitExceeded) when calling the AssociateVpcCidrBlock "
"operation: This network '{}' has met its maximum number of allowed CIDRs: 1".format(vpc.id))
# Test associate ipv6 cidr block after vpc created
vpc = ec2.create_vpc(CidrBlock='10.10.50.0/24')
response = ec2.meta.client.associate_vpc_cidr_block(VpcId=vpc.id, AmazonProvidedIpv6CidrBlock=True)
response['Ipv6CidrBlockAssociation']['Ipv6CidrBlockState']['State'].should.equal('associating')
response['Ipv6CidrBlockAssociation']['Ipv6CidrBlock'].should.contain('::/56')
response['Ipv6CidrBlockAssociation']['AssociationId'].should.contain('vpc-cidr-assoc-')
# Check on describe vpc that has ipv6 cidr block association
vpc = ec2.Vpc(vpc.id)
vpc.ipv6_cidr_block_association_set.should.be.length_of(1)
@mock_ec2
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']
response = ec2.meta.client.disassociate_vpc_cidr_block(AssociationId=assoc_id)
response['Ipv6CidrBlockAssociation']['Ipv6CidrBlockState']['State'].should.equal('disassociating')
response['Ipv6CidrBlockAssociation']['Ipv6CidrBlock'].should.contain('::/56')
response['Ipv6CidrBlockAssociation']['AssociationId'].should.equal(assoc_id)
@mock_ec2
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]}]))
filtered_vpcs.should.be.length_of(1)
filtered_vpcs[0].id.should.equal(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]}]))
filtered_vpcs.should.be.length_of(1)
filtered_vpcs[0].id.should.equal(vpc2.id)
# Test filter for association state in VPC - this will never show anything in this test
filtered_vpcs = list(ec2.vpcs.filter(Filters=[{'Name': 'ipv6-cidr-block-association.state',
'Values': ['associated']}]))
filtered_vpcs.should.be.length_of(2) # 2 of 4 VPCs