Fix: Ec2 - add destinationIpv6CIDR support. (#3106)

* Fix: Ec2 - add destinationIpv6CIDR support.

* removing unneccessary debug statements

* modifying existing test case

* Linting

Co-authored-by: usmankb <usman@krazybee.com>
Co-authored-by: Bert Blommers <info@bertblommers.nl>
This commit is contained in:
usmangani1 2020-07-07 19:02:55 +05:30 committed by GitHub
parent 1299fef8b8
commit 81be4b37a1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 27 additions and 5 deletions

View File

@ -3547,6 +3547,7 @@ class Route(object):
self,
route_table,
destination_cidr_block,
destination_ipv6_cidr_block,
local=False,
gateway=None,
instance=None,
@ -3554,9 +3555,12 @@ class Route(object):
interface=None,
vpc_pcx=None,
):
self.id = generate_route_id(route_table.id, destination_cidr_block)
self.id = generate_route_id(
route_table.id, destination_cidr_block, destination_ipv6_cidr_block
)
self.route_table = route_table
self.destination_cidr_block = destination_cidr_block
self.destination_ipv6_cidr_block = destination_ipv6_cidr_block
self.local = local
self.gateway = gateway
self.instance = instance
@ -3632,6 +3636,7 @@ class RouteBackend(object):
self,
route_table_id,
destination_cidr_block,
destination_ipv6_cidr_block=None,
local=False,
gateway_id=None,
instance_id=None,
@ -3656,9 +3661,10 @@ class RouteBackend(object):
gateway = self.get_internet_gateway(gateway_id)
try:
ipaddress.IPv4Network(
six.text_type(destination_cidr_block), strict=False
)
if destination_cidr_block:
ipaddress.IPv4Network(
six.text_type(destination_cidr_block), strict=False
)
except ValueError:
raise InvalidDestinationCIDRBlockParameterError(destination_cidr_block)
@ -3668,6 +3674,7 @@ class RouteBackend(object):
route = Route(
route_table,
destination_cidr_block,
destination_ipv6_cidr_block,
local=local,
gateway=gateway,
instance=self.get_instance(instance_id) if instance_id else None,

View File

@ -16,6 +16,7 @@ class RouteTables(BaseResponse):
def create_route(self):
route_table_id = self._get_param("RouteTableId")
destination_cidr_block = self._get_param("DestinationCidrBlock")
destination_ipv6_cidr_block = self._get_param("DestinationIpv6CidrBlock")
gateway_id = self._get_param("GatewayId")
instance_id = self._get_param("InstanceId")
nat_gateway_id = self._get_param("NatGatewayId")
@ -25,6 +26,7 @@ class RouteTables(BaseResponse):
self.ec2_backend.create_route(
route_table_id,
destination_cidr_block,
destination_ipv6_cidr_block,
gateway_id=gateway_id,
instance_id=instance_id,
nat_gateway_id=nat_gateway_id,

View File

@ -189,7 +189,9 @@ 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):
def generate_route_id(route_table_id, cidr_block, ipv6_cidr_block=None):
if ipv6_cidr_block and not cidr_block:
cidr_block = ipv6_cidr_block
return "%s~%s" % (route_table_id, cidr_block)

View File

@ -582,6 +582,17 @@ def test_create_route_with_invalid_destination_cidr_block_parameter():
)
)
route_table.create_route(
DestinationIpv6CidrBlock="2001:db8::/125", GatewayId=internet_gateway.id
)
new_routes = [
route
for route in route_table.routes
if route.destination_cidr_block != vpc.cidr_block
]
new_routes.should.have.length_of(1)
new_routes[0].route_table_id.shouldnt.be.equal(None)
@mock_ec2
def test_create_route_with_network_interface_id():