~ubuntu-branches/ubuntu/quantal/ubuntuone-control-panel/quantal-proposed
» Revision
1
: ubuntuone/controlpanel/integrationtests/test_dbus_service.py
« back to all changes in this revision
Viewing changes to ubuntuone/controlpanel/integrationtests/test_dbus_service.py
- browse files at revision 1
- compare with another revision
- download diff
- download tarball
- view history from revision 1
- Committer: Bazaar Package Importer
- Author(s): Natalia Bidart (nessita)
- Date: 2010-12-06 12:27:11 UTC
- Revision ID: james.westby@ubuntu.com-20101206122711-0wvvlliao34bjztf
Tags: upstream-0.0.9
Import upstream version 0.0.9
- files added:
- bin
- data
- docs
- po
- ubuntuone
- ubuntuone/controlpanel
- ubuntuone/controlpanel/gtk
- ubuntuone/controlpanel/gtk/tests
- ubuntuone/controlpanel/integrationtests
- ubuntuone/controlpanel/tests
added
removed
ubuntuone/controlpanel/integrationtests/test_dbus_service.py
1
# -*- coding: utf-8 -*-
2
3
# Authors: Alejandro J. Cura <alecu@canonical.com>
4
# Authors: Natalia B. Bidart <nataliabidart@canonical.com>
5
#
6
# Copyright 2010 Canonical Ltd.
7
#
8
# This program is free software: you can redistribute it and/or modify it
9
# under the terms of the GNU General Public License version 3, as published
10
# by the Free Software Foundation.
11
#
12
# This program is distributed in the hope that it will be useful, but
13
# WITHOUT ANY WARRANTY; without even the implied warranties of
14
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
15
# PURPOSE. See the GNU General Public License for more details.
16
#
17
# You should have received a copy of the GNU General Public License along
18
# with this program. If not, see <http://www.gnu.org/licenses/>.
19
20
"""Tests for the control panel backend DBus service."""
21
22
import dbus
23
import mocker
24
25
from twisted.internet import defer
26
from twisted.python.failure import Failure
27
28
from ubuntuone.controlpanel import dbus_service
29
from ubuntuone.controlpanel import (DBUS_BUS_NAME, DBUS_PREFERENCES_PATH,
30
DBUS_PREFERENCES_IFACE)
31
from ubuntuone.controlpanel.integrationtests import TestCase
32
33
34
SAMPLE_ACCOUNT_INFO = {
35
"quota_used": "12345",
36
"quota_total": "54321",
37
"type": "paying customer",
38
"name": "john carlos",
39
"email": "john@carlos.com",
40
}
41
42
SAMPLE_DEVICES_INFO = [
43
{
44
"device_token": "token-1",
45
"name": "Ubuntu One @ darkstar",
46
"date_added": "2008-12-06T18:15:38.0",
47
"type": "computer",
48
"configurable": "1",
49
"limit_bandwidth": "1",
50
"max_upload_speed": "12345",
51
"max_download_speed": "54321",
52
"available_services": "files, contacts, music, bookmarks",
53
"enabled_services": "files, music",
54
},
55
{
56
"device_token": "token-2",
57
"name": "Ubuntu One @ brightmoon",
58
"date_added": "2010-09-22T20:45:38.0",
59
"type": "computer",
60
"configurable": "0",
61
"available_services": "files, contacts, bookmarks",
62
"enabled_services": "files, bookmarks",
63
},
64
]
65
66
SAMPLE_VOLUMES_INFO = [
67
{
68
"volume_id": "volume-0",
69
"path": "/home/user/Documents",
70
"suggested_path": "~/Documents",
71
"subscribed": "1",
72
},
73
{
74
"volume_id": "volume-1",
75
"path": "/home/user/Music",
76
"suggested_path": "~/Music",
77
"subscribed": "0",
78
},
79
{
80
"volume_id": "volume-2",
81
"path": "/home/user/Pictures/Photos",
82
"suggested_path": "~/Pictures/Photos",
83
"subscribed": "0",
84
},
85
]
86
87
88
class DBusServiceMainTestCase(mocker.MockerTestCase):
89
"""Tests for the main function."""
90
91
def test_dbus_service_main(self):
92
"""The main method starts the loop and hooks up to DBus."""
93
rs_name = "ubuntuone.controlpanel.dbus_service.register_service"
94
rs = self.mocker.replace(rs_name)
95
rs()
96
self.mocker.result(True)
97
rml_name = "ubuntuone.controlpanel.dbus_service.run_mainloop"
98
rml = self.mocker.replace(rml_name)
99
rml()
100
pb_name = "ubuntuone.controlpanel.dbus_service.publish_backend"
101
pb = self.mocker.replace(pb_name)
102
pb()
103
self.mocker.replay()
104
dbus_service.main()
105
106
def test_dbus_service_cant_register(self):
107
"""The main method can't start the loop."""
108
rs_name = "ubuntuone.controlpanel.dbus_service.register_service"
109
rs = self.mocker.replace(rs_name)
110
rs()
111
self.mocker.result(False)
112
self.mocker.replay()
113
dbus_service.main()
114
115
116
class MockBackend(object):
117
"""A mock backend."""
118
exception = None
119
120
def _process(self, result):
121
"""Process the request with the given result."""
122
if self.exception:
123
# pylint: disable=E1102
124
return defer.fail(self.exception(result))
125
return defer.succeed(result)
126
127
def account_info(self):
128
"""Get the user account info."""
129
return self._process(SAMPLE_ACCOUNT_INFO)
130
131
def devices_info(self):
132
"""Get the user devices info."""
133
return self._process(SAMPLE_DEVICES_INFO)
134
135
def change_device_settings(self, token, settings):
136
"""Configure a given device."""
137
return self._process(token)
138
139
def remove_device(self, token):
140
"""Remove a device's tokens from the sso server."""
141
return self._process(token)
142
143
def file_sync_status(self):
144
"""Return the status of the file sync service."""
145
return self._process((True, "Synchronizing"))
146
147
def volumes_info(self):
148
"""Get the user volumes info."""
149
return self._process(SAMPLE_VOLUMES_INFO)
150
151
def change_volume_settings(self, volume_id, settings):
152
"""Configure a given volume."""
153
return self._process(volume_id)
154
155
def query_bookmark_extension(self):
156
"""True if the bookmark extension has been installed."""
157
return self._process(False)
158
159
def install_bookmarks_extension(self):
160
"""Install the extension to sync bookmarks."""
161
return self._process(None)
162
163
164
class DBusServiceTestCase(TestCase):
165
"""Test for the DBus service."""
166
167
timeout = 5
168
169
def setUp(self):
170
"""Initialize each test run."""
171
super(DBusServiceTestCase, self).setUp()
172
dbus_service.init_mainloop()
173
174
def test_register_service(self):
175
"""The DBus service is successfully registered."""
176
ret = dbus_service.register_service()
177
self.assertTrue(ret)
178
179
def test_cant_register_twice(self):
180
"""The DBus service can't register if it already did."""
181
ret = dbus_service.register_service()
182
self.assertTrue(ret)
183
ret = dbus_service.register_service()
184
self.assertFalse(ret)
185
#pylint: disable=W0612
186
test_cant_register_twice.skip = "Must run 2nd check in another process."
187
188
def test_dbus_busname_created(self):
189
"""The DBus BusName is created."""
190
busname = dbus_service.get_busname()
191
self.assertEqual(busname.get_name(), DBUS_BUS_NAME)
192
193
def test_error_handler_with_failure(self):
194
"""Ensure to build a string-string dict to pass to error signals."""
195
error = dbus_service.Failure(TypeError('oh no!'))
196
expected = dbus_service.utils.failure_to_error_dict(error)
197
198
result = dbus_service.error_handler(error)
199
200
self.assertEqual(expected, result)
201
202
def test_error_handler_with_exception(self):
203
"""Ensure to build a string-string dict to pass to error signals."""
204
error = TypeError('oh no, no again!')
205
expected = dbus_service.utils.exception_to_error_dict(error)
206
207
result = dbus_service.error_handler(error)
208
209
self.assertEqual(expected, result)
210
211
def test_error_handler_with_string_dict(self):
212
"""Ensure to build a string-string dict to pass to error signals."""
213
expected = {'test': 'me'}
214
215
result = dbus_service.error_handler(expected)
216
217
self.assertEqual(expected, result)
218
219
def test_error_handler_with_non_string_dict(self):
220
"""Ensure to build a string-string dict to pass to error signals."""
221
expected = {'test': 0, 'qué?': None,
222
10: 'foo\xffbar', True: u'ñoño'}
223
224
result = dbus_service.error_handler(expected)
225
expected = dict(map(lambda x: x if isinstance(x, unicode) else
226
str(x).decode('utf8', 'replace'), i)
227
for i in expected.iteritems())
228
229
self.assertEqual(expected, result)
230
231
def test_error_handler_default(self):
232
"""Ensure to build a string-string dict to pass to error signals."""
233
msg = 'Got unexpected error argument %r' % None
234
expected = {dbus_service.utils.ERROR_TYPE: 'UnknownError',
235
dbus_service.utils.ERROR_MESSAGE: msg}
236
237
result = dbus_service.error_handler(None)
238
239
self.assertEqual(expected, result)
240
241
242
class OperationsTestCase(TestCase):
243
"""Test for the DBus service operations."""
244
245
timeout = 3
246
247
def setUp(self):
248
super(OperationsTestCase, self).setUp()
249
dbus_service.init_mainloop()
250
be = dbus_service.publish_backend(MockBackend())
251
self.addCleanup(be.remove_from_connection)
252
bus = dbus.SessionBus()
253
obj = bus.get_object(bus_name=DBUS_BUS_NAME,
254
object_path=DBUS_PREFERENCES_PATH,
255
follow_name_owner_changes=True)
256
self.backend = dbus.Interface(object=obj,
257
dbus_interface=DBUS_PREFERENCES_IFACE)
258
self.deferred = defer.Deferred()
259
260
def tearDown(self):
261
self.backend = None
262
self.deferred = None
263
super(OperationsTestCase, self).tearDown()
264
265
def got_error(self, *a):
266
"""Some error happened in the DBus call."""
267
self.deferred.errback(*a)
268
269
def ignore(self, *a):
270
"""Do nothing with the returned value."""
271
272
def errback_on_error(self, f):
273
"""Call the given 'f' but errback self.deferred on any error."""
274
275
def inner(*a, **kw):
276
"""Call the given 'f' but errback self.deferred on any error."""
277
try:
278
return f(*a, **kw)
279
except Exception, e: # pylint: disable=W0703
280
self.deferred.errback(Failure(e))
281
282
return inner
283
284
def assert_correct_method_call(self, success_sig, error_sig, success_cb,
285
method, *args):
286
"""Connect 'success_cb' with 'success_sig', and call 'method'.
287
288
'error_sig' will be connected to the class' error handler.
289
290
'success_cb' should fire 'self.deferred' on success. If 'success_cb'
291
fails due to an failed assertion, the deferred will be errback'd.
292
293
"""
294
success_cb = self.errback_on_error(success_cb)
295
res = self.backend.connect_to_signal(success_sig, success_cb)
296
self.addCleanup(res.remove)
297
298
res = self.backend.connect_to_signal(error_sig, self.got_error)
299
self.addCleanup(res.remove)
300
301
method = self.errback_on_error(method)
302
method(*args, reply_handler=self.ignore, error_handler=self.got_error)
303
304
return self.deferred
305
306
def test_account_info_returned(self):
307
"""The account info is successfully returned."""
308
309
def got_signal(account_info):
310
"""The correct signal was fired."""
311
self.assertIn("quota_used", account_info)
312
self.assertIn("quota_total", account_info)
313
self.assertIn("type", account_info)
314
self.assertIn("name", account_info)
315
self.assertIn("email", account_info)
316
self.deferred.callback("success")
317
318
args = ("AccountInfoReady", "AccountInfoError", got_signal,
319
self.backend.account_info)
320
return self.assert_correct_method_call(*args)
321
322
def test_devices_info_returned(self):
323
"""The devices info is successfully returned."""
324
325
def got_signal(devices_list):
326
"""The correct signal was fired."""
327
for device_info in devices_list:
328
self.assertIn("device_token", device_info)
329
self.assertIn("name", device_info)
330
self.assertIn("date_added", device_info)
331
self.assertIn("type", device_info)
332
self.assertIn("configurable", device_info)
333
if int(device_info["configurable"]):
334
self.assertIn("limit_bandwidth", device_info)
335
self.assertIn("max_upload_speed", device_info)
336
self.assertIn("max_download_speed", device_info)
337
self.assertIn("available_services", device_info)
338
self.assertIn("enabled_services", device_info)
339
self.deferred.callback("success")
340
341
args = ("DevicesInfoReady", "DevicesInfoError", got_signal,
342
self.backend.devices_info)
343
return self.assert_correct_method_call(*args)
344
345
def test_change_device_settings(self):
346
"""The device settings are successfully changed."""
347
sample_token = "token-1"
348
349
def got_signal(token):
350
"""The correct token was received."""
351
self.assertEqual(token, sample_token)
352
self.deferred.callback("success")
353
354
settings = {
355
"enabled_services": "files, contacts",
356
}
357
args = ("DeviceSettingsChanged", "DeviceSettingsChangeError",
358
got_signal, self.backend.change_device_settings,
359
sample_token, settings)
360
return self.assert_correct_method_call(*args)
361
362
def test_remove_device(self):
363
"""The device is removed."""
364
sample_token = "token-1"
365
366
def got_signal(token):
367
"""The correct token was received."""
368
self.assertEqual(token, sample_token)
369
self.deferred.callback("success")
370
371
args = ("DeviceRemoved", "DeviceRemovalError", got_signal,
372
self.backend.remove_device, sample_token)
373
return self.assert_correct_method_call(*args)
374
375
def test_file_sync_status(self):
376
"""The file sync status is reported."""
377
378
def got_signal(enabled, status):
379
"""The correct status was received."""
380
self.assertEqual(enabled, True)
381
self.assertEqual(status, "Synchronizing")
382
self.deferred.callback("success")
383
384
args = ("FileSyncStatusReady", "FileSyncStatusError", got_signal,
385
self.backend.file_sync_status)
386
return self.assert_correct_method_call(*args)
387
388
def test_volumes_info(self):
389
"""The volumes info is reported."""
390
391
def got_signal(volumes_dict):
392
"""The correct info was received."""
393
self.assertEqual(volumes_dict, SAMPLE_VOLUMES_INFO)
394
self.deferred.callback("success")
395
396
args = ("VolumesInfoReady", "VolumesInfoError", got_signal,
397
self.backend.volumes_info)
398
return self.assert_correct_method_call(*args)
399
400
def test_change_volume_settings(self):
401
"""The volume settings are successfully changed."""
402
expected_volume_id = SAMPLE_VOLUMES_INFO[0]['volume_id']
403
404
def got_signal(volume_id):
405
"""The correct volume was changed."""
406
self.assertEqual(volume_id, expected_volume_id)
407
self.deferred.callback("success")
408
409
args = ("VolumeSettingsChanged", "VolumeSettingsChangeError",
410
got_signal, self.backend.change_volume_settings,
411
expected_volume_id, {'subscribed': '0'})
412
return self.assert_correct_method_call(*args)
413
414
def test_query_bookmarks_extension(self):
415
"""The bookmarks extension is queried."""
416
417
def got_signal(enabled):
418
"""The correct status was received."""
419
self.assertEqual(enabled, False)
420
self.deferred.callback("success")
421
422
args = ("QueryBookmarksResult", "QueryBookmarksError", got_signal,
423
self.backend.query_bookmark_extension)
424
return self.assert_correct_method_call(*args)
425
426
def test_install_bookmarks_extension(self):
427
"""The bookmarks extension is installed."""
428
429
def got_signal():
430
"""The extension was installed."""
431
self.deferred.callback("success")
432
433
args = ("InstallBookmarksSuccess", "InstallBookmarksError", got_signal,
434
self.backend.install_bookmarks_extension)
435
return self.assert_correct_method_call(*args)
436
437
438
class OperationsErrorTestCase(OperationsTestCase):
439
"""Test for the DBus service operations when there is an error."""
440
441
def setUp(self):
442
super(OperationsErrorTestCase, self).setUp()
443
self.patch(MockBackend, 'exception', AssertionError)
444
445
def assert_correct_method_call(self, success_sig, error_sig, success_cb,
446
method, *args):
447
"""Call parent instance swapping success_sig with error_sig.
448
449
This is because we want to succeed the test when the error signal was
450
received.
451
452
"""
453
454
def got_error_signal(error_dict):
455
"""The error signal was received."""
456
self.assertEqual(error_dict[dbus_service.utils.ERROR_TYPE],
457
'AssertionError')
458
self.deferred.callback("success")
459
460
return super(OperationsErrorTestCase, self).assert_correct_method_call(
461
error_sig, success_sig, got_error_signal, method, *args)
Loggerhead is a web-based interface for Breezy
Version: 2.0.1