twisted/names/test/test_srvconnect.py

   1 # Copyright (c) Twisted Matrix Laboratories.
   2 # See LICENSE for details.
   3
   4 """
   5 Test cases for L{twisted.names.srvconnect}.
   6 """
   7
   8 from twisted.internet import defer, protocol
   9 from twisted.names import client, dns, srvconnect
  10 from twisted.names.common import ResolverBase
  11 from twisted.names.error import DNSNameError
  12 from twisted.internet.error import DNSLookupError
  13 from twisted.trial import unittest
  14 from twisted.test.proto_helpers import MemoryReactor
  15
  16
  17 class FakeResolver(ResolverBase):
  18     """
  19     Resolver that only gives out one given result.
  20
  21     Either L{results} or L{failure} must be set and will be used for
  22     the return value of L{_lookup}
  23
  24     @ivar results: List of L{dns.RRHeader} for the desired result.
  25     @type results: C{list}
  26     @ivar failure: Failure with an exception from L{twisted.names.error}.
  27     @type failure: L{Failure<twisted.python.failure.Failure>}
  28     """
  29
  30     def __init__(self, results=None, failure=None):
  31         self.results = results
  32         self.failure = failure
  33
  34     def _lookup(self, name, cls, qtype, timeout):
  35         """
  36         Return the result or failure on lookup.
  37         """
  38         if self.results is not None:
  39             return defer.succeed((self.results, [], []))
  40         else:
  41             return defer.fail(self.failure)
  42
  43
  44
  45 class DummyFactory(protocol.ClientFactory):
  46     """
  47     Dummy client factory that stores the reason of connection failure.
  48     """
  49     def __init__(self):
  50         self.reason = None
  51
  52     def clientConnectionFailed(self, connector, reason):
  53         self.reason = reason
  54
  55 class SRVConnectorTest(unittest.TestCase):
  56
  57     def setUp(self):
  58         self.patch(client, 'theResolver', FakeResolver())
  59         self.reactor = MemoryReactor()
  60         self.factory = DummyFactory()
  61         self.connector = srvconnect.SRVConnector(self.reactor, 'xmpp-server',
  62                                                  'example.org', self.factory)
  63
  64
  65     def test_SRVPresent(self):
  66         """
  67         Test connectTCP gets called with the address from the SRV record.
  68         """
  69         payload = dns.Record_SRV(port=6269, target='host.example.org', ttl=60)
  70         client.theResolver.results = [dns.RRHeader(name='example.org',
  71                                                    type=dns.SRV,
  72                                                    cls=dns.IN, ttl=60,
  73                                                    payload=payload)]
  74         self.connector.connect()
  75
  76         self.assertIdentical(None, self.factory.reason)
  77         self.assertEqual(
  78             self.reactor.tcpClients.pop()[:2], ('host.example.org', 6269))
  79
  80
  81     def test_SRVNotPresent(self):
  82         """
  83         Test connectTCP gets called with fallback parameters on NXDOMAIN.
  84         """
  85         client.theResolver.failure = DNSNameError('example.org')
  86         self.connector.connect()
  87
  88         self.assertIdentical(None, self.factory.reason)
  89         self.assertEqual(
  90             self.reactor.tcpClients.pop()[:2], ('example.org', 'xmpp-server'))
  91
  92
  93     def test_SRVNoResult(self):
  94         """
  95         Test connectTCP gets called with fallback parameters on empty result.
  96         """
  97         client.theResolver.results = []
  98         self.connector.connect()
  99
 100         self.assertIdentical(None, self.factory.reason)
 101         self.assertEqual(
 102             self.reactor.tcpClients.pop()[:2], ('example.org', 'xmpp-server'))
 103
 104
 105     def test_SRVBadResult(self):
 106         """
 107         Test connectTCP gets called with fallback parameters on bad result.
 108         """
 109         client.theResolver.results = [dns.RRHeader(name='example.org',
 110                                                    type=dns.CNAME,
 111                                                    cls=dns.IN, ttl=60,
 112                                                    payload=None)]
 113         self.connector.connect()
 114
 115         self.assertIdentical(None, self.factory.reason)
 116         self.assertEqual(
 117             self.reactor.tcpClients.pop()[:2], ('example.org', 'xmpp-server'))
 118
 119
 120     def test_SRVNoService(self):
 121         """
 122         Test that connecting fails when no service is present.
 123         """
 124         payload = dns.Record_SRV(port=5269, target='.', ttl=60)
 125         client.theResolver.results = [dns.RRHeader(name='example.org',
 126                                                    type=dns.SRV,
 127                                                    cls=dns.IN, ttl=60,
 128                                                    payload=payload)]
 129         self.connector.connect()
 130
 131         self.assertNotIdentical(None, self.factory.reason)
 132         self.factory.reason.trap(DNSLookupError)
 133         self.assertEqual(self.reactor.tcpClients, [])