third_party/twisted_8_1/twisted/names/test/test_names.py - Issue 12261012: Remove third_party/twisted_8_1

Side by Side Diff: third_party/twisted_8_1/twisted/names/test/test_names.py

Issue 12261012: Remove third_party/twisted_8_1 (Closed) Base URL: svn://svn.chromium.org/chrome/trunk/tools/build

Patch Set: Created 7 years, 10 months ago

Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.

Jump to:

View unified diff | Download patch | Annotate | Revision Log

OLD	NEW
	(Empty)
1 # -- test-case-name: twisted.names.test.test_names --

2 # Copyright (c) 2001-2004 Twisted Matrix Laboratories.

3 # See LICENSE for details.

4

5 """

6 Test cases for twisted.names.

7 """

8 from __future__ import nested_scopes

9

10 import socket, operator, copy

11

12 from twisted.trial import unittest

13

14 from twisted.internet import reactor, defer, error

15 from twisted.internet.defer import succeed

16 from twisted.names import client, server, common, authority, hosts, dns

17 from twisted.python import failure

18 from twisted.names.error import DNSFormatError, DNSServerError, DNSNameError

19 from twisted.names.error import DNSNotImplementedError, DNSQueryRefusedError

20 from twisted.names.error import DNSUnknownError

21 from twisted.names.dns import EFORMAT, ESERVER, ENAME, ENOTIMP, EREFUSED

22 from twisted.names.dns import Message

23 from twisted.names.client import Resolver

24

25

26 def justPayload(results):

27 return [r.payload for r in results[0]]

28

29 class NoFileAuthority(authority.FileAuthority):

30 def __init__(self, soa, records):

31 # Yes, skip FileAuthority

32 common.ResolverBase.__init__(self)

33 self.soa, self.records = soa, records

34

35

36 soa_record = dns.Record_SOA(

37 mname = 'test-domain.com',

38 rname = 'root.test-domain.com',

39 serial = 100,

40 refresh = 1234,

41 minimum = 7654,

42 expire = 19283784,

43 retry = 15,

44 ttl=1

45 )

46

47 reverse_soa = dns.Record_SOA(

48 mname = '93.84.28.in-addr.arpa',

49 rname = '93.84.28.in-addr.arpa',

50 serial = 120,

51 refresh = 54321,

52 minimum = 382,

53 expire = 11193983,

54 retry = 30,

55 ttl=3

56 )

57

58 my_soa = dns.Record_SOA(

59 mname = 'my-domain.com',

60 rname = 'postmaster.test-domain.com',

61 serial = 130,

62 refresh = 12345,

63 minimum = 1,

64 expire = 999999,

65 retry = 100,

66 )

67

68 test_domain_com = NoFileAuthority(

69 soa = ('test-domain.com', soa_record),

70 records = {

71 'test-domain.com': [

72 soa_record,

73 dns.Record_A('127.0.0.1'),

74 dns.Record_NS('39.28.189.39'),

75 dns.Record_MX(10, 'host.test-domain.com'),

76 dns.Record_HINFO(os='Linux', cpu='A Fast One, Dontcha know'),

77 dns.Record_CNAME('canonical.name.com'),

78 dns.Record_MB('mailbox.test-domain.com'),

79 dns.Record_MG('mail.group.someplace'),

80 dns.Record_TXT('A First piece of Text', 'a SecoNd piece'),

81 dns.Record_A6(0, 'ABCD::4321', ''),

82 dns.Record_A6(12, '0:0069::0', 'some.network.tld'),

83 dns.Record_A6(8, '0:5634:1294:AFCB:56AC:48EF:34C3:01FF', 'tra.la.la. net'),

84 dns.Record_TXT('Some more text, haha! Yes. \0 Still here?'),

85 dns.Record_MR('mail.redirect.or.whatever'),

86 dns.Record_MINFO(rmailbx='r mail box', emailbx='e mail box'),

87 dns.Record_AFSDB(subtype=1, hostname='afsdb.test-domain.com'),

88 dns.Record_RP(mbox='whatever.i.dunno', txt='some.more.text'),

89 dns.Record_WKS('12.54.78.12', socket.IPPROTO_TCP, '\x12\x01\x16\xfe\ xc1\x00\x01'),

90 dns.Record_AAAA('AF43:5634:1294:AFCB:56AC:48EF:34C3:01FF')],

91 'http.tcp.test-domain.com': [

92 dns.Record_SRV(257, 16383, 43690, 'some.other.place.fool')

93 ],

94 'host.test-domain.com': [

95 dns.Record_A('123.242.1.5'),

96 dns.Record_A('0.255.0.255'),

97 ],

98 'host-two.test-domain.com': [

99 #

100 # Python bug

101 # dns.Record_A('255.255.255.255'),

102 #

103 dns.Record_A('255.255.255.254'),

104 dns.Record_A('0.0.0.0')

105 ],

106 'cname.test-domain.com': [

107 dns.Record_CNAME('test-domain.com')

108 ],

109 'anothertest-domain.com': [

110 dns.Record_A('1.2.3.4')],

111 }

112 )

