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

EC2: Cross-account VPC peering connections (#6826)

Browse Source
This commit is contained in:
Viren Nadkarni 2023-09-20 12:53:17 +05:30 committed by GitHub
parent 643759bd64
commit 5f1fb65038
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 634 additions and 255 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
moto/ec2/exceptions.py
View File

@ -632,21 +632,23 @@ class InvalidVPCRangeError(EC2ClientError):
super().__init__("InvalidVpc.Range", f"The CIDR '{cidr_block}' is invalid.") super().__init__("InvalidVpc.Range", f"The CIDR '{cidr_block}' is invalid.")
# accept exception # Raised when attempting to accept a VPC peering connection request in own account but in the requester region
class OperationNotPermitted2(EC2ClientError): class OperationNotPermitted2(EC2ClientError):
def __init__(self, client_region: str, pcx_id: str, acceptor_region: str): def __init__(self, client_region: str, pcx_id: str, acceptor_region: str):
super().__init__( super().__init__(
"OperationNotPermitted", "OperationNotPermitted",
f"Incorrect region ({client_region}) specified for this request.VPC peering connection {pcx_id} must be accepted in region {acceptor_region}", f"Incorrect region ({client_region}) specified for this request. "
f"VPC peering connection {pcx_id} must be accepted in region {acceptor_region}",
) )
# reject exception # Raised when attempting to reject a VPC peering connection request in own account but in the requester region
class OperationNotPermitted3(EC2ClientError): class OperationNotPermitted3(EC2ClientError):
def __init__(self, client_region: str, pcx_id: str, acceptor_region: str): def __init__(self, client_region: str, pcx_id: str, acceptor_region: str):
super().__init__( super().__init__(
"OperationNotPermitted", "OperationNotPermitted",
f"Incorrect region ({client_region}) specified for this request.VPC peering connection {pcx_id} must be accepted or rejected in region {acceptor_region}", f"Incorrect region ({client_region}) specified for this request. "
f"VPC peering connection {pcx_id} must be accepted or rejected in region {acceptor_region}",
) )
@ -658,6 +660,15 @@ class OperationNotPermitted4(EC2ClientError):
) )
# Raised when attempting to accept or reject a VPC peering connection request for a VPC not belonging to self
class OperationNotPermitted5(EC2ClientError):
def __init__(self, account_id: str, pcx_id: str, operation: str):
super().__init__(
"OperationNotPermitted",
f"User ({account_id}) cannot {operation} peering {pcx_id}",
)
class InvalidLaunchTemplateNameAlreadyExistsError(EC2ClientError): class InvalidLaunchTemplateNameAlreadyExistsError(EC2ClientError):
def __init__(self) -> None: def __init__(self) -> None:
super().__init__( super().__init__(

4
moto/ec2/models/transit_gateway_attachments.py
View File

@ -95,7 +95,7 @@ class TransitGatewayPeeringAttachment(TransitGatewayAttachment):
"region": region_name, "region": region_name,
"transitGatewayId": transit_gateway_id, "transitGatewayId": transit_gateway_id,
} }
self.status = PeeringConnectionStatus() self.status = PeeringConnectionStatus(accepter_id=peer_account_id)
class TransitGatewayAttachmentBackend: class TransitGatewayAttachmentBackend:
@ -342,5 +342,5 @@ class TransitGatewayAttachmentBackend:
transit_gateway_attachment_id transit_gateway_attachment_id
] ]
transit_gateway_attachment.state = "deleted" transit_gateway_attachment.state = "deleted"
transit_gateway_attachment.status.deleted() # type: ignore[attr-defined] transit_gateway_attachment.status.deleted(deleter_id=self.account_id) # type: ignore[attr-defined]
return transit_gateway_attachment return transit_gateway_attachment

62
moto/ec2/models/vpc_peering_connections.py
View File

