]>
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 | ||
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) |