Route53 query-logging-config APIs (#4437)

IMPLEMENTATION_COVERAGE.md

@@ -3506,7 +3506,7 @@
 - [X] create_health_check
 - [X] create_hosted_zone
 - [ ] create_key_signing_key
-- [ ] create_query_logging_config
+- [X] create_query_logging_config
 - [ ] create_reusable_delegation_set
 - [ ] create_traffic_policy
 - [ ] create_traffic_policy_instance
@@ -3516,7 +3516,7 @@
 - [X] delete_health_check
 - [X] delete_hosted_zone
 - [ ] delete_key_signing_key
-- [ ] delete_query_logging_config
+- [X] delete_query_logging_config
 - [ ] delete_reusable_delegation_set
 - [ ] delete_traffic_policy
 - [ ] delete_traffic_policy_instance
@@ -3536,7 +3536,7 @@
 - [X] get_hosted_zone
 - [ ] get_hosted_zone_count
 - [ ] get_hosted_zone_limit
-- [ ] get_query_logging_config
+- [X] get_query_logging_config
 - [ ] get_reusable_delegation_set
 - [ ] get_reusable_delegation_set_limit
 - [ ] get_traffic_policy
@@ -3547,7 +3547,7 @@
 - [X] list_hosted_zones
 - [X] list_hosted_zones_by_name
 - [ ] list_hosted_zones_by_vpc
-- [ ] list_query_logging_configs
+- [X] list_query_logging_configs
 - [ ] list_resource_record_sets
 - [ ] list_reusable_delegation_sets
 - [X] list_tags_for_resource

moto/logs/models.py

@@ -652,7 +652,9 @@ class LogsBackend(BaseBackend):
         del self.groups[log_group_name]
 
     @paginate(pagination_model=PAGINATION_MODEL)
-    def describe_log_groups(self, log_group_name_prefix, limit=None, next_token=None):
+    def describe_log_groups(
+        self, log_group_name_prefix=None, limit=None, next_token=None
+    ):
         if log_group_name_prefix is None:
             log_group_name_prefix = ""
 

moto/route53/exceptions.py

@@ -0,0 +1,73 @@
+"""Exceptions raised by the Route53 service."""
+from moto.core.exceptions import RESTError
+
+
+class Route53ClientError(RESTError):
+    """Base class for Route53 errors."""
+
+    def __init__(self, *args, **kwargs):
+        kwargs.setdefault("template", "single_error")
+        super().__init__(*args, **kwargs)
+
+
+class InvalidInput(Route53ClientError):
+    """Malformed ARN for the CloudWatch log group."""
+
+    code = 400
+
+    def __init__(self):
+        message = "The ARN for the CloudWatch Logs log group is invalid"
+        super().__init__("InvalidInput", message)
+
+
+class InvalidPaginationToken(Route53ClientError):
+    """Bad NextToken specified when listing query logging configs."""
+
+    code = 400
+
+    def __init__(self):
+        message = (
+            "Route 53 can't get the next page of query logging configurations "
+            "because the specified value for NextToken is invalid."
+        )
+        super().__init__("InvalidPaginationToken", message)
+
+
+class NoSuchCloudWatchLogsLogGroup(Route53ClientError):
+    """CloudWatch LogGroup has a permissions policy, but does not exist."""
+
+    code = 404
+
+    def __init__(self):
+        message = "The specified CloudWatch Logs log group doesn't exist."
+        super().__init__("NoSuchCloudWatchLogsLogGroup", message)
+
+
+class NoSuchHostedZone(Route53ClientError):
+    """HostedZone does not exist."""
+
+    code = 404
+
+    def __init__(self, host_zone_id):
+        message = f"No hosted zone found with ID: {host_zone_id}"
+        super().__init__("NoSuchHostedZone", message)
+
+
+class NoSuchQueryLoggingConfig(Route53ClientError):
+    """Query log config does not exist."""
+
+    code = 404
+
+    def __init__(self):
+        message = "The query logging configuration does not exist"
+        super().__init__("NoSuchQueryLoggingConfig", message)
+
+
+class QueryLoggingConfigAlreadyExists(Route53ClientError):
+    """Query log config exists for the hosted zone."""
+
+    code = 409
+
+    def __init__(self):
+        message = "A query logging configuration already exists for this hosted zone"
+        super().__init__("QueryLoggingConfigAlreadyExists", message)

moto/route53/models.py

@@ -1,13 +1,23 @@
+"""Route53Backend class with methods for supported APIs."""
 import itertools
 from collections import defaultdict
+import re
 
 import string
 import random
 import uuid
 from jinja2 import Template
 
-from moto.core import BaseBackend, CloudFormationModel
+from moto.route53.exceptions import (
-
+    InvalidInput,
+    NoSuchCloudWatchLogsLogGroup,
+    NoSuchHostedZone,
+    NoSuchQueryLoggingConfig,
+    QueryLoggingConfigAlreadyExists,
+)
+from moto.core import BaseBackend, BaseModel, CloudFormationModel
+from moto.utilities.paginator import paginate
+from .utils import PAGINATION_MODEL
 
 ROUTE53_ID_CHOICE = string.ascii_uppercase + string.digits
 
@@ -342,7 +352,7 @@ class RecordSetGroup(CloudFormationModel):
 
     @property
     def physical_resource_id(self):
-        return "arn:aws:route53:::hostedzone/{0}".format(self.hosted_zone_id)
+        return f"arn:aws:route53:::hostedzone/{self.hosted_zone_id}"
 
     @staticmethod
     def cloudformation_name_type():
@@ -372,11 +382,37 @@ class RecordSetGroup(CloudFormationModel):
         return record_set_group
 
 
+class QueryLoggingConfig(BaseModel):
+
+    """QueryLoggingConfig class; this object isn't part of Cloudformation."""
+
+    def __init__(
+        self, query_logging_config_id, hosted_zone_id, cloudwatch_logs_log_group_arn
+    ):
+        self.hosted_zone_id = hosted_zone_id
+        self.cloudwatch_logs_log_group_arn = cloudwatch_logs_log_group_arn
+        self.query_logging_config_id = query_logging_config_id
+        self.location = f"https://route53.amazonaws.com/2013-04-01/queryloggingconfig/{self.query_logging_config_id}"
+
+    def to_xml(self):
+        template = Template(
+            """<QueryLoggingConfig>
+                <CloudWatchLogsLogGroupArn>{{ query_logging_config.cloudwatch_logs_log_group_arn }}</CloudWatchLogsLogGroupArn>
+                <HostedZoneId>{{ query_logging_config.hosted_zone_id }}</HostedZoneId>
+                <Id>{{ query_logging_config.query_logging_config_id }}</Id>
+            </QueryLoggingConfig>"""
+        )
+        # The "Location" value must be put into the header; that's done in
+        # responses.py.
+        return template.render(query_logging_config=self)
+
+
 class Route53Backend(BaseBackend):
     def __init__(self):
         self.zones = {}
         self.health_checks = {}
         self.resource_tags = defaultdict(dict)
+        self.query_logging_configs = {}
 
     def create_hosted_zone(self, name, private_zone, comment=None):
         new_id = create_route53_zone_id()
@@ -414,13 +450,10 @@ class Route53Backend(BaseBackend):
             cleaned_hosted_zone_name = the_zone.name.strip(".")
 
             if not cleaned_record_name.endswith(cleaned_hosted_zone_name):
-                error_msg = """
+                error_msg = f"""
                 An error occurred (InvalidChangeBatch) when calling the ChangeResourceRecordSets operation:
-                RRSet with DNS name %s is not permitted in zone %s
+                RRSet with DNS name {record_set["Name"]} is not permitted in zone {the_zone.name}
-                """ % (
+                """
-                    record_set["Name"],
-                    the_zone.name,
-                )
                 return error_msg
 
             if not record_set["Name"].endswith("."):