113

114 reverse_domain = NoFileAuthority(

115 soa = ('93.84.28.in-addr.arpa', reverse_soa),

116 records = {

117 '123.93.84.28.in-addr.arpa': [

118 dns.Record_PTR('test.host-reverse.lookup.com'),

119 reverse_soa

120 ]

121 }

122 )

123

124

125 my_domain_com = NoFileAuthority(

126 soa = ('my-domain.com', my_soa),

127 records = {

128 'my-domain.com': [

129 my_soa,

130 dns.Record_A('1.2.3.4', ttl='1S'),

131 dns.Record_NS('ns1.domain', ttl='2M'),

132 dns.Record_NS('ns2.domain', ttl='3H'),

133 dns.Record_SRV(257, 16383, 43690, 'some.other.place.fool', ttl='4D')

134 ]

135 }

136 )

137

138

139 class ServerDNSTestCase(unittest.TestCase):

140 """

141 Test cases for DNS server and client.

142 """

143

144 def setUp(self):

145 self.factory = server.DNSServerFactory([

146 test_domain_com, reverse_domain, my_domain_com

147 ], verbose=2)

148

149 p = dns.DNSDatagramProtocol(self.factory)

150

151 while 1:

152 self.listenerTCP = reactor.listenTCP(0, self.factory, interface="127 .0.0.1")

153 port = self.listenerTCP.getHost().port

154

155 try:

156 self.listenerUDP = reactor.listenUDP(port, p, interface="127.0.0 .1")

157 except error.CannotListenError:

158 self.listenerTCP.stopListening()

159 else:

160 break

161

162 self.resolver = client.Resolver(servers=[('127.0.0.1', port)])

163

164

165 def tearDown(self):

166 """

167 Asynchronously disconnect listenerTCP, listenerUDP and resolver.

168 """

169 d1 = self.listenerTCP.loseConnection()

170 d2 = defer.maybeDeferred(self.listenerUDP.stopListening)

171 d = defer.gatherResults([d1, d2])

172 def disconnectTransport(ignored):

173 if getattr(self.resolver.protocol, 'transport', None) is not None:

174 return self.resolver.protocol.transport.stopListening()

175 d.addCallback(disconnectTransport)

176 d.addCallback(lambda x : self.failUnless(

177 self.listenerUDP.disconnected

178 and self.listenerTCP.disconnected))

179 return d

180

181

182 def namesTest(self, d, r):

183 self.response = None

184 def setDone(response):

185 self.response = response

186

187 def checkResults(ignored):

188 if isinstance(self.response, failure.Failure):

189 raise self.response

190 results = justPayload(self.response)

191 assert len(results) == len(r), "%s != %s" % (map(str, results), map( str, r))

192 for rec in results:

193 assert rec in r, "%s not in %s" % (rec, map(str, r))

