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

from rest_framework.reverse import reverse
from rest_framework import status
from rest_framework.test import APITestCase
from desecapi.tests.utils import utils
import httpretty
from django.conf import settings
import json
from django.core.management import call_command
from django.utils import timezone


class UnauthenticatedDomainTests(APITestCase):
    def testExpectUnauthorizedOnGet(self):
        url = reverse('v1:rrsets', args=('example.com',))
        response = self.client.get(url, format='json')
        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)

    def testExpectUnauthorizedOnPost(self):
        url = reverse('v1:rrsets', args=('example.com',))
        response = self.client.post(url, format='json')
        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)

    def testExpectUnauthorizedOnPut(self):
        url = reverse('v1:rrsets', args=('example.com',))
        response = self.client.put(url, format='json')
        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)

    def testExpectUnauthorizedOnDelete(self):
        url = reverse('v1:rrsets', args=('example.com',))
        response = self.client.delete(url, format='json')
        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)


class AuthenticatedRRsetTests(APITestCase):
    dead_types = ('ALIAS', 'DNAME')
    restricted_types = ('SOA', 'RRSIG', 'DNSKEY', 'NSEC3PARAM')

    def setUp(self):
        if not hasattr(self, 'owner'):
            self.owner = utils.createUser()
            self.ownedDomains = [utils.createDomain(self.owner), utils.createDomain(self.owner)]
            self.token = utils.createToken(user=self.owner)
            self.client.credentials(HTTP_AUTHORIZATION='Token ' + self.token)

            self.otherOwner = utils.createUser()
            self.otherDomains = [utils.createDomain(self.otherOwner), utils.createDomain()]
            self.otherToken = utils.createToken(user=self.otherOwner)

            httpretty.reset()
            httpretty.enable(allow_net_connect=False)
            for domain in self.ownedDomains + self.otherDomains:
                httpretty.register_uri(httpretty.PATCH, settings.NSLORD_PDNS_API + '/zones/' + domain.name + '.')
                httpretty.register_uri(httpretty.PUT,
                                       settings.NSLORD_PDNS_API + '/zones/' + domain.name + './notify')

    def testCanGetOwnRRsets(self):
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(len(response.data), 0)  # NS RRset unavailable in mock pdns environment

    def testCantGetForeignRRsets(self):
        url = reverse('v1:rrsets', args=(self.otherDomains[1].name,))
        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)

    def testCanGetOwnRRsetsEmptySubname(self):
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        response = self.client.get(url + '?subname=')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(len(response.data), 0)  # NS RRset unavailable in mock pdns environment

    def testCanGetOwnRRsetsFromSubname(self):
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))

        data = {'records': ['1.2.3.4'], 'ttl': 120, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        data = {'records': ['2.2.3.4'], 'ttl': 120, 'type': 'A', 'subname': 'test'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        data = {'records': ['"test"'], 'ttl': 120, 'type': 'TXT', 'subname': 'test'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(len(response.data), 3)  # NS RRset unavailable in mock pdns environment

        response = self.client.get(url + '?subname=test')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(len(response.data), 2)

    def testCantGetForeignRRsetsFromSubname(self):
        url = reverse('v1:rrsets', args=(self.otherDomains[1].name,))
        response = self.client.get(url + '?subname=test')
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)

    def testCanGetOwnRRsetsFromType(self):
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))

        data = {'records': ['1.2.3.4'], 'ttl': 120, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        data = {'records': ['2.2.3.4'], 'ttl': 120, 'type': 'A', 'subname': 'test'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        data = {'records': ['"test"'], 'ttl': 120, 'type': 'TXT', 'subname': 'test'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(len(response.data), 3)  # NS RRset unavailable in mock pdns environment

        response = self.client.get(url + '?type=A')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(len(response.data), 2)

    def testCantGetForeignRRsetsFromType(self):
        url = reverse('v1:rrsets', args=(self.otherDomains[1].name,))
        response = self.client.get(url + '?test=A')
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)

    def testCanPostOwnRRsets(self):
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(len(response.data), 1)  # NS RRset unavailable in mock pdns environment

        url = reverse('v1:rrset', args=(self.ownedDomains[1].name, '', 'A',))
        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(response.data['records'][0], '1.2.3.4')

        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': ['desec.io.'], 'ttl': 900, 'type': 'PTR'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

    def testCantPostEmptyRRset(self):
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': [], 'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)

        data = {'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)

    def testCantPostDeadTypes(self):
        for type_ in self.dead_types:
            url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
            data = {'records': ['www.example.com.'], 'ttl': 60, 'type': type_}
            response = self.client.post(url, json.dumps(data), content_type='application/json')
            self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)

    def testCantPostRestrictedTypes(self):
        for type_ in self.restricted_types:
            url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
            data = {'records': ['ns1.desec.io. peter.desec.io. 2584 10800 3600 604800 60'], 'ttl': 60, 'type': type_}
            response = self.client.post(url, json.dumps(data), content_type='application/json')
            self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)

    def testCantPostForeignRRsets(self):
        url = reverse('v1:rrsets', args=(self.otherDomains[1].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)

    def testCantPostTwiceRRsets(self):
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': ['3.2.2.1'], 'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_409_CONFLICT)

    def testCantPostFaultyRRsets(self):
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))

        # New record without a value is a syntactical error --> 400
        data = {'records': [], 'ttl': 60, 'type': 'TXT'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)

        # Lower-case type is a syntactical error --> 400
        data = {'records': ['123456'], 'ttl': 60, 'type': 'txt'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)

        # Unknown type is a semantical error --> 422
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': ['123456'], 'ttl': 60, 'type': 'AA'}
        httpretty.register_uri(httpretty.PATCH, settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + '.',
                               body='', status=422)
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_422_UNPROCESSABLE_ENTITY)

    def testCanGetOwnRRset(self):
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        url = reverse('v1:rrset', args=(self.ownedDomains[1].name, '', 'A',))
        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(response.data['records'][0], '1.2.3.4')
        self.assertEqual(response.data['ttl'], 60)

    def testCanGetOwnRRsetApex(self):
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        url = reverse('v1:rrset@', args=(self.ownedDomains[1].name, '', 'A',))
        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(response.data['records'][0], '1.2.3.4')
        self.assertEqual(response.data['ttl'], 60)

    def testCantGetRestrictedTypes(self):
        for type_ in self.restricted_types:
            url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
            response = self.client.get(url + '?type=%s' % type_)
            self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)

            url = reverse('v1:rrset', args=(self.ownedDomains[1].name, '', type_,))
            response = self.client.get(url)
            self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)

    def testCantGetForeignRRset(self):
        self.client.credentials(HTTP_AUTHORIZATION='Token ' + self.otherToken)
        url = reverse('v1:rrsets', args=(self.otherDomains[0].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        self.client.credentials(HTTP_AUTHORIZATION='Token ' + self.token)
        url = reverse('v1:rrset', args=(self.otherDomains[0].name, '', 'A',))
        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)

    def testCanGetOwnRRsetWithSubname(self):
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))

        data = {'records': ['1.2.3.4'], 'ttl': 120, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        data = {'records': ['2.2.3.4'], 'ttl': 120, 'type': 'A', 'subname': 'test'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        data = {'records': ['"test"'], 'ttl': 120, 'type': 'TXT', 'subname': 'test'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(len(response.data), 3, response.data)  # NS RRset unavailable in mock pdns environment

        url = reverse('v1:rrset', args=(self.ownedDomains[1].name, 'test', 'A',))
        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(response.data['records'][0], '2.2.3.4')
        self.assertEqual(response.data['ttl'], 120)
        self.assertEqual(response.data['name'], 'test.' + self.ownedDomains[1].name + '.')

    def testCanGetOwnRRsetWithWildcard(self):
        for subname in ('*', '*.foobar'):
            url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))

            data = {'records': ['"barfoo"'], 'ttl': 120, 'type': 'TXT', 'subname': subname}
            response = self.client.post(url, json.dumps(data), content_type='application/json')
            self.assertEqual(response.status_code, status.HTTP_201_CREATED)

            response1 = self.client.get(url + '?subname=' + subname)
            self.assertEqual(response1.status_code, status.HTTP_200_OK)
            self.assertEqual(response1.data[0]['records'][0], '"barfoo"')
            self.assertEqual(response1.data[0]['ttl'], 120)
            self.assertEqual(response1.data[0]['name'], subname + '.' + self.ownedDomains[1].name + '.')

            url = reverse('v1:rrset', args=(self.ownedDomains[1].name, subname, 'TXT',))
            response2 = self.client.get(url)
            self.assertEqual(response2.data, response1.data[0])

    def testCanPutOwnRRset(self):
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        url = reverse('v1:rrset', args=(self.ownedDomains[1].name, '', 'A',))

        data = {'records': ['2.2.3.4'], 'ttl': 30}
        response = self.client.put(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_200_OK)

        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(response.data['records'][0], '2.2.3.4')
        self.assertEqual(response.data['ttl'], 30)

        data = {'records': ['3.2.3.4']}
        response = self.client.put(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)

        data = {'ttl': 37}
        response = self.client.put(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)

    def testCanPutOwnRRsetApex(self):
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        url = reverse('v1:rrset@', args=(self.ownedDomains[1].name, '', 'A',))

        data = {'records': ['2.2.3.4'], 'ttl': 30}
        response = self.client.put(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_200_OK)

        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(response.data['records'][0], '2.2.3.4')
        self.assertEqual(response.data['ttl'], 30)

        data = {'records': ['3.2.3.4']}
        response = self.client.put(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)

        data = {'ttl': 37}
        response = self.client.put(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)

    def testCanPatchOwnRRset(self):
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        # Change records and TTL
        url = reverse('v1:rrset', args=(self.ownedDomains[1].name, '', 'A',))
        data = {'records': ['3.2.3.4'], 'ttl': 32}
        response = self.client.patch(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_200_OK)

        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(response.data['records'][0], '3.2.3.4')
        self.assertEqual(response.data['ttl'], 32)

        # Change records alone
        data = {'records': ['5.2.3.4']}
        response = self.client.patch(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(response.data['records'][0], '5.2.3.4')
        self.assertEqual(response.data['ttl'], 32)

        # Change TTL alone
        data = {'ttl': 37}
        response = self.client.patch(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(response.data['records'][0], '5.2.3.4')
        self.assertEqual(response.data['ttl'], 37)

        # Change nothing
        data = {}
        response = self.client.patch(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(response.data['records'][0], '5.2.3.4')
        self.assertEqual(response.data['ttl'], 37)

    def testCanPatchOwnRRsetApex(self):
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        # Change records and TTL
        url = reverse('v1:rrset@', args=(self.ownedDomains[1].name, '', 'A',))
        data = {'records': ['3.2.3.4'], 'ttl': 32}
        response = self.client.patch(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_200_OK)

        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(response.data['records'][0], '3.2.3.4')
        self.assertEqual(response.data['ttl'], 32)

        # Change records alone
        data = {'records': ['5.2.3.4']}
        response = self.client.patch(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(response.data['records'][0], '5.2.3.4')
        self.assertEqual(response.data['ttl'], 32)

        # Change TTL alone
        data = {'ttl': 37}
        response = self.client.patch(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(response.data['records'][0], '5.2.3.4')
        self.assertEqual(response.data['ttl'], 37)

        # Change nothing
        data = {}
        response = self.client.patch(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(response.data['records'][0], '5.2.3.4')
        self.assertEqual(response.data['ttl'], 37)

    def testCantChangeForeignRRset(self):
        self.client.credentials(HTTP_AUTHORIZATION='Token ' + self.otherToken)
        url = reverse('v1:rrsets', args=(self.otherDomains[0].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        self.client.credentials(HTTP_AUTHORIZATION='Token ' + self.token)
        url = reverse('v1:rrset', args=(self.otherDomains[0].name, '', 'A',))
        data = {'records': ['3.2.3.4'], 'ttl': 30, 'type': 'A'}

        response = self.client.patch(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)

        response = self.client.put(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)

    def testCantChangeForeignRRsetApex(self):
        self.client.credentials(HTTP_AUTHORIZATION='Token ' + self.otherToken)
        url = reverse('v1:rrsets', args=(self.otherDomains[0].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        self.client.credentials(HTTP_AUTHORIZATION='Token ' + self.token)
        url = reverse('v1:rrset@', args=(self.otherDomains[0].name, '', 'A',))
        data = {'records': ['3.2.3.4'], 'ttl': 30, 'type': 'A'}

        response = self.client.patch(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)

        response = self.client.put(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)

    def testCantChangeEssentialProperties(self):
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A', 'subname': 'test1'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        # Changing the subname is expected to cause an error
        url = reverse('v1:rrset', args=(self.ownedDomains[1].name, 'test1', 'A',))
        data = {'records': ['3.2.3.4'], 'ttl': 120, 'subname': 'test2'}
        response = self.client.patch(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
        response = self.client.put(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)

        # Changing the type is expected to cause an error
        data = {'records': ['3.2.3.4'], 'ttl': 120, 'type': 'TXT'}
        response = self.client.patch(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
        response = self.client.put(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)

        # Check that nothing changed
        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(response.data['records'][0], '1.2.3.4')
        self.assertEqual(response.data['ttl'], 60)
        self.assertEqual(response.data['name'], 'test1.' + self.ownedDomains[1].name + '.')
        self.assertEqual(response.data['subname'], 'test1')
        self.assertEqual(response.data['type'], 'A')

        # This is expected to work, but the fields are ignored
        data = {'records': ['3.2.3.4'], 'name': 'example.com.', 'domain': 'example.com'}
        response = self.client.patch(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_200_OK)

        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(response.data['records'][0], '3.2.3.4')
        self.assertEqual(response.data['domain'], self.ownedDomains[1].name)
        self.assertEqual(response.data['name'], 'test1.' + self.ownedDomains[1].name + '.')

    def testCanDeleteOwnRRset(self):
        # Try PATCH with empty records
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        url = reverse('v1:rrset', args=(self.ownedDomains[1].name, '', 'A',))
        data = {'records': []}
        response = self.client.patch(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)

        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)

        # Try DELETE
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        url = reverse('v1:rrset', args=(self.ownedDomains[1].name, '', 'A',))
        response = self.client.delete(url)
        self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)

        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)

    def testCanDeleteOwnRRsetApex(self):
        # Try PATCH with empty records
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        url = reverse('v1:rrset@', args=(self.ownedDomains[1].name, '', 'A',))
        data = {'records': []}
        response = self.client.patch(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)

        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)

        # Try DELETE
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        url = reverse('v1:rrset@', args=(self.ownedDomains[1].name, '', 'A',))
        response = self.client.delete(url)
        self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)

        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)

    def testCantDeleteForeignRRset(self):
        self.client.credentials(HTTP_AUTHORIZATION='Token ' + self.otherToken)
        url = reverse('v1:rrsets', args=(self.otherDomains[0].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        self.client.credentials(HTTP_AUTHORIZATION='Token ' + self.token)
        url = reverse('v1:rrset', args=(self.otherDomains[0].name, '', 'A',))

        # Try PATCH with empty records
        data = {'records': []}
        response = self.client.patch(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)

        # Try DELETE
        response = self.client.delete(url)
        self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)

        # Make sure it actually is still there
        self.client.credentials(HTTP_AUTHORIZATION='Token ' + self.otherToken)
        url = reverse('v1:rrset@', args=(self.otherDomains[0].name, '', 'A',))
        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(response.data['records'][0], '1.2.3.4')

    def testCantCreateRRsetWhileAccountIsLocked(self):
        self.owner.locked = timezone.now()
        self.owner.save()

        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)

    def testCantModifyRRsetWhileAccountIsLocked(self):
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        self.owner.locked = timezone.now()
        self.owner.save()

        url = reverse('v1:rrset', args=(self.ownedDomains[1].name, '', 'A',))

        # Try PUT
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        response = self.client.patch(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)

        # Try PATCH
        data = {'records': ['4.3.2.1']}
        response = self.client.patch(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)

        # Try PATCH to delete
        data = {'records': []}
        response = self.client.patch(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)

        # Try DELETE
        response = self.client.delete(url)
        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)

    def testPostCausesPdnsAPICall(self):
        httpretty.enable(allow_net_connect=False)
        httpretty.register_uri(httpretty.PATCH, settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + '.')
        httpretty.register_uri(httpretty.PUT, settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + './notify')

        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        self.client.post(url, json.dumps(data), content_type='application/json')

        result = json.loads(httpretty.httpretty.latest_requests[-2].parsed_body)
        self.assertEqual(result['rrsets'][0]['name'], self.ownedDomains[1].name + '.')
        self.assertEqual(result['rrsets'][0]['records'][0]['content'], '1.2.3.4')
        self.assertEqual(httpretty.last_request().method, 'PUT')

    def testDeleteCausesPdnsAPICall(self):
        httpretty.enable(allow_net_connect=False)
        httpretty.register_uri(httpretty.PATCH, settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + '.')
        httpretty.register_uri(httpretty.PUT, settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + './notify')

        # Create record, should cause a pdns PATCH request and a notify
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        self.client.post(url, json.dumps(data), content_type='application/json')

        # Delete record, should cause a pdns PATCH request and a notify
        url = reverse('v1:rrset', args=(self.ownedDomains[1].name, '', 'A',))
        response = self.client.delete(url)
        self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)

        # Check pdns requests from creation
        result = json.loads(httpretty.httpretty.latest_requests[-4].parsed_body)
        self.assertEqual(result['rrsets'][0]['name'], self.ownedDomains[1].name + '.')
        self.assertEqual(result['rrsets'][0]['records'][0]['content'], '1.2.3.4')
        self.assertEqual(httpretty.httpretty.latest_requests[-3].method, 'PUT')

        # Check pdns requests from deletion
        result = json.loads(httpretty.httpretty.latest_requests[-2].parsed_body)
        self.assertEqual(result['rrsets'][0]['name'], self.ownedDomains[1].name + '.')
        self.assertEqual(result['rrsets'][0]['records'], [])
        self.assertEqual(httpretty.httpretty.latest_requests[-1].method, 'PUT')

    def testImportRRsets(self):
        url = reverse('v1:rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        response = self.client.post(url, json.dumps(data), content_type='application/json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        # Not checking anything here; errors will raise an exception
        httpretty.register_uri(httpretty.GET, settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + '.',
                               status=200, body='{"rrsets":[{"name":"asdf","type":"A",' +
                                                '"records":[{"content":"1.1.1.1"},{"content":"2.2.2.2"}],"ttl":20}]}')
        call_command('sync-from-pdns', self.ownedDomains[1].name)