Fix incorrect DnsEntries XML for describe_vpc_endpoints (#4621)

This commit is contained in:
Philip D'Amore 2021-11-23 04:55:47 -05:00 committed by GitHub
parent 8b0a6f3d27
commit 68ffb303b5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 63 additions and 7 deletions

View File

@ -691,13 +691,16 @@ DESCRIBE_VPC_ENDPOINT_RESPONSE = """<DescribeVpcEndpointsResponse xmlns="http://
{% endfor %}
</networkInterfaceIdSet>
{% endif %}
{% if vpc_end_point.dns_entries %}
<dnsEntries>
{% for dns_entry in vpc_end_point.dns_entries %}
<item>{{ dns_entry }}</item>
{% endfor %}
</dnsEntries>
<dnsEntrySet>
{% if vpc_end_point.dns_entries %}
{% for entry in vpc_end_point.dns_entries %}
<item>
<hostedZoneId>{{ entry["hosted_zone_id"] }}</hostedZoneId>
<dnsName>{{ entry["dns_name"] }}</dnsName>
</item>
{% endfor %}
{% endif %}
</dnsEntrySet>
{% if vpc_end_point.security_group_ids %}
<groupSet>
{% for group_id in vpc_end_point.security_group_ids %}

View File

@ -1200,7 +1200,7 @@ def test_describe_classic_link_dns_support_multiple():
@mock_ec2
def test_describe_vpc_end_points():
def test_describe_vpc_gateway_end_points():
ec2 = boto3.client("ec2", region_name="us-west-1")
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]
@ -1242,6 +1242,59 @@ def test_describe_vpc_end_points():
err["Code"].should.equal("InvalidVpcEndpointId.NotFound")
@mock_ec2
def test_describe_vpc_interface_end_points():
ec2 = boto3.client("ec2", region_name="us-west-1")
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]
subnet = ec2.create_subnet(VpcId=vpc["VpcId"], CidrBlock="10.0.1.0/24")["Subnet"]
route_table = ec2.create_route_table(VpcId=vpc["VpcId"])["RouteTable"]
vpc_end_point = ec2.create_vpc_endpoint(
VpcId=vpc["VpcId"],
ServiceName="com.tester.my-test-endpoint",
VpcEndpointType="interface",
SubnetIds=[subnet["SubnetId"]],
)["VpcEndpoint"]
our_id = vpc_end_point["VpcEndpointId"]
vpc_end_point["DnsEntries"].should.have.length_of(1)
vpc_end_point["DnsEntries"][0].should.have.key("DnsName").should.match(
r".*com\.tester\.my-test-endpoint$"
)
vpc_end_point["DnsEntries"][0].should.have.key("HostedZoneId")
all_endpoints = retrieve_all_endpoints(ec2)
[e["VpcEndpointId"] for e in all_endpoints].should.contain(our_id)
our_endpoint = [e for e in all_endpoints if e["VpcEndpointId"] == our_id][0]
vpc_end_point["PrivateDnsEnabled"].should.be.true
our_endpoint["PrivateDnsEnabled"].should.be.true
our_endpoint["VpcId"].should.equal(vpc["VpcId"])
our_endpoint.should_not.have.key("RouteTableIds")
our_endpoint["DnsEntries"].should.equal(vpc_end_point["DnsEntries"])
our_endpoint.should.have.key("VpcEndpointType").equal("interface")
our_endpoint.should.have.key("ServiceName").equal("com.tester.my-test-endpoint")
our_endpoint.should.have.key("State").equal("available")
endpoint_by_id = ec2.describe_vpc_endpoints(VpcEndpointIds=[our_id])[
"VpcEndpoints"
][0]
endpoint_by_id["VpcEndpointId"].should.equal(our_id)
endpoint_by_id["VpcId"].should.equal(vpc["VpcId"])
endpoint_by_id.should_not.have.key("RouteTableIds")
endpoint_by_id["VpcEndpointType"].should.equal("interface")
endpoint_by_id["ServiceName"].should.equal("com.tester.my-test-endpoint")
endpoint_by_id["State"].should.equal("available")
endpoint_by_id["DnsEntries"].should.equal(vpc_end_point["DnsEntries"])
with pytest.raises(ClientError) as ex:
ec2.describe_vpc_endpoints(VpcEndpointIds=[route_table["RouteTableId"]])
err = ex.value.response["Error"]
err["Code"].should.equal("InvalidVpcEndpointId.NotFound")
def retrieve_all_endpoints(ec2):
resp = ec2.describe_vpc_endpoints()
all_endpoints = resp["VpcEndpoints"]