194

195 d.addBoth(setDone)

196 d.addCallback(checkResults)

197 return d

198

199 def testAddressRecord1(self):

200 """Test simple DNS 'A' record queries"""

201 return self.namesTest(

202 self.resolver.lookupAddress('test-domain.com'),

203 [dns.Record_A('127.0.0.1', ttl=19283784)]

204 )

205

206

207 def testAddressRecord2(self):

208 """Test DNS 'A' record queries with multiple answers"""

209 return self.namesTest(

210 self.resolver.lookupAddress('host.test-domain.com'),

211 [dns.Record_A('123.242.1.5', ttl=19283784), dns.Record_A('0.255.0.25 5', ttl=19283784)]

212 )

213

214

215 def testAdressRecord3(self):

216 """Test DNS 'A' record queries with edge cases"""

217 return self.namesTest(

218 self.resolver.lookupAddress('host-two.test-domain.com'),

219 [dns.Record_A('255.255.255.254', ttl=19283784), dns.Record_A('0.0.0. 0', ttl=19283784)]

220 )

221

222 def testAuthority(self):

223 """Test DNS 'SOA' record queries"""

224 return self.namesTest(

225 self.resolver.lookupAuthority('test-domain.com'),

226 [soa_record]

227 )

228

229

230 def testMailExchangeRecord(self):

231 """Test DNS 'MX' record queries"""

232 return self.namesTest(

233 self.resolver.lookupMailExchange('test-domain.com'),

234 [dns.Record_MX(10, 'host.test-domain.com', ttl=19283784)]

235 )

236

237

238 def testNameserver(self):

239 """Test DNS 'NS' record queries"""

240 return self.namesTest(

241 self.resolver.lookupNameservers('test-domain.com'),

242 [dns.Record_NS('39.28.189.39', ttl=19283784)]

243 )

244

245

246 def testHINFO(self):

247 """Test DNS 'HINFO' record queries"""

248 return self.namesTest(

249 self.resolver.lookupHostInfo('test-domain.com'),

250 [dns.Record_HINFO(os='Linux', cpu='A Fast One, Dontcha know', ttl=19 283784)]

251 )

252

253 def testPTR(self):

254 """Test DNS 'PTR' record queries"""

255 return self.namesTest(

256 self.resolver.lookupPointer('123.93.84.28.in-addr.arpa'),

257 [dns.Record_PTR('test.host-reverse.lookup.com', ttl=11193983)]

258 )

259

260

261 def testCNAME(self):

262 """Test DNS 'CNAME' record queries"""

263 return self.namesTest(

264 self.resolver.lookupCanonicalName('test-domain.com'),

265 [dns.Record_CNAME('canonical.name.com', ttl=19283784)]

266 )

267

268 def testCNAMEAdditional(self):

269 """Test additional processing for CNAME records"""

270 return self.namesTest(

271 self.resolver.lookupAddress('cname.test-domain.com'),

272 [dns.Record_CNAME('test-domain.com', ttl=19283784), dns.Record_A('127.0. 0.1', ttl=19283784)]

273 )

274

275 def testMB(self):

276 """Test DNS 'MB' record queries"""

277 return self.namesTest(

278 self.resolver.lookupMailBox('test-domain.com'),

279 [dns.Record_MB('mailbox.test-domain.com', ttl=19283784)]

280 )

281

282

283 def testMG(self):

284 """Test DNS 'MG' record queries"""

285 return self.namesTest(

286 self.resolver.lookupMailGroup('test-domain.com'),

287 [dns.Record_MG('mail.group.someplace', ttl=19283784)]

288 )

289

290

291 def testMR(self):

292 """Test DNS 'MR' record queries"""

293 return self.namesTest(

294 self.resolver.lookupMailRename('test-domain.com'),

295 [dns.Record_MR('mail.redirect.or.whatever', ttl=19283784)]

296 )

297

298

299 def testMINFO(self):

