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