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, cls._toResponderQueue, cls._fromResponderQueue, True])
  22         cls._UDPResponder.setDaemon(True)
  23         cls._UDPResponder.start()
  24
  25         cls._TCPResponder = threading.Thread(name='TCP Responder', target=cls.TCPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue, True])
  26         cls._TCPResponder.setDaemon(True)
  27         cls._TCPResponder.start()
  28
  29     def testTrailingAllowed(self):
  30         """
  31         Trailing: Allowed
  32
  33         """
  34         name = 'allowed.trailing.tests.powerdns.com.'
  35         query = dns.message.make_query(name, 'A', 'IN')
  36         response = dns.message.make_response(query)
  37         rrset = dns.rrset.from_text(name,
  38                                     3600,
  39                                     dns.rdataclass.IN,
  40                                     dns.rdatatype.A,
  41                                     '127.0.0.1')
  42         response.answer.append(rrset)
  43
  44         raw = query.to_wire()
  45         raw = raw + b'A'* 20
  46         (receivedQuery, receivedResponse) = self.sendUDPQuery(raw, response, rawQuery=True)
  47         self.assertTrue(receivedQuery)
  48         self.assertTrue(receivedResponse)
  49         receivedQuery.id = query.id
  50         self.assertEquals(query, receivedQuery)
  51         self.assertEquals(response, receivedResponse)
  52
  53         (receivedQuery, receivedResponse) = self.sendTCPQuery(raw, response, rawQuery=True)
  54         self.assertTrue(receivedQuery)
  55         self.assertTrue(receivedResponse)
  56         receivedQuery.id = query.id
  57         self.assertEquals(query, receivedQuery)
  58         self.assertEquals(response, receivedResponse)
  59
  60     def testTrailingDropped(self):
  61         """
  62         Trailing: dropped
  63
  64         """
  65         name = 'dropped.trailing.tests.powerdns.com.'
  66         query = dns.message.make_query(name, 'AAAA', 'IN')
  67
  68         raw = query.to_wire()
  69         raw = raw + b'A'* 20
  70
  71         (_, receivedResponse) = self.sendUDPQuery(raw, response=None, rawQuery=True)
  72         self.assertEquals(receivedResponse, None)
  73         (_, receivedResponse) = self.sendTCPQuery(raw, response=None, rawQuery=True)
  74         self.assertEquals(receivedResponse, None)