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

Implement list_hosted_zones_by_vpc() functionality (#4771)

Browse Source
This commit is contained in:
Paul 2022-01-29 06:11:24 -05:00 committed by GitHub
parent 8b39233426
commit e1dbec1dff
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 145 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
docs/docs/services/route53.rst
View File

@ -78,7 +78,7 @@ route53
- [X] list_health_checks
- [X] list_hosted_zones
- [X] list_hosted_zones_by_name
- [ ] list_hosted_zones_by_vpc
- [X] list_hosted_zones_by_vpc
- [X] list_query_logging_configs
Return a list of query logging configs.

20
moto/route53/models.py
View File

@ -15,7 +15,7 @@ from moto.route53.exceptions import (
NoSuchQueryLoggingConfig,
QueryLoggingConfigAlreadyExists,
)
from moto.core import BaseBackend, BaseModel, CloudFormationModel
from moto.core import BaseBackend, BaseModel, CloudFormationModel, ACCOUNT_ID
from moto.utilities.paginator import paginate
from .utils import PAGINATION_MODEL
@ -474,6 +474,24 @@ class Route53Backend(BaseBackend):
zones = sorted(zones, key=sort_key)
return dnsname, zones
def list_hosted_zones_by_vpc(self, VPCId, VPCRegion, MaxItems=None, NextToken=None):
zone_list = []
for zone in self.list_hosted_zones():
if zone.private_zone == "true":
this_zone = self.get_hosted_zone(zone.id)
if this_zone.vpcid == VPCId:
this_id = f"/hostedzone/{zone.id}"
zone_list.append(
{
"HostedZoneId": this_id,
"Name": zone.name,
"Owner": {"OwningAccount": ACCOUNT_ID},
}
)
return zone_list
def get_hosted_zone(self, id_):
the_zone = self.zones.get(id_.replace("/hostedzone/", ""))
if not the_zone:

41
moto/route53/responses.py
View File

@ -6,7 +6,7 @@ from jinja2 import Template
import xmltodict
from moto.core.responses import BaseResponse
from moto.route53.exceptions import Route53ClientError, InvalidChangeBatch, InvalidVPCId
from moto.route53.exceptions import Route53ClientError, InvalidChangeBatch
from moto.route53.models import route53_backend
XMLNS = "https://route53.amazonaws.com/doc/2013-04-01/"
@ -45,12 +45,12 @@ class Route53(BaseResponse):
comment = None
private_zone = False
# It is possible to create a Private Hosted Zone without
# associating VPC at the time of creation.
if private_zone == "true":
try:
vpcid = zone_request["VPC"]["VPCId"]
vpcregion = zone_request["VPC"]["VPCRegion"]
except KeyError:
raise InvalidVPCId()
if zone_request.get("VPC", None) is not None:
vpcid = zone_request["VPC"].get("VPCId", None)
vpcregion = zone_request["VPC"].get("VPCRegion", None)
name = zone_request["Name"]
@ -83,6 +83,16 @@ class Route53(BaseResponse):
template = Template(LIST_HOSTED_ZONES_BY_NAME_RESPONSE)
return 200, headers, template.render(zones=zones, dnsname=dnsname, xmlns=XMLNS)
def list_hosted_zones_by_vpc_response(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
parsed_url = urlparse(full_url)
query_params = parse_qs(parsed_url.query)
vpc_id = query_params.get("vpcid")[0]
vpc_region = query_params.get("vpcregion")[0]
zones = route53_backend.list_hosted_zones_by_vpc(vpc_id, vpc_region)
template = Template(LIST_HOSTED_ZONES_BY_VPC_RESPONSE)
return 200, headers, template.render(zones=zones, xmlns=XMLNS)
@error_handler
def get_or_delete_hostzone_response(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
@ -494,6 +504,25 @@ LIST_HOSTED_ZONES_BY_NAME_RESPONSE = """<ListHostedZonesByNameResponse xmlns="{{
<IsTruncated>false</IsTruncated>
</ListHostedZonesByNameResponse>"""
LIST_HOSTED_ZONES_BY_VPC_RESPONSE = """<ListHostedZonesByVpcResponse xmlns="{{xmlns}}">
<HostedZoneSummaries>
{% for zone in zones -%}
<HostedZoneSummary>
<HostedZoneId>{{zone["HostedZoneId"]}}</HostedZoneId>
<Name>{{zone["Name"]}}</Name>
<Owner>
{% if zone["Owner"]["OwningAccount"] -%}
<OwningAccount>{{zone["Owner"]["OwningAccount"]}}</OwningAccount>
{% endif -%}
{% if zone["Owner"]["OwningService"] -%}
<OwningService>zone["Owner"]["OwningService"]</OwningService>
{% endif -%}
</Owner>
</HostedZoneSummary>
{% endfor -%}
</HostedZoneSummaries>
</ListHostedZonesByVpcResponse>"""
CREATE_HEALTH_CHECK_RESPONSE = """<?xml version="1.0" encoding="UTF-8"?>
<CreateHealthCheckResponse xmlns="{{ xmlns }}">
{{ health_check.to_xml() }}

1
moto/route53/urls.py
View File

@ -17,6 +17,7 @@ url_paths = {
r"{0}/(?P<api_version>[\d_-]+)/hostedzone/(?P<zone_id>[^/]+)$": Route53().get_or_delete_hostzone_response,
r"{0}/(?P<api_version>[\d_-]+)/hostedzone/(?P<zone_id>[^/]+)/rrset/?$": Route53().rrset_response,
r"{0}/(?P<api_version>[\d_-]+)/hostedzonesbyname": Route53().list_hosted_zones_by_name_response,
r"{0}/(?P<api_version>[\d_-]+)/hostedzonesbyvpc": Route53().list_hosted_zones_by_vpc_response,
r"{0}/(?P<api_version>[\d_-]+)/healthcheck": Route53().health_check_response,
r"{0}/(?P<api_version>[\d_-]+)/healthcheck/(?P<health_check_id>[^/]+)$": Route53().health_check_response,
r"{0}/(?P<api_version>[\d_-]+)/tags/healthcheck/(?P<zone_id>[^/]+)$": tag_response1,

101
tests/test_route53/test_route53.py
View File

@ -400,21 +400,37 @@ def test_hosted_zone_private_zone_preserved_boto3():
hosted_zones["HostedZones"][0]["Config"]["PrivateZone"].should.equal(True)
hosted_zones = conn.list_hosted_zones_by_name(DNSName="testdns.aws.com.")
len(hosted_zones["HostedZones"]).should.equal(1)
hosted_zones["HostedZones"].should.have.length_of(1)
hosted_zones["HostedZones"][0]["Config"]["PrivateZone"].should.equal(True)
# create_hosted_zone statements with PrivateZone=True,
# but without a _valid_ vpc-id should fail.
conn = boto3.client("route53", region_name=region)
with pytest.raises(ClientError) as exc:
conn.create_hosted_zone(
Name="testdns.aws.com.",
CallerReference=str(hash("foo")),
HostedZoneConfig=dict(PrivateZone=True, Comment="Test"),
)
err = exc.value.response["Error"]
err["Code"].should.equal("InvalidVPCId")
err["Message"].should.equal("Invalid or missing VPC Id.")
# but without a _valid_ vpc-id should NOT fail.
zone2_name = "testdns2.aws.com."
no_vpc_zone = conn.create_hosted_zone(
Name=zone2_name,
CallerReference=str(hash("foo")),
HostedZoneConfig=dict(PrivateZone=True, Comment="Test without VPC"),
)
zone_id = no_vpc_zone["HostedZone"]["Id"].split("/")[-1]
hosted_zone = conn.get_hosted_zone(Id=zone_id)
hosted_zone["HostedZone"]["Config"]["PrivateZone"].should.equal(True)
hosted_zone.should.have.key("VPCs")
hosted_zone["VPCs"].should.have.length_of(1)
hosted_zone["VPCs"][0].should.have.key("VPCId")
hosted_zone["VPCs"][0].should.have.key("VPCRegion")
hosted_zone["VPCs"][0]["VPCId"].should.be.empty
hosted_zone["VPCs"][0]["VPCRegion"].should.be.empty
hosted_zones = conn.list_hosted_zones()
hosted_zones["HostedZones"].should.have.length_of(2)
hosted_zones["HostedZones"][0]["Config"]["PrivateZone"].should.equal(True)
hosted_zones["HostedZones"][1]["Config"]["PrivateZone"].should.equal(True)
hosted_zones = conn.list_hosted_zones_by_name(DNSName=zone2_name)
hosted_zones["HostedZones"].should.have.length_of(1)
hosted_zones["HostedZones"][0]["Config"]["PrivateZone"].should.equal(True)
hosted_zones["HostedZones"][0]["Name"].should.equal(zone2_name)
return
@ -635,6 +651,67 @@ def test_list_hosted_zones_by_dns_name():
zones["HostedZones"][3]["Name"].should.equal("test.a.org.")
@mock_ec2
@mock_route53
def test_list_hosted_zones_by_vpc():
# Create mock VPC so we can get a VPC ID
ec2c = boto3.client("ec2", region_name="us-east-1")
vpc_id = ec2c.create_vpc(CidrBlock="10.1.0.0/16").get("Vpc").get("VpcId")
region = "us-east-1"
conn = boto3.client("route53", region_name=region)
zone_b = conn.create_hosted_zone(
Name="test.b.com.",
CallerReference=str(hash("foo")),
HostedZoneConfig=dict(PrivateZone=True, Comment="test com"),
VPC={"VPCRegion": region, "VPCId": vpc_id},
)
response = conn.list_hosted_zones_by_vpc(VPCId=vpc_id, VPCRegion=region)
response.should.have.key("ResponseMetadata")
response.should.have.key("HostedZoneSummaries")
response["HostedZoneSummaries"].should.have.length_of(1)
response["HostedZoneSummaries"][0].should.have.key("HostedZoneId")
retured_zone = response["HostedZoneSummaries"][0]
retured_zone["HostedZoneId"].should.equal(zone_b["HostedZone"]["Id"])
retured_zone["Name"].should.equal(zone_b["HostedZone"]["Name"])
@mock_ec2
@mock_route53
def test_list_hosted_zones_by_vpc_with_multiple_vpcs():
# Create mock VPC so we can get a VPC ID
ec2c = boto3.client("ec2", region_name="us-east-1")
vpc_id = ec2c.create_vpc(CidrBlock="10.1.0.0/16").get("Vpc").get("VpcId")
region = "us-east-1"
# Create 3 Zones associate with the VPC.
zones = {}
conn = boto3.client("route53", region_name=region)
for zone in ["a", "b", "c"]:
zone_name = f"test.{zone}.com."
zones[zone] = conn.create_hosted_zone(
Name=zone_name,
CallerReference=str(hash("foo")),
HostedZoneConfig=dict(PrivateZone=True, Comment=f"test {zone} com"),
VPC={"VPCRegion": region, "VPCId": vpc_id},
)
# List the zones associated with this vpc
response = conn.list_hosted_zones_by_vpc(VPCId=vpc_id, VPCRegion=region)
response.should.have.key("ResponseMetadata")
response.should.have.key("HostedZoneSummaries")
response["HostedZoneSummaries"].should.have.length_of(3)
# Loop through all zone summaries and verify they match what was created
for summary in response["HostedZoneSummaries"]:
# use the zone name as the index
index = summary["Name"].split(".")[1]
summary.should.have.key("HostedZoneId")
summary["HostedZoneId"].should.equal(zones[index]["HostedZone"]["Id"])
summary.should.have.key("Name")
summary["Name"].should.equal(zones[index]["HostedZone"]["Name"])
@mock_route53
def test_change_resource_record_sets_crud_valid():
conn = boto3.client("route53", region_name="us-east-1")