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

Merge pull request #2555 from catherinetcai/feature/ec2-vpc-classic-link

Browse Source 
Adds basic implementations for vpc-classic-link and vpc-classic-link-dns-support - Redo
This commit is contained in:
Mike Grima 2019-11-16 10:21:29 -08:00 committed by GitHub
commit 028aafd05b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 312 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
IMPLEMENTATION_COVERAGE.md
View File

@ -2217,8 +2217,8 @@
- [X] describe_volumes
- [ ] describe_volumes_modifications
- [X] describe_vpc_attribute
- [ ] describe_vpc_classic_link
- [ ] describe_vpc_classic_link_dns_support
- [X] describe_vpc_classic_link
- [X] describe_vpc_classic_link_dns_support
- [ ] describe_vpc_endpoint_connection_notifications
- [ ] describe_vpc_endpoint_connections
- [ ] describe_vpc_endpoint_service_configurations
@ -2237,8 +2237,8 @@
- [ ] disable_ebs_encryption_by_default
- [ ] disable_transit_gateway_route_table_propagation
- [ ] disable_vgw_route_propagation
- [ ] disable_vpc_classic_link
- [ ] disable_vpc_classic_link_dns_support
- [X] disable_vpc_classic_link
- [X] disable_vpc_classic_link_dns_support
- [X] disassociate_address
- [ ] disassociate_client_vpn_target_network
- [ ] disassociate_iam_instance_profile
@ -2250,8 +2250,8 @@
- [ ] enable_transit_gateway_route_table_propagation
- [ ] enable_vgw_route_propagation
- [ ] enable_volume_io
- [ ] enable_vpc_classic_link
- [ ] enable_vpc_classic_link_dns_support
- [X] enable_vpc_classic_link
- [X] enable_vpc_classic_link_dns_support
- [ ] export_client_vpn_client_certificate_revocation_list
- [ ] export_client_vpn_client_configuration
- [ ] export_image

43
moto/ec2/models.py
View File

