diff --git a/moto/ec2/responses/vpc_peering_connections.py b/moto/ec2/responses/vpc_peering_connections.py
index 1c0b6b983e8b..cae1f7df3569 100644
--- a/moto/ec2/responses/vpc_peering_connections.py
+++ b/moto/ec2/responses/vpc_peering_connections.py
@@ -202,7 +202,6 @@ def modify_vpc_peering_connection_options(self) -> str:
{{ vpc_pcx.accepter_options.AllowEgressFromLocalVpcToRemoteClassicLink or '' }}
{{ vpc_pcx.accepter_options.AllowDnsResolutionFromRemoteVpc or '' }}
- {{ vpc_pcx.peer_vpc.ec2_backend.region_name }}
{{ vpc_pcx._status.code }}
diff --git a/tests/test_ec2/test_vpc_peering.py b/tests/test_ec2/test_vpc_peering.py
index 36da5588917d..e0914a0ee3f6 100644
--- a/tests/test_ec2/test_vpc_peering.py
+++ b/tests/test_ec2/test_vpc_peering.py
@@ -1,3 +1,6 @@
+import os
+from unittest import mock
+
import boto3
import pytest
@@ -115,24 +118,45 @@ def test_vpc_peering_connections_delete_boto3():
@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"),
+ ],
+)
+def test_vpc_peering_connections_cross_region(account1, account2):
# create vpc in us-west-1 and ap-northeast-1
- ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
- vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16")
- ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
- vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+ ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
+ vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.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
- vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
- VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1"
- )
- assert vpc_pcx_usw1.status["Code"] == "initiating-request"
- assert vpc_pcx_usw1.requester_vpc.id == vpc_usw1.id
- assert vpc_pcx_usw1.accepter_vpc.id == vpc_apn1.id
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+ 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["VpcPeeringConnection"]["RequesterVpcInfo"]["VpcId"] == vpc_usw1.id
+ assert vpc_pcx_usw1["VpcPeeringConnection"]["AccepterVpcInfo"]["VpcId"] == vpc_apn1.id
+ # TODO add more assertions
+
# test cross region vpc peering connection exist
- vpc_pcx_apn1 = ec2_apn1.VpcPeeringConnection(vpc_pcx_usw1.id)
- assert vpc_pcx_apn1.id == vpc_pcx_usw1.id
- assert vpc_pcx_apn1.requester_vpc.id == vpc_usw1.id
- assert vpc_pcx_apn1.accepter_vpc.id == vpc_apn1.id
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
+ vpc_pcx_apn1 = ec2_apn1.VpcPeeringConnection(vpc_pcx_usw1["VpcPeeringConnection"]["VpcPeeringConnectionId"])
+ 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
accepter_options = vpc_pcx_apn1.accepter_vpc_info["PeeringOptions"]
assert accepter_options["AllowDnsResolutionFromRemoteVpc"] is False
@@ -145,28 +169,47 @@ def test_vpc_peering_connections_cross_region():
@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"),
+ ],
+)
+def test_modify_vpc_peering_connections_accepter_only(account1, account2):
# create vpc in us-west-1 and ap-northeast-1
- ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
- client = boto3.client("ec2", region_name="us-west-1")
- vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16")
- ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
- vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+ ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
+ client = boto3.client("ec2", region_name="us-west-1")
+ vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.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
- vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
- VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1"
- )
- #
- client.modify_vpc_peering_connection_options(
- VpcPeeringConnectionId=vpc_pcx_usw1.id,
- AccepterPeeringConnectionOptions={"AllowDnsResolutionFromRemoteVpc": True},
- )
- # Accepter options are different
- vpc_pcx_usw1.reload()
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+ vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
+ VpcId=vpc_usw1.id,
+ PeerVpcId=vpc_apn1.id,
+ PeerRegion="ap-northeast-1",
+ PeerOwnerId=account2,
+ )
+
+ # modify peering connection options
+ 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"]
assert accepter_options["AllowDnsResolutionFromRemoteVpc"] is True
assert accepter_options["AllowEgressFromLocalClassicLinkToRemoteVpc"] is False
assert accepter_options["AllowEgressFromLocalVpcToRemoteClassicLink"] is False
+
# Requester options are untouched
requester_options = vpc_pcx_usw1.requester_vpc_info["PeeringOptions"]
assert requester_options["AllowDnsResolutionFromRemoteVpc"] is False
@@ -175,30 +218,47 @@ def test_modify_vpc_peering_connections_accepter_only():
@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"),
+ ],
+)
+def test_modify_vpc_peering_connections_requester_only(account1, account2):
# create vpc in us-west-1 and ap-northeast-1
- ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
- client = boto3.client("ec2", region_name="us-west-1")
- vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16")
- ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
- vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+ ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
+ client = boto3.client("ec2", region_name="us-west-1")
+ vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.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
- vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
- VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1"
- )
- #
- client.modify_vpc_peering_connection_options(
- VpcPeeringConnectionId=vpc_pcx_usw1.id,
- RequesterPeeringConnectionOptions={
- "AllowEgressFromLocalVpcToRemoteClassicLink": True,
- },
- )
- # Requester options are different
- vpc_pcx_usw1.reload()
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+ vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
+ VpcId=vpc_usw1.id,
+ PeerVpcId=vpc_apn1.id,
+ PeerRegion="ap-northeast-1",
+ PeerOwnerId=account2,
+ )
+ #
+ client.modify_vpc_peering_connection_options(
+ VpcPeeringConnectionId=vpc_pcx_usw1.id,
+ RequesterPeeringConnectionOptions={
+ "AllowEgressFromLocalVpcToRemoteClassicLink": True,
+ },
+ )
+ # Requester options are different
+ vpc_pcx_usw1.reload()
+
requester_options = vpc_pcx_usw1.requester_vpc_info["PeeringOptions"]
assert requester_options["AllowDnsResolutionFromRemoteVpc"] is False
assert requester_options["AllowEgressFromLocalClassicLinkToRemoteVpc"] is False
assert requester_options["AllowEgressFromLocalVpcToRemoteClassicLink"] is True
+
# Accepter options are untouched
accepter_options = vpc_pcx_usw1.accepter_vpc_info["PeeringOptions"]
assert accepter_options["AllowDnsResolutionFromRemoteVpc"] is False
@@ -207,97 +267,159 @@ def test_modify_vpc_peering_connections_requester_only():
@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"),
+ ],
+)
+def test_modify_vpc_peering_connections_unknown_vpc(account1, account2):
# create vpc in us-west-1 and ap-northeast-1
- ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
- client = boto3.client("ec2", region_name="us-west-1")
- vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16")
- ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
- vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+ ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
+ client = boto3.client("ec2", region_name="us-west-1")
+ vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.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
- ec2_usw1.create_vpc_peering_connection(
- VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1"
- )
- #
- with pytest.raises(ClientError) as ex:
- client.modify_vpc_peering_connection_options(
- VpcPeeringConnectionId="vpx-unknown", RequesterPeeringConnectionOptions={}
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+ ec2_usw1.create_vpc_peering_connection(
+ VpcId=vpc_usw1.id,
+ PeerVpcId=vpc_apn1.id,
+ PeerRegion="ap-northeast-1",
+ PeerOwnerId=account2,
)
+
+ with pytest.raises(ClientError) as ex:
+ client.modify_vpc_peering_connection_options(
+ VpcPeeringConnectionId="vpx-unknown",
+ RequesterPeeringConnectionOptions={},
+ )
err = ex.value.response["Error"]
assert err["Code"] == "InvalidVpcPeeringConnectionId.NotFound"
assert err["Message"] == "VpcPeeringConnectionID vpx-unknown does not exist."
@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"),
+ ],
+)
+def test_vpc_peering_connections_cross_region_fail(account1, account2):
# create vpc in us-west-1 and ap-northeast-1
- ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
- vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16")
- ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
- vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+ ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
+ vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.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
with pytest.raises(ClientError) as cm:
- ec2_usw1.create_vpc_peering_connection(
- VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-2"
- )
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+ 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"
@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"),
+ ],
+)
+def test_describe_vpc_peering_connections_only_returns_requested_id(account1, account2):
# create vpc in us-west-1 and ap-northeast-1
- ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
- vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16")
- ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
- vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+ ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
+ vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.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
- vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
- VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1"
- )
- vpc_pcx_usw2 = ec2_usw1.create_vpc_peering_connection(
- VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1"
- )
- # describe peering
- ec2_usw1 = boto3.client("ec2", region_name="us-west-1")
- our_vpcx = [vpcx["VpcPeeringConnectionId"] for vpcx in retrieve_all(ec2_usw1)]
- assert vpc_pcx_usw1.id in our_vpcx
- assert vpc_pcx_usw2.id in our_vpcx
- assert vpc_apn1.id not in our_vpcx
-
- both_pcx = ec2_usw1.describe_vpc_peering_connections(
- VpcPeeringConnectionIds=[vpc_pcx_usw1.id, vpc_pcx_usw2.id]
- )["VpcPeeringConnections"]
- assert len(both_pcx) == 2
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+ vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
+ VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1", PeerOwnerId=account2,
+ )
+ vpc_pcx_usw2 = ec2_usw1.create_vpc_peering_connection(
+ VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1", PeerOwnerId=account2,
+ )
- one_pcx = ec2_usw1.describe_vpc_peering_connections(
- VpcPeeringConnectionIds=[vpc_pcx_usw1.id]
- )["VpcPeeringConnections"]
- assert len(one_pcx) == 1
+ # describe peering
+ ec2_usw1 = boto3.client("ec2", region_name="us-west-1")
+ our_vpcx = [vpcx["VpcPeeringConnectionId"] for vpcx in retrieve_all(ec2_usw1)]
+
+ assert vpc_pcx_usw1.id in our_vpcx
+ assert vpc_pcx_usw2.id in our_vpcx
+ assert vpc_apn1.id not in our_vpcx
+
+ 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
-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"),
+ ],
+)
+def test_vpc_peering_connections_cross_region_accept(account1, account2):
# create vpc in us-west-1 and ap-northeast-1
- ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
- vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16")
- ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
- vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+ ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
+ vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.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
- vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
- VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1"
- )
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+ 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
- ec2_apn1 = boto3.client("ec2", region_name="ap-northeast-1")
- ec2_usw1 = boto3.client("ec2", region_name="us-west-1")
- acp_pcx_apn1 = ec2_apn1.accept_vpc_peering_connection(
- VpcPeeringConnectionId=vpc_pcx_usw1.id
- )
- 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]
- )
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
+ ec2_apn1 = boto3.client("ec2", region_name="ap-northeast-1")
+ acp_pcx_apn1 = ec2_apn1.accept_vpc_peering_connection(
+ VpcPeeringConnectionId=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 acp_pcx_apn1["VpcPeeringConnection"]["Status"]["Code"] == "active"
assert (
acp_pcx_apn1["VpcPeeringConnection"]["AccepterVpcInfo"]["Region"]
@@ -328,102 +450,190 @@ def test_vpc_peering_connections_cross_region_accept():
@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"),
+ ],
+)
+def test_vpc_peering_connections_cross_region_reject(account1, account2):
# create vpc in us-west-1 and ap-northeast-1
- ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
- vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16")
- ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
- vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+ ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
+ vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.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
- vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
- VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1"
- )
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+ 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
- ec2_apn1 = boto3.client("ec2", region_name="ap-northeast-1")
- ec2_usw1 = boto3.client("ec2", region_name="us-west-1")
- rej_pcx_apn1 = ec2_apn1.reject_vpc_peering_connection(
- VpcPeeringConnectionId=vpc_pcx_usw1.id
- )
- 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]
- )
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
+ ec2_apn1 = boto3.client("ec2", region_name="ap-northeast-1")
+ rej_pcx_apn1 = ec2_apn1.reject_vpc_peering_connection(
+ VpcPeeringConnectionId=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 rej_pcx_apn1["Return"] is True
assert des_pcx_apn1["VpcPeeringConnections"][0]["Status"]["Code"] == "rejected"
assert des_pcx_usw1["VpcPeeringConnections"][0]["Status"]["Code"] == "rejected"
@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"),
+ ],
+)
+def test_vpc_peering_connections_cross_region_delete(account1, account2):
# create vpc in us-west-1 and ap-northeast-1
- ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
- vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16")
- ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
- vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+ ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
+ vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.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
- vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
- VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1"
- )
- # reject peering from ap-northeast-1
- ec2_apn1 = boto3.client("ec2", region_name="ap-northeast-1")
- ec2_usw1 = boto3.client("ec2", region_name="us-west-1")
- del_pcx_apn1 = ec2_apn1.delete_vpc_peering_connection(
- VpcPeeringConnectionId=vpc_pcx_usw1.id
- )
- 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]
- )
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+ vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
+ VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1", PeerOwnerId=account2
+ )
+
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
+ # reject peering from ap-northeast-1
+ ec2_apn1 = boto3.client("ec2", region_name="ap-northeast-1")
+ del_pcx_apn1 = ec2_apn1.delete_vpc_peering_connection(
+ VpcPeeringConnectionId=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 des_pcx_apn1["VpcPeeringConnections"][0]["Status"]["Code"] == "deleted"
assert des_pcx_usw1["VpcPeeringConnections"][0]["Status"]["Code"] == "deleted"
@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"),
+ ],
+)
+def test_vpc_peering_connections_cross_region_accept_wrong_region(account1, account2):
# create vpc in us-west-1 and ap-northeast-1
- ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
- vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16")
- ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
- vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+ ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
+ vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.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
- vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
- VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1"
- )
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+ 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
- ec2_apn1 = boto3.client("ec2", region_name="ap-northeast-1")
- 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"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
+ # only applicable for cross-region intra-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"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
-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"),
+ ],
+)
+def test_vpc_peering_connections_cross_region_reject_wrong_region(account1, account2):
# create vpc in us-west-1 and ap-northeast-1
- ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
- vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.0.0/16")
- ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
- vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+ ec2_usw1 = boto3.resource("ec2", region_name="us-west-1")
+ vpc_usw1 = ec2_usw1.create_vpc(CidrBlock="10.90.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
- vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
- VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1"
- )
- # reject wrong peering from us-west-1 which will raise error
- ec2_apn1 = boto3.client("ec2", region_name="ap-northeast-1")
- 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
+ with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+ vpc_pcx_usw1 = ec2_usw1.create_vpc_peering_connection(
+ VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-1", PeerOwnerId=account2
+ )
+
+ # reject wrong peering from us-west-1 which will raise error.
+ # only applicable for cross-region intra-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"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):