Browse Source

[1372] added a higher level test for AXFR-style IXFR response with an actual
SQLite3 DB.

JINMEI Tatuya 13 years ago
parent
commit
ecd9c5fc4b
2 changed files with 33 additions and 5 deletions
  1. BIN
      src/bin/xfrout/tests/testdata/test.sqlite3
  2. 33 5
      src/bin/xfrout/tests/xfrout_test.py.in

BIN
src/bin/xfrout/tests/testdata/test.sqlite3


+ 33 - 5
src/bin/xfrout/tests/xfrout_test.py.in

@@ -815,19 +815,47 @@ class TestXfroutSessionWithSQLite3(TestXfroutSessionBase):
         self.xfrsess._server.get_db_file = lambda : TESTDATA_SRCDIR + \
             'test.sqlite3'
 
-    def test_axfr_normal_session(self):
-        XfroutSession._handle(self.xfrsess)
-        response = self.sock.read_msg(Message.PRESERVE_ORDER);
-        self.assertEqual(Rcode.NOERROR(), response.get_rcode())
+    def check_axfr_stream(self, response):
+        '''Common checks for AXFR(-style) response for the test zone.
+        '''
         # This zone contains two A RRs for the same name with different TTLs.
         # These TTLs should be preseved in the AXFR stream.
+        # We'll check some important points as a valid AXFR response:
+        # the first and last RR must be SOA, and these should be the only
+        # SOAs in the response.  The total number of response RRs
+        # must be 5 (zone has 4 RRs, SOA is duplicated)
+        actual_records = response.get_section(Message.SECTION_ANSWER)
+        self.assertEqual(5, len(actual_records))
+        self.assertEqual(RRType.SOA(), actual_records[0].get_type())
+        self.assertEqual(RRType.SOA(), actual_records[-1].get_type())
         actual_ttls = []
-        for rr in response.get_section(Message.SECTION_ANSWER):
+        num_soa = 0
+        for rr in actual_records:
+            if rr.get_type() == RRType.SOA():
+                num_soa += 1
             if rr.get_type() == RRType.A() and \
                     not rr.get_ttl() in actual_ttls:
                 actual_ttls.append(rr.get_ttl().get_value())
+        self.assertEqual(2, num_soa)
         self.assertEqual([3600, 7200], sorted(actual_ttls))
 
+    def test_axfr_normal_session(self):
+        XfroutSession._handle(self.xfrsess)
+        response = self.sock.read_msg(Message.PRESERVE_ORDER);
+        self.assertEqual(Rcode.NOERROR(), response.get_rcode())
+        self.check_axfr_stream(response)
+
+    def test_ixfr_to_axfr(self):
+        self.xfrsess._request_data = \
+            self.create_request_data(ixfr=IXFR_NG_VERSION)
+        XfroutSession._handle(self.xfrsess)
+        response = self.sock.read_msg(Message.PRESERVE_ORDER);
+        self.assertEqual(Rcode.NOERROR(), response.get_rcode())
+        # This is an AXFR-style IXFR.  So the question section should indicate
+        # that it's an IXFR resposne.
+        self.assertEqual(RRType.IXFR(), response.get_question()[0].get_type())
+        self.check_axfr_stream(response)
+
 class MyUnixSockServer(UnixSockServer):
     def __init__(self):
         self._shutdown_event = threading.Event()