@ -2444,6 +2444,7 @@ class VPC(TaggedEC2Resource):
self.instance_tenancy = instance_tenancy
self.is_default = "true" if is_default else "false"
self.enable_dns_support = "true"
self.classic_link_enabled = "false"
# This attribute is set to 'true' only for default VPCs
# or VPCs created using the wizard of the VPC console
self.enable_dns_hostnames = "true" if is_default else "false"
@ -2540,6 +2541,32 @@ class VPC(TaggedEC2Resource):
self.cidr_block_association_set[association_id] = association_set
return association_set
def enable_vpc_classic_link(self):
# Check if current cidr block doesn't fall within the 10.0.0.0/8 block, excluding 10.0.0.0/16 and 10.1.0.0/16.
# Doesn't check any route tables, maybe something for in the future?
# See https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/vpc-classiclink.html#classiclink-limitations
network_address = ipaddress.ip_network(self.cidr_block).network_address
if (
network_address not in ipaddress.ip_network("10.0.0.0/8")
or network_address in ipaddress.ip_network("10.0.0.0/16")
or network_address in ipaddress.ip_network("10.1.0.0/16")
):
self.classic_link_enabled = "true"
return self.classic_link_enabled
def disable_vpc_classic_link(self):
self.classic_link_enabled = "false"
return self.classic_link_enabled
def enable_vpc_classic_link_dns_support(self):
self.classic_link_dns_supported = "true"
return self.classic_link_dns_supported
def disable_vpc_classic_link_dns_support(self):
self.classic_link_dns_supported = "false"
return self.classic_link_dns_supported
def disassociate_vpc_cidr_block(self, association_id):
if self.cidr_block == self.cidr_block_association_set.get(
association_id, {}
@ -2670,6 +2697,22 @@ class VPCBackend(object):
else:
raise InvalidParameterValueError(attr_name)
def enable_vpc_classic_link(self, vpc_id):
vpc = self.get_vpc(vpc_id)
return vpc.enable_vpc_classic_link()
def disable_vpc_classic_link(self, vpc_id):
vpc = self.get_vpc(vpc_id)
return vpc.disable_vpc_classic_link()
def enable_vpc_classic_link_dns_support(self, vpc_id):
vpc = self.get_vpc(vpc_id)
return vpc.enable_vpc_classic_link_dns_support()
def disable_vpc_classic_link_dns_support(self, vpc_id):
vpc = self.get_vpc(vpc_id)
return vpc.disable_vpc_classic_link_dns_support()
def modify_vpc_attribute(self, vpc_id, attr_name, attr_value):
vpc = self.get_vpc(vpc_id)
if attr_name in ("enable_dns_support", "enable_dns_hostnames"):

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

@ -5,6 +5,13 @@ from moto.ec2.utils import filters_from_querystring
class VPCs(BaseResponse):
def _get_doc_date(self):
return (
"2013-10-15"
if "Boto/" in self.headers.get("user-agent", "")
else "2016-11-15"
)
def create_vpc(self):
cidr_block = self._get_param("CidrBlock")
instance_tenancy = self._get_param("InstanceTenancy", if_none="default")
@ -16,11 +23,7 @@ class VPCs(BaseResponse):
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"
)
doc_date = self._get_doc_date()
template = self.response_template(CREATE_VPC_RESPONSE)
return template.render(vpc=vpc, doc_date=doc_date)
@ -50,6 +53,64 @@ class VPCs(BaseResponse):
template = self.response_template(DESCRIBE_VPC_ATTRIBUTE_RESPONSE)
return template.render(vpc_id=vpc_id, attribute=attribute, value=value)
def describe_vpc_classic_link_dns_support(self):
vpc_ids = self._get_multi_param("VpcIds")
filters = filters_from_querystring(self.querystring)
vpcs = self.ec2_backend.get_all_vpcs(vpc_ids=vpc_ids, filters=filters)
doc_date = self._get_doc_date()
template = self.response_template(
DESCRIBE_VPC_CLASSIC_LINK_DNS_SUPPORT_RESPONSE
)
return template.render(vpcs=vpcs, doc_date=doc_date)
def enable_vpc_classic_link_dns_support(self):
vpc_id = self._get_param("VpcId")
classic_link_dns_supported = self.ec2_backend.enable_vpc_classic_link_dns_support(
vpc_id=vpc_id
)
doc_date = self._get_doc_date()
template = self.response_template(ENABLE_VPC_CLASSIC_LINK_DNS_SUPPORT_RESPONSE)
return template.render(
classic_link_dns_supported=classic_link_dns_supported, doc_date=doc_date
)
def disable_vpc_classic_link_dns_support(self):
vpc_id = self._get_param("VpcId")
classic_link_dns_supported = self.ec2_backend.disable_vpc_classic_link_dns_support(
vpc_id=vpc_id
)
doc_date = self._get_doc_date()
template = self.response_template(DISABLE_VPC_CLASSIC_LINK_DNS_SUPPORT_RESPONSE)
return template.render(
classic_link_dns_supported=classic_link_dns_supported, doc_date=doc_date
)
def describe_vpc_classic_link(self):
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 = self._get_doc_date()
template = self.response_template(DESCRIBE_VPC_CLASSIC_LINK_RESPONSE)
return template.render(vpcs=vpcs, doc_date=doc_date)
def enable_vpc_classic_link(self):
vpc_id = self._get_param("VpcId")
classic_link_enabled = self.ec2_backend.enable_vpc_classic_link(vpc_id=vpc_id)
doc_date = self._get_doc_date()
template = self.response_template(ENABLE_VPC_CLASSIC_LINK_RESPONSE)
return template.render(
classic_link_enabled=classic_link_enabled, doc_date=doc_date
)
def disable_vpc_classic_link(self):
vpc_id = self._get_param("VpcId")
classic_link_enabled = self.ec2_backend.disable_vpc_classic_link(vpc_id=vpc_id)
doc_date = self._get_doc_date()
template = self.response_template(DISABLE_VPC_CLASSIC_LINK_RESPONSE)
return template.render(
classic_link_enabled=classic_link_enabled, doc_date=doc_date
)
def modify_vpc_attribute(self):
vpc_id = self._get_param("VpcId")
@ -149,6 +210,56 @@ CREATE_VPC_RESPONSE = """
</vpc>
</CreateVpcResponse>"""
DESCRIBE_VPC_CLASSIC_LINK_DNS_SUPPORT_RESPONSE = """
<DescribeVpcClassicLinkDnsSupportResponse xmlns="http://ec2.amazonaws.com/doc/{{doc_date}}/">
<requestId>7a62c442-3484-4f42-9342-6942EXAMPLE</requestId>
<vpcs>
{% for vpc in vpcs %}
<item>
<vpcId>{{ vpc.id }}</vpcId>
<classicLinkDnsSupported>{{ vpc.classic_link_dns_supported }}</classicLinkDnsSupported>
</item>
{% endfor %}
</vpcs>
</DescribeVpcClassicLinkDnsSupportResponse>"""
ENABLE_VPC_CLASSIC_LINK_DNS_SUPPORT_RESPONSE = """
<EnableVpcClassicLinkDnsSupportResponse xmlns="http://ec2.amazonaws.com/doc/{{doc_date}}/">
<requestId>7a62c442-3484-4f42-9342-6942EXAMPLE</requestId>
<return>{{ classic_link_dns_supported }}</return>
</EnableVpcClassicLinkDnsSupportResponse>"""
DISABLE_VPC_CLASSIC_LINK_DNS_SUPPORT_RESPONSE = """
<DisableVpcClassicLinkDnsSupportResponse xmlns="http://ec2.amazonaws.com/doc/{{doc_date}}/">
<requestId>7a62c442-3484-4f42-9342-6942EXAMPLE</requestId>
<return>{{ classic_link_dns_supported }}</return>
</DisableVpcClassicLinkDnsSupportResponse>"""
DESCRIBE_VPC_CLASSIC_LINK_RESPONSE = """
<DescribeVpcClassicLinkResponse 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>
<classicLinkEnabled>{{ vpc.classic_link_enabled }}</classicLinkEnabled>
</item>
{% endfor %}
</vpcSet>
</DescribeVpcClassicLinkResponse>"""
ENABLE_VPC_CLASSIC_LINK_RESPONSE = """
<EnableVpcClassicLinkResponse xmlns="http://ec2.amazonaws.com/doc/{{doc_date}}/">
<requestId>7a62c442-3484-4f42-9342-6942EXAMPLE</requestId>
<return>{{ classic_link_enabled }}</return>
</EnableVpcClassicLinkResponse>"""
DISABLE_VPC_CLASSIC_LINK_RESPONSE = """
<DisableVpcClassicLinkResponse xmlns="http://ec2.amazonaws.com/doc/{{doc_date}}/">
<requestId>7a62c442-3484-4f42-9342-6942EXAMPLE</requestId>
<return>{{ classic_link_enabled }}</return>
</DisableVpcClassicLinkResponse>"""
DESCRIBE_VPCS_RESPONSE = """
<DescribeVpcsResponse xmlns="http://ec2.amazonaws.com/doc/{{doc_date}}/">
<requestId>7a62c442-3484-4f42-9342-6942EXAMPLE</requestId>

147
tests/test_ec2/test_vpcs.py
View File

@ -678,3 +678,150 @@ def test_create_vpc_with_invalid_cidr_range():
"An error occurred (InvalidVpc.Range) when calling the CreateVpc "
"operation: The CIDR '{}' is invalid.".format(vpc_cidr_block)
)
@mock_ec2
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").should.be.true
@mock_ec2
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").should.be.false
@mock_ec2
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").should.be.false
@mock_ec2
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").should.be.true
@mock_ec2
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").should.be.false
@mock_ec2
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_ec2
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").should.be.true
@mock_ec2
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").should.be.false
@mock_ec2
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").should.be.true
@mock_ec2
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").should.be.false
@mock_ec2
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"]
)