/usr/lib/python2.7/dist-packages/twisted/names/test/test_srvconnect.py is in python-twisted-names 13.2.0-1ubuntu1.
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 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 | # 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, ServiceNameUnknownError
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):
"""
Tests for L{srvconnect.SRVConnector}.
"""
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_SRVNoResultUnknownServiceDefaultPort(self):
"""
connectTCP gets called with default port if the service is not defined.
"""
self.connector = srvconnect.SRVConnector(self.reactor,
'thisbetternotexist',
'example.org', self.factory,
defaultPort=5222)
client.theResolver.failure = ServiceNameUnknownError()
self.connector.connect()
self.assertIdentical(None, self.factory.reason)
self.assertEqual(
self.reactor.tcpClients.pop()[:2], ('example.org', 5222))
def test_SRVNoResultUnknownServiceNoDefaultPort(self):
"""
Connect fails on no result, unknown service and no default port.
"""
self.connector = srvconnect.SRVConnector(self.reactor,
'thisbetternotexist',
'example.org', self.factory)
client.theResolver.failure = ServiceNameUnknownError()
self.connector.connect()
self.assertTrue(self.factory.reason.check(ServiceNameUnknownError))
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, [])
def test_unicodeDomain(self):
"""
L{srvconnect.SRVConnector} automatically encodes unicode domain using
C{idna} encoding.
"""
self.connector = srvconnect.SRVConnector(
self.reactor, 'xmpp-client', u'\u00e9chec.example.org',
self.factory)
self.assertIsInstance(self.connector.domain, bytes)
self.assertEqual(b'xn--chec-9oa.example.org', self.connector.domain)
|