300 """Test DNS 'MINFO' record queries"""

301 return self.namesTest(

302 self.resolver.lookupMailboxInfo('test-domain.com'),

303 [dns.Record_MINFO(rmailbx='r mail box', emailbx='e mail box', ttl=19 283784)]

304 )

305

306

307 def testSRV(self):

308 """Test DNS 'SRV' record queries"""

309 return self.namesTest(

310 self.resolver.lookupService('http.tcp.test-domain.com'),

311 [dns.Record_SRV(257, 16383, 43690, 'some.other.place.fool', ttl=1928 3784)]

312 )

313

314 def testAFSDB(self):

315 """Test DNS 'AFSDB' record queries"""

316 return self.namesTest(

317 self.resolver.lookupAFSDatabase('test-domain.com'),

318 [dns.Record_AFSDB(subtype=1, hostname='afsdb.test-domain.com', ttl=1 9283784)]

319 )

320

321

322 def testRP(self):

323 """Test DNS 'RP' record queries"""

324 return self.namesTest(

325 self.resolver.lookupResponsibility('test-domain.com'),

326 [dns.Record_RP(mbox='whatever.i.dunno', txt='some.more.text', ttl=19 283784)]

327 )

328

329

330 def testTXT(self):

331 """Test DNS 'TXT' record queries"""

332 return self.namesTest(

333 self.resolver.lookupText('test-domain.com'),

334 [dns.Record_TXT('A First piece of Text', 'a SecoNd piece', ttl=19283 784),

335 dns.Record_TXT('Some more text, haha! Yes. \0 Still here?', ttl= 19283784)]

336 )

337

338

339 def testWKS(self):

340 """Test DNS 'WKS' record queries"""

341 return self.namesTest(

342 self.resolver.lookupWellKnownServices('test-domain.com'),

343 [dns.Record_WKS('12.54.78.12', socket.IPPROTO_TCP, '\x12\x01\x16\xfe \xc1\x00\x01', ttl=19283784)]

344 )

345

346

347 def testSomeRecordsWithTTLs(self):

348 result_soa = copy.copy(my_soa)

349 result_soa.ttl = my_soa.expire

350 return self.namesTest(

351 self.resolver.lookupAllRecords('my-domain.com'),

352 [result_soa,

353 dns.Record_A('1.2.3.4', ttl='1S'),

354 dns.Record_NS('ns1.domain', ttl='2M'),

355 dns.Record_NS('ns2.domain', ttl='3H'),

356 dns.Record_SRV(257, 16383, 43690, 'some.other.place.fool', ttl='4D' )]

357 )

358

359

360 def testAAAA(self):

361 """Test DNS 'AAAA' record queries (IPv6)"""

362 return self.namesTest(

363 self.resolver.lookupIPV6Address('test-domain.com'),

364 [dns.Record_AAAA('AF43:5634:1294:AFCB:56AC:48EF:34C3:01FF', ttl=1928 3784)]

365 )

366

367 def testA6(self):

368 """Test DNS 'A6' record queries (IPv6)"""

369 return self.namesTest(

370 self.resolver.lookupAddress6('test-domain.com'),

371 [dns.Record_A6(0, 'ABCD::4321', '', ttl=19283784),

372 dns.Record_A6(12, '0:0069::0', 'some.network.tld', ttl=19283784),

373 dns.Record_A6(8, '0:5634:1294:AFCB:56AC:48EF:34C3:01FF', 'tra.la.la .net', ttl=19283784)]

374 )

375

376

377 def testZoneTransfer(self):

378 """Test DNS 'AXFR' queries (Zone transfer)"""

379 default_ttl = soa_record.expire

380 results = [copy.copy(r) for r in reduce(operator.add, test_domain_com.re cords.values())]

381 for r in results:

382 if r.ttl is None:

383 r.ttl = default_ttl

384 return self.namesTest(

385 self.resolver.lookupZone('test-domain.com').addCallback(lambda r: (r [0][:-1],)),

386 results

387 )

388

389 def testSimilarZonesDontInterfere(self):

390 """Tests that unrelated zones don't mess with each other."""

391 return self.namesTest(

392 self.resolver.lookupAddress("anothertest-domain.com"),

393 [dns.Record_A('1.2.3.4', ttl=19283784)]

394 )

395

396

397

398 class DNSServerFactoryTests(unittest.TestCase):

399 """

400 Tests for L{server.DNSServerFactory}.

401 """

402 def _messageReceivedTest(self, methodName, message):

403 """

404 Assert that the named method is called with the given message when

405 it is passed to L{DNSServerFactory.messageReceived}.

406 """

407 # Make it appear to have some queries so that

408 # DNSServerFactory.allowQuery allows it.

409 message.queries = [None]

410

411 receivedMessages = []

412 def fakeHandler(message, protocol, address):

413 receivedMessages.append((message, protocol, address))

414

415 class FakeProtocol(object):

416 def writeMessage(self, message):

417 pass

418

419 protocol = FakeProtocol()

420 factory = server.DNSServerFactory(None)

421 setattr(factory, methodName, fakeHandler)

422 factory.messageReceived(message, protocol)

423 self.assertEqual(receivedMessages, [(message, protocol, None)])

424

425

426 def test_notifyMessageReceived(self):

427 """

428 L{DNSServerFactory.messageReceived} passes messages with an opcode

429 of C{OP_NOTIFY} on to L{DNSServerFactory.handleNotify}.

430 """

431 # RFC 1996, section 4.5

432 opCode = 4

433 self._messageReceivedTest('handleNotify', Message(opCode=opCode))

434

435

436 def test_updateMessageReceived(self):

437 """

438 L{DNSServerFactory.messageReceived} passes messages with an opcode

439 of C{OP_UPDATE} on to L{DNSServerFactory.handleOther}.

440

441 This may change if the implementation ever covers update messages.

442 """

443 # RFC 2136, section 1.3

444 opCode = 5

445 self._messageReceivedTest('handleOther', Message(opCode=opCode))

446

447

448

449 class HelperTestCase(unittest.TestCase):

450 def testSerialGenerator(self):

451 f = self.mktemp()

452 a = authority.getSerial(f)

453 for i in range(20):

454 b = authority.getSerial(f)

455 self.failUnless(a < b)

456 a = b

457

458

459 class AXFRTest(unittest.TestCase):

460 def setUp(self):

461 self.results = None

462 self.d = defer.Deferred()

463 self.d.addCallback(self._gotResults)

464 self.controller = client.AXFRController('fooby.com', self.d)

465

466 self.soa = dns.RRHeader(name='fooby.com', type=dns.SOA, cls=dns.IN, ttl= 86400, auth=False,

467 payload=dns.Record_SOA(mname='fooby.com',

468 rname='hooj.fooby.com',

469 serial=100,

470 refresh=200,

471 retry=300,

472 expire=400,

473 minimum=500,

474 ttl=600))

475

476 self.records = [

477 self.soa,

478 dns.RRHeader(name='fooby.com', type=dns.NS, cls=dns.IN, ttl=700, aut h=False,

479 payload=dns.Record_NS(name='ns.twistedmatrix.com', ttl= 700)),

480

481 dns.RRHeader(name='fooby.com', type=dns.MX, cls=dns.IN, ttl=700, aut h=False,

482 payload=dns.Record_MX(preference=10, exchange='mail.mv3 d.com', ttl=700)),

483

484 dns.RRHeader(name='fooby.com', type=dns.A, cls=dns.IN, ttl=700, auth =False,

485 payload=dns.Record_A(address='64.123.27.105', ttl=700)) ,

486 self.soa

487 ]

488

489 def _makeMessage(self):

490 # hooray they all have the same message format

491 return dns.Message(id=999, answer=1, opCode=0, recDes=0, recAv=1, auth=1 , rCode=0, trunc=0, maxSize=0)

492

493 def testBindAndTNamesStyle(self):

494 # Bind style = One big single message

495 m = self._makeMessage()

496 m.queries = [dns.Query('fooby.com', dns.AXFR, dns.IN)]

497 m.answers = self.records

498 self.controller.messageReceived(m, None)

499 self.assertEquals(self.results, self.records)

500

501 def _gotResults(self, result):

502 self.results = result

503

504 def testDJBStyle(self):

505 # DJB style = message per record

506 records = self.records[:]

507 while records:

508 m = self._makeMessage()

509 m.queries = [] # DJB doesn't specify any queries.. hmm..

510 m.answers = [records.pop(0)]

511 self.controller.messageReceived(m, None)

512 self.assertEquals(self.results, self.records)

513

514 class HostsTestCase(unittest.TestCase):

515 def setUp(self):

516 f = open('EtcHosts', 'w')

517 f.write('''

518 1.1.1.1 EXAMPLE EXAMPLE.EXAMPLETHING

519 1.1.1.2 HOOJY

520 ::1 ip6thingy

521 ''')

522 f.close()

523 self.resolver = hosts.Resolver('EtcHosts')

524

525 def testGetHostByName(self):

526 data = [('EXAMPLE', '1.1.1.1'),

527 ('EXAMPLE.EXAMPLETHING', '1.1.1.1'),

528 ('HOOJY', '1.1.1.2'),

529 ]

530 ds = [self.resolver.getHostByName(n).addCallback(self.assertEqual, ip)

531 for n, ip in data]

532 return defer.gatherResults(ds)

533

534 def testLookupAddress(self):

535 d = self.resolver.lookupAddress('HOOJY')

536 d.addCallback(lambda x: self.assertEqual(x[0][0].payload.dottedQuad(),

537 '1.1.1.2'))

538 return d

539

540 def testIPv6(self):

541 d = self.resolver.lookupIPV6Address('ip6thingy')

542 d.addCallback(self.assertEqual, '::1')

543 return d

544

545 testIPv6.skip = 'IPv6 support is not in our hosts resolver yet'

546

547 def testNotImplemented(self):

548 return self.assertFailure(self.resolver.lookupMailExchange('EXAMPLE'),

549 NotImplementedError)

550

551 def testQuery(self):

552 d = self.resolver.query(dns.Query('EXAMPLE'))

553 d.addCallback(lambda x: self.assertEqual(x[0][0].payload.dottedQuad(),

554 '1.1.1.1'))

555 return d

556

557 def testNotFound(self):

558 return self.assertFailure(self.resolver.lookupAddress('foueoa'),

559 dns.DomainError)

560

561

562 class FakeDNSDatagramProtocol(object):

563 transport = object()

564

565 def __init__(self):

566 self.queries = []

567

568 def query(self, address, queries, timeout=10, id=None):

569 self.queries.append((address, queries, timeout, id))

570 return defer.fail(dns.DNSQueryTimeoutError(queries))

571

572 def removeResend(self, id):

573 # Ignore this for the time being.

574 pass

575

576 class RetryLogic(unittest.TestCase):

577 testServers = [

578 '1.2.3.4',

579 '4.3.2.1',

580 'a.b.c.d',

581 'z.y.x.w']

582

583 def testRoundRobinBackoff(self):

584 addrs = [(x, 53) for x in self.testServers]

585 r = client.Resolver(resolv=None, servers=addrs)

586 r.protocol = proto = FakeDNSDatagramProtocol()

587 return r.lookupAddress("foo.example.com"

588 ).addCallback(self._cbRoundRobinBackoff

589 ).addErrback(self._ebRoundRobinBackoff, proto

590 )

591

592 def _cbRoundRobinBackoff(self, result):

593 raise unittest.FailTest("Lookup address succeeded, should have timed out ")

594

595 def _ebRoundRobinBackoff(self, failure, fakeProto):

596 failure.trap(defer.TimeoutError)