@@ -477,6 +510,7 @@ class Route53Backend(BaseBackend):
         for zone in self.list_hosted_zones():
             if zone.name == name:
                 return zone
+        return None
 
     def delete_hosted_zone(self, id_):
         return self.zones.pop(id_.replace("/hostedzone/", ""), None)
@@ -493,5 +527,92 @@ class Route53Backend(BaseBackend):
     def delete_health_check(self, health_check_id):
         return self.health_checks.pop(health_check_id, None)
 
+    @staticmethod
+    def _validate_arn(region, arn):
+        match = re.match(fr"arn:aws:logs:{region}:\d{{12}}:log-group:.+", arn)
+        if not arn or not match:
+            raise InvalidInput()
+
+        # The CloudWatch Logs log group must be in the "us-east-1" region.
+        match = re.match(r"^(?:[^:]+:){3}(?P<region>[^:]+).*", arn)
+        if match.group("region") != "us-east-1":
+            raise InvalidInput()
+
+    def create_query_logging_config(self, region, hosted_zone_id, log_group_arn):
+        """Process the create_query_logging_config request."""
+        # Does the hosted_zone_id exist?
+        response = self.list_hosted_zones()
+        zones = list(response) if response else []
+        for zone in zones:
+            if zone.id == hosted_zone_id:
+                break
+        else:
+            raise NoSuchHostedZone(hosted_zone_id)
+
+        # Ensure CloudWatch Logs log ARN is valid, otherwise raise an error.
+        self._validate_arn(region, log_group_arn)
+
+        # Note:  boto3 checks the resource policy permissions before checking
+        # whether the log group exists.  moto doesn't have a way of checking
+        # the resource policy, so in some instances moto will complain
+        # about a log group that doesn't exist whereas boto3 will complain
+        # that "The resource policy that you're using for Route 53 query
+        # logging doesn't grant Route 53 sufficient permission to create
+        # a log stream in the specified log group."
+
+        from moto.logs import logs_backends  # pylint: disable=import-outside-toplevel
+
+        response = logs_backends[region].describe_log_groups()
+        log_groups = response[0] if response else []
+        for entry in log_groups:
+            if log_group_arn == entry["arn"]:
+                break
+        else:
+            # There is no CloudWatch Logs log group with the specified ARN.
+            raise NoSuchCloudWatchLogsLogGroup()
+
+        # Verify there is no existing query log config using the same hosted
+        # zone.
+        for query_log in self.query_logging_configs.values():
+            if query_log.hosted_zone_id == hosted_zone_id:
+                raise QueryLoggingConfigAlreadyExists()
+
+        # Create an instance of the query logging config.
+        query_logging_config_id = str(uuid.uuid4())
+        query_logging_config = QueryLoggingConfig(
+            query_logging_config_id, hosted_zone_id, log_group_arn
+        )
+        self.query_logging_configs[query_logging_config_id] = query_logging_config
+        return query_logging_config
+
+    def delete_query_logging_config(self, query_logging_config_id):
+        """Delete query logging config, if it exists."""
+        if query_logging_config_id not in self.query_logging_configs:
+            raise NoSuchQueryLoggingConfig()
+        self.query_logging_configs.pop(query_logging_config_id)
+
+    def get_query_logging_config(self, query_logging_config_id):
+        """Return query logging config, if it exists."""
+        if query_logging_config_id not in self.query_logging_configs:
+            raise NoSuchQueryLoggingConfig()
+        return self.query_logging_configs[query_logging_config_id]
+
+    @paginate(pagination_model=PAGINATION_MODEL)
+    def list_query_logging_configs(
+        self, hosted_zone_id=None, next_token=None, max_results=None,
+    ):  # pylint: disable=unused-argument
+        """Return a list of query logging configs."""
+        if hosted_zone_id:
+            # Does the hosted_zone_id exist?
+            response = self.list_hosted_zones()
+            zones = list(response) if response else []
+            for zone in zones:
+                if zone.id == hosted_zone_id:
+                    break
+            else:
+                raise NoSuchHostedZone(hosted_zone_id)
+
+        return list(self.query_logging_configs.values())
+
 
 route53_backend = Route53Backend()

