Limit the number of queries sent out to get NS addresses per query.

[thirdparty/pdns.git] / regression-tests.ixfrdist / test_IXFR.py

diff --git a/regression-tests.ixfrdist/test_IXFR.py b/regression-tests.ixfrdist/test_IXFR.py
index 8f99316dda3c40ad7f6c096202b2397b12513116..986a7413853877d9db045d8e5ffa627c46797554 100644 (file)
--- a/regression-tests.ixfrdist/test_IXFR.py
+++ b/regression-tests.ixfrdist/test_IXFR.py
@@ -35,7 +35,10 @@ class IXFRDistBasicTest(IXFRDistTest):
 
     global xfrServerPort
     _xfrDone = 0
-    _config_domains = { 'example': '127.0.0.1:' + str(xfrServerPort) }
+    _config_domains = { 'example': '127.0.0.1:' + str(xfrServerPort),   # zone for actual XFR testing
+                        'example2': '127.0.0.1:1',                      # bogus port is intentional - zone is intentionally unloadable
+                        # example3                                      # intentionally absent for 'unconfigured zone' testing
+                        'example4': '127.0.0.1:' + str(xfrServerPort) } # for testing how ixfrdist deals with getting the wrong zone on XFR
 
     @classmethod
     def setUpClass(cls):
@@ -103,12 +106,50 @@ class IXFRDistBasicTest(IXFRDistTest):
 
         # answers[1].sort(key=lambda rrset: (rrset.name, rrset.rdtype))
         self.assertEqual(answers, expected)
-
-    def testXFR(self):
+        # check the TTLs
+        answerPos = 0
+        for expectedAnswer in expected:
+            pos = 0
+            for rec in expectedAnswer:
+                self.assertEquals(rec.ttl, answers[answerPos][pos].ttl)
+                pos = pos + 1
+            answerPos = answerPos + 1
+
+    def test_a_XFR(self):
         self.waitUntilCorrectSerialIsLoaded(1)
         self.checkFullZone(1)
 
         self.waitUntilCorrectSerialIsLoaded(2)
         self.checkFullZone(2)
 
-        self.checkIXFR(1,2)
\ No newline at end of file
+        self.checkIXFR(1,2)
+
+    # _b_ because we expect post-XFR testing state
+    def test_b_UDP_SOA_existing(self):
+        query = dns.message.make_query('example.', 'SOA')
+        expected = dns.message.make_response(query)
+        expected.answer.append(xfrServer._getSOAForSerial(2))
+
+        response = self.sendUDPQuery(query)
+        self.assertEquals(expected, response)
+        # check the TTLs
+        pos = 0
+        for rec in expected.answer:
+            self.assertEquals(rec.ttl, response.answer[pos].ttl)
+            pos = pos + 1
+
+    def test_b_UDP_SOA_not_loaded(self):
+        query = dns.message.make_query('example2.', 'SOA')
+        expected = dns.message.make_response(query)
+        expected.set_rcode(dns.rcode.REFUSED)
+
+        response = self.sendUDPQuery(query)
+        self.assertEquals(expected, response)
+
+    def test_b_UDP_SOA_not_configured(self):
+        query = dns.message.make_query('example3.', 'SOA')
+        expected = dns.message.make_response(query)
+        expected.set_rcode(dns.rcode.REFUSED)
+
+        response = self.sendUDPQuery(query)
+        self.assertEquals(expected, response)