]>
Commit | Line | Data |
---|---|---|
55baa1f2 RG |
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 | ||
5df86a8a | 21 | cls._UDPResponder = threading.Thread(name='UDP Responder', target=cls.UDPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue, True]) |
55baa1f2 RG |
22 | cls._UDPResponder.setDaemon(True) |
23 | cls._UDPResponder.start() | |
5df86a8a RG |
24 | |
25 | cls._TCPResponder = threading.Thread(name='TCP Responder', target=cls.TCPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue, True]) | |
55baa1f2 RG |
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 + '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 + '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) |