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

EC2:DescribeVpcPeeringConnection() - use VpcPeeringConnectionIds-parameter if provided (#4230)

Browse Source
This commit is contained in:
Bert Blommers 2021-08-27 14:56:31 +01:00 committed by GitHub
parent 020257904e
commit 8b7df4da44
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 35 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
moto/ec2/models.py
View File

@ -3563,7 +3563,9 @@ class VPCPeeringConnectionBackend(object):
vpc_pcx_cx.vpc_pcxs[vpc_pcx_id] = vpc_pcx vpc_pcx_cx.vpc_pcxs[vpc_pcx_id] = vpc_pcx
return vpc_pcx return vpc_pcx
def get_all_vpc_peering_connections(self): def get_all_vpc_peering_connections(self, vpc_peering_ids=None):
if vpc_peering_ids:
return [pcx for pcx in self.vpc_pcxs.values() if pcx.id in vpc_peering_ids]
return self.vpc_pcxs.values() return self.vpc_pcxs.values()
def get_vpc_peering_connection(self, vpc_pcx_id): def get_vpc_peering_connection(self, vpc_pcx_id):

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

@ -29,7 +29,8 @@ class VPCPeeringConnections(BaseResponse):
return template.render(vpc_pcx=vpc_pcx) return template.render(vpc_pcx=vpc_pcx)
def describe_vpc_peering_connections(self): def describe_vpc_peering_connections(self):
vpc_pcxs = self.ec2_backend.get_all_vpc_peering_connections() ids = self._get_multi_param("VpcPeeringConnectionId")
vpc_pcxs = self.ec2_backend.get_all_vpc_peering_connections(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(vpc_pcxs=vpc_pcxs) return template.render(vpc_pcxs=vpc_pcxs)

30
tests/test_ec2/test_vpc_peering.py
View File

@ -227,6 +227,36 @@ def test_vpc_peering_connections_cross_region_fail():
cm.value.response["Error"]["Code"].should.equal("InvalidVpcID.NotFound") cm.value.response["Error"]["Code"].should.equal("InvalidVpcID.NotFound")
@mock_ec2
def test_describe_vpc_peering_connections_only_returns_requested_id():
# 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")
# 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")
all_pcx = ec2_usw1.describe_vpc_peering_connections()["VpcPeeringConnections"]
all_pcx.should.have.length_of(2)
both_pcx = ec2_usw1.describe_vpc_peering_connections(
VpcPeeringConnectionIds=[vpc_pcx_usw1.id, vpc_pcx_usw2.id]
)["VpcPeeringConnections"]
both_pcx.should.have.length_of(2)
one_pcx = ec2_usw1.describe_vpc_peering_connections(
VpcPeeringConnectionIds=[vpc_pcx_usw1.id]
)["VpcPeeringConnections"]
one_pcx.should.have.length_of(1)
@mock_ec2 @mock_ec2
def test_vpc_peering_connections_cross_region_accept(): def test_vpc_peering_connections_cross_region_accept():
# create vpc in us-west-1 and ap-northeast-1 # create vpc in us-west-1 and ap-northeast-1