← Back to branch summary

~ubuntu-branches/ubuntu/lucid/landscape-client/lucid-updates

« back to all changes in this revision

Viewing changes to landscape/lib/tests/test_dns.py

  • Committer: Package Import Robot
  • Author(s): Andreas Hasenack
  • Date: 2012-04-10 14:28:48 UTC
  • mfrom: (1.1.27)
  • mto: This revision was merged to the branch mainline in revision 35.
  • Revision ID: package-import@ubuntu.com-20120410142848-7xsy4g2xii7y7ntc
Import upstream version 12.04.3

Show diffs side-by-side

added added

removed removed

Lines of Context:
landscape/lib/tests/test_dns.py
 
1
from landscape.tests.helpers import LandscapeTest
 
2
from landscape.lib.dns import (
 
3
    _lookup_server_record, _lookup_hostname, discover_server)
 
4
 
 
5
from twisted.internet import defer
 
6
from twisted.names import dns
 
7
from twisted.names.error import ResolverError
 
8
 
 
9
 
 
10
class FakeResolverResult(object):
 
11
    """
 
12
    A fake resolver result returned by L{FakeResolver}.
 
13
 
 
14
    @param type: The result type L{twisted.names.dns.SRV}
 
15
    @param payload: The result contents
 
16
    """
 
17
    def __init__(self):
 
18
        self.type = None
 
19
 
 
20
        class Payload(object):
 
21
            """
 
22
            A payload result returned by fake resolver.
 
23
 
 
24
            @param target: The result of the lookup
 
25
            """
 
26
            def __init__(self):
 
27
                self.target = ""
 
28
 
 
29
        class Target(object):
 
30
            """
 
31
            A payload target returned by fake resolver.
 
32
 
 
33
            @param name: The name contained by the target.
 
34
            """
 
35
            def __init__(self):
 
36
                self.name = ""
 
37
 
 
38
        self.payload = Payload()
 
39
        self.payload.target = Target()
 
40
 
 
41
 
 
42
class FakeResolver(object):
 
43
    """
 
44
    A fake resolver that mimics L{twisted.names.client.Resolver}
 
45
    """
 
46
    def __init__(self):
 
47
        self.results = None
 
48
        self.name = None
 
49
        self.queried = None
 
50
 
 
51
    def lookupService(self, arg1):
 
52
        self.queried = arg1
 
53
        deferred = defer.Deferred()
 
54
        deferred.callback(self.results)
 
55
        return deferred
 
56
 
 
57
    def getHostByName(self, arg1):
 
58
        self.queried = arg1
 
59
        deferred = defer.Deferred()
 
60
        deferred.callback(self.name)
 
61
        return deferred
 
62
 
 
63
 
 
64
class BadResolver(object):
 
65
    """
 
66
    A resolver that mimics L{twisted.names.client.Resolver} and always returns
 
67
    an error.
 
68
    """
 
69
    def lookupService(self, arg1):
 
70
        deferred = defer.Deferred()
 
71
        deferred.errback(ResolverError("Couldn't connect"))
 
72
        return deferred
 
73
 
 
74
    def getHostByName(self, arg1):
 
75
        deferred = defer.Deferred()
 
76
        deferred.errback(ResolverError("Couldn't connect"))
 
77
        return deferred
 
78
 
 
79
 
 
80
class DnsSrvLookupTest(LandscapeTest):
 
81
    def test_with_server_found(self):
 
82
        """
 
83
        Looking up a DNS SRV record should return the result of the lookup.
 
84
        """
 
85
        fake_result = FakeResolverResult()
 
86
        fake_result.type = dns.SRV
 
87
        fake_result.payload.target.name = "a.b.com"
 
88
        fake_resolver = FakeResolver()
 
89
        fake_resolver.results = [[fake_result]]
 
90
        query_string = "_landscape._tcp.mylandscapehost.com"
 
91
 
 
92
        def check(result):
 
93
            self.assertEqual(fake_resolver.queried, query_string)
 
94
            self.assertEqual("a.b.com", result)
 
95
 
 
96
        d = _lookup_server_record(fake_resolver, query_string)
 
97
        d.addCallback(check)
 
98
        return d
 
99
 
 
100
    def test_with_server_not_found(self):
 
