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
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
|
# test_auth - Tests for canonical.ubuntuone.oauthdesktop.auth module
#
# Author: Stuart Langridge <stuart.langridge@canonical.com>
#
# Copyright 2009 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License version 3, as published
# by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see <http://www.gnu.org/licenses/>.
"""Tests for the OAuth client code for StorageFS."""
import gnomekeyring
from contrib.mocker import ANY, IN, MockerTestCase
import testresources
from canonical.ubuntuone.storage.protocol import oauth
from canonical.ubuntuone.oauthdesktop.auth import (AuthorisationClient,
NoAccessToken)
class AuthorisationClientTests(MockerTestCase):
"""Test the GNOME keyring integration portions of the auth code."""
def setUp(self):
"""Sets up a mock keyring."""
MockerTestCase.setUp(self)
self.keyring = self.mocker.mock()
self.client = None
self.item = self.mocker.mock(gnomekeyring.Found)
self.item_id = 999
ex = self.expect(self.item.item_id)
ex.result(self.item_id)
ex.count(0, None)
ex = self.expect(self.item.secret)
ex.result('oauth_token=access_key&oauth_token_secret=access_secret')
ex.count(0, None)
def expect_token_query(self):
"""Expects the keyring to be queried for a token."""
return self.expect(
self.keyring.find_items_sync(
gnomekeyring.ITEM_GENERIC_SECRET,
{'ubuntuone-realm': 'realm',
'oauth-consumer-key': 'consumer_key'})
)
def expect_token_store(self):
"""Expects the token to be stored in the keyring."""
self.expect(self.keyring.item_create_sync(
None, gnomekeyring.ITEM_GENERIC_SECRET,
'UbuntuOne token for realm',
{'ubuntuone-realm': 'realm', 'oauth-consumer-key': 'consumer_key'},
# Either order for the token and secret is valid in the item secret
IN(['oauth_token=access_key&oauth_token_secret=access_secret',
'oauth_token_secret=access_secret&oauth_token=access_key']),
True)).result(self.item_id)
def mock_has_token(self):
"""Mocks a cached token in the keyring."""
self.expect_token_query().result([self.item])
def mock_no_token(self, exception):
"""Mocks no token in the keyring."""
self.expect_token_query().throw(exception)
def replay(self, callback_parent=None, callback_denied=None,
do_login=True):
"""Starts the replay phase and sets up a client object to be tested,
wired up to the mock keyring.
"""
self.mocker.replay()
self.client = AuthorisationClient(
'realm', 'request_token_url', 'user_authorisation_url',
'access_token_url', 'consumer_key', 'consumer_secret',
callback_parent, callback_denied, do_login, keyring=self.keyring)
def test_get_access_token(self):
"""The get_access_token method returns the access token"""
self.mock_has_token()
self.replay()
token = self.client.get_access_token()
self.assertTrue(isinstance(token, oauth.OAuthToken))
self.assertEqual(token.key, 'access_key')
self.assertEqual(token.secret, 'access_secret')
def test_get_access_token_no_match(self):
"""The get_access_token method fails if there are no matching items"""
self.mock_no_token(gnomekeyring.NoMatchError)
self.replay()
self.assertRaises(NoAccessToken, self.client.get_access_token)
def test_get_access_token_denied(self):
"""The get_access_token method fails if access is denied"""
self.mock_no_token(gnomekeyring.DeniedError)
self.replay()
self.assertRaises(NoAccessToken, self.client.get_access_token)
def test_have_access_token(self):
"""The `have_access_token` method returns True if a the
keyring contains a token."""
self.mock_has_token()
self.replay()
self.assertEqual(self.client.have_access_token(), True)
def test_have_access_token_fail(self):
"""The `have_access_token` method returns False if the keyring
does not contain a token."""
self.mock_no_token(gnomekeyring.NoMatchError)
self.replay()
self.assertEqual(self.client.have_access_token(), False)
def test_store_token(self):
"""The store_token method correctly stores an item in the keyring,
and correctly sets an ACL on it."""
self.expect_token_store()
saka = self.mocker.replace(
"canonical.ubuntuone.oauthdesktop.key_acls.set_all_key_acls")
saka(item_id=self.item_id)
self.mocker.result(None)
sleep = self.mocker.replace("time.sleep")
sleep(4)
self.mocker.result(None)
self.replay()
self.client.store_token(oauth.OAuthToken('access_key', 'access_secret'))
def test_clear_existing_token(self):
"""Makes sure that clear token clears an existing token."""
self.mock_has_token()
self.expect(self.keyring.item_delete_sync(None, self.item_id))
self.replay()
self.client.clear_token()
def test_clear_no_existing_token(self):
"""Makes sure that clear with no existing token still works."""
self.mock_no_token(gnomekeyring.NoMatchError)
self.replay()
self.client.clear_token()
def test_ensure_access_token(self):
"""If the user already has a token, no new token is requested."""
self.mock_has_token()
callback_function = self.mocker.mock()
callback_function(ANY)
self.replay(callback_parent=callback_function)
self.client.ensure_access_token()
def test_ensure_access_token_no_token(self):
"""If the user has no token, a new one is requested and stored, via
an OAuth callback to the internal webserver."""
self.mock_no_token(gnomekeyring.NoMatchError)
request_token = self.mocker.mock()
self.expect(request_token.key).result('access_key').count(0, None)
ex = self.expect(request_token.secret)
ex.result('access_secret').count(0, None)
make_token_request = self.mocker.mock()
self.expect(make_token_request(ANY)).result(request_token)
open_in_browser = self.mocker.mock()
open_in_browser(ANY)
ri = self.mocker.replace("random.randint")
ri(1000000, 10000000)
self.mocker.result(12345678)
get_temporary_httpd = self.mocker.mock()
get_temporary_httpd(12345678, ANY, True)
self.mocker.result("http://callbackurl")
self.replay(callback_parent=lambda a: None)
self.client.make_token_request = make_token_request
self.client.open_in_browser = open_in_browser
self.client.get_temporary_httpd = get_temporary_httpd
# skip the "are we online, via networkmanager" bit
self.client.acquire_access_token_if_online = \
self.client.acquire_access_token
self.client.ensure_access_token()
class AcquireAccessTokenTests(testresources.ResourcedTestCase):
"""OAuth token acquisition tests."""
def setUp(self):
super(AcquireAccessTokenTests, self).setUp()
def tearDown(self):
super(AcquireAccessTokenTests, self).tearDown()
def test_acquire_access_token(self):
"""Test that acquire_access_token() can acquire the access token."""
def make_token_request(oauth_request):
"""Make an OAuth token request via the test browser."""
return oauth.OAuthToken.from_string("oauth_token=access_token&" +
"oauth_token_secret=" +
"access_secret")
def open_in_browser(url):
"""Just return as we aren't subscribed to the page."""
return
def got_token(token):
"""Called with the token once auth is completed"""
self.assertTrue(isinstance(token, oauth.OAuthToken))
def get_temporary_httpd(nonce, retrieve_function, store):
"""Mock the temporary httpd and return a callback URL"""
# returns callback URL; this is an invalid URL, of course,
# (port is too high) but we check later that the mechanize
# browser tries to navigate there
return "http://127.0.0.1:99999/?nonce=99999"
def store_token(token):
"""Don't use the keyring; do nothing"""
pass
client = AuthorisationClient(
'http://ubuntuone/',
'http://ubuntuone/oauth/request/',
'http://ubuntuone/oauth/authorize/',
'http://ubuntuone/oauth/access/',
'consumer_key', 'consumer_secret',
callback_parent=got_token)
client.make_token_request = make_token_request
client.open_in_browser = open_in_browser
client.get_temporary_httpd = get_temporary_httpd
client.store_token = store_token
client.acquire_access_token('token description')
|