← Back to branch summary

~rvb/maas/parent-child-relationship

« back to all changes in this revision

Viewing changes to src/provisioningserver/tests/test_cobbler_xmlrpc.py

  • Committer: Jeroen Vermeulen
  • Date: 2012-01-25 09:43:49 UTC
  • mto: This revision was merged to the branch mainline in revision 51.
  • Revision ID: jeroen.vermeulen@canonical.com-20120125094349-8u6aw3xbqawm6rpa
Cobbler XMLRPC API and beginning of test fake.

Show diffs side-by-side

added added

removed removed

Lines of Context:
src/provisioningserver/tests/test_cobbler_xmlrpc.py
 
1
# Copyright 2012 Canonical Ltd.  This software is licensed under the
 
2
# GNU Affero General Public License version 3 (see the file LICENSE).
 
3
 
 
4
from __future__ import (
 
5
    print_function,
 
6
    unicode_literals,
 
7
    )
 
8
 
 
9
"""Tests for `CobblerSession`."""
 
10
 
 
11
__metaclass__ = type
 
12
__all__ = []
 
13
 
 
14
from random import Random
 
15
from xmlrpclib import Fault
 
16
from twisted.internet.defer import inlineCallbacks, returnValue
 
17
 
 
18
from testtools.deferredruntest import AsynchronousDeferredRunTest
 
19
from unittest import TestCase
 
20
from provisioningserver import cobblerclient
 
21
from provisioningserver.fakecobbler import fake_token
 
22
 
 
23
 
 
24
randomizer = Random()
 
25
 
 
26
 
 
27
def pick_number():
 
28
    """Pick an arbitrary number."""
 
29
    return randomizer.randint(0, 10**9)
 
30
 
 
31
 
 
32
class FakeAuthFailure(Fault):
 
33
    """Imitated Cobbler authentication failure."""
 
34
 
 
35
    def __init__(self, token):
 
36
        super(FakeAuthFailure, self).__init__(1, "invalid token: %s" % token)
 
37
 
 
38
 
 
39
def make_auth_failure(broken_token=None):
 
40
    """Mimick a Cobbler authentication failure."""
 
41
    if broken_token is None:
 
42
        broken_token = fake_token()
 
43
    return FakeAuthFailure(broken_token)
 
44
 
 
45
 
 
46
class InstrumentedSession(cobblerclient.CobblerSession):
 
47
    """Instrumented `CobblerSession` with a fake XMLRPC proxy.
 
48
 
 
49
    :ivar fake_proxy: Test double for the XMLRPC proxy.
 
50
    :ivar fake_token: Auth token that login will pretend to receive.
 
51
    """
 
52
 
 
53
    def __init__(self, *args, **kwargs):
 
54
        """Create and instrument a session.
 
55
 
 
56
        In addition to the arguments for `CobblerSession.__init__`, pass a
 
57
        keyword argument `fake_proxy` to set a test double that the session
 
58
        will use for its proxy; and `fake_token` to provide a login token
 
59
        that the session should pretend it gets from the server on login.
 
60
        """
 
61
        self.fake_proxy = kwargs['fake_proxy']
 
62
        self.fake_token = kwargs['fake_token']
 
63
        del kwargs['fake_proxy']
 
64
        del kwargs['fake_token']
 
65
        super(InstrumentedSession, self).__init__(*args, **kwargs)
 
66
 
 
67
    def _make_twisted_proxy(self):
 
68
        return self.fake_proxy
 
69
 
 
70
    def _login(self):
 
71
        self.token = self.fake_token
 
72
 
 
73
 
 
74
class RecordingFakeProxy:
 
75
    """Simple fake Twisted XMLRPC proxy.
 
76
 
 
77
    Records XMLRPC calls, and returns predetermined values.
 
78
    """
 
79
    def __init__(self):
 
80
        self.calls = []
 
81
        self.return_values = None
 
82
 
 
83
    def set_return_values(self, values):
 
84
        """Set predetermined value to return on following call(s).
 
85
 
 
86
        If any return value is an `Exception`, it will be raised instead.
 
87
        """
 
88
        self.return_values = values
 
89
 
 
90
    @inlineCallbacks
 
91
    def callRemote(self, method, *args):
 
92
        self.calls.append((method, ) + tuple(args))
 
93
        if self.return_values:
 
94
            value = self.return_values.pop(0)
 
95
        else:
 
96
            value = None
 
97
        if isinstance(value, Exception):
 
98
            raise value
 
99
        else:
 
100
            value = yield value
 
101
            returnValue(value)
 
102
 
 
103
 
 
104
class TestCobblerSession(TestCase):
 
105
    """Test session management against a fake XMLRPC session."""
 
106
 
 
107
    run_tests_with = AsynchronousDeferredRunTest.make_factory()
 
108
 
 
109
    def make_url_user_password(self):
 
110
        """Produce arbitrary API URL, username, and password."""
 
111
        return (
 
112
            'http://api.example.com/%d' % pick_number(),
 
113
            'username%d' % pick_number(),
 
114
            'password%d' % pick_number(),
 
115
            )
 
116
 
 
117
    def make_recording_session(self, session_args=None, token=None):
 
118
        """Create an `InstrumentedSession` with a `RecordingFakeProxy`."""
 
119
        if session_args is None:
 
120
            session_args = self.make_url_user_password()
 
121
        if token is None:
 
122
            token = fake_token()
 
123
        fake_proxy = RecordingFakeProxy()
 