597

598 # Assert that each server is tried with a particular timeout

599 # before the timeout is increased and the attempts are repeated.

600

601 for t in (1, 3, 11, 45):

602 tries = fakeProto.queries[:len(self.testServers)]

603 del fakeProto.queries[:len(self.testServers)]

604

605 tries.sort()

606 expected = list(self.testServers)

607 expected.sort()

608

609 for ((addr, query, timeout, id), expectedAddr) in zip(tries, expecte d):

610 self.assertEquals(addr, (expectedAddr, 53))

611 self.assertEquals(timeout, t)

612

613 self.failIf(fakeProto.queries)

614

615 class ResolvConfHandling(unittest.TestCase):

616 def testMissing(self):

617 resolvConf = self.mktemp()

618 r = client.Resolver(resolv=resolvConf)

619 self.assertEquals(r.dynServers, [('127.0.0.1', 53)])

620 r._parseCall.cancel()

621

622 def testEmpty(self):

623 resolvConf = self.mktemp()

624 fObj = file(resolvConf, 'w')

625 fObj.close()

626 r = client.Resolver(resolv=resolvConf)

627 self.assertEquals(r.dynServers, [('127.0.0.1', 53)])

628 r._parseCall.cancel()

629

630

631

632 class FilterAnswersTests(unittest.TestCase):

633 """

634 Test L{twisted.names.client.Resolver.filterAnswers}'s handling of various

635 error conditions it might encounter.

636 """

637 def setUp(self):

638 # Create a resolver pointed at an invalid server - we won't be hitting

639 # the network in any of these tests.

640 self.resolver = Resolver(servers=[('0.0.0.0', 0)])

641

642

643 def test_truncatedMessage(self):

644 """

645 Test that a truncated message results in an equivalent request made via

646 TCP.

647 """

648 m = Message(trunc=True)

649 m.addQuery('example.com')

650

651 def queryTCP(queries):

652 self.assertEqual(queries, m.queries)

653 response = Message()

654 response.answers = ['answer']

655 response.authority = ['authority']

656 response.additional = ['additional']

657 return succeed(response)

658 self.resolver.queryTCP = queryTCP

659 d = self.resolver.filterAnswers(m)

660 d.addCallback(

661 self.assertEqual, (['answer'], ['authority'], ['additional']))

662 return d

663

664

665 def _rcodeTest(self, rcode, exc):

666 m = Message(rCode=rcode)

667 err = self.resolver.filterAnswers(m)

668 err.trap(exc)

669

670

671 def test_formatError(self):

672 """

673 Test that a message with a result code of C{EFORMAT} results in a

674 failure wrapped around L{DNSFormatError}.

675 """

676 return self._rcodeTest(EFORMAT, DNSFormatError)

677

678

679 def test_serverError(self):

680 """

681 Like L{test_formatError} but for C{ESERVER}/L{DNSServerError}.

682 """

683 return self._rcodeTest(ESERVER, DNSServerError)

684

685

686 def test_nameError(self):

687 """

688 Like L{test_formatError} but for C{ENAME}/L{DNSNameError}.

689 """

690 return self._rcodeTest(ENAME, DNSNameError)

691

692

693 def test_notImplementedError(self):

694 """

695 Like L{test_formatError} but for C{ENOTIMP}/L{DNSNotImplementedError}.

696 """

697 return self._rcodeTest(ENOTIMP, DNSNotImplementedError)

698

699

700 def test_refusedError(self):

701 """

702 Like L{test_formatError} but for C{EREFUSED}/L{DNSQueryRefusedError}.

703 """

704 return self._rcodeTest(EREFUSED, DNSQueryRefusedError)

705

706

707 def test_refusedErrorUnknown(self):

708 """

709 Like L{test_formatError} but for an unrecognized error code and

710 L{DNSUnknownError}.

711 """

712 return self._rcodeTest(EREFUSED + 1, DNSUnknownError)

OLD	NEW