@ -7,6 +7,7 @@ from ..exceptions import (
InvalidVPCPeeringConnectionStateTransitionError, InvalidVPCPeeringConnectionStateTransitionError,
OperationNotPermitted2, OperationNotPermitted2,
OperationNotPermitted3, OperationNotPermitted3,
OperationNotPermitted5,
) )
from .core import TaggedEC2Resource from .core import TaggedEC2Resource
from .vpcs import VPC from .vpcs import VPC
@ -14,21 +15,24 @@ from ..utils import random_vpc_peering_connection_id
class PeeringConnectionStatus: class PeeringConnectionStatus:
def __init__(self, code: str = "initiating-request", message: str = ""): def __init__(
self, accepter_id: str, code: str = "initiating-request", message: str = ""
):
self.accepter_id = accepter_id
self.code = code self.code = code
self.message = message self.message = message
def deleted(self) -> None: def deleted(self, deleter_id: str) -> None:
self.code = "deleted" self.code = "deleted"
self.message = "Deleted by {deleter ID}" self.message = f"Deleted by {deleter_id}"
def initiating(self) -> None: def initiating(self) -> None:
self.code = "initiating-request" self.code = "initiating-request"
self.message = "Initiating Request to {accepter ID}" self.message = f"Initiating Request to {self.accepter_id}"
def pending(self) -> None: def pending(self) -> None:
self.code = "pending-acceptance" self.code = "pending-acceptance"
self.message = "Pending Acceptance by {accepter ID}" self.message = f"Pending Acceptance by {self.accepter_id}"
def accept(self) -> None: def accept(self) -> None:
self.code = "active" self.code = "active"
@ -61,7 +65,7 @@ class VPCPeeringConnection(TaggedEC2Resource, CloudFormationModel):
self.requester_options = self.DEFAULT_OPTIONS.copy() self.requester_options = self.DEFAULT_OPTIONS.copy()
self.accepter_options = self.DEFAULT_OPTIONS.copy() self.accepter_options = self.DEFAULT_OPTIONS.copy()
self.add_tags(tags or {}) self.add_tags(tags or {})
self._status = PeeringConnectionStatus() self._status = PeeringConnectionStatus(accepter_id=peer_vpc.owner_id)
@staticmethod @staticmethod
def cloudformation_name_type() -> str: def cloudformation_name_type() -> str:
@ -79,7 +83,7 @@ class VPCPeeringConnection(TaggedEC2Resource, CloudFormationModel):
cloudformation_json: Any, cloudformation_json: Any,
account_id: str, account_id: str,
region_name: str, region_name: str,
**kwargs: Any **kwargs: Any,
) -> "VPCPeeringConnection": ) -> "VPCPeeringConnection":
from ..models import ec2_backends from ..models import ec2_backends
@ -120,11 +124,15 @@ class VPCPeeringConnectionBackend:
vpc_pcx = VPCPeeringConnection(self, vpc_pcx_id, vpc, peer_vpc, tags) vpc_pcx = VPCPeeringConnection(self, vpc_pcx_id, vpc, peer_vpc, tags)
vpc_pcx._status.pending() vpc_pcx._status.pending()
self.vpc_pcxs[vpc_pcx_id] = vpc_pcx self.vpc_pcxs[vpc_pcx_id] = vpc_pcx
# insert cross region peering info # insert cross-account/cross-region peering info
if vpc.ec2_backend.region_name != peer_vpc.ec2_backend.region_name: if vpc.owner_id != peer_vpc.owner_id or vpc.region != peer_vpc.region:
for vpc_pcx_cx in peer_vpc.ec2_backend.get_vpc_pcx_refs(): for backend in peer_vpc.ec2_backend.get_vpc_pcx_refs():
if vpc_pcx_cx.region_name == peer_vpc.ec2_backend.region_name: if (
vpc_pcx_cx.vpc_pcxs[vpc_pcx_id] = vpc_pcx backend.account_id == peer_vpc.owner_id
and backend.region_name == peer_vpc.region
):
backend.vpc_pcxs[vpc_pcx_id] = vpc_pcx
return vpc_pcx return vpc_pcx
def describe_vpc_peering_connections( def describe_vpc_peering_connections(
@ -142,16 +150,24 @@ class VPCPeeringConnectionBackend:
def delete_vpc_peering_connection(self, vpc_pcx_id: str) -> VPCPeeringConnection: def delete_vpc_peering_connection(self, vpc_pcx_id: str) -> VPCPeeringConnection:
deleted = self.get_vpc_peering_connection(vpc_pcx_id) deleted = self.get_vpc_peering_connection(vpc_pcx_id)
deleted._status.deleted() deleted._status.deleted(deleter_id=self.account_id) # type: ignore[attr-defined]
return deleted return deleted
def accept_vpc_peering_connection(self, vpc_pcx_id: str) -> VPCPeeringConnection: def accept_vpc_peering_connection(self, vpc_pcx_id: str) -> VPCPeeringConnection:
vpc_pcx = self.get_vpc_peering_connection(vpc_pcx_id) vpc_pcx = self.get_vpc_peering_connection(vpc_pcx_id)
# if cross region need accepter from another region
pcx_req_region = vpc_pcx.vpc.ec2_backend.region_name # validate cross-account acceptance
pcx_acp_region = vpc_pcx.peer_vpc.ec2_backend.region_name req_account_id = vpc_pcx.vpc.owner_id
acp_account_id = vpc_pcx.peer_vpc.owner_id
if req_account_id != acp_account_id and self.account_id != acp_account_id: # type: ignore[attr-defined]
raise OperationNotPermitted5(self.account_id, vpc_pcx_id, "accept") # type: ignore[attr-defined]
# validate cross-region acceptance
pcx_req_region = vpc_pcx.vpc.region
pcx_acp_region = vpc_pcx.peer_vpc.region
if pcx_req_region != pcx_acp_region and self.region_name == pcx_req_region: # type: ignore[attr-defined] if pcx_req_region != pcx_acp_region and self.region_name == pcx_req_region: # type: ignore[attr-defined]
raise OperationNotPermitted2(self.region_name, vpc_pcx.id, pcx_acp_region) # type: ignore[attr-defined] raise OperationNotPermitted2(self.region_name, vpc_pcx.id, pcx_acp_region) # type: ignore[attr-defined]
if vpc_pcx._status.code != "pending-acceptance": if vpc_pcx._status.code != "pending-acceptance":
raise InvalidVPCPeeringConnectionStateTransitionError(vpc_pcx.id) raise InvalidVPCPeeringConnectionStateTransitionError(vpc_pcx.id)
vpc_pcx._status.accept() vpc_pcx._status.accept()
@ -159,11 +175,19 @@ class VPCPeeringConnectionBackend:
def reject_vpc_peering_connection(self, vpc_pcx_id: str) -> VPCPeeringConnection: def reject_vpc_peering_connection(self, vpc_pcx_id: str) -> VPCPeeringConnection:
vpc_pcx = self.get_vpc_peering_connection(vpc_pcx_id) vpc_pcx = self.get_vpc_peering_connection(vpc_pcx_id)
# if cross region need accepter from another region
pcx_req_region = vpc_pcx.vpc.ec2_backend.region_name # validate cross-account rejection
pcx_acp_region = vpc_pcx.peer_vpc.ec2_backend.region_name req_account_id = vpc_pcx.vpc.owner_id
acp_account_id = vpc_pcx.peer_vpc.owner_id
if req_account_id != acp_account_id and self.account_id != acp_account_id: # type: ignore[attr-defined]
raise OperationNotPermitted5(self.account_id, vpc_pcx_id, "reject") # type: ignore[attr-defined]
# validate cross-region acceptance
pcx_req_region = vpc_pcx.vpc.region
pcx_acp_region = vpc_pcx.peer_vpc.region
if pcx_req_region != pcx_acp_region and self.region_name == pcx_req_region: # type: ignore[attr-defined] if pcx_req_region != pcx_acp_region and self.region_name == pcx_req_region: # type: ignore[attr-defined]
raise OperationNotPermitted3(self.region_name, vpc_pcx.id, pcx_acp_region) # type: ignore[attr-defined] raise OperationNotPermitted3(self.region_name, vpc_pcx.id, pcx_acp_region) # type: ignore[attr-defined]
if vpc_pcx._status.code != "pending-acceptance": if vpc_pcx._status.code != "pending-acceptance":
raise InvalidVPCPeeringConnectionStateTransitionError(vpc_pcx.id) raise InvalidVPCPeeringConnectionStateTransitionError(vpc_pcx.id)
vpc_pcx._status.reject() vpc_pcx._status.reject()

4
moto/ec2/models/vpcs.py
View File

@ -205,6 +205,10 @@ class VPC(TaggedEC2Resource, CloudFormationModel):
def owner_id(self) -> str: def owner_id(self) -> str:
return self.ec2_backend.account_id return self.ec2_backend.account_id
@property
def region(self) -> str:
return self.ec2_backend.region_name
@staticmethod @staticmethod
def cloudformation_name_type() -> str: def cloudformation_name_type() -> str:
return "" return ""

114
moto/ec2/responses/vpc_peering_connections.py
View File

@ -3,21 +3,23 @@ from ._base_response import EC2BaseResponse
class VPCPeeringConnections(EC2BaseResponse): class VPCPeeringConnections(EC2BaseResponse):
def create_vpc_peering_connection(self) -> str: def create_vpc_peering_connection(self) -> str:
peer_region = self._get_param("PeerRegion")
tags = self._parse_tag_specification().get("vpc-peering-connection", {}) tags = self._parse_tag_specification().get("vpc-peering-connection", {})
if peer_region == self.region or peer_region is None: account_id = self._get_param("PeerOwnerId") or self.current_account
peer_vpc = self.ec2_backend.get_vpc(self._get_param("PeerVpcId")) region_name = self._get_param("PeerRegion") or self.region
else:
from moto.ec2.models import ec2_backends
peer_vpc = ec2_backends[self.current_account][peer_region].get_vpc(
self._get_param("PeerVpcId")
)
vpc = self.ec2_backend.get_vpc(self._get_param("VpcId")) vpc = self.ec2_backend.get_vpc(self._get_param("VpcId"))
# Peer VPC could belong to another account or region
from moto.ec2.models import ec2_backends
peer_vpc = ec2_backends[account_id][region_name].get_vpc(
self._get_param("PeerVpcId")
)
vpc_pcx = self.ec2_backend.create_vpc_peering_connection(vpc, peer_vpc, tags) vpc_pcx = self.ec2_backend.create_vpc_peering_connection(vpc, peer_vpc, tags)
template = self.response_template(CREATE_VPC_PEERING_CONNECTION_RESPONSE) template = self.response_template(CREATE_VPC_PEERING_CONNECTION_RESPONSE)
return template.render(account_id=self.current_account, vpc_pcx=vpc_pcx) return template.render(vpc_pcx=vpc_pcx)
def delete_vpc_peering_connection(self) -> str: def delete_vpc_peering_connection(self) -> str:
vpc_pcx_id = self._get_param("VpcPeeringConnectionId") vpc_pcx_id = self._get_param("VpcPeeringConnectionId")
@ -31,13 +33,13 @@ class VPCPeeringConnections(EC2BaseResponse):
vpc_peering_ids=ids vpc_peering_ids=ids
) )
template = self.response_template(DESCRIBE_VPC_PEERING_CONNECTIONS_RESPONSE) template = self.response_template(DESCRIBE_VPC_PEERING_CONNECTIONS_RESPONSE)
return template.render(account_id=self.current_account, vpc_pcxs=vpc_pcxs) return template.render(vpc_pcxs=vpc_pcxs)
def accept_vpc_peering_connection(self) -> str: def accept_vpc_peering_connection(self) -> str:
vpc_pcx_id = self._get_param("VpcPeeringConnectionId") vpc_pcx_id = self._get_param("VpcPeeringConnectionId")
vpc_pcx = self.ec2_backend.accept_vpc_peering_connection(vpc_pcx_id) vpc_pcx = self.ec2_backend.accept_vpc_peering_connection(vpc_pcx_id)
template = self.response_template(ACCEPT_VPC_PEERING_CONNECTION_RESPONSE) template = self.response_template(ACCEPT_VPC_PEERING_CONNECTION_RESPONSE)
return template.render(account_id=self.current_account, vpc_pcx=vpc_pcx) return template.render(vpc_pcx=vpc_pcx)
def reject_vpc_peering_connection(self) -> str: def reject_vpc_peering_connection(self) -> str:
vpc_pcx_id = self._get_param("VpcPeeringConnectionId") vpc_pcx_id = self._get_param("VpcPeeringConnectionId")
@ -68,39 +70,46 @@ CREATE_VPC_PEERING_CONNECTION_RESPONSE = """
<CreateVpcPeeringConnectionResponse xmlns="http://ec2.amazonaws.com/doc/2016-11-15/"> <CreateVpcPeeringConnectionResponse xmlns="http://ec2.amazonaws.com/doc/2016-11-15/">
<requestId>7a62c49f-347e-4fc4-9331-6e8eEXAMPLE</requestId> <requestId>7a62c49f-347e-4fc4-9331-6e8eEXAMPLE</requestId>
<vpcPeeringConnection> <vpcPeeringConnection>
<vpcPeeringConnectionId>{{ vpc_pcx.id }}</vpcPeeringConnectionId> <vpcPeeringConnectionId>{{ vpc_pcx.id }}</vpcPeeringConnectionId>
<requesterVpcInfo> <requesterVpcInfo>
<ownerId>{{ account_id }}</ownerId> <ownerId>{{ vpc_pcx.vpc.owner_id }}</ownerId>
<region>{{ vpc_pcx.vpc.region }}</region>
<vpcId>{{ vpc_pcx.vpc.id }}</vpcId> <vpcId>{{ vpc_pcx.vpc.id }}</vpcId>
<cidrBlock>{{ vpc_pcx.vpc.cidr_block }}</cidrBlock> <cidrBlock>{{ vpc_pcx.vpc.cidr_block }}</cidrBlock>
<cidrBlockSet></cidrBlockSet>
<ipv6CidrBlockSet></ipv6CidrBlockSet>
<peeringOptions> <peeringOptions>
<allowEgressFromLocalClassicLinkToRemoteVpc>{{ vpc_pcx.requester_options.AllowEgressFromLocalClassicLinkToRemoteVpc or '' }}</allowEgressFromLocalClassicLinkToRemoteVpc> <allowEgressFromLocalClassicLinkToRemoteVpc>{{ vpc_pcx.requester_options.AllowEgressFromLocalClassicLinkToRemoteVpc or '' }}</allowEgressFromLocalClassicLinkToRemoteVpc>
<allowEgressFromLocalVpcToRemoteClassicLink>{{ vpc_pcx.requester_options.AllowEgressFromLocalVpcToRemoteClassicLink or '' }}</allowEgressFromLocalVpcToRemoteClassicLink> <allowEgressFromLocalVpcToRemoteClassicLink>{{ vpc_pcx.requester_options.AllowEgressFromLocalVpcToRemoteClassicLink or '' }}</allowEgressFromLocalVpcToRemoteClassicLink>
<allowDnsResolutionFromRemoteVpc>{{ vpc_pcx.requester_options.AllowDnsResolutionFromRemoteVpc or '' }}</allowDnsResolutionFromRemoteVpc> <allowDnsResolutionFromRemoteVpc>{{ vpc_pcx.requester_options.AllowDnsResolutionFromRemoteVpc or '' }}</allowDnsResolutionFromRemoteVpc>
</peeringOptions> </peeringOptions>
</requesterVpcInfo> </requesterVpcInfo>
<accepterVpcInfo> <accepterVpcInfo>
<ownerId>{{ account_id }}</ownerId> <ownerId>{{ vpc_pcx.peer_vpc.owner_id }}</ownerId>
<vpcId>{{ vpc_pcx.peer_vpc.id }}</vpcId> <region>{{ vpc_pcx.peer_vpc.region }}</region>
<peeringOptions> <vpcId>{{ vpc_pcx.peer_vpc.id }}</vpcId>
<allowEgressFromLocalClassicLinkToRemoteVpc>{{ vpc_pcx.accepter_options.AllowEgressFromLocalClassicLinkToRemoteVpc or '' }}</allowEgressFromLocalClassicLinkToRemoteVpc> <cidrBlock>{{ vpc_pcx.peer_vpc.cidr_block }}</cidrBlock>
<allowEgressFromLocalVpcToRemoteClassicLink>{{ vpc_pcx.accepter_options.AllowEgressFromLocalVpcToRemoteClassicLink or '' }}</allowEgressFromLocalVpcToRemoteClassicLink> <cidrBlockSet></cidrBlockSet>
<allowDnsResolutionFromRemoteVpc>{{ vpc_pcx.accepter_options.AllowDnsResolutionFromRemoteVpc or '' }}</allowDnsResolutionFromRemoteVpc> <ipv6CidrBlockSet></ipv6CidrBlockSet>
</peeringOptions> <peeringOptions>
</accepterVpcInfo> <allowEgressFromLocalClassicLinkToRemoteVpc>{{ vpc_pcx.accepter_options.AllowEgressFromLocalClassicLinkToRemoteVpc or '' }}</allowEgressFromLocalClassicLinkToRemoteVpc>
<status> <allowEgressFromLocalVpcToRemoteClassicLink>{{ vpc_pcx.accepter_options.AllowEgressFromLocalVpcToRemoteClassicLink or '' }}</allowEgressFromLocalVpcToRemoteClassicLink>
<allowDnsResolutionFromRemoteVpc>{{ vpc_pcx.accepter_options.AllowDnsResolutionFromRemoteVpc or '' }}</allowDnsResolutionFromRemoteVpc>
</peeringOptions>
</accepterVpcInfo>
<status>
<code>initiating-request</code> <code>initiating-request</code>
<message>Initiating Request to {accepter ID}</message> <message>Initiating Request to {{ vpc_pcx.peer_vpc.owner_id }}</message>
</status> </status>
<expirationTime>2014-02-18T14:37:25.000Z</expirationTime> <expirationTime>2014-02-18T14:37:25.000Z</expirationTime>
<tagSet> <tagSet>
{% for tag in vpc_pcx.get_tags() %} {% for tag in vpc_pcx.get_tags() %}
<item> <item>
<key>{{ tag.key }}</key> <key>{{ tag.key }}</key>
<value>{{ tag.value }}</value> <value>{{ tag.value }}</value>
</item> </item>
{% endfor %} {% endfor %}
</tagSet> </tagSet>
</vpcPeeringConnection> </vpcPeeringConnection>
</CreateVpcPeeringConnectionResponse> </CreateVpcPeeringConnectionResponse>
""" """
@ -113,10 +122,12 @@ DESCRIBE_VPC_PEERING_CONNECTIONS_RESPONSE = """
<item> <item>
<vpcPeeringConnectionId>{{ vpc_pcx.id }}</vpcPeeringConnectionId> <vpcPeeringConnectionId>{{ vpc_pcx.id }}</vpcPeeringConnectionId>
<requesterVpcInfo> <requesterVpcInfo>
<ownerId>{{ account_id }}</ownerId> <ownerId>{{ vpc_pcx.vpc.owner_id }}</ownerId>
<region>{{ vpc_pcx.vpc.region }}</region>
<vpcId>{{ vpc_pcx.vpc.id }}</vpcId> <vpcId>{{ vpc_pcx.vpc.id }}</vpcId>
<cidrBlock>{{ vpc_pcx.vpc.cidr_block }}</cidrBlock> <cidrBlock>{{ vpc_pcx.vpc.cidr_block }}</cidrBlock>
<region>{{ vpc_pcx.vpc.ec2_backend.region_name }}</region> <cidrBlockSet></cidrBlockSet>
<ipv6CidrBlockSet></ipv6CidrBlockSet>
<peeringOptions> <peeringOptions>
<allowEgressFromLocalClassicLinkToRemoteVpc>{{ vpc_pcx.requester_options.AllowEgressFromLocalClassicLinkToRemoteVpc or '' }}</allowEgressFromLocalClassicLinkToRemoteVpc> <allowEgressFromLocalClassicLinkToRemoteVpc>{{ vpc_pcx.requester_options.AllowEgressFromLocalClassicLinkToRemoteVpc or '' }}</allowEgressFromLocalClassicLinkToRemoteVpc>
<allowEgressFromLocalVpcToRemoteClassicLink>{{ vpc_pcx.requester_options.AllowEgressFromLocalVpcToRemoteClassicLink or '' }}</allowEgressFromLocalVpcToRemoteClassicLink> <allowEgressFromLocalVpcToRemoteClassicLink>{{ vpc_pcx.requester_options.AllowEgressFromLocalVpcToRemoteClassicLink or '' }}</allowEgressFromLocalVpcToRemoteClassicLink>
@ -124,10 +135,12 @@ DESCRIBE_VPC_PEERING_CONNECTIONS_RESPONSE = """
</peeringOptions> </peeringOptions>
</requesterVpcInfo> </requesterVpcInfo>
<accepterVpcInfo> <accepterVpcInfo>
<ownerId>{{ account_id }}</ownerId> <ownerId>{{ vpc_pcx.peer_vpc.owner_id }}</ownerId>
<region>{{ vpc_pcx.peer_vpc.region }}</region>
<vpcId>{{ vpc_pcx.peer_vpc.id }}</vpcId> <vpcId>{{ vpc_pcx.peer_vpc.id }}</vpcId>
<cidrBlock>{{ vpc_pcx.peer_vpc.cidr_block }}</cidrBlock> <cidrBlock>{{ vpc_pcx.peer_vpc.cidr_block }}</cidrBlock>
<region>{{ vpc_pcx.peer_vpc.ec2_backend.region_name }}</region> <cidrBlockSet></cidrBlockSet>
<ipv6CidrBlockSet></ipv6CidrBlockSet>
<peeringOptions> <peeringOptions>
<allowEgressFromLocalClassicLinkToRemoteVpc>{{ vpc_pcx.accepter_options.AllowEgressFromLocalClassicLinkToRemoteVpc or '' }}</allowEgressFromLocalClassicLinkToRemoteVpc> <allowEgressFromLocalClassicLinkToRemoteVpc>{{ vpc_pcx.accepter_options.AllowEgressFromLocalClassicLinkToRemoteVpc or '' }}</allowEgressFromLocalClassicLinkToRemoteVpc>
<allowEgressFromLocalVpcToRemoteClassicLink>{{ vpc_pcx.accepter_options.AllowEgressFromLocalVpcToRemoteClassicLink or '' }}</allowEgressFromLocalVpcToRemoteClassicLink> <allowEgressFromLocalVpcToRemoteClassicLink>{{ vpc_pcx.accepter_options.AllowEgressFromLocalVpcToRemoteClassicLink or '' }}</allowEgressFromLocalVpcToRemoteClassicLink>
@ -165,21 +178,30 @@ ACCEPT_VPC_PEERING_CONNECTION_RESPONSE = """
<vpcPeeringConnection> <vpcPeeringConnection>
<vpcPeeringConnectionId>{{ vpc_pcx.id }}</vpcPeeringConnectionId> <vpcPeeringConnectionId>{{ vpc_pcx.id }}</vpcPeeringConnectionId>
<requesterVpcInfo> <requesterVpcInfo>
<ownerId>{{ account_id }}</ownerId> <ownerId>{{ vpc_pcx.vpc.owner_id }}</ownerId>
<vpcId>{{ vpc_pcx.vpc.id }}</vpcId> <region>{{ vpc_pcx.vpc.region }}</region>
<cidrBlock>{{ vpc_pcx.vpc.cidr_block }}</cidrBlock> <vpcId>{{ vpc_pcx.vpc.id }}</vpcId>
<region>{{ vpc_pcx.vpc.ec2_backend.region_name }}</region> <cidrBlock>{{ vpc_pcx.vpc.cidr_block }}</cidrBlock>
<cidrBlockSet></cidrBlockSet>
<ipv6CidrBlockSet></ipv6CidrBlockSet>
<peeringOptions>
<allowEgressFromLocalClassicLinkToRemoteVpc>{{ vpc_pcx.requester_options.AllowEgressFromLocalClassicLinkToRemoteVpc or '' }}</allowEgressFromLocalClassicLinkToRemoteVpc>
<allowEgressFromLocalVpcToRemoteClassicLink>{{ vpc_pcx.requester_options.AllowEgressFromLocalVpcToRemoteClassicLink or '' }}</allowEgressFromLocalVpcToRemoteClassicLink>
<allowDnsResolutionFromRemoteVpc>{{ vpc_pcx.requester_options.AllowDnsResolutionFromRemoteVpc or '' }}</allowDnsResolutionFromRemoteVpc>
</peeringOptions>
</requesterVpcInfo> </requesterVpcInfo>
<accepterVpcInfo> <accepterVpcInfo>
<ownerId>{{ account_id }}</ownerId> <ownerId>{{ vpc_pcx.peer_vpc.owner_id }}</ownerId>
<region>{{ vpc_pcx.peer_vpc.region }}</region>
<vpcId>{{ vpc_pcx.peer_vpc.id }}</vpcId> <vpcId>{{ vpc_pcx.peer_vpc.id }}</vpcId>
<cidrBlock>{{ vpc_pcx.peer_vpc.cidr_block }}</cidrBlock> <cidrBlock>{{ vpc_pcx.peer_vpc.cidr_block }}</cidrBlock>
<cidrBlockSet></cidrBlockSet>
<ipv6CidrBlockSet></ipv6CidrBlockSet>
<peeringOptions> <peeringOptions>
<allowEgressFromLocalClassicLinkToRemoteVpc>{{ vpc_pcx.accepter_options.AllowEgressFromLocalClassicLinkToRemoteVpc or '' }}</allowEgressFromLocalClassicLinkToRemoteVpc> <allowEgressFromLocalClassicLinkToRemoteVpc>{{ vpc_pcx.accepter_options.AllowEgressFromLocalClassicLinkToRemoteVpc or '' }}</allowEgressFromLocalClassicLinkToRemoteVpc>
<allowEgressFromLocalVpcToRemoteClassicLink>{{ vpc_pcx.accepter_options.AllowEgressFromLocalVpcToRemoteClassicLink or '' }}</allowEgressFromLocalVpcToRemoteClassicLink> <allowEgressFromLocalVpcToRemoteClassicLink>{{ vpc_pcx.accepter_options.AllowEgressFromLocalVpcToRemoteClassicLink or '' }}</allowEgressFromLocalVpcToRemoteClassicLink>
<allowDnsResolutionFromRemoteVpc>{{ vpc_pcx.accepter_options.AllowDnsResolutionFromRemoteVpc or '' }}</allowDnsResolutionFromRemoteVpc> <allowDnsResolutionFromRemoteVpc>{{ vpc_pcx.accepter_options.AllowDnsResolutionFromRemoteVpc or '' }}</allowDnsResolutionFromRemoteVpc>
</peeringOptions> </peeringOptions>
<region>{{ vpc_pcx.peer_vpc.ec2_backend.region_name }}</region>
</accepterVpcInfo> </accepterVpcInfo>
<status> <status>
<code>{{ vpc_pcx._status.code }}</code> <code>{{ vpc_pcx._status.code }}</code>