124
        return InstrumentedSession(
 
125
            *session_args, fake_proxy=fake_proxy, fake_token=token)
 
126
 
 
127
    def test_initializes_but_does_not_authenticate_on_creation(self):
 
128
        url, user, password = self.make_url_user_password()
 
129
        session = self.make_recording_session(token=fake_token())
 
130
        self.assertEqual(None, session.token)
 
131
 
 
132
    def test_authenticate_authenticates_initially(self):
 
133
        token = fake_token()
 
134
        session = self.make_recording_session(token=token)
 
135
        self.assertEqual(None, session.token)
 
136
        session.authenticate()
 
137
        self.assertEqual(token, session.token)
 
138
 
 
139
    def test_state_cookie_stays_constant_during_normal_use(self):
 
140
        session = self.make_recording_session()
 
141
        state = session.record_state()
 
142
        self.assertEqual(state, session.record_state())
 
143
        session.call("some_method")
 
144
        self.assertEqual(state, session.record_state())
 
145
 
 
146
    def test_authentication_changes_state_cookie(self):
 
147
        session = self.make_recording_session()
 
148
        old_cookie = session.record_state()
 
149
        session.authenticate()
 
150
        self.assertNotEqual(old_cookie, session.record_state())
 
151
 
 
152
    def test_authenticate_backs_off_from_overwriting_concurrent_auth(self):
 
153
        session = self.make_recording_session()
 
154
        # Two requests are made concurrently.
 
155
        cookie_before_request_1 = session.record_state()
 
156
        cookie_before_request_2 = session.record_state()
 
157
        # Request 1 comes back with an authentication failure, and its
 
158
        # callback refreshes the session's auth token.
 
159
        session.authenticate(cookie_before_request_1)
 
160
        token_for_retrying_request_1 = session.token
 
161
        # Request 2 also comes back an authentication failure, and its
 
162
        # callback also asks the session to ensure that it is
 
163
        # authenticated.
 
164
        session.authenticate(cookie_before_request_2)
 
165
        token_for_retrying_request_2 = session.token
 
166
 
 
167
        # The double authentication does not confuse the session; both
 
168
        # callbacks get the same auth token for their retries.
 
169
        self.assertEqual(
 
170
            token_for_retrying_request_1, token_for_retrying_request_2)
 
171
        # The token they get is a new token, not the one they started
 
172
        # with.
 
173
        self.assertNotEqual(cookie_before_request_1, session.token)
 
174
        self.assertNotEqual(cookie_before_request_2, session.token)
 
175
 
 
176
    def test_substitute_token_substitutes_only_placeholder(self):
 
177
        token = fake_token()
 
178
        session = self.make_recording_session(token=token)
 
179
        session.authenticate()
 
180
        arbitrary_number = pick_number()
 
181
        inputs = [
 
182
            arbitrary_number,
 
183
            cobblerclient.CobblerSession.token_placeholder,
 
184
            None,
 
185
            ]
 
186
        outputs = [
 
187
            arbitrary_number,
 
188
            token,
 
189
            None]
 
190
        self.assertEqual(outputs, map(session.substitute_token, inputs))
 
191
 
 
192
    @inlineCallbacks
 
193
    def test_call_calls_xmlrpc(self):
 
194
        session = self.make_recording_session()
 
195
        return_value = pick_number()
 
196
        method = 'method%d' % pick_number()
 
197
        arg = pick_number()
 
198
        session.fake_proxy.set_return_values([return_value])
 
199
        actual_return_value = yield session.call(method, arg)
 
200
        self.assertEqual(return_value, actual_return_value)
 
201
        self.assertEqual([(method, arg)], session.fake_proxy.calls)
 
202
 
 
203
    @inlineCallbacks
 
204
    def test_call_reauthenticates_and_retries_on_auth_failure(self):
 
205
        session = self.make_recording_session()
 
206
        fake_proxy = session.fake_proxy
 
207
        fake_proxy.set_return_values([make_auth_failure()])
 
208
        fake_proxy.calls = []
 
209
        yield session.call("failing_method")
 
210
        self.assertEqual(
 
211
            [
 
212
                # Initial call to failing_method: auth failure.
 
213
                ('failing_method', ),
 
214
                # But call() re-authenticates, and retries.
 
215
                ('failing_method', ),
 
216
            ],
 
217
            fake_proxy.calls)
 
218
 
 
219
    @inlineCallbacks
 
220
    def test_call_raises_repeated_auth_failure(self):
 
221
        session = self.make_recording_session()
 
222
        session.fake_proxy.set_return_values([
 
223
            # Initial operation fails: not authenticated.
 
224
            make_auth_failure(),
 
225
            # But retry still raises authentication failure.
 
226
            make_auth_failure(),
 
227
            ])
 
228
        try:
 
229
            d = session.call('double_fail')
 
230
            return_value = yield d
 
231
        except Exception:
 
232
            pass
 
233
        else:
 
234
            self.fail("Returned %s instead of raising." % return_value)
 
235
 
 
236
    @inlineCallbacks
 
237
    def test_call_raises_general_failure(self):
 
238
        session = self.make_recording_session()
 
239
        session.fake_proxy.set_return_values([
 
240
            Exception("Memory error.  Where did I put it?"),
 
241
            ])
 
242
        try:
 
243
            d = session.call('failing_method')
 
244
            return_value = yield d
 
245
        except Exception:
 
246
            pass
 
247
        else:
 
248
            self.assertTrue(
 
249
                False, "Returned %s instead of raising." % return_value)