moto/route53/responses.py

@@ -1,14 +1,20 @@
-from jinja2 import Template
+"""Handles Route53 API requests, invokes method and returns response."""
 from urllib.parse import parse_qs, urlparse
 
-from moto.core.responses import BaseResponse
+from jinja2 import Template
-from .models import route53_backend
 import xmltodict
 
+from moto.core.responses import BaseResponse
+from moto.core.exceptions import InvalidToken
+from moto.route53.exceptions import Route53ClientError, InvalidPaginationToken
+from moto.route53.models import route53_backend
+
 XMLNS = "https://route53.amazonaws.com/doc/2013-04-01/"
 
 
 class Route53(BaseResponse):
+    """Handler for Route53 requests and responses."""
+
     def list_or_create_hostzone_response(self, request, full_url, headers):
         self.setup_class(request, full_url, headers)
 
@@ -56,7 +62,7 @@ class Route53(BaseResponse):
         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)
+        return 200, headers, template.render(zones=zones, dnsname=dnsname, xmlns=XMLNS)
 
     def get_or_delete_hostzone_response(self, request, full_url, headers):
         self.setup_class(request, full_url, headers)
@@ -148,15 +154,20 @@ class Route53(BaseResponse):
                 caller_reference, health_check_args
             )
             template = Template(CREATE_HEALTH_CHECK_RESPONSE)
-            return 201, headers, template.render(health_check=health_check)
+            return 201, headers, template.render(health_check=health_check, xmlns=XMLNS)
         elif method == "DELETE":
             health_check_id = parsed_url.path.split("/")[-1]
             route53_backend.delete_health_check(health_check_id)
-            return 200, headers, DELETE_HEALTH_CHECK_RESPONSE
+            template = Template(DELETE_HEALTH_CHECK_RESPONSE)
+            return 200, headers, template.render(xmlns=XMLNS)
         elif method == "GET":
             template = Template(LIST_HEALTH_CHECKS_RESPONSE)
             health_checks = route53_backend.list_health_checks()
