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
|
# -*- coding: utf-8 -*-
#
# Author: Alejandro J. Cura <alecu@canonical.com>
#
# Copyright 2011 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 Ubuntu One credentials management IPC service."""
from twisted.internet import defer
from twisted.trial.unittest import TestCase
from ubuntuone.platform.credentials.windows import (
CredentialsManagement,
RemovableSignal,
)
TEST_APP_NAME = "test application"
TEST_ERROR_DICT = {}
TEST_CREDENTIALS = {
"token": "1234token",
}
class FakeSSOProxy(object):
"""A fake SSOProxy."""
def __init__(self):
"""Initialize this fake."""
signals = [
"credentials_stored",
"credentials_cleared",
"credentials_found",
"credentials_not_found",
"authorization_denied",
"credentials_error",
]
for s in signals:
handler_name = "on_%s" % s
callback_name = "on_%s_cb" % s
def make_handler(callback_name):
"""Create a handler for a given callback_name."""
def handler(*args):
"""A signal was called."""
callback = getattr(self, callback_name, None)
if callback is not None:
callback(*args)
return handler
setattr(self, handler_name, make_handler(callback_name))
setattr(self, callback_name, None)
def find_credentials(self, app_name, options):
"""Ask the U1 credentials."""
return defer.succeed(TEST_CREDENTIALS)
def clear_credentials(self, app_name, options):
"""Clear the U1 credentials."""
return defer.succeed(None)
def store_credentials(self, app_name, options):
"""Store the U1 credentials."""
return defer.succeed(None)
def register(self, app_name, options):
"""Store the U1 credentials."""
return defer.succeed(None)
def login(self, app_name, options):
"""Store the U1 credentials."""
return defer.succeed(None)
class RemovableSignalTestCase(TestCase):
"""Tests for RemovableSignal."""
def setUp(self):
"""Initialize this fake instance."""
self.proxy = FakeSSOProxy()
def test_dunder_callable(self):
"""__call__ works as expected."""
sample_store = []
test_cb = lambda app_name, creds: sample_store.append(creds)
rs = RemovableSignal(self.proxy, "on_credentials_found_cb", test_cb)
rs(TEST_APP_NAME, TEST_CREDENTIALS)
self.assertEqual(sample_store[0], TEST_CREDENTIALS)
def test_remove(self):
"""The signal has a .remove that removes the callback."""
sample_store = []
test_cb = lambda app_name, creds: sample_store.append(creds)
rs = RemovableSignal(self.proxy, "on_credentials_found_cb", test_cb)
rs.remove()
rs(TEST_APP_NAME, TEST_CREDENTIALS)
self.assertEqual(len(sample_store), 0)
class CredentialsManagementTestCase(TestCase):
"""Tests for CredentialsManagement."""
timeout = 5
def setUp(self):
"""Initialize these tests."""
self.proxy = FakeSSOProxy()
self.cm = CredentialsManagement(self.proxy)
def test_find_credentials(self):
"""Test the find_credentials method."""
d = defer.Deferred()
ok = lambda: d.callback("ok")
error = lambda *args: d.errback(args)
self.cm.find_credentials(reply_handler=ok, error_handler=error)
return d
def test_clear_credentials(self):
"""Test the clear_credentials method."""
d = defer.Deferred()
ok = lambda: d.callback("ok")
error = lambda *args: d.errback(args)
self.cm.clear_credentials(reply_handler=ok, error_handler=error)
return d
def test_store_credentials(self):
"""Test the store_credentials method."""
d = defer.Deferred()
ok = lambda: d.callback("ok")
error = lambda *args: d.errback(args)
self.cm.store_credentials(TEST_CREDENTIALS, reply_handler=ok,
error_handler=error)
return d
def test_register(self):
"""Test the register method."""
d = defer.Deferred()
ok = lambda: d.callback("ok")
error = lambda *args: d.errback(args)
self.cm.register({}, reply_handler=ok, error_handler=error)
return d
def test_login(self):
"""Test the login method."""
d = defer.Deferred()
ok = lambda: d.callback("ok")
error = lambda *args: d.errback(args)
self.cm.login({}, reply_handler=ok, error_handler=error)
return d
def test_register_to_credentials_found(self):
"""Test the register_to_credentials_found method."""
cb = lambda creds: self.assertEqual(creds, TEST_CREDENTIALS)
other_cb = self.cm.register_to_credentials_found(cb)
other_cb(TEST_APP_NAME, TEST_CREDENTIALS)
def test_register_to_credentials_not_found(self):
"""Test the register_to_credentials_not_found method."""
cb = lambda: None
other_cb = self.cm.register_to_credentials_not_found(cb)
other_cb(TEST_APP_NAME)
def _verify_not_called_twice(self, signal_name, *args):
"""Test that the callback is not called twice."""
d = defer.Deferred()
def signal_handler(*args):
"""Fake the behaviour of CredentialsManagementTool."""
d.callback(args[0] if len(args) > 0 else None)
register = getattr(self.cm, "register_to_" + signal_name)
signal = register(signal_handler)
proxy_cb = getattr(self.proxy, "on_" + signal_name)
proxy_cb(*args)
if getattr(signal, "remove", False):
signal.remove()
proxy_cb(*args)
def test_not_called_twice_credentials_stored(self):
"""Test that on_credentials_stored is not called twice."""
self._verify_not_called_twice("credentials_stored")
def test_not_called_twice_credentials_cleared(self):
"""Test that on_credentials_cleared is not called twice."""
self._verify_not_called_twice("credentials_cleared")
def test_not_called_twice_credentials_found(self):
"""Test that on_credentials_found is not called twice."""
self._verify_not_called_twice("credentials_found", TEST_APP_NAME,
TEST_CREDENTIALS)
def test_not_called_twice_credentials_not_found(self):
"""Test that on_credentials_not_found is not called twice."""
self._verify_not_called_twice("credentials_not_found")
def test_not_called_twice_authorization_denied(self):
"""Test that on_authorization_denied is not called twice."""
self._verify_not_called_twice("authorization_denied")
def test_not_called_twice_credentials_error(self):
"""Test that on_credentials_error is not called twice."""
self._verify_not_called_twice("credentials_error", TEST_ERROR_DICT)
|