Fix #1830 Add support for cross-region VPC peering

Add a class level store in models/VPCBackend of ec2
for saving vpcs of all regions info. Any instance can correctly find vpc in another region
when connecting vpc of cross-region or vpc of same region.

Modify vpc_peering_connections in ec2/responses to handle
vpc peering of same region or cross region.

Update vpc_peering_connections response
template content to latest (2016-11-15) .

Add vpc cross region peering successful test case.
Add vpc cross region peering fail test case.

Related: https://github.com/spulec/moto/issues/1830

Reference
CreateVpcPeeringConnection Sample Response
https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_CreateVpcPeeringConnection.html
This commit is contained in:
hans 2018-09-21 23:29:04 +08:00
commit cedb89dc3b
3 changed files with 76 additions and 12 deletions

View file

@ -13,6 +13,7 @@ from pkg_resources import resource_filename
import boto.ec2
from collections import defaultdict
import weakref
from datetime import datetime
from boto.ec2.instance import Instance as BotoInstance, Reservation
from boto.ec2.blockdevicemapping import BlockDeviceMapping, BlockDeviceType
@ -2115,10 +2116,20 @@ class VPC(TaggedEC2Resource):
class VPCBackend(object):
__refs__ = defaultdict(list)
def __init__(self):
self.vpcs = {}
self.__refs__[self.__class__].append(weakref.ref(self))
super(VPCBackend, self).__init__()
@classmethod
def get_instances(cls):
for inst_ref in cls.__refs__[cls]:
inst = inst_ref()
if inst is not None:
yield inst
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, amazon_provided_ipv6_cidr_block)
@ -2142,6 +2153,13 @@ class VPCBackend(object):
raise InvalidVPCIdError(vpc_id)
return self.vpcs.get(vpc_id)
# get vpc by vpc id and aws region
def get_cross_vpc(self, vpc_id, peer_region):
for vpcs in self.get_instances():
if vpcs.region_name == peer_region:
match_vpc = vpcs.get_vpc(vpc_id)
return match_vpc
def get_all_vpcs(self, vpc_ids=None, filters=None):
matches = self.vpcs.values()
if vpc_ids: