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

Route53 - Update impl cov to show change_resource_record_sets (#4401)

Browse Source
This commit is contained in:
Bert Blommers 2021-10-12 22:04:43 +00:00 committed by GitHub
parent 5f6ca27bb8
commit 36e66d32b9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 78 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
IMPLEMENTATION_COVERAGE.md
View File

@ -3473,11 +3473,11 @@
## route53
<details>
<summary>10% implemented</summary>
<summary>17% implemented</summary>
- [ ] activate_key_signing_key
- [ ] associate_vpc_with_hosted_zone
- [ ] change_resource_record_sets
- [X] change_resource_record_sets
- [X] change_tags_for_resource
- [X] create_health_check
- [X] create_hosted_zone
@ -3519,9 +3519,9 @@
- [ ] get_traffic_policy_instance
- [ ] get_traffic_policy_instance_count
- [ ] list_geo_locations
- [ ] list_health_checks
- [ ] list_hosted_zones
- [ ] list_hosted_zones_by_name
- [X] list_health_checks
- [X] list_hosted_zones
- [X] list_hosted_zones_by_name
- [ ] list_hosted_zones_by_vpc
- [ ] list_query_logging_configs
- [ ] list_resource_record_sets

68
moto/route53/models.py
View File

@ -407,14 +407,76 @@ class Route53Backend(BaseBackend):
return self.resource_tags[resource_id]
return {}
def get_all_hosted_zones(self):
def change_resource_record_sets(self, the_zone, change_list):
for value in change_list:
action = value["Action"]
record_set = value["ResourceRecordSet"]
cleaned_record_name = record_set["Name"].strip(".")
cleaned_hosted_zone_name = the_zone.name.strip(".")
if not cleaned_record_name.endswith(cleaned_hosted_zone_name):
error_msg = """
An error occurred (InvalidChangeBatch) when calling the ChangeResourceRecordSets operation:
RRSet with DNS name %s is not permitted in zone %s
""" % (
record_set["Name"],
the_zone.name,
)
return error_msg
if not record_set["Name"].endswith("."):
record_set["Name"] += "."
if action in ("CREATE", "UPSERT"):
if "ResourceRecords" in record_set:
resource_records = list(record_set["ResourceRecords"].values())[0]
if not isinstance(resource_records, list):
# Depending on how many records there are, this may
# or may not be a list
resource_records = [resource_records]
record_set["ResourceRecords"] = [
x["Value"] for x in resource_records
]
if action == "CREATE":
the_zone.add_rrset(record_set)
else:
the_zone.upsert_rrset(record_set)
elif action == "DELETE":
if "SetIdentifier" in record_set:
the_zone.delete_rrset_by_id(record_set["SetIdentifier"])
else:
the_zone.delete_rrset(record_set)
return None
def list_hosted_zones(self):
return self.zones.values()
def list_hosted_zones_by_name(self, dnsname):
if dnsname:
dnsname = dnsname[0]
if dnsname[-1] != ".":
dnsname += "."
zones = [zone for zone in self.list_hosted_zones() if zone.name == dnsname]
else:
# sort by names, but with domain components reversed
# see http://boto3.readthedocs.io/en/latest/reference/services/route53.html#Route53.Client.list_hosted_zones_by_name
def sort_key(zone):
domains = zone.name.split(".")
if domains[-1] == "":
domains = domains[-1:] + domains[:-1]
return ".".join(reversed(domains))
zones = self.list_hosted_zones()
zones = sorted(zones, key=sort_key)
return dnsname, zones
def get_hosted_zone(self, id_):
return self.zones.get(id_.replace("/hostedzone/", ""))
def get_hosted_zone_by_name(self, name):
for zone in self.get_all_hosted_zones():
for zone in self.list_hosted_zones():
if zone.name == name:
return zone
@ -427,7 +489,7 @@ class Route53Backend(BaseBackend):
self.health_checks[health_check_id] = health_check
return health_check
def get_health_checks(self):
def list_health_checks(self):
return self.health_checks.values()
def delete_health_check(self, health_check_id):

72
moto/route53/responses.py
View File

@ -44,7 +44,7 @@ class Route53(BaseResponse):
return 201, headers, template.render(zone=new_zone)
elif request.method == "GET":
all_zones = route53_backend.get_all_hosted_zones()
all_zones = route53_backend.list_hosted_zones()
template = Template(LIST_HOSTED_ZONES_RESPONSE)
return 200, headers, template.render(zones=all_zones)
@ -54,27 +54,7 @@ class Route53(BaseResponse):
query_params = parse_qs(parsed_url.query)
dnsname = query_params.get("dnsname")
if dnsname:
dnsname = dnsname[0]
if dnsname[-1] != ".":
dnsname += "."
zones = [
zone
for zone in route53_backend.get_all_hosted_zones()
if zone.name == dnsname
]
else:
# sort by names, but with domain components reversed
# see http://boto3.readthedocs.io/en/latest/reference/services/route53.html#Route53.Client.list_hosted_zones_by_name
def sort_key(zone):
domains = zone.name.split(".")
if domains[-1] == "":
domains = domains[-1:] + domains[:-1]
return ".".join(reversed(domains))
zones = route53_backend.get_all_hosted_zones()
zones = sorted(zones, key=sort_key)
dnsname, zones = route53_backend.list_hosted_zones_by_name(dnsname)
template = Template(LIST_HOSTED_ZONES_BY_NAME_RESPONSE)
return 200, headers, template.render(zones=zones, dnsname=dnsname)
@ -119,47 +99,11 @@ class Route53(BaseResponse):
]["Change"]
]
for value in change_list:
action = value["Action"]
record_set = value["ResourceRecordSet"]
cleaned_record_name = record_set["Name"].strip(".")
cleaned_hosted_zone_name = the_zone.name.strip(".")
if not cleaned_record_name.endswith(cleaned_hosted_zone_name):
error_msg = """
An error occurred (InvalidChangeBatch) when calling the ChangeResourceRecordSets operation:
RRSet with DNS name %s is not permitted in zone %s
""" % (
record_set["Name"],
the_zone.name,
)
return 400, headers, error_msg
if not record_set["Name"].endswith("."):
record_set["Name"] += "."
if action in ("CREATE", "UPSERT"):
if "ResourceRecords" in record_set:
resource_records = list(record_set["ResourceRecords"].values())[
0
]
if not isinstance(resource_records, list):
# Depending on how many records there are, this may
# or may not be a list
resource_records = [resource_records]
record_set["ResourceRecords"] = [
x["Value"] for x in resource_records
]
if action == "CREATE":
the_zone.add_rrset(record_set)
else:
the_zone.upsert_rrset(record_set)
elif action == "DELETE":
if "SetIdentifier" in record_set:
the_zone.delete_rrset_by_id(record_set["SetIdentifier"])
else:
the_zone.delete_rrset(record_set)
error_msg = route53_backend.change_resource_record_sets(
the_zone, change_list
)
if error_msg:
return 400, headers, error_msg
return 200, headers, CHANGE_RRSET_RESPONSE
@ -212,7 +156,7 @@ class Route53(BaseResponse):
return 200, headers, DELETE_HEALTH_CHECK_RESPONSE
elif method == "GET":
template = Template(LIST_HEALTH_CHECKS_RESPONSE)
health_checks = route53_backend.get_health_checks()
health_checks = route53_backend.list_health_checks()
return 200, headers, template.render(health_checks=health_checks)
def not_implemented_response(self, request, full_url, headers):