regression-tests.recursor-dnssec/test_OOOTCP.py

   1 import dns
   2 import os
   3 import time
   4 from recursortests import RecursorTest
   5
   6 class testOOOTCP(RecursorTest):
   7     _confdir = 'OOOTCP'
   8
   9     _config_template = """dnssec=validate
  10 """
  11
  12     @classmethod
  13     def generateRecursorConfig(cls, confdir):
  14         super(testOOOTCP, cls).generateRecursorConfig(confdir)
  15
  16     def testOOOVeryBasic(self):
  17         expected = {}
  18         queries = []
  19         for zone in ['5.delay1.example.', '0.delay2.example.']:
  20             expected[zone] = dns.rrset.from_text(zone, 0, dns.rdataclass.IN, 'TXT', 'a')
  21             query = dns.message.make_query(zone, 'TXT', want_dnssec=True)
  22             query.flags |= dns.flags.AD
  23             queries.append(query)
  24
  25         ress = self.sendTCPQueries(queries)
  26
  27         self.assertEqual(len(ress), len(expected))
  28
  29         i = 0
  30         for exp in [expected['0.delay2.example.'], expected['5.delay1.example.']]:
  31             print('ress0')
  32             print(ress[i].answer[0].to_text())
  33             print('exp')
  34             print(exp.to_text())
  35             self.assertMessageIsAuthenticated(ress[i])
  36             self.assertRRsetInAnswer(ress[i], exp)
  37             self.assertMatchingRRSIGInAnswer(ress[i], exp)
  38             i = i + 1
  39
  40     def testOOOTimeout(self):
  41         expected = {}
  42         queries = []
  43         for zone in ['25.delay1.example.', '1.delay2.example.']:
  44             query = dns.message.make_query(zone, 'TXT', want_dnssec=True)
  45             query.flags |= dns.flags.AD
  46             queries.append(query)
  47
  48         ress = self.sendTCPQueries(queries)
  49
  50         self.assertEqual(len(ress), 2)
  51         exp = dns.rrset.from_text('1.delay2.example.', 0, dns.rdataclass.IN, 'TXT', 'a')
  52         self.assertRRsetInAnswer(ress[0], exp)
  53         self.assertMatchingRRSIGInAnswer(ress[0], exp)
  54         self.assertRcodeEqual(ress[1], dns.rcode.SERVFAIL)
  55
  56         # Let the auth timeout happen to not disturb other tests
  57         # this can happen if the auth is single-threaded
  58         time.sleep(1)
  59