← Back to branch summary

~vishvananda/nova/network-refactor

« back to all changes in this revision

Viewing changes to vendor/Twisted-10.0.0/twisted/names/test/test_srvconnect.py

  • Committer: Jesse Andrews
  • Date: 2010-05-28 06:05:26 UTC
  • Revision ID: git-v1:bf6e6e718cdc7488e2da87b21e258ccc065fe499
initial commit

Show diffs side-by-side

added added

removed removed

Lines of Context:
vendor/Twisted-10.0.0/twisted/names/test/test_srvconnect.py
 
1
# Copyright (c) 2007-2009 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.assertEquals(
 
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.assertEquals(
 
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.assertEquals(
 
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.assertEquals(
 
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.assertEquals(self.reactor.tcpClients, [])