desec-stack/api/desecapi/tests/test_rrsets.py

from ipaddress import IPv4Network
import re
from itertools import product
from math import ceil, floor

from django.conf import settings
from django.core.exceptions import ValidationError
from django.core.management import call_command
from rest_framework import status

from desecapi.models import Domain, RR, RRset
from desecapi.models.records import RR_SET_TYPES_AUTOMATIC, RR_SET_TYPES_UNSUPPORTED
from desecapi.tests.base import DesecTestCase, AuthenticatedRRSetBaseTestCase


class UnauthenticatedRRSetTestCase(DesecTestCase):
    def test_unauthorized_access(self):
        url = self.reverse("v1:rrsets", name="example.com")
        for method in [
            self.client.get,
            self.client.post,
            self.client.put,
            self.client.delete,
            self.client.patch,
        ]:
            response = method(url)
            self.assertStatus(response, status.HTTP_401_UNAUTHORIZED)


class AuthenticatedRRSetTestCase(AuthenticatedRRSetBaseTestCase):
    def test_subname_validity(self):
        for subname in [
            "aEroport",
            "AEROPORT",
            "aéroport",
            "a" * 64,
        ]:
            with self.assertRaises(ValidationError):
                RRset(domain=self.my_domain, subname=subname, ttl=60, type="A").save()
        for subname in [
            "aeroport",
            "a" * 63,
        ]:
            RRset(domain=self.my_domain, subname=subname, ttl=60, type="A").save()

    def test_retrieve_my_rr_sets(self):
        for response in [
            self.client.get_rr_sets(self.my_domain.name),
            self.client.get_rr_sets(self.my_domain.name, subname=""),
        ]:
            self.assertStatus(response, status.HTTP_200_OK)
            self.assertEqual(len(response.data), 1, response.data)

    def test_retrieve_my_rr_sets_pagination(self):
        def convert_links(links):
            mapping = {}
            for link in links.split(", "):
                _url, label = link.split("; ")
                label = re.search('rel="(.*)"', label).group(1)
                _url = _url[1:-1]
                assert label not in mapping
                mapping[label] = _url
            return mapping

        def assertPaginationResponse(
            response, expected_length, expected_directional_links=[]
        ):
            self.assertStatus(response, status.HTTP_200_OK)
            self.assertEqual(len(response.data), expected_length)

            _links = convert_links(response["Link"])
            self.assertEqual(
                len(_links), len(expected_directional_links) + 1
            )  # directional links, plus "first"
            self.assertTrue(_links["first"].endswith("/?cursor="))
            for directional_link in expected_directional_links:
                self.assertEqual(
                    _links["first"].find("/?cursor="),
                    _links[directional_link].find("/?cursor="),
                )
                self.assertTrue(len(_links[directional_link]) > len(_links["first"]))

        # Prepare extra records so that we get three pages (total: n + 1)
        n = int(settings.REST_FRAMEWORK["PAGE_SIZE"] * 2.5)
        RRset.objects.bulk_create(
            [
                RRset(domain=self.my_domain, subname=str(i), ttl=123, type="A")
                for i in range(n)
            ]
        )

        # No pagination
        response = self.client.get_rr_sets(self.my_domain.name)
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
        self.assertEqual(
            response.data["detail"],
            f'Pagination required. You can query up to {settings.REST_FRAMEWORK["PAGE_SIZE"]} items at a time ({n+1} total). '
            "Please use the `first` page link (see Link header).",
        )
        links = convert_links(response["Link"])
        self.assertEqual(len(links), 1)
        self.assertTrue(links["first"].endswith("/?cursor="))

        # First page
        url = links["first"]
        response = self.client.get(url)
        assertPaginationResponse(
            response, settings.REST_FRAMEWORK["PAGE_SIZE"], ["next"]
        )

        # Next
        url = convert_links(response["Link"])["next"]
        response = self.client.get(url)
        assertPaginationResponse(
            response, settings.REST_FRAMEWORK["PAGE_SIZE"], ["next", "prev"]
        )
        data_next = response.data.copy()

        # Next-next (last) page
        url = convert_links(response["Link"])["next"]
        response = self.client.get(url)
        assertPaginationResponse(response, n / 5 + 1, ["prev"])

        # Prev
        url = convert_links(response["Link"])["prev"]
        response = self.client.get(url)
        assertPaginationResponse(
            response, settings.REST_FRAMEWORK["PAGE_SIZE"], ["next", "prev"]
        )

        # Make sure that one step forward equals two steps forward and one step back
        self.assertEqual(response.data, data_next)

    def test_retrieve_other_rr_sets(self):
        self.assertStatus(
            self.client.get_rr_sets(self.other_domain.name), status.HTTP_404_NOT_FOUND
        )
        self.assertStatus(
            self.client.get_rr_sets(self.other_domain.name, subname="test"),
            status.HTTP_404_NOT_FOUND,
        )
        self.assertStatus(
            self.client.get_rr_sets(self.other_domain.name, type="A"),
            status.HTTP_404_NOT_FOUND,
        )

    def test_retrieve_my_rr_sets_filter(self):
        response = self.client.get_rr_sets(self.my_rr_set_domain.name, query="?cursor=")
        self.assertStatus(response, status.HTTP_200_OK)
        expected_number_of_rrsets = min(
            len(self._test_rr_sets()), settings.REST_FRAMEWORK["PAGE_SIZE"]
        )
        self.assertEqual(len(response.data), expected_number_of_rrsets)

        for subname in self.SUBNAMES:
            response = self.client.get_rr_sets(
                self.my_rr_set_domain.name, subname=subname
            )
            self.assertStatus(response, status.HTTP_200_OK)
            self.assertRRSetsCount(
                response.data,
                [dict(subname=subname)],
                count=len(self._test_rr_sets(subname=subname)),
            )

        for type_ in self.ALLOWED_TYPES:
            response = self.client.get_rr_sets(self.my_rr_set_domain.name, type=type_)
            self.assertStatus(response, status.HTTP_200_OK)

    def test_create_my_rr_sets(self):
        for subname in [
            None,
            "create-my-rr-sets",
            "foo.create-my-rr-sets",
            "bar.baz.foo.create-my-rr-sets",
        ]:
            for data in [
                {"subname": subname, "records": ["1.2.3.4"], "ttl": 3660, "type": "A"},
                {
                    "subname": "" if subname is None else subname,
                    "records": ["desec.io."],
                    "ttl": 36900,
                    "type": "PTR",
                },
                {
                    "subname": "" if subname is None else subname,
                    "ttl": 3650,
                    "type": "TXT",
                    "records": ['"foo"'],
                },
                {
                    "subname": f"{subname}.cname".lower(),
                    "ttl": 3600,
                    "type": "CNAME",
                    "records": ["example.com."],
                },
            ]:
                # Try POST with missing subname
                if data["subname"] is None:
                    data.pop("subname")

                with self.assertPdnsRequests(
                    self.requests_desec_rr_sets_update(name=self.my_empty_domain.name)
                ):
                    response = self.client.post_rr_set(
                        domain_name=self.my_empty_domain.name, **data
                    )
                    self.assertStatus(response, status.HTTP_201_CREATED)
                    self.assertTrue(
                        all(
                            field in response.data
                            for field in [
                                "created",
                                "domain",
                                "subname",
                                "name",
                                "records",
                                "ttl",
                                "type",
                                "touched",
                            ]
                        )
                    )
                    self.assertEqual(
                        self.my_empty_domain.touched,
                        max(
                            rrset.touched
                            for rrset in self.my_empty_domain.rrset_set.all()
                        ),
                    )

                # Check for uniqueness on second attempt
                response = self.client.post_rr_set(
                    domain_name=self.my_empty_domain.name, **data
                )
                self.assertContains(
                    response,
                    "Another RRset with the same subdomain and type exists for this domain.",
                    status_code=status.HTTP_400_BAD_REQUEST,
                )

                response = self.client.get_rr_sets(self.my_empty_domain.name)
                self.assertStatus(response, status.HTTP_200_OK)
                self.assertRRSetsCount(response.data, [data])

                response = self.client.get_rr_set(
                    self.my_empty_domain.name, data.get("subname", ""), data["type"]
                )
                self.assertStatus(response, status.HTTP_200_OK)
                self.assertRRSet(response.data, **data)

    def test_create_my_rr_sets_type_restriction(self):
        for subname in [
            "",
            "create-my-rr-sets",
            "foo.create-my-rr-sets",
            "bar.baz.foo.create-my-rr-sets",
        ]:
            for data in (
                [
                    {"subname": subname, "ttl": 60, "type": "a"},
                    {
                        "subname": subname,
                        "records": ["10 example.com."],
                        "ttl": 60,
                        "type": "txt",
                    },
                ]
                + [
                    {
                        "subname": subname,
                        "records": ["10 example.com."],
                        "ttl": 60,
                        "type": type_,
                    }
                    for type_ in self.UNSUPPORTED_TYPES
                ]
                + [
                    {
                        "subname": subname,
                        "records": [
                            "get.desec.io. get.desec.io. 2584 10800 3600 604800 60"
                        ],
                        "ttl": 60,
                        "type": type_,
                    }
                    for type_ in self.AUTOMATIC_TYPES
                ]
            ):
                response = self.client.post_rr_set(self.my_domain.name, **data)
                self.assertStatus(response, status.HTTP_400_BAD_REQUEST)

                response = self.client.get_rr_sets(self.my_domain.name)
                self.assertStatus(response, status.HTTP_200_OK)
                self.assertRRSetsCount(response.data, [data], count=0)

    def test_create_my_rr_sets_only_at_apex(self):
        for type_, records in {
            "DNSKEY": ["257 3 15 l02Woi0iS8Aa25FQkUd9RMzZHJpBoRQwAQEX1SxZJA4="],
        }.items():
            data = {
                "subname": "non-apex",
                "ttl": 3600,
                "type": type_,
                "records": records,
            }
            r = self.client.post_rr_set(self.my_empty_domain.name, **data)
            self.assertContains(
                r,
                f"{type_} RRset must have empty subname",
                status_code=status.HTTP_400_BAD_REQUEST,
            )

    def test_create_my_rr_sets_restricted_at_apex(self):
        for type_, records in {
            "CNAME": ["foobar.com."],
            "DS": ["45586 5 1 D0FDF996D1AF2CCDBDC942B02CB02D379629E20B"],
        }.items():
            data = {"subname": "", "ttl": 3600, "type": type_, "records": records}
            r = self.client.post_rr_set(self.my_empty_domain.name, **data)
            self.assertContains(
                r,
                f"{type_} RRset cannot have empty subname",
                status_code=status.HTTP_400_BAD_REQUEST,
            )

    def test_create_my_rr_sets_cname_multiple_records(self):
        for records in (["foobar.com.", "foobar.com."], ["foobar.com.", "foobar.org."]):
            data = {"subname": "asdf", "ttl": 3600, "type": "CNAME", "records": records}
            response = self.client.post_rr_set(self.my_empty_domain.name, **data)
            self.assertContains(
                response,
                "CNAME RRset cannot have multiple records",
                status_code=status.HTTP_400_BAD_REQUEST,
            )

    def test_create_my_rr_sets_dname_multiple_records(self):
        for records in (["foobar.com.", "foobar.com."], ["foobar.com.", "foobar.org."]):
            data = {"subname": "asdf", "ttl": 3600, "type": "DNAME", "records": records}
            response = self.client.post_rr_set(self.my_empty_domain.name, **data)
            self.assertContains(
                response,
                "DNAME RRset cannot have multiple records",
                status_code=status.HTTP_400_BAD_REQUEST,
            )

    def test_create_my_rr_sets_cname_exclusivity(self):
        self.create_rr_set(self.my_domain, ["1.2.3.4"], type="A", ttl=3600, subname="a")
        self.create_rr_set(
            self.my_domain, ["example.com."], type="CNAME", ttl=3600, subname="cname"
        )

        # Can't add a CNAME where something else is
        data = {
            "subname": "a",
            "ttl": 3600,
            "type": "CNAME",
            "records": ["foobar.com."],
        }
        response = self.client.post_rr_set(self.my_domain.name, **data)
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)

        # Can't add something else where a CNAME is
        data = {"subname": "cname", "ttl": 3600, "type": "A", "records": ["4.3.2.1"]}
        response = self.client.post_rr_set(self.my_domain.name, **data)
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)

    def test_create_my_rr_sets_without_records(self):
        for subname in [
            "",
            "create-my-rr-sets",
            "foo.create-my-rr-sets",
            "bar.baz.foo.create-my-rr-sets",
        ]:
            for data in [
                {"subname": subname, "records": [], "ttl": 60, "type": "A"},
                {"subname": subname, "ttl": 60, "type": "A"},
            ]:
                response = self.client.post_rr_set(self.my_empty_domain.name, **data)
                self.assertStatus(response, status.HTTP_400_BAD_REQUEST)

                response = self.client.get_rr_sets(self.my_empty_domain.name)
                self.assertStatus(response, status.HTTP_200_OK)
                self.assertRRSetsCount(response.data, [], count=0)

    def test_create_other_rr_sets(self):
        data = {"records": ["1.2.3.4"], "ttl": 60, "type": "A"}
        response = self.client.post_rr_set(self.other_domain.name, **data)
        self.assertStatus(response, status.HTTP_404_NOT_FOUND)

    @staticmethod
    def _create_test_txt_record(record, type_="TXT"):
        return {
            "records": [f"{record}"],
            "ttl": 3600,
            "type": type_,
            "subname": f"name{len(record)}",
        }

    def test_create_my_rr_sets_chunk_too_long(self):
        for l, t in product([1, 255, 256, 498], ["TXT", "SPF"]):
            with self.assertPdnsRequests(
                self.requests_desec_rr_sets_update(self.my_empty_domain.name)
            ):
                response = self.client.post_rr_set(
                    self.my_empty_domain.name,
                    **self._create_test_txt_record(f'"{"A" * l}"', t),
                )
                self.assertStatus(response, status.HTTP_201_CREATED)
            with self.assertPdnsRequests(
                self.requests_desec_rr_sets_update(self.my_empty_domain.name)
            ):
                self.client.delete_rr_set(
                    self.my_empty_domain.name, type_=t, subname=f"name{l+2}"
                )

    def test_create_my_rr_sets_too_long_content(self):
        def token(length):
            if length == 0:
                return ""
            if length <= 255:
                return f'"{"A" * length}"'
            return f"{token(255)} " * (length // 255) + token(length % 255)

        def p2w_length(length):
            return ceil(length / 255 * 256)

        def w2p_length(length):
            return floor(length / 256 * 255)

        max_wirelength = 64000
        max_preslength = w2p_length(max_wirelength)

        assert max_preslength == 63750
        assert p2w_length(max_preslength) == 64000
        assert p2w_length(max_preslength + 1) == 64002

        for t in ["SPF", "TXT"]:
            response = self.client.post_rr_set(
                self.my_empty_domain.name,
                # record of wire length 501 bytes in chunks of max 255 each (RFC 4408)
                **self._create_test_txt_record(token(max_preslength + 1), t),
            )
            self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
            self.assertIn(
                f"Ensure this value has no more than {max_wirelength} byte in wire format (it has {p2w_length(max_preslength + 1)}).",
                str(response.data),
            )

        with self.assertPdnsRequests(
            self.requests_desec_rr_sets_update(self.my_empty_domain.name)
        ):
            response = self.client.post_rr_set(
                self.my_empty_domain.name,
                # record of wire length 500 bytes in chunks of max 255 each (RFC 4408)
                **self._create_test_txt_record(token(max_preslength)),
            )
            self.assertStatus(response, status.HTTP_201_CREATED)

    def test_create_my_rr_sets_too_large_rrset(self):
        network = IPv4Network("127.0.0.0/20")  # size: 4096 IP addresses
        data = {
            "records": [str(ip) for ip in network],
            "ttl": 3600,
            "type": "A",
            "subname": "name",
        }
        response = self.client.post_rr_set(self.my_empty_domain.name, **data)
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
        excess_length = 28743 + len(self.my_empty_domain.name)
        self.assertIn(
            f"Total length of RRset exceeds limit by {excess_length} bytes.",
            str(response.data),
        )

    def test_create_my_rr_sets_twice(self):
        data = {"records": ["1.2.3.4"], "ttl": 3660, "type": "A"}
        with self.assertPdnsRequests(
            self.requests_desec_rr_sets_update(self.my_empty_domain.name)
        ):
            response = self.client.post_rr_set(self.my_empty_domain.name, **data)
            self.assertStatus(response, status.HTTP_201_CREATED)

        data["records"][0] = "3.2.2.1"
        response = self.client.post_rr_set(self.my_empty_domain.name, **data)
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)

    def test_create_my_rr_sets_duplicate_content(self):
        for records in [
            ["::1", "0::1"],
            # TODO add more examples
        ]:
            data = {"records": records, "ttl": 3660, "type": "AAAA"}
            response = self.client.post_rr_set(self.my_empty_domain.name, **data)
            self.assertContains(
                response, "Duplicate", status_code=status.HTTP_400_BAD_REQUEST
            )

    def test_create_my_rr_sets_upper_case(self):
        for subname in ["asdF", "cAse", "asdf.FOO", "--F", "ALLCAPS"]:
            data = {"records": ["1.2.3.4"], "ttl": 60, "type": "A", "subname": subname}
            response = self.client.post_rr_set(self.my_empty_domain.name, **data)
            self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
            self.assertIn("Subname can only use (lowercase)", str(response.data))

    def test_create_my_rr_sets_subname_too_long(self):
        name = (
            "9.1.f.1.0.6.1.d.f.2.2.b.9.1.6.3.c.2.f.7.9.5.2.4.7.8.3.2.6.6.c.6.9.3.5.6.9.2.4.d.c.e.a.e.f.1.8"
            ".b.a.c.9.0.6.2.c.b.c.1.6.3.8.2.7.9.0.5.2.a.c.f.f.2.6.a.c.3.c.e.3.0.6.1.8.0.7.4.0.1.0.0.2.ip6.test"
        )
        domain = self.create_domain(name=name, owner=self.owner)

        subname = (
            "e.8.c.f.e.0.9.f.4.9.1.f.1.0.6.1.d.f.2.2.b.9.1.6.3.c.9.3.5.6.9.2.4.d.c.e.a.e.f.1.8.b.a.c.9.0"
            ".6.2.c.b.c.1.6.3.8.2.6.7.9.0.5.2.a.c.f.f.2.6"
        )
        data = {"subname": subname, "records": ["1.2.3.4"], "ttl": 3600, "type": "A"}
        response = self.client.post_rr_set(domain.name, **data)
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
        self.assertEqual(response.data["subname"][0].code, "name_too_long")

    def test_create_my_rr_sets_subname_too_many_dots(self):
        for subname in ["dottest.", ".dottest", "dot..test"]:
            data = {
                "subname": subname,
                "records": ["10 example.com."],
                "ttl": 3600,
                "type": "MX",
            }
            response = self.client.post_rr_set(self.my_domain.name, **data)
            self.assertStatus(response, status.HTTP_400_BAD_REQUEST)

        response = self.client.get_rr_sets(self.my_domain.name)
        self.assertStatus(response, status.HTTP_200_OK)
        self.assertRRSetsCount(response.data, [data], count=0)

    def test_create_my_rr_sets_empty_payload(self):
        response = self.client.post_rr_set(self.my_empty_domain.name)
        self.assertContains(
            response, "No data provided", status_code=status.HTTP_400_BAD_REQUEST
        )

    def test_create_my_rr_sets_cname_two_records(self):
        data = {
            "subname": "sub",
            "records": ["example.com.", "example.org."],
            "ttl": 3600,
            "type": "CNAME",
        }
        response = self.client.post_rr_set(self.my_domain.name, **data)
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)

    def test_create_my_rr_sets_canonical_content(self):
        # TODO fill in more examples
        datas = [
            # record type: (non-canonical input, canonical output expectation)
            ("A", ("127.0.0.1", "127.0.0.1")),
            ("AAAA", ("0000::0000:0001", "::1")),
            ("AAAA", ("::ffff:127.0.0.1", "::ffff:7f00:1")),
            ("AAAA", ("2001:db8::128.2.129.4", "2001:db8::8002:8104")),
            ("AFSDB", ("02 turquoise.FEMTO.edu.", "2 turquoise.femto.edu.")),
            (
                "APL",
                (
                    "2:FF00:0:0:0:0::/8  !1:192.168.38.0/28",
                    "2:ff00::/8 !1:192.168.38.0/28",
                ),
            ),
            ("CAA", ('0128 "issue" "letsencrypt.org"', '128 issue "letsencrypt.org"')),
            (
                "CDNSKEY",
                (
                    "0256  03  08  AwEAAday3UX323uVzQqtOMQ7EHQYfD5Ofv4akjQGN2zY5AgB/2jmdR/+ 1PvXFqzKCAGJv4wjABEBNWLLFm7ew1hHMDZEKVL17aml0EBKI6Dsz6Mx t6n7ScvLtHaFRKaxT4i2JxiuVhKdQR9XGMiWAPQKrRM5SLG0P+2F+TLK l3D0L/cD",
                    "256 3 8 AwEAAday3UX323uVzQqtOMQ7EHQYfD5Ofv4akjQGN2zY5AgB/2jmdR/+1PvXFqzKCAGJv4wjABEBNWLLFm7ew1hHMDZEKVL17aml0EBKI6Dsz6Mxt6n7ScvLtHaFRKaxT4i2JxiuVhKdQR9XGMiWAPQKrRM5SLG0P+2F+TLKl3D0L/cD",
                ),
            ),
            (
                "CDNSKEY",
                (
                    "257 3 8 AwEAAcw5QLr0IjC0wKbGoBPQv4qmeqHy9mvL5qGQTuaG5TSrNqEAR6b/ qvxDx6my4JmEmjUPA1JeEI9YfTUieMr2UZflu7aIbZFLw0vqiYrywCGr CHXLalOrEOmrvAxLvq4vHtuTlH7JIszzYBSes8g1vle6KG7xXiP3U5Ll 96Qiu6bZ31rlMQSPB20xbqJJh6psNSrQs41QvdcXAej+K2Hl1Wd8kPri ec4AgiBEh8sk5Pp8W9ROLQ7PcbqqttFaW2m7N/Wy4qcFU13roWKDEAst bxH5CHPoBfZSbIwK4KM6BK/uDHpSPIbiOvOCW+lvu9TAiZPc0oysY6as lO7jXv16Gws=",
                    "257 3 8 AwEAAcw5QLr0IjC0wKbGoBPQv4qmeqHy9mvL5qGQTuaG5TSrNqEAR6b/qvxDx6my4JmEmjUPA1JeEI9YfTUieMr2UZflu7aIbZFLw0vqiYrywCGrCHXLalOrEOmrvAxLvq4vHtuTlH7JIszzYBSes8g1vle6KG7xXiP3U5Ll96Qiu6bZ31rlMQSPB20xbqJJh6psNSrQs41QvdcXAej+K2Hl1Wd8kPriec4AgiBEh8sk5Pp8W9ROLQ7PcbqqttFaW2m7N/Wy4qcFU13roWKDEAstbxH5CHPoBfZSbIwK4KM6BK/uDHpSPIbiOvOCW+lvu9TAiZPc0oysY6aslO7jXv16Gws=",
                ),
            ),
            (
                "CDNSKEY",
                (
                    "257 3 13 aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryq uB78Pyk/NTEoai5bxoipVQQXzHlzyg==",
                    "257 3 13 aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryquB78Pyk/NTEoai5bxoipVQQXzHlzyg==",
                ),
            ),
            ("CDNSKEY", ("0 3 0 AA==", "0 3 0 AA==")),
            (
                "CDS",
                (
                    "047883  013  02  43BD262211B2A748335149408F67BC95B9A4A3174FD86E6A83830380 446E7AFD",
                    "47883 13 2 43BD262211B2A748335149408F67BC95B9A4A3174FD86E6A83830380446E7AFD".lower(),
                ),
            ),
            ("CDS", ("0 0 0 00", "0 0 0 00")),
            ("CERT", ("06 00 00 sadfdd==", "6 0 0 sadfdQ==")),
            ("CNAME", ("EXAMPLE.COM.", "example.com.")),
            ("CSYNC", ("066 03  NS  AAAA A", "66 3 A NS AAAA")),
            ("DHCID", ("xxxx", "xxxx")),
            (
                "DLV",
                (
                    "6454 8 2 5CBA665A006F6487625C6218522F09BD3673C25FA10F25CB18459AA1 0DF1F520",
                    "6454 8 2 5CBA665A006F6487625C6218522F09BD3673C25FA10F25CB18459AA10DF1F520".lower(),
                ),
            ),
            (
                "DLV",
                (
                    "6454 8 2 5C BA665A006F6487625C6218522F09BD3673C25FA10F25CB18459AA1 0DF1F520",
                    "6454 8 2 5CBA665A006F6487625C6218522F09BD3673C25FA10F25CB18459AA10DF1F520".lower(),
                ),
            ),
            ("DNAME", ("EXAMPLE.COM.", "example.com.")),
            (
                "DNSKEY",
                (
                    "0256  03  08  AwEAAday3UX323uVzQqtOMQ7EHQYfD5Ofv4akjQGN2zY5AgB/2jmdR/+ 1PvXFqzKCAGJv4wjABEBNWLLFm7ew1hHMDZEKVL17aml0EBKI6Dsz6Mx t6n7ScvLtHaFRKaxT4i2JxiuVhKdQR9XGMiWAPQKrRM5SLG0P+2F+TLK l3D0L/cD",
                    "256 3 8 AwEAAday3UX323uVzQqtOMQ7EHQYfD5Ofv4akjQGN2zY5AgB/2jmdR/+1PvXFqzKCAGJv4wjABEBNWLLFm7ew1hHMDZEKVL17aml0EBKI6Dsz6Mxt6n7ScvLtHaFRKaxT4i2JxiuVhKdQR9XGMiWAPQKrRM5SLG0P+2F+TLKl3D0L/cD",
                ),
            ),
            (
                "DNSKEY",
                (
                    "257 3 8 AwEAAcw5QLr0IjC0wKbGoBPQv4qmeqHy9mvL5qGQTuaG5TSrNqEAR6b/ qvxDx6my4JmEmjUPA1JeEI9YfTUieMr2UZflu7aIbZFLw0vqiYrywCGr CHXLalOrEOmrvAxLvq4vHtuTlH7JIszzYBSes8g1vle6KG7xXiP3U5Ll 96Qiu6bZ31rlMQSPB20xbqJJh6psNSrQs41QvdcXAej+K2Hl1Wd8kPri ec4AgiBEh8sk5Pp8W9ROLQ7PcbqqttFaW2m7N/Wy4qcFU13roWKDEAst bxH5CHPoBfZSbIwK4KM6BK/uDHpSPIbiOvOCW+lvu9TAiZPc0oysY6as lO7jXv16Gws=",
                    "257 3 8 AwEAAcw5QLr0IjC0wKbGoBPQv4qmeqHy9mvL5qGQTuaG5TSrNqEAR6b/qvxDx6my4JmEmjUPA1JeEI9YfTUieMr2UZflu7aIbZFLw0vqiYrywCGrCHXLalOrEOmrvAxLvq4vHtuTlH7JIszzYBSes8g1vle6KG7xXiP3U5Ll96Qiu6bZ31rlMQSPB20xbqJJh6psNSrQs41QvdcXAej+K2Hl1Wd8kPriec4AgiBEh8sk5Pp8W9ROLQ7PcbqqttFaW2m7N/Wy4qcFU13roWKDEAstbxH5CHPoBfZSbIwK4KM6BK/uDHpSPIbiOvOCW+lvu9TAiZPc0oysY6aslO7jXv16Gws=",
                ),
            ),
            (
                "DNSKEY",
                (
                    "257 3 13 aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryq uB78Pyk/NTEoai5bxoipVQQXzHlzyg==",
                    "257 3 13 aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryquB78Pyk/NTEoai5bxoipVQQXzHlzyg==",
                ),
            ),
            (
                "DS",
                (
                    "6454 8 2 5CBA665A006F6487625C6218522F09BD3673C25FA10F25CB18459AA1 0DF1F520",
                    "6454 8 2 5CBA665A006F6487625C6218522F09BD3673C25FA10F25CB18459AA10DF1F520".lower(),
                ),
            ),
            (
                "DS",
                (
                    "6454 8 2 5C BA665A006F6487625C6218522F09BD3673C25FA10F25CB18459AA1 0DF1F520",
                    "6454 8 2 5CBA665A006F6487625C6218522F09BD3673C25FA10F25CB18459AA10DF1F520".lower(),
                ),
            ),
            ("EUI48", ("AA-BB-CC-DD-EE-FF", "aa-bb-cc-dd-ee-ff")),
            ("EUI64", ("AA-BB-CC-DD-EE-FF-aa-aa", "aa-bb-cc-dd-ee-ff-aa-aa")),
            ("HINFO", ("cpu os", '"cpu" "os"')),
            ("HINFO", ('"cpu" "os"', '"cpu" "os"')),
            (
                "HTTPS",
                ("01 h3POOL.exaMPLe. aLPn=h2,h3", "1 h3POOL.exaMPLe. alpn=h2,h3"),
            ),
            (
                "HTTPS",
                (
                    "01 h3POOL.exaMPLe. aLPn=h2,h3 ECH=MTIzLi4uCg==",
                    '1 h3POOL.exaMPLe. alpn=h2,h3 ech="MTIzLi4uCg=="',
                ),
            ),
            # ('IPSECKEY', ('01 00 02 . ASDFAF==', '1 0 2 . ASDFAA==')),
            # ('IPSECKEY', ('01 00 02 . 000000==', '1 0 2 . 00000w==')),
            ("KX", ("010 example.com.", "10 example.com.")),
            ("L32", ("010  10.1.2.0", "10 10.1.2.0")),
            ("L64", ("010   2001:0Db8:2140:2000", "10 2001:0db8:2140:2000")),
            (
                "LOC",
                (
                    "023 012 59 N 042 022 48.500 W 65.00m 20.00m 10.00m 10.00m",
                    "23 12 59.000 N 42 22 48.500 W 65.00m 20.00m 10.00m 10.00m",
                ),
            ),
            ("LP", ("010   l64-subnet1.example.com.", "10 l64-subnet1.example.com.")),
            ("MX", ("10 010.1.1.1.", "10 010.1.1.1.")),
            ("MX", ("010 010.1.1.2.", "10 010.1.1.2.")),
            ("MX", ("0 .", "0 .")),
            (
                "NAPTR",
                (
                    '100  50  "s"  "z3950+I2L+I2C"     ""  _z3950._tcp.gatech.edu.',
                    '100 50 "s" "z3950+I2L+I2C" "" _z3950._tcp.gatech.edu.',
                ),
            ),
            ("NID", ("010 0014:4fff:ff20:Ee64", "10 0014:4fff:ff20:ee64")),
            ("NS", ("EXaMPLE.COM.", "example.com.")),
            (
                "OPENPGPKEY",
                (
                    "mG8EXtVIsRMFK4EEACIDAwQSZPNqE4tS xLFJYhX+uabSgMrhOqUizJhkLx82",
                    "mG8EXtVIsRMFK4EEACIDAwQSZPNqE4tSxLFJYhX+uabSgMrhOqUizJhkLx82",
                ),
            ),
            ("PTR", ("EXAMPLE.COM.", "example.com.")),
            ("RP", ("hostmaster.EXAMPLE.com. .", "hostmaster.example.com. .")),
            ("SMIMEA", ("3 01 0 aaBBccddeeff", "3 1 0 aabbccddeeff")),
            (
                "SPF",
                (
                    '"v=spf1 ip4:10.1" ".1.1 ip4:127" ".0.0.0/16 ip4:192.168.0.0/27 include:example.com -all"',
                    '"v=spf1 ip4:10.1" ".1.1 ip4:127" ".0.0.0/16 ip4:192.168.0.0/27 include:example.com -all"',
                ),
            ),
            ("SPF", ('"foo" "bar"', '"foo" "bar"')),
            ("SPF", ('"foobar"', '"foobar"')),
            ("SRV", ("0 000 0 .", "0 0 0 .")),
            ("SRV", ("100 1 5061 EXAMPLE.com.", "100 1 5061 example.com.")),
            ("SRV", ("100 1 5061 example.com.", "100 1 5061 example.com.")),
            ("SSHFP", ("2 2 aabbccEEddff", "2 2 aabbcceeddff")),
            (
                "SVCB",
                (
                    "2 sVc2.example.NET. IPV6hint=2001:db8:00:0::2 port=01234",
                    "2 sVc2.example.NET. port=1234 ipv6hint=2001:db8::2",
                ),
            ),
            (
                "SVCB",
                (
                    "2 sVc2.example.NET. ECH=MjIyLi4uCg== IPV6hint=2001:db8:00:0::2 port=01234",
                    '2 sVc2.example.NET. port=1234 ech="MjIyLi4uCg==" ipv6hint=2001:db8::2',
                ),
            ),
            (
                "TLSA",
                (
                    "3 0001 1 000AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA",
                    "3 1 1 000aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
                ),
            ),
            (
                "TLSA",
                (
                    "003 00 002 696B8F6B92A913560b23ef5720c378881faffe74432d04eb35db957c0a93987b47adf26abb5dac10ba482597ae16edb069b511bec3e26010d1927bf6392760dd",
                    "3 0 2 696b8f6b92a913560b23ef5720c378881faffe74432d04eb35db957c0a93987b47adf26abb5dac10ba482597ae16edb069b511bec3e26010d1927bf6392760dd",
                ),
            ),
            (
                "TLSA",
                (
                    "3 0 2 696B8F6B92A913560b23ef5720c378881faffe74432d04eb35db957c0a93987b47adf26abb5dac10ba482597ae16edb069b511bec3e26010d1927bf6392760dd696B8F6B92A913560b23ef5720c378881faffe74432d04eb35db957c0a93987b47adf26abb5dac10ba482597ae16edb069b511bec3e26010d1927bf6392760dd",
                    "3 0 2 696b8f6b92a913560b23ef5720c378881faffe74432d04eb35db957c0a93987b47adf26abb5dac10ba482597ae16edb069b511bec3e26010d1927bf6392760dd696b8f6b92a913560b23ef5720c378881faffe74432d04eb35db957c0a93987b47adf26abb5dac10ba482597ae16edb069b511bec3e26010d1927bf6392760dd",
                ),
            ),
            ("TXT", ('"foo" "bar"', '"foo" "bar"')),
            ("TXT", ('"foobar"', '"foobar"')),
            ("TXT", ('"foo" "" "bar"', '"foo" "" "bar"')),
            ("TXT", ('"" "" "foo" "" "bar"', '"" "" "foo" "" "bar"')),
            (
                "TXT",
                (
                    r'"\130\164name\164Boss\164type\1611"',
                    r'"\130\164name\164Boss\164type\1611"',
                ),
            ),
            (
                "TXT",
                (
                    '"' + "".join(rf"\{n:03}" for n in range(256)) + '"',  # all bytes
                    r'"\000\001\002\003\004\005\006\007\008\009\010\011\012\013\014\015\016\017\018\019\020\021\022\023\024\025\026\027\028\029\030\031 !\"#$%&'
                    + "'"
                    + r'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~\127\128\129\130\131\132\133\134\135\136\137\138\139\140\141\142\143\144\145\146\147\148\149\150\151\152\153\154\155\156\157\158\159\160\161\162\163\164\165\166\167\168\169\170\171\172\173\174\175\176\177\178\179\180\181\182\183\184\185\186\187\188\189\190\191\192\193\194\195\196\197\198\199\200\201\202\203\204\205\206\207\208\209\210\211\212\213\214\215\216\217\218\219\220\221\222\223\224\225\226\227\228\229\230\231\232\233\234\235\236\237\238\239\240\241\242\243\244\245\246\247\248\249\250\251\252\253\254" "\255"',
                ),
            ),
            (
                "URI",
                (
                    '10 01 "ftp://ftp1.example.com/public"',
                    '10 1 "ftp://ftp1.example.com/public"',
                ),
            ),
        ]
        for t, (record, canonical_record) in datas:
            if not record:
                continue
            subname = "" if t == "DNSKEY" else "test"
            data = {"records": [record], "ttl": 3660, "type": t, "subname": subname}
            with self.assertPdnsRequests(
                self.requests_desec_rr_sets_update(name=self.my_empty_domain.name)
            ):
                response = self.client.post_rr_set(self.my_empty_domain.name, **data)
                self.assertStatus(response, status.HTTP_201_CREATED)
                self.assertEqual(
                    canonical_record,
                    response.data["records"][0],
                    f"For RR set type {t}, expected '{canonical_record}' to be the canonical form of "
                    f'\'{record}\', but saw \'{response.data["records"][0]}\'.',
                )
            with self.assertPdnsRequests(
                self.requests_desec_rr_sets_update(name=self.my_empty_domain.name)
            ):
                response = self.client.delete_rr_set(
                    self.my_empty_domain.name, subname=subname, type_=t
                )
                self.assertStatus(response, status.HTTP_204_NO_CONTENT)
        self.assertAllSupportedRRSetTypes(set(t for t, _ in datas))

    def test_create_my_rr_sets_known_type_benign(self):
        # TODO fill in more examples
        datas = {
            "A": ["127.0.0.1", "127.0.0.2"],
            "AAAA": ["::1", "::2"],
            "AFSDB": ["2 turquoise.femto.edu."],
            "APL": [
                # from RFC 3123 Sec. 4
                "1:192.168.32.0/21 !1:192.168.38.0/28",
                "1:192.168.42.0/26 1:192.168.42.64/26 1:192.168.42.128/25",
                "1:127.0.0.1/32 1:172.16.64.0/22",
                "1:224.0.0.0/4  2:FF00:0:0:0:0:0:0:0/8",
                # made-up (not from RFC)
                "1:1.2.3.4/32 2:::/128",
                "2:FF00::/8 !1:192.168.38.0/28",
            ],
            "CAA": [
                '128 issue "letsencrypt.org"',
                '128 iodef "mailto:desec@example.com"',
                '1 issue "letsencrypt.org"',
            ],
            "CERT": ["06 0 0 sadfdd==", "IPGP 0 0 sadfdd=="],
            "CDNSKEY": [
                "256 3 8 AwEAAday3UX323uVzQqtOMQ7EHQYfD5Ofv4akjQGN2zY5AgB/2jmdR/+ 1PvXFqzKCAGJv4wjABEBNWLLFm7ew1hHMDZEKVL17aml0EBKI6Dsz6Mx t6n7ScvLtHaFRKaxT4i2JxiuVhKdQR9XGMiWAPQKrRM5SLG0P+2F+TLK l3D0L/cD",
                "257 3 8 AwEAAcw5QLr0IjC0wKbGoBPQv4qmeqHy9mvL5qGQTuaG5TSrNqEAR6b/ qvxDx6my4JmEmjUPA1JeEI9YfTUieMr2UZflu7aIbZFLw0vqiYrywCGr CHXLalOrEOmrvAxLvq4vHtuTlH7JIszzYBSes8g1vle6KG7xXiP3U5Ll 96Qiu6bZ31rlMQSPB20xbqJJh6psNSrQs41QvdcXAej+K2Hl1Wd8kPri ec4AgiBEh8sk5Pp8W9ROLQ7PcbqqttFaW2m7N/Wy4qcFU13roWKDEAst bxH5CHPoBfZSbIwK4KM6BK/uDHpSPIbiOvOCW+lvu9TAiZPc0oysY6as lO7jXv16Gws=",
                "257 3 13 aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryq uB78Pyk/NTEoai5bxoipVQQXzHlzyg==",
            ],
            "CDS": [
                "6454 8 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 2 5CBA665A006F6487625C6218522F09BD3673C25FA10F25CB18459AA1 0DF1F520",
                "62703 13 2 085BF1EE0ADBBC99D4D9328229EBDCAEC5FAB20E38610072AD055474 4C7AF4A0",
                "61655 13 4 C838A5C66FCBF83B8B6B50C3CEEC3524777FE4AF8A9FE0172ECAD242 48B0CA1A216DD0D538F20C130DD3059538204B04",
                "6454 8 5 24396E17E36D031F71C354B06A979A67A01F503E",
            ],
            "CNAME": ["example.com."],
            "CSYNC": ["0 0", "66 1 A", "66 2 AAAA", "66 3 A NS AAAA", "66 15 NSEC"],
            "DHCID": ["aaaaaaaaaaaa", "aa aaa  aaaa a a a"],
            "DLV": [
                "6454 8 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 2 5CBA665A006F6487625C6218522F09BD3673C25FA10F25CB18459AA1 0DF1F520",
                "62703 13 2 085BF1EE0ADBBC99D4D9328229EBDCAEC5FAB20E38610072AD055474 4C7AF4A0",
                "61655 13 4 C838A5C66FCBF83B8B6B50C3CEEC3524777FE4AF8A9FE0172ECAD242 48B0CA1A216DD0D538F20C130DD3059538204B04",
                "6454 8 5 24396E17E36D031F71C354B06A979A67A01F503E",
            ],
            "DNAME": ["example.com."],
            "DNSKEY": [
                "256 3 8 AwEAAday3UX323uVzQqtOMQ7EHQYfD5Ofv4akjQGN2zY5AgB/2jmdR/+ 1PvXFqzKCAGJv4wjABEBNWLLFm7ew1hHMDZEKVL17aml0EBKI6Dsz6Mx t6n7ScvLtHaFRKaxT4i2JxiuVhKdQR9XGMiWAPQKrRM5SLG0P+2F+TLK l3D0L/cD",
                "257 3 8 AwEAAcw5QLr0IjC0wKbGoBPQv4qmeqHy9mvL5qGQTuaG5TSrNqEAR6b/ qvxDx6my4JmEmjUPA1JeEI9YfTUieMr2UZflu7aIbZFLw0vqiYrywCGr CHXLalOrEOmrvAxLvq4vHtuTlH7JIszzYBSes8g1vle6KG7xXiP3U5Ll 96Qiu6bZ31rlMQSPB20xbqJJh6psNSrQs41QvdcXAej+K2Hl1Wd8kPri ec4AgiBEh8sk5Pp8W9ROLQ7PcbqqttFaW2m7N/Wy4qcFU13roWKDEAst bxH5CHPoBfZSbIwK4KM6BK/uDHpSPIbiOvOCW+lvu9TAiZPc0oysY6as lO7jXv16Gws=",
                "257 3 13 aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryq uB78Pyk/NTEoai5bxoipVQQXzHlzyg==",
            ],
            "DS": [
                "6454 8 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 2 5CBA665A006F6487625C6218522F09BD3673C25FA10F25CB18459AA1 0DF1F520",
                "62703 13 2 085BF1EE0ADBBC99D4D9328229EBDCAEC5FAB20E38610072AD055474 4C7AF4A0",
                "61655 13 4 C838A5C66FCBF83B8B6B50C3CEEC3524777FE4AF8A9FE0172ECAD242 48B0CA1A216DD0D538F20C130DD3059538204B04",
                "6454 8 5 24396E17E36D031F71C354B06A979A67A01F503E",
            ],
            "EUI48": ["aa-bb-cc-dd-ee-ff", "AA-BB-CC-DD-EE-FF"],
            "EUI64": ["aa-bb-cc-dd-ee-ff-00-11", "AA-BB-CC-DD-EE-FF-00-11"],
            "HINFO": ['"ARMv8-A" "Linux"'],
            "HTTPS": [
                # from https://www.ietf.org/archive/id/draft-ietf-dnsop-svcb-https-06.html#name-examples, with ech base64'd
                "1 . alpn=h3",
                "0 pool.svc.example.",
                '1 h3pool.example. alpn=h2,h3 ech="MTIzLi4uCg=="',
                '2 .      alpn=h2 ech="YWJjLi4uCg=="',
                # made-up (not from RFC)
                "1 pool.svc.example. no-default-alpn alpn=h2 port=1234 ipv4hint=192.168.123.1",
                "2 . ech=... key65333=ex1 key65444=ex2 mandatory=key65444,ech",  # see #section-7
            ],
            # 'IPSECKEY': [
            #     '12 0 2 . asdfdf==',
            #     '03 1 1 127.0.0.1 asdfdf==',
            #     '10 02 2 bade::affe AQNRU3mG7TVTO2BkR47usntb102uFJtugbo6BSGvgqt4AQ==',
            #     '12 3 01 example.com. asdfdf==',
            # ],
            "KX": ["4 example.com.", "28 io."],
            "L32": ["010   10.1.2.0", "65535 1.2.3.4"],
            "L64": ["010   2001:0DB8:1140:1000", "10 2001:0DB8:1140:1000"],
            "LOC": ["23 12 59.000 N 42 22 48.500 W 65.00m 20.00m 10.00m 10.00m"],
            "LP": ["10 l64-subnet1.example.com.", "65535 ."],
            "MX": ["10 example.com.", "20 1.1.1.1."],
            "NAPTR": ['100  50  "s"  "z3950+I2L+I2C"     ""  _z3950._tcp.gatech.edu.'],
            "NID": ["65535   0014:4fff:ff20:ee64"],
            "NS": ["ns1.example.com."],
            "OPENPGPKEY": [
                "mG8EXtVIsRMFK4EEACIDAwQSZPNqE4tSxLFJYhX+uabSgMrhOqUizJhkLx82",  # key incomplete
                "YWFh\xf0\x9f\x92\xa9YWFh",  # valid as non-alphabet bytes will be ignored
            ],
            "PTR": ["example.com.", "*.example.com."],
            "RP": ["hostmaster.example.com. ."],
            "SMIMEA": ["3 1 0 aabbccddeeff"],
            "SPF": [
                '"v=spf1 include:example.com ~all"',
                '"v=spf1 ip4:10.1.1.1 ip4:127.0.0.0/16 ip4:192.168.0.0/27 include:example.com -all"',
                '"spf2.0/pra,mfrom ip6:2001:558:fe14:76:68:87:28:0/120 -all"',
            ],
            "SRV": ["0 0 0 .", "100 1 5061 example.com."],
            "SSHFP": ["2 2 aabbcceeddff"],
            "SVCB": [
                "0 svc4-baz.example.net.",
                "1 . key65333=...",
                '2 svc2.example.net. ech="MjIyLi4uCg==" ipv6hint=2001:db8::2 port=1234',
            ],
            "TLSA": [
                "3 1 1 AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA",
                "3 0 2 696b8f6b92a913560b23ef5720c378881faffe74432d04eb35db957c0a93987b47adf26abb5dac10ba482597ae16edb069b511bec3e26010d1927bf6392760dd",
                "3 0 2 696b8f6b92a913560b23ef5720c378881faffe74432d04eb35db957c0a93987b47adf26abb5dac10ba482597ae16edb069b511bec3e26010d1927bf6392760dd696b8f6b92a913560b23ef5720c378881faffe74432d04eb35db957c0a93987b47adf26abb5dac10ba482597ae16edb069b511bec3e26010d1927bf6392760dd",
            ],
            "TXT": [
                '"foobar"',
                '"foo" "bar"',
                '"“红色联合”对“四·二八兵团”总部大楼的攻击已持续了两天"',
                '"new\\010line"'
                '"🧥 👚 👕 👖 👔 👗 👙 👘 👠 👡 👢 👞 👟 🥾 🥿  🧦 🧤 🧣 🎩 🧢 👒 🎓 ⛑ 👑 👝 👛 👜 💼 🎒 👓 🕶 🥽 🥼 🌂 🧵"',
            ],
            "URI": ['10 1 "ftp://ftp1.example.com/public"'],
        }
        self.assertAllSupportedRRSetTypes(set(datas.keys()))
        for t, records in datas.items():
            subname = "" if t == "DNSKEY" else "test"
            for r in records:
                data = {"records": [r], "ttl": 3660, "type": t, "subname": subname}
                with self.assertPdnsRequests(
                    self.requests_desec_rr_sets_update(name=self.my_empty_domain.name)
                ):
                    response = self.client.post_rr_set(
                        self.my_empty_domain.name, **data
                    )
                    self.assertStatus(response, status.HTTP_201_CREATED)
                with self.assertPdnsRequests(
                    self.requests_desec_rr_sets_update(name=self.my_empty_domain.name)
                ):
                    response = self.client.delete_rr_set(
                        self.my_empty_domain.name, subname=subname, type_=t
                    )
                    self.assertStatus(response, status.HTTP_204_NO_CONTENT)

    def test_create_my_rr_sets_known_type_invalid(self):
        # TODO fill in more examples
        datas = {
            # recordtype: [list of examples expected to be rejected, individually]
            "A": [
                "127.0.0.999",
                "127.000.0.01",
                "127.0.0.256",
                "::1",
                "foobar",
                "10.0.1",
                "10!",
            ],
            "AAAA": ["::g", "1:1:1:1:1:1:1:1:", "1:1:1:1:1:1:1:1:1"],
            "AFSDB": ["example.com.", "1 1", "1 de"],
            "APL": [
                "0:192.168.32.0/21 !1:192.168.38.0/28",
                "1:192.168.32.0/21 !!1:192.168.38.0/28",
                "1:192.168.32.0/33",
                "18:12345/2",
                "1:127.0.0.1",
                "2:FF00:0:0:0:0:0:0:0:0/8" "2:::/129",
            ],
            "CAA": ['43235 issue "letsencrypt.org"'],
            "CERT": ["6 0 sadfdd=="],
            "CDNSKEY": [
                "a 3 13 aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryq uB78Pyk/NTEoai5bxoipVQQXzHlzyg=="
                "257 b 13 aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryq uB78Pyk/NTEoai5bxoipVQQXzHlzyg=="
                "257 3 c aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryq uB78Pyk/NTEoai5bxoipVQQXzHlzyg=="
                "257 3 13 d",
                "0 3 0 0",
            ],
            "CDS": [
                "a 8 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "-6454 8 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 b 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 c 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 1 d",
                "6454 8 0 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 1 aabbccddeeff",
                "0 0 0 0",
            ],
            "CNAME": ["example.com", "10 example.com."],
            "CSYNC": ["0 -1 A", "444 65536 A", "0 3 AAA"],
            "DHCID": ["x", "xx", "xxx"],
            "DLV": [
                "a 8 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "-6454 8 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 b 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 c 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 1 d",
                "6454 8 0 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 1 aabbccddeeff",
            ],
            "DNAME": ["example.com", "10 example.com."],
            "DNSKEY": [
                "a 3 13 aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryq uB78Pyk/NTEoai5bxoipVQQXzHlzyg=="
                "257 b 13 aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryq uB78Pyk/NTEoai5bxoipVQQXzHlzyg=="
                "257 3 c aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryq uB78Pyk/NTEoai5bxoipVQQXzHlzyg=="
                "257 3 13 d"
            ],
            "DS": [
                "a 8 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "-6454 8 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 b 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 c 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 1 d",
                "6454 8 0 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 1 aabbccddeeff",
            ],
            "EUI48": ["aa-bb-ccdd-ee-ff", "AA-BB-CC-DD-EE-GG"],
            "EUI64": ["aa-bb-cc-dd-ee-ff-gg-11", "AA-BB-C C-DD-EE-FF-00-11"],
            "HINFO": ['"ARMv8-A"', f'"a" "{"b"*256}"'],
            "HTTPS": [
                # from https://tools.ietf.org/html/draft-ietf-dnsop-svcb-https-02#section-10.3, with ech base64'd
                '1 h3pool alpn=h2,h3 ech="MTIzLi4uCg=="',
                # made-up (not from RFC)
                "0 pool.svc.example. no-default-alpn port=1234 ipv4hint=192.168.123.1",  # no keys in alias mode
                "1 pool.svc.example. no-default-alpn port=1234 ipv4hint=192.168.123.1 ipv4hint=192.168.123.2",  # dup
            ],
            # 'IPSECKEY': [],
            "KX": ["-1 example.com", "10 example.com"],
            "L32": ["65536 10.1.2.0", "5 a.1.2.0", "10 10.1.02.0"],
            "L64": ["65536 2001:0DB8:4140:4000", "5 01:0DB8:4140:4000"],
            "LOC": [
                "23 12 61.000 N 42 22 48.500 W 65.00m 20.00m 10.00m 10.00m",
                "foo",
                "1.1.1.1",
            ],
            "LP": [
                "10 l64-subnet1.example.com",
                "-3 l64-subnet1.example.com.",
                "65536 l64-subnet1.example.com.",
            ],
            "MX": [
                "10 example.com",
                "example.com.",
                "-5 asdf.",
                "65537 asdf.",
                "10 _foo.example.com.",
                "10 $url.",
            ],
            "NAPTR": [
                '100  50  "s"  "z3950+I2L+I2C"     ""  _z3950._tcp.gatech.edu',
                '100  50  "s"     ""  _z3950._tcp.gatech.edu.',
                '100  50  3 2  "z3950+I2L+I2C"     ""  _z3950._tcp.gatech.edu.',
            ],
            "NID": ["010 14:4fff:ff20:Ee64", "d 0014:4fff:ff20:ee64", "20 ::14::ee64"],
            "NS": ["ns1.example.com", "127.0.0.1", "_foobar.example.dedyn.io.", "."],
            "OPENPGPKEY": ["1 2 3"],
            "PTR": ['"example.com."', "10 *.example.com."],
            "RP": ["hostmaster.example.com.", "10 foo."],
            "SMIMEA": ["3 1 0 aGVsbG8gd29ybGQh"],
            "SPF": ['"v=spf1', "v=spf1 include:example.com ~all"],
            "SRV": [
                "0 0 0 0",
                "100 5061 example.com.",
                "0 0 16920 _foo.example.com.",
                "0 0 16920 $url.",
            ],
            "SSHFP": ["aabbcceeddff"],
            "SVCB": [
                "0 svc4-baz.example.net. keys=val",
                "1 not.fully.qualified key65333=...",
                '2 duplicate.key. ech="MjIyLi4uCg==" ech="MjIyLi4uCg=="',
            ],
            "TLSA": ["3 1 1 AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"],
            "TXT": [
                'foob"ar',
                "v=spf1 include:example.com ~all",
                '"foo\nbar"',
                '"\x00" "NUL byte yo"',
                '"'
                + "".join(rf"\{n:03}" for n in range(257))
                + '"',  # \256 does not exist
            ],
            "URI": ['"1" "2" "3"'],
        }
        self.assertAllSupportedRRSetTypes(set(datas.keys()))
        for t, records in datas.items():
            for r in records:
                data = {"records": [r], "ttl": 3660, "type": t, "subname": ""}
                response = self.client.post_rr_set(self.my_empty_domain.name, **data)
                self.assertNotContains(
                    response, "Duplicate", status_code=status.HTTP_400_BAD_REQUEST
                )

    def test_create_my_rr_sets_txt_splitting(self):
        for t in ["TXT", "SPF"]:
            for l in [200, 255, 256, 300, 400]:
                data = {
                    "records": [f'"{"a"*l}"'],
                    "ttl": 3660,
                    "type": t,
                    "subname": f"x{l}",
                }
                with self.assertPdnsRequests(
                    self.requests_desec_rr_sets_update(name=self.my_empty_domain.name)
                ):
                    response = self.client.post_rr_set(
                        self.my_empty_domain.name, **data
                    )
                    self.assertStatus(response, status.HTTP_201_CREATED)
                response = self.client.get_rr_set(self.my_empty_domain.name, f"x{l}", t)
                num_tokens = response.data["records"][0].count(" ") + 1
                num_tokens_expected = l // 256 + 1
                self.assertEqual(
                    num_tokens,
                    num_tokens_expected,
                    f"For a {t} record with a token of length of {l}, expected to see "
                    f"{num_tokens_expected} tokens in the canonical format, but saw {num_tokens}.",
                )
                self.assertEqual(
                    "".join(r.strip('" ') for r in response.data["records"][0]), "a" * l
                )

    def test_create_my_rr_sets_unknown_type(self):
        for _type in ["AA", "ASDF"] + list(
            RR_SET_TYPES_AUTOMATIC | RR_SET_TYPES_UNSUPPORTED
        ):
            response = self.client.post_rr_set(
                self.my_domain.name, records=["1234"], ttl=3660, type=_type
            )
            self.assertContains(
                response,
                text="managed automatically"
                if _type in RR_SET_TYPES_AUTOMATIC
                else "type is currently unsupported",
                status_code=status.HTTP_400_BAD_REQUEST,
            )

    def test_create_my_rr_sets_ttl_too_small(self):
        ttl = settings.MINIMUM_TTL_DEFAULT - 1
        response = self.client.post_rr_set(
            self.my_empty_domain.name, records=["1.2.3.4"], ttl=ttl, type="A"
        )
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
        detail = f"Ensure this value is greater than or equal to {self.my_empty_domain.minimum_ttl}."
        self.assertEqual(response.data["ttl"][0], detail)

        ttl += 1
        with self.assertPdnsRequests(
            self.requests_desec_rr_sets_update(name=self.my_empty_domain.name)
        ):
            response = self.client.post_rr_set(
                self.my_empty_domain.name, records=["1.2.23.4"], ttl=ttl, type="A"
            )
        self.assertStatus(response, status.HTTP_201_CREATED)

    def test_create_my_rr_sets_ttl_too_large(self):
        max_ttl = 24 * 3600
        response = self.client.post_rr_set(
            self.my_empty_domain.name, records=["1.2.3.4"], ttl=max_ttl + 1, type="A"
        )
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
        detail = f"Ensure this value is less than or equal to {max_ttl}."
        self.assertEqual(response.data["ttl"][0], detail)

        with self.assertPdnsRequests(
            self.requests_desec_rr_sets_update(name=self.my_empty_domain.name)
        ):
            response = self.client.post_rr_set(
                self.my_empty_domain.name, records=["1.2.23.4"], ttl=max_ttl, type="A"
            )
        self.assertStatus(response, status.HTTP_201_CREATED)

    def test_retrieve_my_rr_sets_apex(self):
        response = self.client.get_rr_set(
            self.my_rr_set_domain.name, subname="", type_="A"
        )
        self.assertStatus(response, status.HTTP_200_OK)
        self.assertEqual(response.data["records"][0], "1.2.3.4")
        self.assertEqual(response.data["ttl"], 3620)

    def test_retrieve_my_rr_sets_restricted_types(self):
        for type_ in self.AUTOMATIC_TYPES:
            response = self.client.get_rr_sets(self.my_domain.name, type=type_)
            self.assertStatus(response, status.HTTP_403_FORBIDDEN)
            response = self.client.get_rr_sets(
                self.my_domain.name, type=type_, subname=""
            )
            self.assertStatus(response, status.HTTP_403_FORBIDDEN)

    def test_update_my_rr_sets(self):
        for subname in self.SUBNAMES:
            with self.assertPdnsRequests(
                self.requests_desec_rr_sets_update(name=self.my_rr_set_domain.name)
            ):
                data = {
                    "records": ["2.2.3.4"],
                    "ttl": 3630,
                    "type": "A",
                    "subname": subname,
                }
                response = self.client.put_rr_set(
                    self.my_rr_set_domain.name, subname, "A", data
                )
                self.assertStatus(response, status.HTTP_200_OK)

            response = self.client.get_rr_set(self.my_rr_set_domain.name, subname, "A")
            self.assertStatus(response, status.HTTP_200_OK)
            self.assertEqual(response.data["records"], ["2.2.3.4"])
            self.assertEqual(response.data["ttl"], 3630)

            response = self.client.put_rr_set(
                self.my_rr_set_domain.name, subname, "A", {"records": ["2.2.3.5"]}
            )
            self.assertStatus(response, status.HTTP_400_BAD_REQUEST)

            response = self.client.put_rr_set(
                self.my_rr_set_domain.name, subname, "A", {"ttl": 3637}
            )
            self.assertStatus(response, status.HTTP_400_BAD_REQUEST)

    def test_update_my_rr_set_with_invalid_payload_type(self):
        for subname in self.SUBNAMES:
            data = [
                {"records": ["2.2.3.4"], "ttl": 30, "type": "A", "subname": subname}
            ]
            response = self.client.put_rr_set(
                self.my_rr_set_domain.name, subname, "A", data
            )
            self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
            self.assertEquals(
                response.data["non_field_errors"][0],
                "Invalid data. Expected a dictionary, but got list.",
            )

            data = "foobar"
            response = self.client.put_rr_set(
                self.my_rr_set_domain.name, subname, "A", data
            )
            self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
            self.assertEquals(
                response.data["non_field_errors"][0],
                "Invalid data. Expected a dictionary, but got str.",
            )

    def test_partially_update_my_rr_sets(self):
        for subname in self.SUBNAMES:
            current_rr_set = self.client.get_rr_set(
                self.my_rr_set_domain.name, subname, "A"
            ).data
            for data in [
                {"records": ["2.2.3.4"], "ttl": 3630},
                {"records": ["3.2.3.4"]},
                {"records": ["3.2.3.4", "9.8.8.7"]},
                {"ttl": 3637},
            ]:
                with self.assertPdnsRequests(
                    self.requests_desec_rr_sets_update(name=self.my_rr_set_domain.name)
                ):
                    response = self.client.patch_rr_set(
                        self.my_rr_set_domain.name, subname, "A", data
                    )
                    self.assertStatus(response, status.HTTP_200_OK)

                response = self.client.get_rr_set(
                    self.my_rr_set_domain.name, subname, "A"
                )
                self.assertStatus(response, status.HTTP_200_OK)
                current_rr_set.update(data)
                self.assertEqual(
                    set(response.data["records"]), set(current_rr_set["records"])
                )
                self.assertEqual(response.data["ttl"], current_rr_set["ttl"])

            response = self.client.patch_rr_set(
                self.my_rr_set_domain.name, subname, "A", {}
            )
            self.assertStatus(response, status.HTTP_200_OK)

    def test_rr_sets_touched_if_noop(self):
        for subname in self.SUBNAMES:
            touched_old = RRset.objects.get(
                domain=self.my_rr_set_domain, type="A", subname=subname
            ).touched
            response = self.client.patch_rr_set(
                self.my_rr_set_domain.name, subname, "A", {}
            )
            self.assertStatus(response, status.HTTP_200_OK)

            touched_new = RRset.objects.get(
                domain=self.my_rr_set_domain, type="A", subname=subname
            ).touched
            self.assertGreater(touched_new, touched_old)
            self.assertEqual(
                Domain.objects.get(name=self.my_rr_set_domain.name).touched, touched_new
            )

    def test_partially_update_other_rr_sets(self):
        data = {"records": ["3.2.3.4"], "ttl": 334}
        for subname in self.SUBNAMES:
            response = self.client.patch_rr_set(
                self.other_rr_set_domain.name, subname, "A", data
            )
            self.assertStatus(response, status.HTTP_404_NOT_FOUND)

    def test_update_other_rr_sets(self):
        data = {"ttl": 305}
        for subname in self.SUBNAMES:
            response = self.client.patch_rr_set(
                self.other_rr_set_domain.name, subname, "A", data
            )
            self.assertStatus(response, status.HTTP_404_NOT_FOUND)

    def test_update_essential_properties(self):
        # Changing the subname is expected to cause an error
        url = self.reverse(
            "v1:rrset", name=self.my_rr_set_domain.name, subname="test", type="A"
        )
        data = {"records": ["3.2.3.4"], "ttl": 3620, "subname": "test2", "type": "A"}
        response = self.client.patch(url, data)
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
        self.assertEquals(response.data["subname"][0].code, "read-only-on-update")
        response = self.client.put(url, data)
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
        self.assertEquals(response.data["subname"][0].code, "read-only-on-update")

        # Changing the type is expected to cause an error
        data = {"records": ["3.2.3.4"], "ttl": 3620, "subname": "test", "type": "TXT"}
        response = self.client.patch(url, data)
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
        self.assertEquals(response.data["type"][0].code, "read-only-on-update")
        response = self.client.put(url, data)
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
        self.assertEquals(response.data["type"][0].code, "read-only-on-update")

        # Changing "created" is no-op
        response = self.client.get(url)
        data = response.data
        created = data["created"]
        data["created"] = "2019-07-19T17:22:49.575717Z"
        response = self.client.patch(url, data)
        self.assertStatus(response, status.HTTP_200_OK)
        response = self.client.put(url, data)
        self.assertStatus(response, status.HTTP_200_OK)

        # Check that nothing changed
        response = self.client.get(url)
        self.assertStatus(response, status.HTTP_200_OK)
        self.assertEqual(response.data["records"][0], "2.2.3.4")
        self.assertEqual(response.data["ttl"], 3620)
        self.assertEqual(
            response.data["name"], "test." + self.my_rr_set_domain.name + "."
        )
        self.assertEqual(response.data["subname"], "test")
        self.assertEqual(response.data["type"], "A")
        self.assertEqual(response.data["created"], created)

        # This is expected to work, but the fields are ignored
        data = {"records": ["3.2.3.4"], "name": "example.com.", "domain": "example.com"}
        with self.assertPdnsRequests(
            self.requests_desec_rr_sets_update(name=self.my_rr_set_domain.name)
        ):
            response = self.client.patch(url, data)
            self.assertStatus(response, status.HTTP_200_OK)

        response = self.client.get(url)
        self.assertStatus(response, status.HTTP_200_OK)
        self.assertEqual(response.data["records"][0], "3.2.3.4")
        self.assertEqual(response.data["domain"], self.my_rr_set_domain.name)
        self.assertEqual(
            response.data["name"], "test." + self.my_rr_set_domain.name + "."
        )

    def test_update_unknown_rrset(self):
        url = self.reverse(
            "v1:rrset",
            name=self.my_rr_set_domain.name,
            subname="doesnotexist",
            type="A",
        )
        data = {"records": ["3.2.3.4"], "ttl": 3620}

        response = self.client.patch(url, data)
        self.assertStatus(response, status.HTTP_404_NOT_FOUND)

        response = self.client.put(url, data)
        self.assertStatus(response, status.HTTP_404_NOT_FOUND)

    def test_delete_my_rr_sets_with_patch(self):
        data = {"records": []}
        for subname in self.SUBNAMES:
            with self.assertPdnsRequests(
                self.requests_desec_rr_sets_update(name=self.my_rr_set_domain.name)
            ):
                response = self.client.patch_rr_set(
                    self.my_rr_set_domain.name, subname, "A", data
                )
                self.assertStatus(response, status.HTTP_204_NO_CONTENT)

            # Deletion is only idempotent via DELETE. For PATCH/PUT, the view raises 404 if the instance does not
            # exist. By that time, the view has not parsed the payload yet and does not know it is a deletion.
            response = self.client.patch_rr_set(
                self.my_rr_set_domain.name, subname, "A", data
            )
            self.assertStatus(response, status.HTTP_404_NOT_FOUND)

            response = self.client.get_rr_set(self.my_rr_set_domain.name, subname, "A")
            self.assertStatus(response, status.HTTP_404_NOT_FOUND)

    def test_delete_my_rr_sets_with_delete(self):
        for subname in self.SUBNAMES:
            with self.assertPdnsRequests(
                self.requests_desec_rr_sets_update(name=self.my_rr_set_domain.name)
            ):
                response = self.client.delete_rr_set(
                    self.my_rr_set_domain.name, subname=subname, type_="A"
                )
                self.assertStatus(response, status.HTTP_204_NO_CONTENT)
                domain = Domain.objects.get(name=self.my_rr_set_domain.name)
                self.assertEqual(domain.touched, domain.published)

            response = self.client.delete_rr_set(
                self.my_rr_set_domain.name, subname=subname, type_="A"
            )
            self.assertStatus(response, status.HTTP_204_NO_CONTENT)

            response = self.client.get_rr_set(
                self.my_rr_set_domain.name, subname=subname, type_="A"
            )
            self.assertStatus(response, status.HTTP_404_NOT_FOUND)

    def test_delete_other_rr_sets(self):
        data = {"records": []}
        for subname in self.SUBNAMES:
            # Try PATCH empty
            response = self.client.patch_rr_set(
                self.other_rr_set_domain.name, subname, "A", data
            )
            self.assertStatus(response, status.HTTP_404_NOT_FOUND)

            # Try DELETE
            response = self.client.delete_rr_set(
                self.other_rr_set_domain.name, subname, "A"
            )
            self.assertStatus(response, status.HTTP_404_NOT_FOUND)

            # Make sure it actually is still there
            self.assertGreater(
                len(
                    self.other_rr_set_domain.rrset_set.filter(subname=subname, type="A")
                ),
                0,
            )

    def test_import_rr_sets(self):
        with self.assertPdnsRequests(
            self.request_pdns_zone_retrieve(name=self.my_domain.name)
        ):
            call_command("sync-from-pdns", self.my_domain.name)
        for response in [
            self.client.get_rr_sets(self.my_domain.name),
            self.client.get_rr_sets(self.my_domain.name, subname=""),
        ]:
            self.assertStatus(response, status.HTTP_200_OK)
            self.assertEqual(len(response.data), 1, response.data)
            self.assertContainsRRSets(
                response.data,
                [dict(subname="", records=settings.DEFAULT_NS, type="NS")],
            )

    def test_extra_dnskeys(self):
        name = "ietf.org"
        dnskeys = [
            "256 3 5 AwEAAdDECajHaTjfSoNTY58WcBah1BxPKVIHBz4IfLjfqMvium4lgKtKZLe97DgJ5/NQrNEGGQmr6fKvUj67cfrZUojZ2cGRiz"
            "VhgkOqZ9scaTVXNuXLM5Tw7VWOVIceeXAuuH2mPIiEV6MhJYUsW6dvmNsJ4XwCgNgroAmXhoMEiWEjBB+wjYZQ5GtZHBFKVXACSWTiCtdd"
            "HcueOeSVPi5WH94VlubhHfiytNPZLrObhUCHT6k0tNE6phLoHnXWU+6vpsYpz6GhMw/R9BFxW5PdPFIWBgoWk2/XFVRSKG9Lr61b2z1R12"
            "6xeUwvw46RVy3hanV3vNO7LM5HniqaYclBbhk=",
            "257 3 5 AwEAAavjQ1H6pE8FV8LGP0wQBFVL0EM9BRfqxz9p/sZ+8AByqyFHLdZcHoOGF7CgB5OKYMvGOgysuYQloPlwbq7Ws5WywbutbX"
            "yG24lMWy4jijlJUsaFrS5EvUu4ydmuRc/TGnEXnN1XQkO+waIT4cLtrmcWjoY8Oqud6lDaJdj1cKr2nX1NrmMRowIu3DIVtGbQJmzpukpD"
            "VZaYMMAm8M5vz4U2vRCVETLgDoQ7rhsiD127J8gVExjO8B0113jCajbFRcMtUtFTjH4z7jXP2ZzDcXsgpe4LYFuenFQAcRBRlE6oaykHR7"
            "rlPqqmw58nIELJUFoMcb/BdRLgbyTeurFlnxs=",
        ]
        expected_ds = [
            "45586 5 2 67fcd7e0b9e0366309f3b6f7476dff931d5226edc5348cd80fd82a081dfcf6ee",
            "45586 5 4 aee6931c7790c428bca35dab9179cb27f042715e38e5a8adb6bb24c57c21c65dbd02a5b09887787f30128bfac8b6f0b5",
        ]

        domain = Domain.objects.create(name=name, owner=self.owner)
        rrset = domain.rrset_set.create(subname="", type="DNSKEY", ttl=3600)
        rrset.records.bulk_create(
            [RR(rrset=rrset, content=dnskey) for dnskey in dnskeys]
        )

        url = self.reverse("v1:domain-detail", name=domain.name)
        with self.assertPdnsRequests(
            self.request_pdns_zone_retrieve_crypto_keys(name=domain.name)
        ):
            response = self.client.get(url)
            self.assertStatus(response, status.HTTP_200_OK)
            self.assertEqual(
                response.data["keys"],
                [
                    {
                        "dnskey": key["dnskey"],
                        "ds": key["cds"] if key["flags"] & 1 else [],
                        "flags": key["flags"],
                        "keytype": key["keytype"],
                        "managed": True,
                    }
                    for key in self.get_body_pdns_zone_retrieve_crypto_keys()
                ]
                + [
                    {
                        "dnskey": dnskeys[0],
                        "ds": [],
                        "flags": 256,
                        "keytype": None,
                        "managed": False,
                    },
                    {
                        "dnskey": dnskeys[1],
                        "ds": expected_ds,
                        "flags": 257,
                        "keytype": None,
                        "managed": False,
                    },
                ],
            )