]>
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,
192 # dns.rdatatype.AAAA,
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,
202 # dns.rdatatype.AAAA,
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,
249 # dns.rdatatype.AAAA,
251 # response.answer.append(rrset)
252 # responses.append(response)
254 # response = dns.message.make_response(query)
255 # rrset = dns.rrset.from_text('dummy.' + name,
258 # dns.rdatatype.AAAA,
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)