-            return 200, headers, template.render(health_checks=health_checks)
+            return (
+                200,
+                headers,
+                template.render(health_checks=health_checks, xmlns=XMLNS),
+            )
 
     def not_implemented_response(self, request, full_url, headers):
         self.setup_class(request, full_url, headers)
@@ -167,7 +178,7 @@ class Route53(BaseResponse):
         elif "trafficpolicyinstances" in full_url:
             action = "policies"
         raise NotImplementedError(
-            "The action for {0} has not been implemented for route 53".format(action)
+            f"The action for {action} has not been implemented for route 53"
         )
 
     def list_or_change_tags_for_resource_request(self, request, full_url, headers):
@@ -196,7 +207,6 @@ class Route53(BaseResponse):
 
             route53_backend.change_tags_for_resource(id_, tags)
             template = Template(CHANGE_TAGS_FOR_RESOURCE_RESPONSE)
-
             return 200, headers, template.render()
 
     def get_change(self, request, full_url, headers):
@@ -206,20 +216,100 @@ class Route53(BaseResponse):
             parsed_url = urlparse(full_url)
             change_id = parsed_url.path.rstrip("/").rsplit("/", 1)[1]
             template = Template(GET_CHANGE_RESPONSE)
-            return 200, headers, template.render(change_id=change_id)
+            return 200, headers, template.render(change_id=change_id, xmlns=XMLNS)
+
+    def list_or_create_query_logging_config_response(self, request, full_url, headers):
+        self.setup_class(request, full_url, headers)
+
+        if request.method == "POST":
+            json_body = xmltodict.parse(self.body)["CreateQueryLoggingConfigRequest"]
+            hosted_zone_id = json_body["HostedZoneId"]
+            log_group_arn = json_body["CloudWatchLogsLogGroupArn"]
+            try:
+                query_logging_config = route53_backend.create_query_logging_config(
+                    self.region, hosted_zone_id, log_group_arn
+                )
+            except Route53ClientError as r53error:
+                return r53error.code, {}, r53error.description
+
+            template = Template(CREATE_QUERY_LOGGING_CONFIG_RESPONSE)
+            headers["Location"] = query_logging_config.location
+            return (
+                201,
+                headers,
+                template.render(query_logging_config=query_logging_config, xmlns=XMLNS),
+            )
+
+        elif request.method == "GET":
+            hosted_zone_id = self._get_param("hostedzoneid")
+            next_token = self._get_param("nexttoken")
+            max_results = self._get_int_param("maxresults")
+            try:
+                try:
+                    # The paginator picks up named arguments, returns tuple.
+                    # pylint: disable=unbalanced-tuple-unpacking
+                    (
+                        all_configs,
+                        next_token,
+                    ) = route53_backend.list_query_logging_configs(
+                        hosted_zone_id=hosted_zone_id,
+                        next_token=next_token,
+                        max_results=max_results,
+                    )
+                except InvalidToken as exc:
+                    raise InvalidPaginationToken() from exc
+            except Route53ClientError as r53error:
+                return r53error.code, {}, r53error.description
+
+            template = Template(LIST_QUERY_LOGGING_CONFIGS_RESPONSE)
+            return (
+                200,
+                headers,
+                template.render(
+                    query_logging_configs=all_configs,
+                    next_token=next_token,
+                    xmlns=XMLNS,
+                ),
+            )
+
+    def get_or_delete_query_logging_config_response(self, request, full_url, headers):
+        self.setup_class(request, full_url, headers)
+        parsed_url = urlparse(full_url)
+        query_logging_config_id = parsed_url.path.rstrip("/").rsplit("/", 1)[1]
+
+        if request.method == "GET":
+            try:
+                query_logging_config = route53_backend.get_query_logging_config(
+                    query_logging_config_id
+                )
+            except Route53ClientError as r53error:
+                return r53error.code, {}, r53error.description
+            template = Template(GET_QUERY_LOGGING_CONFIG_RESPONSE)
+            return (
+                200,
+                headers,
+                template.render(query_logging_config=query_logging_config, xmlns=XMLNS),
+            )
+
+        elif request.method == "DELETE":
+            try:
+                route53_backend.delete_query_logging_config(query_logging_config_id)
+            except Route53ClientError as r53error:
+                return r53error.code, {}, r53error.description
+            return 200, headers, ""
 
 
-def no_such_hosted_zone_error(zoneid, headers={}):
+def no_such_hosted_zone_error(zoneid, headers=None):
+    if not headers:
+        headers = {}
     headers["X-Amzn-ErrorType"] = "NoSuchHostedZone"
     headers["Content-Type"] = "text/xml"