101
        """
 
102
        Looking up a DNS SRV record and finding nothing exists should return
 
103
        an empty string.
 
104
        """
 
105
        fake_resolver = FakeResolver()
 
106
        fake_resolver.results = [[]]
 
107
 
 
108
        def check(result):
 
109
            self.assertEqual("", result)
 
110
 
 
111
        d = _lookup_server_record(fake_resolver,
 
112
                                 "_landscape._tcp.mylandscapehost.com")
 
113
        d.addCallback(check)
 
114
        return d
 
115
 
 
116
    def test_with_resolver_error(self):
 
117
        """A resolver error triggers error handling code."""
 
118
        # The failure should be properly logged
 
119
        logging_mock = self.mocker.replace("logging.info")
 
120
        logging_mock("SRV lookup of _landscape._tcp.mylandscapehost.com "
 
121
                     "failed.")
 
122
        self.mocker.replay()
 
123
 
 
124
        d = _lookup_server_record(BadResolver(),
 
125
                                 "_landscape._tcp.mylandscapehost.com")
 
126
        self.assertFailure(d, ResolverError)
 
127
        return d
 
128
 
 
129
 
 
130
class DnsNameLookupTest(LandscapeTest):
 
131
    def test_with_name_found(self):
 
132
        """
 
133
        Looking up a DNS name record should return the result of the lookup.
 
134
        """
 
135
        fake_resolver = FakeResolver()
 
136
        fake_resolver.name = "a.b.com"
 
137
        query_string = "landscape.localdomain"
 
138
 
 
139
        def check(result):
 
140
            self.assertEqual(fake_resolver.queried, query_string)
 
141
            self.assertEqual("a.b.com", result)
 
142
 
 
143
        d = _lookup_hostname(None, fake_resolver, query_string)
 
144
        d.addCallback(check)
 
145
        return d
 
146
 
 
147
    def test_with_name_not_found(self):
 
148
        """
 
149
        Looking up a DNS NAME record and not finding a result should return
 
150
        None.
 
151
        """
 
152
        fake_resolver = FakeResolver()
 
153
        fake_resolver.name = None
 
154
 
 
155
        def check(result):
 
156
            self.assertEqual(None, result)
 
157
 
 
158
        d = _lookup_hostname(None, fake_resolver, "landscape.localdomain")
 
159
        d.addCallback(check)
 
160
        return d
 
161
 
 
162
    def test_with_resolver_error(self):
 
163
        """A resolver error triggers error handling code."""
 
164
        # The failure should be properly logged
 
165
        logging_mock = self.mocker.replace("logging.info")
 
166
        logging_mock("Name lookup of landscape.localdomain failed.")
 
167
        self.mocker.replay()
 
168
 
 
169
        d = _lookup_hostname(None, BadResolver(), "landscape.localdomain")
 
170
        self.assertFailure(d, ResolverError)
 
171
        return d
 
172
 
 
173
 
 
174
class DiscoverServerTest(LandscapeTest):
 
175
    def test_srv_lookup(self):
 
176
        """The DNS name of the server is found using a SRV lookup."""
 
177
        fake_result = FakeResolverResult()
 
178
        fake_result.type = dns.SRV
 
179
        fake_result.payload.target.name = "a.b.com"
 
180
        fake_resolver = FakeResolver()
 
181
        fake_resolver.results = [[fake_result]]
 
182
 
 
183
        d = discover_server(resolver=fake_resolver)
 
184
 
 
185
        def check(result):
 
186
            self.assertEqual("a.b.com", result)
 
187
 
 
188
        d.addCallback(check)
 
189
        return d
 
190
 
 
191
    def test_a_name_lookup(self):
 
192
        """The DNS name of the server is found using an A name lookup."""
 
193
        fake_resolver = FakeResolver()
 
194
        fake_resolver.name = "x.y.com"
 
195
 
 
196
        d = discover_server(resolver=fake_resolver)
 
197
 
 
198
        def check(result):
 
199
            self.assertEqual("x.y.com", result)
 
200
 
 
201
        d.addCallback(check)
 
202
        return d
 
203
 
 
204
    def test_failed_lookup(self):
 
205
        """A resolver error is returned when server autodiscovery fails."""
 
206
        d = _lookup_server_record(BadResolver(), "landscape.localdomain")
 
207
        self.assertFailure(d, ResolverError)
 
208
        return d