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

import json

from rest_framework import status

from desecapi.tests.base import DesecTestCase


class ReplicationTest(DesecTestCase):
    def test_serials(self):
        url = self.reverse("v1:serial")
        zones = [
            {"name": "test.example.", "edited_serial": 12345},
            {"name": "example.org.", "edited_serial": 54321},
        ]
        serials = {zone["name"]: zone["edited_serial"] for zone in zones}
        pdns_requests = [
            {
                "method": "GET",
                "uri": self.get_full_pdns_url(r"/zones", ns="MASTER"),
                "status": 200,
                "body": json.dumps(zones),
            }
        ]

        # Run twice to make sure cache output varies on remote address
        for i in range(2):
            response = self.client.get(path=url, REMOTE_ADDR="123.8.0.2")
            self.assertStatus(response, status.HTTP_401_UNAUTHORIZED)

            with self.assertPdnsRequests(pdns_requests):
                response = self.client.get(path=url, REMOTE_ADDR="10.8.0.2")
            self.assertStatus(response, status.HTTP_200_OK)
            self.assertEqual(response.data, serials)

            # Do not expect pdns request in next iteration (result will be cached)
            pdns_requests = []