-    message = "Zone %s Not Found" % zoneid
+    error_response = f"""<ErrorResponse xmlns="{XMLNS}">
-    error_response = (
+        <Error>
-        "<Error><Code>NoSuchHostedZone</Code><Message>%s</Message></Error>" % message
+            <Code>NoSuchHostedZone</Code>
-    )
+            <Message>Zone {zoneid} Not Found</Message>
-    error_response = '<ErrorResponse xmlns="%s">%s</ErrorResponse>' % (
+        </Error>
-        XMLNS,
+    </ErrorResponse>"""
-        error_response,
-    )
     return 404, headers, error_response
 
 
@@ -323,7 +413,7 @@ LIST_HOSTED_ZONES_RESPONSE = """<ListHostedZonesResponse xmlns="https://route53.
    <IsTruncated>false</IsTruncated>
 </ListHostedZonesResponse>"""
 
-LIST_HOSTED_ZONES_BY_NAME_RESPONSE = """<ListHostedZonesByNameResponse xmlns="https://route53.amazonaws.com/doc/2013-04-01/">
+LIST_HOSTED_ZONES_BY_NAME_RESPONSE = """<ListHostedZonesByNameResponse xmlns="{{ xmlns }}">
   {% if dnsname %}
   <DNSName>{{ dnsname }}</DNSName>
   {% endif %}
@@ -346,12 +436,12 @@ LIST_HOSTED_ZONES_BY_NAME_RESPONSE = """<ListHostedZonesByNameResponse xmlns="ht
 </ListHostedZonesByNameResponse>"""
 
 CREATE_HEALTH_CHECK_RESPONSE = """<?xml version="1.0" encoding="UTF-8"?>
-<CreateHealthCheckResponse xmlns="https://route53.amazonaws.com/doc/2013-04-01/">
+<CreateHealthCheckResponse xmlns="{{ xmlns }}">
   {{ health_check.to_xml() }}
 </CreateHealthCheckResponse>"""
 
 LIST_HEALTH_CHECKS_RESPONSE = """<?xml version="1.0" encoding="UTF-8"?>
-<ListHealthChecksResponse xmlns="https://route53.amazonaws.com/doc/2013-04-01/">
+<ListHealthChecksResponse xmlns="{{ xmlns }}">
    <HealthChecks>
    {% for health_check in health_checks %}
       {{ health_check.to_xml() }}
@@ -362,14 +452,36 @@ LIST_HEALTH_CHECKS_RESPONSE = """<?xml version="1.0" encoding="UTF-8"?>
 </ListHealthChecksResponse>"""
 
 DELETE_HEALTH_CHECK_RESPONSE = """<?xml version="1.0" encoding="UTF-8"?>
-    <DeleteHealthCheckResponse xmlns="https://route53.amazonaws.com/doc/2013-04-01/">
+    <DeleteHealthCheckResponse xmlns="{{ xmlns }}">
 </DeleteHealthCheckResponse>"""
 
 GET_CHANGE_RESPONSE = """<?xml version="1.0" encoding="UTF-8"?>