686
tests/test_ec2/test_vpc_peering.py
View File

@ -1,9 +1,12 @@
import os
from unittest import mock
import boto3 import boto3
import pytest import pytest
from botocore.exceptions import ClientError from botocore.exceptions import ClientError
from moto import mock_ec2 from moto import mock_ec2, settings
def create_vpx_pcx(ec2, client): def create_vpx_pcx(ec2, client):
@ -40,6 +43,7 @@ def test_vpc_peering_connections_get_all_boto3():
if vpc_pcx["VpcPeeringConnectionId"] == vpc_pcx_id if vpc_pcx["VpcPeeringConnectionId"] == vpc_pcx_id
][0] ][0]
assert my_vpc_pcx["Status"]["Code"] == "pending-acceptance" assert my_vpc_pcx["Status"]["Code"] == "pending-acceptance"
assert my_vpc_pcx["Status"]["Message"] == "Pending Acceptance by 123456789012"
@mock_ec2 @mock_ec2
@ -115,24 +119,84 @@ def test_vpc_peering_connections_delete_boto3():
@mock_ec2 @mock_ec2
def test_vpc_peering_connections_cross_region(): @pytest.mark.parametrize(
"account1,account2",
[
pytest.param("111111111111", "111111111111", id="within account"),
pytest.param("111111111111", "222222222222", id="across accounts"),
],
)
@pytest.mark.skipif(
settings.TEST_SERVER_MODE, reason="Cannot set account ID in server mode"
)
def test_vpc_peering_connections_cross_region(account1, account2):
# create vpc in us-west-1 and ap-northeast-1 # create vpc in us-west-1 and ap-northeast-1
ec2_usw1 = boto3.resource("ec2", region_name="us-west-1") with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16") ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1") vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
# create peering # create peering
vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection( with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1" ec2_usw1 = boto3.client("ec2", region_name="us-west-1")
vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
VpcId=vpc_usw1.id,
PeerVpcId=vpc_apn1.id,
PeerRegion="ap-northeast-1",
PeerOwnerId=account2,
)
assert (
vpc_pcx_usw1["VpcPeeringConnection"]["Status"]["Code"] == "initiating-request"
) )
assert vpc_pcx_usw1.status["Code"] == "initiating-request" assert (
assert vpc_pcx_usw1.requester_vpc.id == vpc_usw1.id vpc_pcx_usw1["VpcPeeringConnection"]["Status"]["Message"]
assert vpc_pcx_usw1.accepter_vpc.id == vpc_apn1.id == f"Initiating Request to {account2}"
)
assert (
vpc_pcx_usw1["VpcPeeringConnection"]["RequesterVpcInfo"]["VpcId"] == vpc_usw1.id
)
assert (
vpc_pcx_usw1["VpcPeeringConnection"]["RequesterVpcInfo"]["CidrBlock"]
== "10.90.0.0/16"
)
assert (
vpc_pcx_usw1["VpcPeeringConnection"]["RequesterVpcInfo"]["OwnerId"] == account1
)
assert (
vpc_pcx_usw1["VpcPeeringConnection"]["RequesterVpcInfo"]["Region"]
== "us-west-1"
)
assert (
vpc_pcx_usw1["VpcPeeringConnection"]["AccepterVpcInfo"]["VpcId"] == vpc_apn1.id
)
assert (
vpc_pcx_usw1["VpcPeeringConnection"]["AccepterVpcInfo"]["CidrBlock"]
== "10.20.0.0/16"
)
assert (
vpc_pcx_usw1["VpcPeeringConnection"]["AccepterVpcInfo"]["OwnerId"] == account2
)
assert (
vpc_pcx_usw1["VpcPeeringConnection"]["AccepterVpcInfo"]["Region"]
== "ap-northeast-1"
)
# test cross region vpc peering connection exist # test cross region vpc peering connection exist
vpc_pcx_apn1 = ec2_apn1.VpcPeeringConnection(vpc_pcx_usw1.id) with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
assert vpc_pcx_apn1.id == vpc_pcx_usw1.id vpc_pcx_apn1 = ec2_apn1.VpcPeeringConnection(
assert vpc_pcx_apn1.requester_vpc.id == vpc_usw1.id vpc_pcx_usw1["VpcPeeringConnection"]["VpcPeeringConnectionId"]
assert vpc_pcx_apn1.accepter_vpc.id == vpc_apn1.id )
assert (
vpc_pcx_apn1.id
== vpc_pcx_usw1["VpcPeeringConnection"]["VpcPeeringConnectionId"]
)
assert vpc_pcx_apn1.requester_vpc.id == vpc_usw1.id
assert vpc_pcx_apn1.accepter_vpc.id == vpc_apn1.id
# Quick check to verify the options have a default value # Quick check to verify the options have a default value
accepter_options = vpc_pcx_apn1.accepter_vpc_info["PeeringOptions"] accepter_options = vpc_pcx_apn1.accepter_vpc_info["PeeringOptions"]
assert accepter_options["AllowDnsResolutionFromRemoteVpc"] is False assert accepter_options["AllowDnsResolutionFromRemoteVpc"] is False
@ -145,28 +209,50 @@ def test_vpc_peering_connections_cross_region():
@mock_ec2 @mock_ec2
def test_modify_vpc_peering_connections_accepter_only(): @pytest.mark.parametrize(
"account1,account2",
[
pytest.param("111111111111", "111111111111", id="within account"),
pytest.param("111111111111", "222222222222", id="across accounts"),
],
)
@pytest.mark.skipif(
settings.TEST_SERVER_MODE, reason="Cannot set account ID in server mode"
)
def test_modify_vpc_peering_connections_accepter_only(account1, account2):
# create vpc in us-west-1 and ap-northeast-1 # create vpc in us-west-1 and ap-northeast-1
ec2_usw1 = boto3.resource("ec2", region_name="us-west-1") with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
client = boto3.client("ec2", region_name="us-west-1") ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16") client = boto3.client("ec2", region_name="us-west-1")
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1") vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
# create peering # create peering
vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection( with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1" vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
) VpcId=vpc_usw1.id,
# PeerVpcId=vpc_apn1.id,
client.modify_vpc_peering_connection_options( PeerRegion="ap-northeast-1",
VpcPeeringConnectionId=vpc_pcx_usw1.id, PeerOwnerId=account2,
AccepterPeeringConnectionOptions={"AllowDnsResolutionFromRemoteVpc": True}, )
)
# Accepter options are different # modify peering connection options
vpc_pcx_usw1.reload() client.modify_vpc_peering_connection_options(
VpcPeeringConnectionId=vpc_pcx_usw1.id,
AccepterPeeringConnectionOptions={"AllowDnsResolutionFromRemoteVpc": True},
)
# Accepter options are different
vpc_pcx_usw1.reload()
accepter_options = vpc_pcx_usw1.accepter_vpc_info["PeeringOptions"] accepter_options = vpc_pcx_usw1.accepter_vpc_info["PeeringOptions"]
assert accepter_options["AllowDnsResolutionFromRemoteVpc"] is True assert accepter_options["AllowDnsResolutionFromRemoteVpc"] is True
assert accepter_options["AllowEgressFromLocalClassicLinkToRemoteVpc"] is False assert accepter_options["AllowEgressFromLocalClassicLinkToRemoteVpc"] is False
assert accepter_options["AllowEgressFromLocalVpcToRemoteClassicLink"] is False assert accepter_options["AllowEgressFromLocalVpcToRemoteClassicLink"] is False
# Requester options are untouched # Requester options are untouched
requester_options = vpc_pcx_usw1.requester_vpc_info["PeeringOptions"] requester_options = vpc_pcx_usw1.requester_vpc_info["PeeringOptions"]
assert requester_options["AllowDnsResolutionFromRemoteVpc"] is False assert requester_options["AllowDnsResolutionFromRemoteVpc"] is False
@ -175,30 +261,50 @@ def test_modify_vpc_peering_connections_accepter_only():
@mock_ec2 @mock_ec2
def test_modify_vpc_peering_connections_requester_only(): @pytest.mark.parametrize(
"account1,account2",
[
pytest.param("111111111111", "111111111111", id="within account"),
pytest.param("111111111111", "222222222222", id="across accounts"),
],
)
@pytest.mark.skipif(
settings.TEST_SERVER_MODE, reason="Cannot set account ID in server mode"
)
def test_modify_vpc_peering_connections_requester_only(account1, account2):
# create vpc in us-west-1 and ap-northeast-1 # create vpc in us-west-1 and ap-northeast-1
ec2_usw1 = boto3.resource("ec2", region_name="us-west-1") with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
client = boto3.client("ec2", region_name="us-west-1") ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16") client = boto3.client("ec2", region_name="us-west-1")
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1") vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
# create peering # create peering
vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection( with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1" vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
) VpcId=vpc_usw1.id,
# PeerVpcId=vpc_apn1.id,
client.modify_vpc_peering_connection_options( PeerRegion="ap-northeast-1",
VpcPeeringConnectionId=vpc_pcx_usw1.id, PeerOwnerId=account2,
RequesterPeeringConnectionOptions={ )
"AllowEgressFromLocalVpcToRemoteClassicLink": True, #
}, client.modify_vpc_peering_connection_options(
) VpcPeeringConnectionId=vpc_pcx_usw1.id,
# Requester options are different RequesterPeeringConnectionOptions={
vpc_pcx_usw1.reload() "AllowEgressFromLocalVpcToRemoteClassicLink": True,
},
)
# Requester options are different
vpc_pcx_usw1.reload()
requester_options = vpc_pcx_usw1.requester_vpc_info["PeeringOptions"] requester_options = vpc_pcx_usw1.requester_vpc_info["PeeringOptions"]
assert requester_options["AllowDnsResolutionFromRemoteVpc"] is False assert requester_options["AllowDnsResolutionFromRemoteVpc"] is False
assert requester_options["AllowEgressFromLocalClassicLinkToRemoteVpc"] is False assert requester_options["AllowEgressFromLocalClassicLinkToRemoteVpc"] is False
assert requester_options["AllowEgressFromLocalVpcToRemoteClassicLink"] is True assert requester_options["AllowEgressFromLocalVpcToRemoteClassicLink"] is True
# Accepter options are untouched # Accepter options are untouched
accepter_options = vpc_pcx_usw1.accepter_vpc_info["PeeringOptions"] accepter_options = vpc_pcx_usw1.accepter_vpc_info["PeeringOptions"]
assert accepter_options["AllowDnsResolutionFromRemoteVpc"] is False assert accepter_options["AllowDnsResolutionFromRemoteVpc"] is False
@ -207,97 +313,180 @@ def test_modify_vpc_peering_connections_requester_only():
@mock_ec2 @mock_ec2
def test_modify_vpc_peering_connections_unknown_vpc(): @pytest.mark.parametrize(
"account1,account2",
[
pytest.param("111111111111", "111111111111", id="within account"),
pytest.param("111111111111", "222222222222", id="across accounts"),
],
)
@pytest.mark.skipif(
settings.TEST_SERVER_MODE, reason="Cannot set account ID in server mode"
)
def test_modify_vpc_peering_connections_unknown_vpc(account1, account2):
# create vpc in us-west-1 and ap-northeast-1 # create vpc in us-west-1 and ap-northeast-1
ec2_usw1 = boto3.resource("ec2", region_name="us-west-1") with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
client = boto3.client("ec2", region_name="us-west-1") ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16") client = boto3.client("ec2", region_name="us-west-1")
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1") vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
# create peering # create peering
ec2_usw1.create_vpc_peering_connection( with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1" ec2_usw1.create_vpc_peering_connection(
) VpcId=vpc_usw1.id,
# PeerVpcId=vpc_apn1.id,
with pytest.raises(ClientError) as ex: PeerRegion="ap-northeast-1",
client.modify_vpc_peering_connection_options( PeerOwnerId=account2,
VpcPeeringConnectionId="vpx-unknown", RequesterPeeringConnectionOptions={}
) )
with pytest.raises(ClientError) as ex:
client.modify_vpc_peering_connection_options(
VpcPeeringConnectionId="vpx-unknown",
RequesterPeeringConnectionOptions={},
)
err = ex.value.response["Error"] err = ex.value.response["Error"]
assert err["Code"] == "InvalidVpcPeeringConnectionId.NotFound" assert err["Code"] == "InvalidVpcPeeringConnectionId.NotFound"
assert err["Message"] == "VpcPeeringConnectionID vpx-unknown does not exist." assert err["Message"] == "VpcPeeringConnectionID vpx-unknown does not exist."
@mock_ec2 @mock_ec2
def test_vpc_peering_connections_cross_region_fail(): @pytest.mark.parametrize(
"account1,account2",
[
pytest.param("111111111111", "111111111111", id="within account"),
pytest.param("111111111111", "222222222222", id="across accounts"),
],
)
@pytest.mark.skipif(
settings.TEST_SERVER_MODE, reason="Cannot set account ID in server mode"
)
def test_vpc_peering_connections_cross_region_fail(account1, account2):
# create vpc in us-west-1 and ap-northeast-1 # create vpc in us-west-1 and ap-northeast-1
ec2_usw1 = boto3.resource("ec2", region_name="us-west-1") with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16") ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1") vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
# create peering wrong region with no vpc # create peering wrong region with no vpc
with pytest.raises(ClientError) as cm: with pytest.raises(ClientError) as cm:
ec2_usw1.create_vpc_peering_connection( with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-2" ec2_usw1.create_vpc_peering_connection(
) VpcId=vpc_usw1.id,
PeerVpcId=vpc_apn1.id,
PeerRegion="ap-northeast-2",
PeerOwnerId=account2,
)
assert cm.value.response["Error"]["Code"] == "InvalidVpcID.NotFound" assert cm.value.response["Error"]["Code"] == "InvalidVpcID.NotFound"
@mock_ec2 @mock_ec2
def test_describe_vpc_peering_connections_only_returns_requested_id(): @pytest.mark.parametrize(
"account1,account2",
[
pytest.param("111111111111", "111111111111", id="within account"),
pytest.param("111111111111", "222222222222", id="across accounts"),
],
)
@pytest.mark.skipif(
settings.TEST_SERVER_MODE, reason="Cannot set account ID in server mode"
)
def test_describe_vpc_peering_connections_only_returns_requested_id(account1, account2):
# create vpc in us-west-1 and ap-northeast-1 # create vpc in us-west-1 and ap-northeast-1
ec2_usw1 = boto3.resource("ec2", region_name="us-west-1") with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16") ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1") vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
# create peering # create peering
vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection( with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1" vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
) VpcId=vpc_usw1.id,
vpc_pcx_usw2 = ec2_usw1.create_vpc_peering_connection( PeerVpcId=vpc_apn1.id,
VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1" PeerRegion="ap-northeast-1",
) PeerOwnerId=account2,
# describe peering )
ec2_usw1 = boto3.client("ec2", region_name="us-west-1") vpc_pcx_usw2 = ec2_usw1.create_vpc_peering_connection(
our_vpcx = [vpcx["VpcPeeringConnectionId"] for vpcx in retrieve_all(ec2_usw1)] VpcId=vpc_usw1.id,
assert vpc_pcx_usw1.id in our_vpcx PeerVpcId=vpc_apn1.id,
assert vpc_pcx_usw2.id in our_vpcx PeerRegion="ap-northeast-1",
assert vpc_apn1.id not in our_vpcx PeerOwnerId=account2,
)
both_pcx = ec2_usw1.describe_vpc_peering_connections( # describe peering
VpcPeeringConnectionIds=[vpc_pcx_usw1.id, vpc_pcx_usw2.id] ec2_usw1 = boto3.client("ec2", region_name="us-west-1")
)["VpcPeeringConnections"] our_vpcx = [vpcx["VpcPeeringConnectionId"] for vpcx in retrieve_all(ec2_usw1)]
assert len(both_pcx) == 2
one_pcx = ec2_usw1.describe_vpc_peering_connections( assert vpc_pcx_usw1.id in our_vpcx
VpcPeeringConnectionIds=[vpc_pcx_usw1.id] assert vpc_pcx_usw2.id in our_vpcx
)["VpcPeeringConnections"] assert vpc_apn1.id not in our_vpcx
assert len(one_pcx) == 1
both_pcx = ec2_usw1.describe_vpc_peering_connections(
VpcPeeringConnectionIds=[vpc_pcx_usw1.id, vpc_pcx_usw2.id]
)["VpcPeeringConnections"]
assert len(both_pcx) == 2
one_pcx = ec2_usw1.describe_vpc_peering_connections(
VpcPeeringConnectionIds=[vpc_pcx_usw1.id]
)["VpcPeeringConnections"]
assert len(one_pcx) == 1
@mock_ec2 @mock_ec2
def test_vpc_peering_connections_cross_region_accept(): @pytest.mark.parametrize(
"account1,account2",
[
pytest.param("111111111111", "111111111111", id="within account"),
pytest.param("111111111111", "222222222222", id="across accounts"),
],
)
@pytest.mark.skipif(
settings.TEST_SERVER_MODE, reason="Cannot set account ID in server mode"
)
def test_vpc_peering_connections_cross_region_accept(account1, account2):
# create vpc in us-west-1 and ap-northeast-1 # create vpc in us-west-1 and ap-northeast-1
ec2_usw1 = boto3.resource("ec2", region_name="us-west-1") with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16") ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1") vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
# create peering # create peering
vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection( with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1" vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
) VpcId=vpc_usw1.id,
PeerVpcId=vpc_apn1.id,
PeerRegion="ap-northeast-1",
PeerOwnerId=account2,
)
# accept peering from ap-northeast-1 # accept peering from ap-northeast-1
ec2_apn1 = boto3.client("ec2", region_name="ap-northeast-1") with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
ec2_usw1 = boto3.client("ec2", region_name="us-west-1") ec2_apn1 = boto3.client("ec2", region_name="ap-northeast-1")
acp_pcx_apn1 = ec2_apn1.accept_vpc_peering_connection( acp_pcx_apn1 = ec2_apn1.accept_vpc_peering_connection(
VpcPeeringConnectionId=vpc_pcx_usw1.id VpcPeeringConnectionId=vpc_pcx_usw1.id
) )
des_pcx_apn1 = ec2_usw1.describe_vpc_peering_connections(
VpcPeeringConnectionIds=[vpc_pcx_usw1.id] with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
) ec2_usw1 = boto3.client("ec2", region_name="us-west-1")
des_pcx_usw1 = ec2_usw1.describe_vpc_peering_connections( des_pcx_apn1 = ec2_usw1.describe_vpc_peering_connections(
VpcPeeringConnectionIds=[vpc_pcx_usw1.id] VpcPeeringConnectionIds=[vpc_pcx_usw1.id]
) )
des_pcx_usw1 = ec2_usw1.describe_vpc_peering_connections(
VpcPeeringConnectionIds=[vpc_pcx_usw1.id]
)
assert acp_pcx_apn1["VpcPeeringConnection"]["Status"]["Code"] == "active" assert acp_pcx_apn1["VpcPeeringConnection"]["Status"]["Code"] == "active"
assert ( assert (
acp_pcx_apn1["VpcPeeringConnection"]["AccepterVpcInfo"]["Region"] acp_pcx_apn1["VpcPeeringConnection"]["AccepterVpcInfo"]["Region"]
@ -328,102 +517,231 @@ def test_vpc_peering_connections_cross_region_accept():
@mock_ec2 @mock_ec2
def test_vpc_peering_connections_cross_region_reject(): @pytest.mark.parametrize(
"account1,account2",
[
pytest.param("111111111111", "111111111111", id="within account"),
pytest.param("111111111111", "222222222222", id="across accounts"),
],
)
@pytest.mark.skipif(
settings.TEST_SERVER_MODE, reason="Cannot set account ID in server mode"
)
def test_vpc_peering_connections_cross_region_reject(account1, account2):
# create vpc in us-west-1 and ap-northeast-1 # create vpc in us-west-1 and ap-northeast-1
ec2_usw1 = boto3.resource("ec2", region_name="us-west-1") with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16") ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1") vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
# create peering # create peering
vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection( with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1" vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
) VpcId=vpc_usw1.id,
PeerVpcId=vpc_apn1.id,
PeerRegion="ap-northeast-1",
PeerOwnerId=account2,
)
# reject peering from ap-northeast-1 # reject peering from ap-northeast-1
ec2_apn1 = boto3.client("ec2", region_name="ap-northeast-1") with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
ec2_usw1 = boto3.client("ec2", region_name="us-west-1") ec2_apn1 = boto3.client("ec2", region_name="ap-northeast-1")
rej_pcx_apn1 = ec2_apn1.reject_vpc_peering_connection( rej_pcx_apn1 = ec2_apn1.reject_vpc_peering_connection(
VpcPeeringConnectionId=vpc_pcx_usw1.id VpcPeeringConnectionId=vpc_pcx_usw1.id
) )
des_pcx_apn1 = ec2_usw1.describe_vpc_peering_connections(
VpcPeeringConnectionIds=[vpc_pcx_usw1.id] with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
) ec2_usw1 = boto3.client("ec2", region_name="us-west-1")
des_pcx_usw1 = ec2_usw1.describe_vpc_peering_connections( des_pcx_apn1 = ec2_usw1.describe_vpc_peering_connections(
VpcPeeringConnectionIds=[vpc_pcx_usw1.id] VpcPeeringConnectionIds=[vpc_pcx_usw1.id]
) )
des_pcx_usw1 = ec2_usw1.describe_vpc_peering_connections(
VpcPeeringConnectionIds=[vpc_pcx_usw1.id]
)
assert rej_pcx_apn1["Return"] is True assert rej_pcx_apn1["Return"] is True
assert des_pcx_apn1["VpcPeeringConnections"][0]["Status"]["Code"] == "rejected" assert des_pcx_apn1["VpcPeeringConnections"][0]["Status"]["Code"] == "rejected"
assert des_pcx_usw1["VpcPeeringConnections"][0]["Status"]["Code"] == "rejected" assert des_pcx_usw1["VpcPeeringConnections"][0]["Status"]["Code"] == "rejected"
@mock_ec2 @mock_ec2
def test_vpc_peering_connections_cross_region_delete(): @pytest.mark.parametrize(
"account1,account2",
[
pytest.param("111111111111", "111111111111", id="within account"),
pytest.param("111111111111", "222222222222", id="across accounts"),
],
)
@pytest.mark.skipif(
settings.TEST_SERVER_MODE, reason="Cannot set account ID in server mode"
)
def test_vpc_peering_connections_cross_region_delete(account1, account2):
# create vpc in us-west-1 and ap-northeast-1 # create vpc in us-west-1 and ap-northeast-1
ec2_usw1 = boto3.resource("ec2", region_name="us-west-1") with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16") ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1") vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
# create peering # create peering
vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection( with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1" vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
) VpcId=vpc_usw1.id,
# reject peering from ap-northeast-1 PeerVpcId=vpc_apn1.id,
ec2_apn1 = boto3.client("ec2", region_name="ap-northeast-1") PeerRegion="ap-northeast-1",
ec2_usw1 = boto3.client("ec2", region_name="us-west-1") PeerOwnerId=account2,
del_pcx_apn1 = ec2_apn1.delete_vpc_peering_connection( )
VpcPeeringConnectionId=vpc_pcx_usw1.id
) with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
des_pcx_apn1 = ec2_usw1.describe_vpc_peering_connections( # reject peering from ap-northeast-1
VpcPeeringConnectionIds=[vpc_pcx_usw1.id] ec2_apn1 = boto3.client("ec2", region_name="ap-northeast-1")
) del_pcx_apn1 = ec2_apn1.delete_vpc_peering_connection(
des_pcx_usw1 = ec2_usw1.describe_vpc_peering_connections( VpcPeeringConnectionId=vpc_pcx_usw1.id
VpcPeeringConnectionIds=[vpc_pcx_usw1.id] )
)
with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
ec2_usw1 = boto3.client("ec2", region_name="us-west-1")
des_pcx_apn1 = ec2_usw1.describe_vpc_peering_connections(
VpcPeeringConnectionIds=[vpc_pcx_usw1.id]
)
des_pcx_usw1 = ec2_usw1.describe_vpc_peering_connections(
VpcPeeringConnectionIds=[vpc_pcx_usw1.id]
)
assert del_pcx_apn1["Return"] is True assert del_pcx_apn1["Return"] is True
assert des_pcx_apn1["VpcPeeringConnections"][0]["Status"]["Code"] == "deleted" assert des_pcx_apn1["VpcPeeringConnections"][0]["Status"]["Code"] == "deleted"
assert (
des_pcx_apn1["VpcPeeringConnections"][0]["Status"]["Message"]
== f"Deleted by {account2}"
)
assert des_pcx_usw1["VpcPeeringConnections"][0]["Status"]["Code"] == "deleted" assert des_pcx_usw1["VpcPeeringConnections"][0]["Status"]["Code"] == "deleted"
assert (
des_pcx_usw1["VpcPeeringConnections"][0]["Status"]["Message"]
== f"Deleted by {account2}"
)
@mock_ec2 @mock_ec2
def test_vpc_peering_connections_cross_region_accept_wrong_region(): @pytest.mark.parametrize(
"account1,account2",
[
pytest.param("111111111111", "111111111111", id="within account"),
pytest.param("111111111111", "222222222222", id="across accounts"),
],
)
@pytest.mark.skipif(
settings.TEST_SERVER_MODE, reason="Cannot set account ID in server mode"
)
def test_vpc_peering_connections_cross_region_accept_wrong_region(account1, account2):
# create vpc in us-west-1 and ap-northeast-1 # create vpc in us-west-1 and ap-northeast-1
ec2_usw1 = boto3.resource("ec2", region_name="us-west-1") with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16") ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1") vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
# create peering # create peering
vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection( with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1" vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
) VpcId=vpc_usw1.id,
PeerVpcId=vpc_apn1.id,
PeerRegion="ap-northeast-1",
PeerOwnerId=account2,
)
# accept wrong peering from us-west-1 which will raise error # accept wrong peering from us-west-1 which will raise error
ec2_apn1 = boto3.client("ec2", region_name="ap-northeast-1") # only applicable for cross-region intra-account peering.
ec2_usw1 = boto3.client("ec2", region_name="us-west-1") if account1 == account2:
with pytest.raises(ClientError) as cm: with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
ec2_usw1.accept_vpc_peering_connection(VpcPeeringConnectionId=vpc_pcx_usw1.id) ec2_usw1 = boto3.client("ec2", region_name="us-west-1")
assert cm.value.response["Error"]["Code"] == "OperationNotPermitted" with pytest.raises(ClientError) as cm:
exp_msg = f"Incorrect region (us-west-1) specified for this request.VPC peering connection {vpc_pcx_usw1.id} must be accepted in region ap-northeast-1" ec2_usw1.accept_vpc_peering_connection(
assert cm.value.response["Error"]["Message"] == exp_msg VpcPeeringConnectionId=vpc_pcx_usw1.id
)
assert cm.value.response["Error"]["Code"] == "OperationNotPermitted"
exp_msg = f"Incorrect region (us-west-1) specified for this request. VPC peering connection {vpc_pcx_usw1.id} must be accepted in region ap-northeast-1"
assert cm.value.response["Error"]["Message"] == exp_msg
# Ensure accepting peering from requester account raises
# only applicable for cross-region inter-account peering.
if account1 != account2:
with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
ec2_usw1 = boto3.client("ec2", region_name="us-west-1")
with pytest.raises(ClientError) as cm:
ec2_usw1.accept_vpc_peering_connection(
VpcPeeringConnectionId=vpc_pcx_usw1.id
)
assert cm.value.response["Error"]["Code"] == "OperationNotPermitted"
exp_msg = f"User ({account1}) cannot accept peering {vpc_pcx_usw1.id}"
assert cm.value.response["Error"]["Message"] == exp_msg
@mock_ec2 @mock_ec2
def test_vpc_peering_connections_cross_region_reject_wrong_region(): @pytest.mark.parametrize(
"account1,account2",
[
pytest.param("111111111111", "111111111111", id="within account"),
pytest.param("111111111111", "222222222222", id="across accounts"),
],
)
@pytest.mark.skipif(
settings.TEST_SERVER_MODE, reason="Cannot set account ID in server mode"
)
def test_vpc_peering_connections_cross_region_reject_wrong_region(account1, account2):
# create vpc in us-west-1 and ap-northeast-1 # create vpc in us-west-1 and ap-northeast-1
ec2_usw1 = boto3.resource("ec2", region_name="us-west-1") with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16") ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1") vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
# create peering # create peering
vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection( with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1" vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
) VpcId=vpc_usw1.id,
# reject wrong peering from us-west-1 which will raise error PeerVpcId=vpc_apn1.id,
ec2_apn1 = boto3.client("ec2", region_name="ap-northeast-1") PeerRegion="ap-northeast-1",
ec2_usw1 = boto3.client("ec2", region_name="us-west-1") PeerOwnerId=account2,
with pytest.raises(ClientError) as cm: )
ec2_usw1.reject_vpc_peering_connection(VpcPeeringConnectionId=vpc_pcx_usw1.id)
assert cm.value.response["Error"]["Code"] == "OperationNotPermitted" # reject wrong peering from us-west-1 which will raise error.
exp_msg = f"Incorrect region (us-west-1) specified for this request.VPC peering connection {vpc_pcx_usw1.id} must be accepted or rejected in region ap-northeast-1" # only applicable for cross-region intra-account peering.
assert cm.value.response["Error"]["Message"] == exp_msg if account1 == account2:
with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
ec2_usw1 = boto3.client("ec2", region_name="us-west-1")
with pytest.raises(ClientError) as cm:
ec2_usw1.reject_vpc_peering_connection(
VpcPeeringConnectionId=vpc_pcx_usw1.id
)
assert cm.value.response["Error"]["Code"] == "OperationNotPermitted"
exp_msg = f"Incorrect region (us-west-1) specified for this request. VPC peering connection {vpc_pcx_usw1.id} must be accepted or rejected in region ap-northeast-1"
assert cm.value.response["Error"]["Message"] == exp_msg
# Ensure rejecting peering from requester account raises
# only applicable for cross-region inter-account peering.
if account1 != account2:
with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
ec2_usw1 = boto3.client("ec2", region_name="us-west-1")
with pytest.raises(ClientError) as cm:
ec2_usw1.reject_vpc_peering_connection(
VpcPeeringConnectionId=vpc_pcx_usw1.id
)
assert cm.value.response["Error"]["Code"] == "OperationNotPermitted"
exp_msg = f"User ({account1}) cannot reject peering {vpc_pcx_usw1.id}"
assert cm.value.response["Error"]["Message"] == exp_msg
def retrieve_all(client): def retrieve_all(client):