1 # -*- test-case-name: twisted.names.test.test_names -*-
2 # Copyright (c) Twisted Matrix Laboratories.
3 # See LICENSE for details.
6 Test cases for twisted.names.
9 import socket, operator, copy
10 from StringIO import StringIO
12 from twisted.trial import unittest
14 from twisted.internet import reactor, defer, error
15 from twisted.internet.task import Clock
16 from twisted.internet.defer import succeed
17 from twisted.names import client, server, common, authority, dns
18 from twisted.python import failure
19 from twisted.names.error import DNSFormatError, DNSServerError, DNSNameError
20 from twisted.names.error import DNSNotImplementedError, DNSQueryRefusedError
21 from twisted.names.error import DNSUnknownError
22 from twisted.names.dns import EFORMAT, ESERVER, ENAME, ENOTIMP, EREFUSED
23 from twisted.names.dns import Message
24 from twisted.names.client import Resolver
25 from twisted.names.secondary import (
26 SecondaryAuthorityService, SecondaryAuthority)
27 from twisted.names.test.test_client import StubPort
29 from twisted.python.compat import reduce
30 from twisted.test.proto_helpers import StringTransport, MemoryReactor
32 def justPayload(results):
33 return [r.payload for r in results[0]]
35 class NoFileAuthority(authority.FileAuthority):
36 def __init__(self, soa, records):
37 # Yes, skip FileAuthority
38 common.ResolverBase.__init__(self)
39 self.soa, self.records = soa, records
42 soa_record = dns.Record_SOA(
43 mname = 'test-domain.com',
44 rname = 'root.test-domain.com',
53 reverse_soa = dns.Record_SOA(
54 mname = '93.84.28.in-addr.arpa',
55 rname = '93.84.28.in-addr.arpa',
64 my_soa = dns.Record_SOA(
65 mname = 'my-domain.com',
66 rname = 'postmaster.test-domain.com',
74 test_domain_com = NoFileAuthority(
75 soa = ('test-domain.com', soa_record),
79 dns.Record_A('127.0.0.1'),
80 dns.Record_NS('39.28.189.39'),
81 dns.Record_SPF('v=spf1 mx/30 mx:example.org/30 -all'),
82 dns.Record_SPF('v=spf1 +mx a:\0colo', '.example.com/28 -all not valid'),
83 dns.Record_MX(10, 'host.test-domain.com'),
84 dns.Record_HINFO(os='Linux', cpu='A Fast One, Dontcha know'),
85 dns.Record_CNAME('canonical.name.com'),
86 dns.Record_MB('mailbox.test-domain.com'),
87 dns.Record_MG('mail.group.someplace'),
88 dns.Record_TXT('A First piece of Text', 'a SecoNd piece'),
89 dns.Record_A6(0, 'ABCD::4321', ''),
90 dns.Record_A6(12, '0:0069::0', 'some.network.tld'),
91 dns.Record_A6(8, '0:5634:1294:AFCB:56AC:48EF:34C3:01FF', 'tra.la.la.net'),
92 dns.Record_TXT('Some more text, haha! Yes. \0 Still here?'),
93 dns.Record_MR('mail.redirect.or.whatever'),
94 dns.Record_MINFO(rmailbx='r mail box', emailbx='e mail box'),
95 dns.Record_AFSDB(subtype=1, hostname='afsdb.test-domain.com'),
96 dns.Record_RP(mbox='whatever.i.dunno', txt='some.more.text'),
97 dns.Record_WKS('12.54.78.12', socket.IPPROTO_TCP,
98 '\x12\x01\x16\xfe\xc1\x00\x01'),
99 dns.Record_NAPTR(100, 10, "u", "sip+E2U",
100 "!^.*$!sip:information@domain.tld!"),
101 dns.Record_AAAA('AF43:5634:1294:AFCB:56AC:48EF:34C3:01FF')],
102 'http.tcp.test-domain.com': [
103 dns.Record_SRV(257, 16383, 43690, 'some.other.place.fool')
105 'host.test-domain.com': [
106 dns.Record_A('123.242.1.5'),
107 dns.Record_A('0.255.0.255'),
109 'host-two.test-domain.com': [
112 # dns.Record_A('255.255.255.255'),
114 dns.Record_A('255.255.255.254'),
115 dns.Record_A('0.0.0.0')
117 'cname.test-domain.com': [
118 dns.Record_CNAME('test-domain.com')
120 'anothertest-domain.com': [
121 dns.Record_A('1.2.3.4')],
125 reverse_domain = NoFileAuthority(
126 soa = ('93.84.28.in-addr.arpa', reverse_soa),
128 '123.93.84.28.in-addr.arpa': [
129 dns.Record_PTR('test.host-reverse.lookup.com'),
136 my_domain_com = NoFileAuthority(
137 soa = ('my-domain.com', my_soa),
141 dns.Record_A('1.2.3.4', ttl='1S'),
142 dns.Record_NS('ns1.domain', ttl='2M'),
143 dns.Record_NS('ns2.domain', ttl='3H'),
144 dns.Record_SRV(257, 16383, 43690, 'some.other.place.fool', ttl='4D')
150 class ServerDNSTestCase(unittest.TestCase):
152 Test cases for DNS server and client.
156 self.factory = server.DNSServerFactory([
157 test_domain_com, reverse_domain, my_domain_com
160 p = dns.DNSDatagramProtocol(self.factory)
163 listenerTCP = reactor.listenTCP(0, self.factory, interface="127.0.0.1")
164 # It's simpler to do the stop listening with addCleanup,
165 # even though we might not end up using this TCP port in
166 # the test (if the listenUDP below fails). Cleaning up
167 # this TCP port sooner than "cleanup time" would mean
168 # adding more code to keep track of the Deferred returned
170 self.addCleanup(listenerTCP.stopListening)
171 port = listenerTCP.getHost().port
174 listenerUDP = reactor.listenUDP(port, p, interface="127.0.0.1")
175 except error.CannotListenError:
178 self.addCleanup(listenerUDP.stopListening)
181 self.listenerTCP = listenerTCP
182 self.listenerUDP = listenerUDP
183 self.resolver = client.Resolver(servers=[('127.0.0.1', port)])
188 Clean up any server connections associated with the
189 L{DNSServerFactory} created in L{setUp}
191 # It'd be great if DNSServerFactory had a method that
192 # encapsulated this task. At least the necessary data is
194 for conn in self.factory.connections[:]:
195 conn.transport.loseConnection()
198 def namesTest(self, d, r):
200 def setDone(response):
201 self.response = response
203 def checkResults(ignored):
204 if isinstance(self.response, failure.Failure):
206 results = justPayload(self.response)
207 assert len(results) == len(r), "%s != %s" % (map(str, results), map(str, r))
209 assert rec in r, "%s not in %s" % (rec, map(str, r))
212 d.addCallback(checkResults)
215 def testAddressRecord1(self):
216 """Test simple DNS 'A' record queries"""
217 return self.namesTest(
218 self.resolver.lookupAddress('test-domain.com'),
219 [dns.Record_A('127.0.0.1', ttl=19283784)]
223 def testAddressRecord2(self):
224 """Test DNS 'A' record queries with multiple answers"""
225 return self.namesTest(
226 self.resolver.lookupAddress('host.test-domain.com'),
227 [dns.Record_A('123.242.1.5', ttl=19283784), dns.Record_A('0.255.0.255', ttl=19283784)]
231 def testAddressRecord3(self):
232 """Test DNS 'A' record queries with edge cases"""
233 return self.namesTest(
234 self.resolver.lookupAddress('host-two.test-domain.com'),
235 [dns.Record_A('255.255.255.254', ttl=19283784), dns.Record_A('0.0.0.0', ttl=19283784)]
239 def testAuthority(self):
240 """Test DNS 'SOA' record queries"""
241 return self.namesTest(
242 self.resolver.lookupAuthority('test-domain.com'),
247 def testMailExchangeRecord(self):
248 """Test DNS 'MX' record queries"""
249 return self.namesTest(
250 self.resolver.lookupMailExchange('test-domain.com'),
251 [dns.Record_MX(10, 'host.test-domain.com', ttl=19283784)]
255 def testNameserver(self):
256 """Test DNS 'NS' record queries"""
257 return self.namesTest(
258 self.resolver.lookupNameservers('test-domain.com'),
259 [dns.Record_NS('39.28.189.39', ttl=19283784)]
264 """Test DNS 'HINFO' record queries"""
265 return self.namesTest(
266 self.resolver.lookupHostInfo('test-domain.com'),
267 [dns.Record_HINFO(os='Linux', cpu='A Fast One, Dontcha know', ttl=19283784)]
271 """Test DNS 'PTR' record queries"""
272 return self.namesTest(
273 self.resolver.lookupPointer('123.93.84.28.in-addr.arpa'),
274 [dns.Record_PTR('test.host-reverse.lookup.com', ttl=11193983)]
279 """Test DNS 'CNAME' record queries"""
280 return self.namesTest(
281 self.resolver.lookupCanonicalName('test-domain.com'),
282 [dns.Record_CNAME('canonical.name.com', ttl=19283784)]
285 def testCNAMEAdditional(self):
286 """Test additional processing for CNAME records"""
287 return self.namesTest(
288 self.resolver.lookupAddress('cname.test-domain.com'),
289 [dns.Record_CNAME('test-domain.com', ttl=19283784), dns.Record_A('127.0.0.1', ttl=19283784)]
293 """Test DNS 'MB' record queries"""
294 return self.namesTest(
295 self.resolver.lookupMailBox('test-domain.com'),
296 [dns.Record_MB('mailbox.test-domain.com', ttl=19283784)]
301 """Test DNS 'MG' record queries"""
302 return self.namesTest(
303 self.resolver.lookupMailGroup('test-domain.com'),
304 [dns.Record_MG('mail.group.someplace', ttl=19283784)]
309 """Test DNS 'MR' record queries"""
310 return self.namesTest(
311 self.resolver.lookupMailRename('test-domain.com'),
312 [dns.Record_MR('mail.redirect.or.whatever', ttl=19283784)]
317 """Test DNS 'MINFO' record queries"""
318 return self.namesTest(
319 self.resolver.lookupMailboxInfo('test-domain.com'),
320 [dns.Record_MINFO(rmailbx='r mail box', emailbx='e mail box', ttl=19283784)]
325 """Test DNS 'SRV' record queries"""
326 return self.namesTest(
327 self.resolver.lookupService('http.tcp.test-domain.com'),
328 [dns.Record_SRV(257, 16383, 43690, 'some.other.place.fool', ttl=19283784)]
332 """Test DNS 'AFSDB' record queries"""
333 return self.namesTest(
334 self.resolver.lookupAFSDatabase('test-domain.com'),
335 [dns.Record_AFSDB(subtype=1, hostname='afsdb.test-domain.com', ttl=19283784)]
340 """Test DNS 'RP' record queries"""
341 return self.namesTest(
342 self.resolver.lookupResponsibility('test-domain.com'),
343 [dns.Record_RP(mbox='whatever.i.dunno', txt='some.more.text', ttl=19283784)]
348 """Test DNS 'TXT' record queries"""
349 return self.namesTest(
350 self.resolver.lookupText('test-domain.com'),
351 [dns.Record_TXT('A First piece of Text', 'a SecoNd piece', ttl=19283784),
352 dns.Record_TXT('Some more text, haha! Yes. \0 Still here?', ttl=19283784)]
358 L{DNSServerFactory} can serve I{SPF} resource records.
360 return self.namesTest(
361 self.resolver.lookupSenderPolicy('test-domain.com'),
362 [dns.Record_SPF('v=spf1 mx/30 mx:example.org/30 -all', ttl=19283784),
363 dns.Record_SPF('v=spf1 +mx a:\0colo', '.example.com/28 -all not valid', ttl=19283784)]
368 """Test DNS 'WKS' record queries"""
369 return self.namesTest(
370 self.resolver.lookupWellKnownServices('test-domain.com'),
371 [dns.Record_WKS('12.54.78.12', socket.IPPROTO_TCP, '\x12\x01\x16\xfe\xc1\x00\x01', ttl=19283784)]
375 def testSomeRecordsWithTTLs(self):
376 result_soa = copy.copy(my_soa)
377 result_soa.ttl = my_soa.expire
378 return self.namesTest(
379 self.resolver.lookupAllRecords('my-domain.com'),
381 dns.Record_A('1.2.3.4', ttl='1S'),
382 dns.Record_NS('ns1.domain', ttl='2M'),
383 dns.Record_NS('ns2.domain', ttl='3H'),
384 dns.Record_SRV(257, 16383, 43690, 'some.other.place.fool', ttl='4D')]
389 """Test DNS 'AAAA' record queries (IPv6)"""
390 return self.namesTest(
391 self.resolver.lookupIPV6Address('test-domain.com'),
392 [dns.Record_AAAA('AF43:5634:1294:AFCB:56AC:48EF:34C3:01FF', ttl=19283784)]
396 """Test DNS 'A6' record queries (IPv6)"""
397 return self.namesTest(
398 self.resolver.lookupAddress6('test-domain.com'),
399 [dns.Record_A6(0, 'ABCD::4321', '', ttl=19283784),
400 dns.Record_A6(12, '0:0069::0', 'some.network.tld', ttl=19283784),
401 dns.Record_A6(8, '0:5634:1294:AFCB:56AC:48EF:34C3:01FF', 'tra.la.la.net', ttl=19283784)]
405 def test_zoneTransfer(self):
407 Test DNS 'AXFR' queries (Zone transfer)
409 default_ttl = soa_record.expire
410 results = [copy.copy(r) for r in reduce(operator.add, test_domain_com.records.values())]
414 return self.namesTest(
415 self.resolver.lookupZone('test-domain.com').addCallback(lambda r: (r[0][:-1],)),
420 def testSimilarZonesDontInterfere(self):
421 """Tests that unrelated zones don't mess with each other."""
422 return self.namesTest(
423 self.resolver.lookupAddress("anothertest-domain.com"),
424 [dns.Record_A('1.2.3.4', ttl=19283784)]
428 def test_NAPTR(self):
430 Test DNS 'NAPTR' record queries.
432 return self.namesTest(
433 self.resolver.lookupNamingAuthorityPointer('test-domain.com'),
434 [dns.Record_NAPTR(100, 10, "u", "sip+E2U",
435 "!^.*$!sip:information@domain.tld!",
440 class DNSServerFactoryTests(unittest.TestCase):
442 Tests for L{server.DNSServerFactory}.
444 def _messageReceivedTest(self, methodName, message):
446 Assert that the named method is called with the given message when
447 it is passed to L{DNSServerFactory.messageReceived}.
449 # Make it appear to have some queries so that
450 # DNSServerFactory.allowQuery allows it.
451 message.queries = [None]
453 receivedMessages = []
454 def fakeHandler(message, protocol, address):
455 receivedMessages.append((message, protocol, address))
457 class FakeProtocol(object):
458 def writeMessage(self, message):
461 protocol = FakeProtocol()
462 factory = server.DNSServerFactory(None)
463 setattr(factory, methodName, fakeHandler)
464 factory.messageReceived(message, protocol)
465 self.assertEqual(receivedMessages, [(message, protocol, None)])
468 def test_notifyMessageReceived(self):
470 L{DNSServerFactory.messageReceived} passes messages with an opcode
471 of C{OP_NOTIFY} on to L{DNSServerFactory.handleNotify}.
473 # RFC 1996, section 4.5
475 self._messageReceivedTest('handleNotify', Message(opCode=opCode))
478 def test_updateMessageReceived(self):
480 L{DNSServerFactory.messageReceived} passes messages with an opcode
481 of C{OP_UPDATE} on to L{DNSServerFactory.handleOther}.
483 This may change if the implementation ever covers update messages.
485 # RFC 2136, section 1.3
487 self._messageReceivedTest('handleOther', Message(opCode=opCode))
490 def test_connectionTracking(self):
492 The C{connectionMade} and C{connectionLost} methods of
493 L{DNSServerFactory} cooperate to keep track of all
494 L{DNSProtocol} objects created by a factory which are
497 protoA, protoB = object(), object()
498 factory = server.DNSServerFactory()
499 factory.connectionMade(protoA)
500 self.assertEqual(factory.connections, [protoA])
501 factory.connectionMade(protoB)
502 self.assertEqual(factory.connections, [protoA, protoB])
503 factory.connectionLost(protoA)
504 self.assertEqual(factory.connections, [protoB])
505 factory.connectionLost(protoB)
506 self.assertEqual(factory.connections, [])
509 class HelperTestCase(unittest.TestCase):
510 def testSerialGenerator(self):
512 a = authority.getSerial(f)
514 b = authority.getSerial(f)
515 self.failUnless(a < b)
519 class AXFRTest(unittest.TestCase):
522 self.d = defer.Deferred()
523 self.d.addCallback(self._gotResults)
524 self.controller = client.AXFRController('fooby.com', self.d)
526 self.soa = dns.RRHeader(name='fooby.com', type=dns.SOA, cls=dns.IN, ttl=86400, auth=False,
527 payload=dns.Record_SOA(mname='fooby.com',
528 rname='hooj.fooby.com',
538 dns.RRHeader(name='fooby.com', type=dns.NS, cls=dns.IN, ttl=700, auth=False,
539 payload=dns.Record_NS(name='ns.twistedmatrix.com', ttl=700)),
541 dns.RRHeader(name='fooby.com', type=dns.MX, cls=dns.IN, ttl=700, auth=False,
542 payload=dns.Record_MX(preference=10, exchange='mail.mv3d.com', ttl=700)),
544 dns.RRHeader(name='fooby.com', type=dns.A, cls=dns.IN, ttl=700, auth=False,
545 payload=dns.Record_A(address='64.123.27.105', ttl=700)),
549 def _makeMessage(self):
550 # hooray they all have the same message format
551 return dns.Message(id=999, answer=1, opCode=0, recDes=0, recAv=1, auth=1, rCode=0, trunc=0, maxSize=0)
553 def testBindAndTNamesStyle(self):
554 # Bind style = One big single message
555 m = self._makeMessage()
556 m.queries = [dns.Query('fooby.com', dns.AXFR, dns.IN)]
557 m.answers = self.records
558 self.controller.messageReceived(m, None)
559 self.assertEqual(self.results, self.records)
561 def _gotResults(self, result):
562 self.results = result
564 def testDJBStyle(self):
565 # DJB style = message per record
566 records = self.records[:]
568 m = self._makeMessage()
569 m.queries = [] # DJB *doesn't* specify any queries.. hmm..
570 m.answers = [records.pop(0)]
571 self.controller.messageReceived(m, None)
572 self.assertEqual(self.results, self.records)
574 class FakeDNSDatagramProtocol(object):
577 self.transport = StubPort()
579 def query(self, address, queries, timeout=10, id=None):
580 self.queries.append((address, queries, timeout, id))
581 return defer.fail(dns.DNSQueryTimeoutError(queries))
583 def removeResend(self, id):
584 # Ignore this for the time being.
587 class RetryLogic(unittest.TestCase):
594 def testRoundRobinBackoff(self):
595 addrs = [(x, 53) for x in self.testServers]
596 r = client.Resolver(resolv=None, servers=addrs)
597 r.protocol = proto = FakeDNSDatagramProtocol()
598 return r.lookupAddress("foo.example.com"
599 ).addCallback(self._cbRoundRobinBackoff
600 ).addErrback(self._ebRoundRobinBackoff, proto
603 def _cbRoundRobinBackoff(self, result):
604 raise unittest.FailTest("Lookup address succeeded, should have timed out")
606 def _ebRoundRobinBackoff(self, failure, fakeProto):
607 failure.trap(defer.TimeoutError)
609 # Assert that each server is tried with a particular timeout
610 # before the timeout is increased and the attempts are repeated.
612 for t in (1, 3, 11, 45):
613 tries = fakeProto.queries[:len(self.testServers)]
614 del fakeProto.queries[:len(self.testServers)]
617 expected = list(self.testServers)
620 for ((addr, query, timeout, id), expectedAddr) in zip(tries, expected):
621 self.assertEqual(addr, (expectedAddr, 53))
622 self.assertEqual(timeout, t)
624 self.failIf(fakeProto.queries)
626 class ResolvConfHandling(unittest.TestCase):
627 def testMissing(self):
628 resolvConf = self.mktemp()
629 r = client.Resolver(resolv=resolvConf)
630 self.assertEqual(r.dynServers, [('127.0.0.1', 53)])
631 r._parseCall.cancel()
634 resolvConf = self.mktemp()
635 fObj = file(resolvConf, 'w')
637 r = client.Resolver(resolv=resolvConf)
638 self.assertEqual(r.dynServers, [('127.0.0.1', 53)])
639 r._parseCall.cancel()
643 class FilterAnswersTests(unittest.TestCase):
645 Test L{twisted.names.client.Resolver.filterAnswers}'s handling of various
646 error conditions it might encounter.
649 # Create a resolver pointed at an invalid server - we won't be hitting
650 # the network in any of these tests.
651 self.resolver = Resolver(servers=[('0.0.0.0', 0)])
654 def test_truncatedMessage(self):
656 Test that a truncated message results in an equivalent request made via
659 m = Message(trunc=True)
660 m.addQuery('example.com')
662 def queryTCP(queries):
663 self.assertEqual(queries, m.queries)
665 response.answers = ['answer']
666 response.authority = ['authority']
667 response.additional = ['additional']
668 return succeed(response)
669 self.resolver.queryTCP = queryTCP
670 d = self.resolver.filterAnswers(m)
672 self.assertEqual, (['answer'], ['authority'], ['additional']))
676 def _rcodeTest(self, rcode, exc):
677 m = Message(rCode=rcode)
678 err = self.resolver.filterAnswers(m)
682 def test_formatError(self):
684 Test that a message with a result code of C{EFORMAT} results in a
685 failure wrapped around L{DNSFormatError}.
687 return self._rcodeTest(EFORMAT, DNSFormatError)
690 def test_serverError(self):
692 Like L{test_formatError} but for C{ESERVER}/L{DNSServerError}.
694 return self._rcodeTest(ESERVER, DNSServerError)
697 def test_nameError(self):
699 Like L{test_formatError} but for C{ENAME}/L{DNSNameError}.
701 return self._rcodeTest(ENAME, DNSNameError)
704 def test_notImplementedError(self):
706 Like L{test_formatError} but for C{ENOTIMP}/L{DNSNotImplementedError}.
708 return self._rcodeTest(ENOTIMP, DNSNotImplementedError)
711 def test_refusedError(self):
713 Like L{test_formatError} but for C{EREFUSED}/L{DNSQueryRefusedError}.
715 return self._rcodeTest(EREFUSED, DNSQueryRefusedError)
718 def test_refusedErrorUnknown(self):
720 Like L{test_formatError} but for an unrecognized error code and
723 return self._rcodeTest(EREFUSED + 1, DNSUnknownError)
727 class AuthorityTests(unittest.TestCase):
729 Tests for the basic response record selection code in L{FileAuthority}
730 (independent of its fileness).
732 def test_recordMissing(self):
734 If a L{FileAuthority} has a zone which includes an I{NS} record for a
735 particular name and that authority is asked for another record for the
736 same name which does not exist, the I{NS} record is not included in the
737 authority section of the response.
739 authority = NoFileAuthority(
740 soa=(str(soa_record.mname), soa_record),
742 str(soa_record.mname): [
744 dns.Record_NS('1.2.3.4'),
746 d = authority.lookupAddress(str(soa_record.mname))
748 d.addCallback(result.append)
749 answer, authority, additional = result[0]
750 self.assertEqual(answer, [])
754 str(soa_record.mname), soa_record.TYPE,
755 ttl=soa_record.expire, payload=soa_record,
757 self.assertEqual(additional, [])
760 def _referralTest(self, method):
762 Create an authority and make a request against it. Then verify that the
763 result is a referral, including no records in the answers or additional
764 sections, but with an I{NS} record in the authority section.
766 subdomain = 'example.' + str(soa_record.mname)
767 nameserver = dns.Record_NS('1.2.3.4')
768 authority = NoFileAuthority(
769 soa=(str(soa_record.mname), soa_record),
774 d = getattr(authority, method)(subdomain)
776 d.addCallback(result.append)
777 answer, authority, additional = result[0]
778 self.assertEqual(answer, [])
780 authority, [dns.RRHeader(
781 subdomain, dns.NS, ttl=soa_record.expire,
782 payload=nameserver, auth=False)])
783 self.assertEqual(additional, [])
786 def test_referral(self):
788 When an I{NS} record is found for a child zone, it is included in the
789 authority section of the response. It is marked as non-authoritative if
790 the authority is not also authoritative for the child zone (RFC 2181,
793 self._referralTest('lookupAddress')
796 def test_allRecordsReferral(self):
798 A referral is also generated for a request of type C{ALL_RECORDS}.
800 self._referralTest('lookupAllRecords')
804 class NoInitialResponseTestCase(unittest.TestCase):
806 def test_no_answer(self):
808 If a request returns a L{dns.NS} response, but we can't connect to the
809 given server, the request fails with the error returned at connection.
812 def query(self, *args):
813 # Pop from the message list, so that it blows up if more queries
814 # are run than expected.
815 return succeed(messages.pop(0))
817 def queryProtocol(self, *args, **kwargs):
818 return defer.fail(socket.gaierror("Couldn't connect"))
820 resolver = Resolver(servers=[('0.0.0.0', 0)])
821 resolver._query = query
823 # Let's patch dns.DNSDatagramProtocol.query, as there is no easy way to
825 self.patch(dns.DNSDatagramProtocol, "query", queryProtocol)
828 dns.RRHeader(name='fooba.com', type=dns.NS, cls=dns.IN, ttl=700,
830 payload=dns.Record_NS(name='ns.twistedmatrix.com',
832 m = dns.Message(id=999, answer=1, opCode=0, recDes=0, recAv=1, auth=1,
833 rCode=0, trunc=0, maxSize=0)
836 return self.assertFailure(
837 resolver.getHostByName("fooby.com"), socket.gaierror)
841 class SecondaryAuthorityServiceTests(unittest.TestCase):
843 Tests for L{SecondaryAuthorityService}, a service which keeps one or more
844 authorities up to date by doing zone transfers from a master.
847 def test_constructAuthorityFromHost(self):
849 L{SecondaryAuthorityService} can be constructed with a C{str} giving a
850 master server address and several domains, causing the creation of a
851 secondary authority for each domain and that master server address and
852 the default DNS port.
854 primary = '192.168.1.2'
855 service = SecondaryAuthorityService(
856 primary, ['example.com', 'example.org'])
857 self.assertEqual(service.primary, primary)
858 self.assertEqual(service._port, 53)
860 self.assertEqual(service.domains[0].primary, primary)
861 self.assertEqual(service.domains[0]._port, 53)
862 self.assertEqual(service.domains[0].domain, 'example.com')
864 self.assertEqual(service.domains[1].primary, primary)
865 self.assertEqual(service.domains[1]._port, 53)
866 self.assertEqual(service.domains[1].domain, 'example.org')
869 def test_constructAuthorityFromHostAndPort(self):
871 L{SecondaryAuthorityService.fromServerAddressAndDomains} constructs a
872 new L{SecondaryAuthorityService} from a C{str} giving a master server
873 address and DNS port and several domains, causing the creation of a secondary
874 authority for each domain and that master server address and the given
877 primary = '192.168.1.3'
879 service = SecondaryAuthorityService.fromServerAddressAndDomains(
880 (primary, port), ['example.net', 'example.edu'])
881 self.assertEqual(service.primary, primary)
882 self.assertEqual(service._port, 5335)
884 self.assertEqual(service.domains[0].primary, primary)
885 self.assertEqual(service.domains[0]._port, port)
886 self.assertEqual(service.domains[0].domain, 'example.net')
888 self.assertEqual(service.domains[1].primary, primary)
889 self.assertEqual(service.domains[1]._port, port)
890 self.assertEqual(service.domains[1].domain, 'example.edu')
894 class SecondaryAuthorityTests(unittest.TestCase):
896 L{twisted.names.secondary.SecondaryAuthority} correctly constructs objects
897 with a specified IP address and optionally specified DNS port.
900 def test_defaultPort(self):
902 When constructed using L{SecondaryAuthority.__init__}, the default port
905 secondary = SecondaryAuthority('192.168.1.1', 'inside.com')
906 self.assertEqual(secondary.primary, '192.168.1.1')
907 self.assertEqual(secondary._port, 53)
908 self.assertEqual(secondary.domain, 'inside.com')
911 def test_explicitPort(self):
913 When constructed using L{SecondaryAuthority.fromServerAddressAndDomain},
914 the specified port is used.
916 secondary = SecondaryAuthority.fromServerAddressAndDomain(
917 ('192.168.1.1', 5353), 'inside.com')
918 self.assertEqual(secondary.primary, '192.168.1.1')
919 self.assertEqual(secondary._port, 5353)
920 self.assertEqual(secondary.domain, 'inside.com')
923 def test_transfer(self):
925 An attempt is made to transfer the zone for the domain the
926 L{SecondaryAuthority} was constructed with from the server address it
927 was constructed with when L{SecondaryAuthority.transfer} is called.
929 class ClockMemoryReactor(Clock, MemoryReactor):
932 MemoryReactor.__init__(self)
934 secondary = SecondaryAuthority.fromServerAddressAndDomain(
935 ('192.168.1.2', 1234), 'example.com')
936 secondary._reactor = reactor = ClockMemoryReactor()
940 # Verify a connection attempt to the server address above
941 host, port, factory, timeout, bindAddress = reactor.tcpClients.pop(0)
942 self.assertEqual(host, '192.168.1.2')
943 self.assertEqual(port, 1234)
945 # See if a zone transfer query is issued.
946 proto = factory.buildProtocol((host, port))
947 transport = StringTransport()
948 proto.makeConnection(transport)
951 # DNSProtocol.writeMessage length encodes the message by prepending a
952 # 2 byte message length to the buffered value.
953 msg.decode(StringIO(transport.value()[2:]))
956 [dns.Query('example.com', dns.AXFR, dns.IN)], msg.queries)