-<GetChangeResponse xmlns="https://route53.amazonaws.com/doc/2013-04-01/">
+<GetChangeResponse xmlns="{{ xmlns }}">
    <ChangeInfo>
       <Status>INSYNC</Status>
       <SubmittedAt>2010-09-10T01:36:41.958Z</SubmittedAt>
       <Id>{{ change_id }}</Id>
    </ChangeInfo>
 </GetChangeResponse>"""
+
+CREATE_QUERY_LOGGING_CONFIG_RESPONSE = """<?xml version="1.0" encoding="UTF-8"?>
+<CreateQueryLoggingConfigResponse xmlns="{{ xmlns }}">
+  {{ query_logging_config.to_xml() }}
+</CreateQueryLoggingConfigResponse>"""
+
+GET_QUERY_LOGGING_CONFIG_RESPONSE = """<?xml version="1.0" encoding="UTF-8"?>
+<CreateQueryLoggingConfigResponse xmlns="{{ xmlns }}">
+  {{ query_logging_config.to_xml() }}
+</CreateQueryLoggingConfigResponse>"""
+
+LIST_QUERY_LOGGING_CONFIGS_RESPONSE = """<?xml version="1.0" encoding="UTF-8"?>
+<ListQueryLoggingConfigsResponse xmlns="{{ xmlns }}">
+   <QueryLoggingConfigs>
+      {% for query_logging_config in query_logging_configs %}
+         {{ query_logging_config.to_xml() }}
+      {% endfor %}
+   </QueryLoggingConfigs>
+   {% if next_token %}
+      <NextToken>{{ next_token }}</NextToken>
+   {% endif %}
+</ListQueryLoggingConfigsResponse>"""

moto/route53/urls.py

@@ -1,3 +1,4 @@
+"""Route53 base URL and path."""
 from .responses import Route53
 
 url_bases = [r"https?://route53(.*)\.amazonaws.com"]
@@ -22,4 +23,6 @@ url_paths = {
     r"{0}/(?P<api_version>[\d_-]+)/tags/hostedzone/(?P<zone_id>[^/]+)$": tag_response2,
     r"{0}/(?P<api_version>[\d_-]+)/trafficpolicyinstances/*": Route53().not_implemented_response,
     r"{0}/(?P<api_version>[\d_-]+)/change/(?P<change_id>[^/]+)$": Route53().get_change,
+    r"{0}/(?P<api_version>[\d_-]+)/queryloggingconfig$": Route53().list_or_create_query_logging_config_response,
+    r"{0}/(?P<api_version>[\d_-]+)/queryloggingconfig/(?P<query_id>[^/]+)$": Route53().get_or_delete_query_logging_config_response,
 }

moto/route53/utils.py

@@ -0,0 +1,10 @@
+"""Pagination control model for Route53."""
+
+PAGINATION_MODEL = {
+    "list_query_logging_configs": {
+        "input_token": "next_token",
+        "limit_key": "max_results",
+        "limit_default": 100,
+        "page_ending_range_keys": ["hosted_zone_id"],
+    },
+}

tests/test_route53/test_route53_query_logging_config.py

@@ -0,0 +1,277 @@
+"""Route53 unit tests specific to query_logging_config APIs."""
+import pytest
+
+import boto3
+from botocore.exceptions import ClientError
+
+from moto import mock_logs
+from moto import mock_route53
+from moto.core import ACCOUNT_ID
+from moto.core.utils import get_random_hex
+
+# The log group must be in the us-east-1 region.
+TEST_REGION = "us-east-1"
+
+
+def create_hosted_zone_id(route53_client, hosted_zone_test_name):
+    """Return ID of a newly created Route53 public hosted zone"""
+    response = route53_client.create_hosted_zone(
+        Name=hosted_zone_test_name,
+        CallerReference=f"test_caller_ref_{get_random_hex(6)}",
+    )
+    assert response["ResponseMetadata"]["HTTPStatusCode"] == 201
+    assert "HostedZone" in response and response["HostedZone"]["Id"]
+    return response["HostedZone"]["Id"]
+
+
+def create_log_group_arn(logs_client, hosted_zone_test_name):
+    """Return ARN of a newly created CloudWatch log group."""
+    log_group_name = f"/aws/route53/{hosted_zone_test_name}"
+    response = logs_client.create_log_group(logGroupName=log_group_name)
+    assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
+
+    log_group_arn = None
+    response = logs_client.describe_log_groups()
+    for entry in response["logGroups"]:
+        if entry["logGroupName"] == log_group_name:
+            log_group_arn = entry["arn"]
+            break
+    return log_group_arn
+
+
+@mock_logs
+@mock_route53
+def test_create_query_logging_config_bad_args():
+    """Test bad arguments to create_query_logging_config()."""
+    client = boto3.client("route53", region_name=TEST_REGION)
+    logs_client = boto3.client("logs", region_name=TEST_REGION)
+
+    hosted_zone_test_name = f"route53_query_log_{get_random_hex(6)}.test"
+    hosted_zone_id = create_hosted_zone_id(client, hosted_zone_test_name)
+    log_group_arn = create_log_group_arn(logs_client, hosted_zone_test_name)
+
+    # Check exception:  NoSuchHostedZone
+    with pytest.raises(ClientError) as exc:
+        client.create_query_logging_config(
+            HostedZoneId="foo", CloudWatchLogsLogGroupArn=log_group_arn,
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "NoSuchHostedZone"
+    assert "No hosted zone found with ID: foo" in err["Message"]
+
+    # Check exception:  InvalidInput (bad CloudWatch Logs log ARN)
+    with pytest.raises(ClientError) as exc:
+        client.create_query_logging_config(
+            HostedZoneId=hosted_zone_id,
+            CloudWatchLogsLogGroupArn=f"arn:aws:logs:{TEST_REGION}:{ACCOUNT_ID}:foo-bar:foo",
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "InvalidInput"
+    assert "The ARN for the CloudWatch Logs log group is invalid" in err["Message"]
+
+    # Check exception:  InvalidInput (CloudWatch Logs log not in us-east-1)
+    with pytest.raises(ClientError) as exc:
+        client.create_query_logging_config(
+            HostedZoneId=hosted_zone_id,
+            CloudWatchLogsLogGroupArn=log_group_arn.replace(TEST_REGION, "us-west-1"),
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "InvalidInput"
+    assert "The ARN for the CloudWatch Logs log group is invalid" in err["Message"]
+
+    # Check exception:  NoSuchCloudWatchLogsLogGroup
+    with pytest.raises(ClientError) as exc:
+        client.create_query_logging_config(
+            HostedZoneId=hosted_zone_id,
+            CloudWatchLogsLogGroupArn=log_group_arn.replace(
+                hosted_zone_test_name, "foo"
+            ),
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "NoSuchCloudWatchLogsLogGroup"
+    assert "The specified CloudWatch Logs log group doesn't exist" in err["Message"]
+
+    # Check exception:  QueryLoggingConfigAlreadyExists
+    client.create_query_logging_config(
+        HostedZoneId=hosted_zone_id, CloudWatchLogsLogGroupArn=log_group_arn,
+    )
+    with pytest.raises(ClientError) as exc:
+        client.create_query_logging_config(
+            HostedZoneId=hosted_zone_id, CloudWatchLogsLogGroupArn=log_group_arn,
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "QueryLoggingConfigAlreadyExists"
+    assert (
+        "A query logging configuration already exists for this hosted zone"
+        in err["Message"]
+    )
+
+
+@mock_logs
+@mock_route53
+def test_create_query_logging_config_good_args():
+    """Test a valid create_logging_config() request."""
+    client = boto3.client("route53", region_name=TEST_REGION)
+    logs_client = boto3.client("logs", region_name=TEST_REGION)
+
+    hosted_zone_test_name = f"route53_query_log_{get_random_hex(6)}.test"
+    hosted_zone_id = create_hosted_zone_id(client, hosted_zone_test_name)
+    log_group_arn = create_log_group_arn(logs_client, hosted_zone_test_name)
+
+    response = client.create_query_logging_config(
+        HostedZoneId=hosted_zone_id, CloudWatchLogsLogGroupArn=log_group_arn,
+    )
+    config = response["QueryLoggingConfig"]
+    assert config["HostedZoneId"] == hosted_zone_id.split("/")[-1]
+    assert config["CloudWatchLogsLogGroupArn"] == log_group_arn
+    assert config["Id"]
+
+    location = response["Location"]
+    assert (
+        location
+        == f"https://route53.amazonaws.com/2013-04-01/queryloggingconfig/{config['Id']}"
+    )
+
+
+@mock_logs
+@mock_route53
+def test_delete_query_logging_config():
+    """Test valid and invalid delete_query_logging_config requests."""
+    client = boto3.client("route53", region_name=TEST_REGION)
+    logs_client = boto3.client("logs", region_name=TEST_REGION)
+
+    # Create a query logging config that can then be deleted.
+    hosted_zone_test_name = f"route53_query_log_{get_random_hex(6)}.test"
+    hosted_zone_id = create_hosted_zone_id(client, hosted_zone_test_name)
+    log_group_arn = create_log_group_arn(logs_client, hosted_zone_test_name)
+
+    query_response = client.create_query_logging_config(
+        HostedZoneId=hosted_zone_id, CloudWatchLogsLogGroupArn=log_group_arn,
+    )
+
+    # Test the deletion.
+    query_id = query_response["QueryLoggingConfig"]["Id"]
+    response = client.delete_query_logging_config(Id=query_id)
+    # There is no response other than the usual ResponseMetadata.
+    assert list(response.keys()) == ["ResponseMetadata"]
+
+    # Test the deletion of a non-existent query logging config, i.e., the
+    # one that was just deleted.
+    with pytest.raises(ClientError) as exc:
+        client.delete_query_logging_config(Id=query_id)
+    err = exc.value.response["Error"]
+    assert err["Code"] == "NoSuchQueryLoggingConfig"
+    assert "The query logging configuration does not exist" in err["Message"]
+
+
+@mock_logs
+@mock_route53
+def test_get_query_logging_config():
+    """Test valid and invalid get_query_logging_config requests."""
+    client = boto3.client("route53", region_name=TEST_REGION)
+    logs_client = boto3.client("logs", region_name=TEST_REGION)
+
+    # Create a query logging config that can then be retrieved.
+    hosted_zone_test_name = f"route53_query_log_{get_random_hex(6)}.test"
+    hosted_zone_id = create_hosted_zone_id(client, hosted_zone_test_name)
+    log_group_arn = create_log_group_arn(logs_client, hosted_zone_test_name)
+
+    query_response = client.create_query_logging_config(
+        HostedZoneId=hosted_zone_id, CloudWatchLogsLogGroupArn=log_group_arn,
+    )
+
+    # Test the retrieval.
+    query_id = query_response["QueryLoggingConfig"]["Id"]
+    response = client.get_query_logging_config(Id=query_id)
+    config = response["QueryLoggingConfig"]
+    assert config["HostedZoneId"] == hosted_zone_id.split("/")[-1]
+    assert config["CloudWatchLogsLogGroupArn"] == log_group_arn
+    assert config["Id"]
+
+    # Test the retrieval of a non-existent query logging config.
+    with pytest.raises(ClientError) as exc:
+        client.get_query_logging_config(Id="1234567890")
+    err = exc.value.response["Error"]
+    assert err["Code"] == "NoSuchQueryLoggingConfig"
+    assert "The query logging configuration does not exist" in err["Message"]
+
+
+@mock_logs
+@mock_route53
+def test_list_query_logging_configs_bad_args():
+    """Test bad arguments to list_query_logging_configs()."""
+    client = boto3.client("route53", region_name=TEST_REGION)
+    logs_client = boto3.client("logs", region_name=TEST_REGION)
+
+    # Check exception:  NoSuchHostedZone
+    with pytest.raises(ClientError) as exc:
+        client.list_query_logging_configs(HostedZoneId="foo", MaxResults="10")
+    err = exc.value.response["Error"]
+    assert err["Code"] == "NoSuchHostedZone"
+    assert "No hosted zone found with ID: foo" in err["Message"]
+
+    # Create a couple of query logging configs to work with.
+    for _ in range(3):
+        hosted_zone_test_name = f"route53_query_log_{get_random_hex(6)}.test"
+        hosted_zone_id = create_hosted_zone_id(client, hosted_zone_test_name)
+        log_group_arn = create_log_group_arn(logs_client, hosted_zone_test_name)
+        client.create_query_logging_config(
+            HostedZoneId=hosted_zone_id, CloudWatchLogsLogGroupArn=log_group_arn,
+        )
+
+    # Retrieve a query logging config, then request more with an invalid token.
+    client.list_query_logging_configs(MaxResults="1")
+    with pytest.raises(ClientError) as exc:
+        client.list_query_logging_configs(NextToken="foo")
+    err = exc.value.response["Error"]
+    assert err["Code"] == "InvalidPaginationToken"
+    assert (
+        "Route 53 can't get the next page of query logging configurations "
+        "because the specified value for NextToken is invalid." in err["Message"]
+    )
+
+
+@mock_logs
+@mock_route53
+def test_list_query_logging_configs_good_args():
+    """Test valid arguments to list_query_logging_configs()."""
+    client = boto3.client("route53", region_name=TEST_REGION)
+    logs_client = boto3.client("logs", region_name=TEST_REGION)
+
+    # Test when there are no query logging configs.
+    response = client.list_query_logging_configs()
+    query_logging_configs = response["QueryLoggingConfigs"]
+    assert len(query_logging_configs) == 0
+
+    # Create a couple of query logging configs to work with.
+    zone_ids = []
+    for _ in range(10):
+        hosted_zone_test_name = f"route53_query_log_{get_random_hex(6)}.test"
+        hosted_zone_id = create_hosted_zone_id(client, hosted_zone_test_name)
+        zone_ids.append(hosted_zone_id)
+
+        log_group_arn = create_log_group_arn(logs_client, hosted_zone_test_name)
+        client.create_query_logging_config(
+            HostedZoneId=hosted_zone_id, CloudWatchLogsLogGroupArn=log_group_arn,
+        )
+
+    # Verify all 10 of the query logging configs can be retrieved in one go.
+    response = client.list_query_logging_configs()
+    query_logging_configs = response["QueryLoggingConfigs"]
+    assert len(query_logging_configs) == 10
+    for idx, query_logging_config in enumerate(query_logging_configs):
+        assert query_logging_config["HostedZoneId"] == zone_ids[idx].split("/")[-1]
+
+    # Request only two of the query logging configs and verify there's a
+    # next_token.
+    response = client.list_query_logging_configs(MaxResults="2")
+    assert len(response["QueryLoggingConfigs"]) == 2
+    assert response["NextToken"]
+
+    # Request the remaining 8 query logging configs and verify there is
+    # no next token.
+    response = client.list_query_logging_configs(
+        MaxResults="8", NextToken=response["NextToken"]
+    )
+    assert len(response["QueryLoggingConfigs"]) == 8
+    assert "NextToken" not in response