]>
git.ipfire.org Git - thirdparty/pdns.git/blob - regression-tests.dnsdist/test_AXFR.py
4 from dnsdisttests
import DNSDistTest
6 class TestAXFR(DNSDistTest
):
8 # this test suite uses a different responder port
9 # because, contrary to the other ones, its
10 # TCP responder allows multiple responses and we don't want
12 _testServerPort
= 5370
13 _config_template
= """
14 newServer{address="127.0.0.1:%s"}
17 def startResponders(cls
):
18 print("Launching responders..")
20 cls
._UDPResponder
= threading
.Thread(name
='UDP Responder', target
=cls
.UDPResponder
, args
=[cls
._testServerPort
, cls
._toResponderQueue
, cls
._fromResponderQueue
])
21 cls
._UDPResponder
.setDaemon(True)
22 cls
._UDPResponder
.start()
23 cls
._TCPResponder
= threading
.Thread(name
='TCP Responder', target
=cls
.TCPResponder
, args
=[cls
._testServerPort
, cls
._toResponderQueue
, cls
._fromResponderQueue
, False, True])
24 cls
._TCPResponder
.setDaemon(True)
25 cls
._TCPResponder
.start()
27 def testOneMessageAXFR(self
):
31 name
= 'one.axfr.tests.powerdns.com.'
32 query
= dns
.message
.make_query(name
, 'AXFR', 'IN')
33 response
= dns
.message
.make_response(query
)
34 soa
= dns
.rrset
.from_text(name
,
38 'ns.' + name
+ ' hostmaster.' + name
+ ' 1 3600 3600 3600 60')
39 response
.answer
.append(soa
)
40 response
.answer
.append(dns
.rrset
.from_text(name
,
45 response
.answer
.append(soa
)
47 (receivedQuery
, receivedResponse
) = self
.sendTCPQuery(query
, response
)
48 receivedQuery
.id = query
.id
49 self
.assertEqual(query
, receivedQuery
)
50 self
.assertEqual(response
, receivedResponse
)
52 def testOneNoSOAAXFR(self
):
54 AXFR: One message, no SOA
56 name
= 'onenosoa.axfr.tests.powerdns.com.'
57 query
= dns
.message
.make_query(name
, 'AXFR', 'IN')
58 response
= dns
.message
.make_response(query
)
59 response
.answer
.append(dns
.rrset
.from_text(name
,
65 (receivedQuery
, receivedResponse
) = self
.sendTCPQuery(query
, response
)
66 receivedQuery
.id = query
.id
67 self
.assertEqual(query
, receivedQuery
)
68 self
.assertEqual(response
, receivedResponse
)
70 def testFourMessagesAXFR(self
):
74 name
= 'four.axfr.tests.powerdns.com.'
75 query
= dns
.message
.make_query(name
, 'AXFR', 'IN')
77 soa
= dns
.rrset
.from_text(name
,
81 'ns.' + name
+ ' hostmaster.' + name
+ ' 1 3600 3600 3600 60')
82 response
= dns
.message
.make_response(query
)
83 response
.answer
.append(soa
)
84 responses
.append(response
)
86 response
= dns
.message
.make_response(query
)
87 response
.answer
.append(dns
.rrset
.from_text(name
,
92 responses
.append(response
)
94 response
= dns
.message
.make_response(query
)
95 rrset
= dns
.rrset
.from_text(name
,
100 response
.answer
.append(rrset
)
101 responses
.append(response
)
103 response
= dns
.message
.make_response(query
)
104 rrset
= dns
.rrset
.from_text(name
,
109 response
.answer
.append(rrset
)
110 responses
.append(response
)
112 response
= dns
.message
.make_response(query
)
113 response
.answer
.append(soa
)
114 responses
.append(response
)
116 (receivedQuery
, receivedResponses
) = self
.sendTCPQueryWithMultipleResponses(query
, responses
)
117 receivedQuery
.id = query
.id
118 self
.assertEqual(query
, receivedQuery
)
119 self
.assertEqual(len(receivedResponses
), len(responses
))
121 def testFourNoFinalSOAAXFR(self
):
123 AXFR: Four messages, no final SOA
125 name
= 'fournosoa.axfr.tests.powerdns.com.'
126 query
= dns
.message
.make_query(name
, 'AXFR', 'IN')
128 soa
= dns
.rrset
.from_text(name
,
132 'ns.' + name
+ ' hostmaster.' + name
+ ' 1 3600 3600 3600 60')
133 response
= dns
.message
.make_response(query
)
134 response
.answer
.append(soa
)
135 responses
.append(response
)
137 response
= dns
.message
.make_response(query
)
138 response
.answer
.append(dns
.rrset
.from_text(name
,
143 responses
.append(response
)
145 response
= dns
.message
.make_response(query
)
146 rrset
= dns
.rrset
.from_text(name
,
151 response
.answer
.append(rrset
)
152 responses
.append(response
)
154 response
= dns
.message
.make_response(query
)
155 rrset
= dns
.rrset
.from_text(name
,
160 response
.answer
.append(rrset
)
161 responses
.append(response
)
163 (receivedQuery
, receivedResponses
) = self
.sendTCPQueryWithMultipleResponses(query
, responses
)
164 receivedQuery
.id = query
.id
165 self
.assertEqual(query
, receivedQuery
)
166 self
.assertEqual(len(receivedResponses
), len(responses
))
168 def testFourNoFirstSOAAXFR(self
):
170 AXFR: Four messages, no SOA in the first one
172 name
= 'fournosoainfirst.axfr.tests.powerdns.com.'
173 query
= dns
.message
.make_query(name
, 'AXFR', 'IN')
175 soa
= dns
.rrset
.from_text(name
,
179 'ns.' + name
+ ' hostmaster.' + name
+ ' 1 3600 3600 3600 60')
180 response
= dns
.message
.make_response(query
)
181 response
.answer
.append(dns
.rrset
.from_text(name
,
186 responses
.append(response
)
188 response
= dns
.message
.make_response(query
)
189 rrset
= dns
.rrset
.from_text(name
,
194 response
.answer
.append(soa
)
195 response
.answer
.append(rrset
)
196 responses
.append(response
)
198 response
= dns
.message
.make_response(query
)
199 rrset
= dns
.rrset
.from_text('dummy.' + name
,
204 response
.answer
.append(rrset
)
205 responses
.append(response
)
207 response
= dns
.message
.make_response(query
)
208 rrset
= dns
.rrset
.from_text(name
,
213 response
.answer
.append(rrset
)
214 response
.answer
.append(soa
)
215 responses
.append(response
)
217 (receivedQuery
, receivedResponses
) = self
.sendTCPQueryWithMultipleResponses(query
, responses
)
218 receivedQuery
.id = query
.id
219 self
.assertEqual(query
, receivedQuery
)
220 self
.assertEqual(len(receivedResponses
), 1)
222 def testFourLastSOAInSecondAXFR(self
):
224 AXFR: Four messages, SOA in the first one and the second one
226 name
= 'foursecondsoainsecond.axfr.tests.powerdns.com.'
227 query
= dns
.message
.make_query(name
, 'AXFR', 'IN')
229 soa
= dns
.rrset
.from_text(name
,
233 'ns.' + name
+ ' hostmaster.' + name
+ ' 1 3600 3600 3600 60')
235 response
= dns
.message
.make_response(query
)
236 response
.answer
.append(soa
)
237 response
.answer
.append(dns
.rrset
.from_text(name
,
242 responses
.append(response
)
244 response
= dns
.message
.make_response(query
)
245 response
.answer
.append(soa
)
246 rrset
= dns
.rrset
.from_text(name
,
251 response
.answer
.append(rrset
)
252 responses
.append(response
)
254 response
= dns
.message
.make_response(query
)
255 rrset
= dns
.rrset
.from_text('dummy.' + name
,
260 response
.answer
.append(rrset
)
261 responses
.append(response
)
263 response
= dns
.message
.make_response(query
)
264 rrset
= dns
.rrset
.from_text(name
,
269 response
.answer
.append(rrset
)
270 responses
.append(response
)
272 (receivedQuery
, receivedResponses
) = self
.sendTCPQueryWithMultipleResponses(query
, responses
)
273 receivedQuery
.id = query
.id
274 self
.assertEqual(query
, receivedQuery
)
275 self
.assertEqual(len(receivedResponses
), 2)