regression-tests.dnsdist/test_Trailing.py

   1 #!/usr/bin/env python
   2 import threading
   3 import dns
   4 from dnsdisttests import DNSDistTest
   5
   6 class TestTrailing(DNSDistTest):
   7
   8     # this test suite uses a different responder port
   9     # because, contrary to the other ones, its
  10     # responders allow trailing data and we don't want
  11     # to mix things up.
  12     _testServerPort = 5360
  13     _config_template = """
  14     newServer{address="127.0.0.1:%s"}
  15     addAction(AndRule({QTypeRule(dnsdist.AAAA), TrailingDataRule()}), DropAction())
  16     """
  17     @classmethod
  18     def startResponders(cls):
  19         print("Launching responders..")
  20
  21         cls._UDPResponder = threading.Thread(name='UDP Responder', target=cls.UDPResponder, args=[cls._testServerPort, True])
  22         cls._UDPResponder.setDaemon(True)
  23         cls._UDPResponder.start()
  24         cls._TCPResponder = threading.Thread(name='TCP Responder', target=cls.TCPResponder, args=[cls._testServerPort, True])
  25         cls._TCPResponder.setDaemon(True)
  26         cls._TCPResponder.start()
  27
  28     def testTrailingAllowed(self):
  29         """
  30         Trailing: Allowed
  31
  32         """
  33         name = 'allowed.trailing.tests.powerdns.com.'
  34         query = dns.message.make_query(name, 'A', 'IN')
  35         response = dns.message.make_response(query)
  36         rrset = dns.rrset.from_text(name,
  37                                     3600,
  38                                     dns.rdataclass.IN,
  39                                     dns.rdatatype.A,
  40                                     '127.0.0.1')
  41         response.answer.append(rrset)
  42
  43         raw = query.to_wire()
  44         raw = raw + 'A'* 20
  45         (receivedQuery, receivedResponse) = self.sendUDPQuery(raw, response, rawQuery=True)
  46         self.assertTrue(receivedQuery)
  47         self.assertTrue(receivedResponse)
  48         receivedQuery.id = query.id
  49         self.assertEquals(query, receivedQuery)
  50         self.assertEquals(response, receivedResponse)
  51
  52         (receivedQuery, receivedResponse) = self.sendTCPQuery(raw, response, rawQuery=True)
  53         self.assertTrue(receivedQuery)
  54         self.assertTrue(receivedResponse)
  55         receivedQuery.id = query.id
  56         self.assertEquals(query, receivedQuery)
  57         self.assertEquals(response, receivedResponse)
  58
  59     def testTrailingDropped(self):
  60         """
  61         Trailing: dropped
  62
  63         """
  64         name = 'dropped.trailing.tests.powerdns.com.'
  65         query = dns.message.make_query(name, 'AAAA', 'IN')
  66
  67         raw = query.to_wire()
  68         raw = raw + 'A'* 20
  69
  70         (_, receivedResponse) = self.sendUDPQuery(raw, response=None, rawQuery=True)
  71         self.assertEquals(receivedResponse, None)
  72         (_, receivedResponse) = self.sendTCPQuery(raw, response=None, rawQuery=True)
  73         self.assertEquals(receivedResponse, None)