1
# -*- coding: utf-8 -*-
3
# Authors: Alejandro J. Cura <alecu@canonical.com>
4
# Authors: Natalia B. Bidart <nataliabidart@canonical.com>
6
# Copyright 2010 Canonical Ltd.
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.
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.
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/>.
20
"""Tests for the control panel backend."""
24
from twisted.internet import defer
25
from twisted.internet.defer import inlineCallbacks
27
from ubuntuone.controlpanel import backend
28
from ubuntuone.controlpanel.backend import (ACCOUNT_API, QUOTA_API,
29
DEVICES_API, DEVICE_REMOVE_API)
30
from ubuntuone.controlpanel.tests import TestCase
31
from ubuntuone.controlpanel.webclient import WebClientError
33
SAMPLE_CREDENTIALS = {"token": "ABC1234DEF"}
35
SAMPLE_ACCOUNT_JSON = """
37
"username": "andrewpz",
38
"openid": "https://login.launchpad.net/+id/abcdefg",
39
"first_name": "Andrew P.",
42
"host": "https://couchdb.one.ubuntu.com",
43
"root": "https://couchdb.one.ubuntu.com/u/abc/def/12345",
44
"dbpath": "u/abc/def/12345"
46
"couchdb_root": "https://couchdb.one.ubuntu.com/u/abc/def/12345",
47
"email": "andrewpz@protocultura.net",
48
"nickname": "Andrew P. Zoilo",
51
"upgrade_available": false,
52
"description": "Paid Plan, 50 GB of storage",
54
"started": "2010-03-24T18:38:38Z",
66
SAMPLE_QUOTA_JSON = """
73
EXPECTED_ACCOUNT_INFO = {
74
"quota_used": "2350345156",
75
"quota_total": "53687091200",
76
"type": "Paid Plan, 50 GB of storage",
77
"name": "Andrew P. Zoilo",
78
"email": "andrewpz@protocultura.net",
81
SAMPLE_DEVICES_JSON = """
84
"token": "ABCDEF01234token",
85
"description": "Ubuntu One @ darkstar",
89
"token": "ABC1234DEF",
90
"description": "Ubuntu One @ localhost",
95
"description": "Nokia E65",
101
EXPECTED_DEVICES_INFO = [
103
"device_id": "ComputerABCDEF01234token",
104
"name": "Ubuntu One @ darkstar",
110
'device_id': 'ComputerABC1234DEF',
111
'limit_bandwidth': "0",
112
'max_download_speed': '-1',
113
'max_upload_speed': '-1',
114
'name': 'Ubuntu One @ localhost',
118
"device_id": "Phone1000",
126
{'volume_id': 'test1', 'path': '~/test1', 'subscribed': False},
127
{'volume_id': 'test2', 'path': '~/test2', 'subscribed': True},
131
class MockWebClient(object):
132
"""A mock webclient."""
136
def __init__(self, get_credentials):
137
"""Initialize this mock instance."""
138
self.get_credentials = get_credentials
140
def call_api(self, method):
141
"""Get a given url from the webservice."""
143
return defer.fail(WebClientError(self.failure))
145
result = simplejson.loads(self.results[method])
146
return defer.succeed(result)
149
class MockDBusClient(object):
150
"""A mock dbus_client module."""
152
creds = SAMPLE_CREDENTIALS
154
limits = {"download": -1, "upload": -1}
156
def get_credentials(self):
157
"""Return the mock credentials."""
158
return defer.succeed(self.creds)
160
def get_throttling_limits(self):
161
"""Return the sample speed limits."""
164
def set_throttling_limits(self, limits):
165
"""Set the sample speed limits."""
166
self.limits["download"] = int(limits["download"])
167
self.limits["upload"] = int(limits["upload"])
169
def bandwidth_throttling_enabled(self):
170
"""Return the state of throttling."""
171
return self.throttling
173
def enable_bandwidth_throttling(self):
174
"""Enable bw throttling."""
175
self.throttling = True
177
def disable_bandwidth_throttling(self):
178
"""Disable bw throttling."""
179
self.throttling = False
181
def get_volumes(self):
182
"""Grab list of folders and shares."""
183
return SAMPLE_VOLUMES
186
class BackendBasicTestCase(TestCase):
187
"""Simple tests for the backend."""
192
self.patch(backend, "WebClient", MockWebClient)
193
self.patch(backend, "dbus_client", MockDBusClient())
194
self.local_token = "Computer" + SAMPLE_CREDENTIALS["token"]
195
self.be = backend.ControlBackend()
197
def test_backend_creation(self):
198
"""The backend instance is successfully created."""
199
self.assertEqual(self.be.wc.__class__, MockWebClient)
202
def test_get_token(self):
203
"""The get_token method returns the right token."""
204
token = yield self.be.get_token()
205
self.assertEqual(token, SAMPLE_CREDENTIALS["token"])
208
class BackendAccountTestCase(BackendBasicTestCase):
209
"""Account tests for the backend."""
212
def test_account_info(self):
213
"""The account_info method exercises its callback."""
214
# pylint: disable=E1101
215
self.be.wc.results[ACCOUNT_API] = SAMPLE_ACCOUNT_JSON
216
self.be.wc.results[QUOTA_API] = SAMPLE_QUOTA_JSON
217
result = yield self.be.account_info()
218
self.assertEqual(result, EXPECTED_ACCOUNT_INFO)
221
def test_account_info_fails(self):
222
"""The account_info method exercises its errback."""
223
# pylint: disable=E1101
224
self.be.wc.failure = 404
225
yield self.assertFailure(self.be.account_info(), WebClientError)
228
class BackendDevicesTestCase(BackendBasicTestCase):
229
"""Devices tests for the backend."""
232
def test_devices_info(self):
233
"""The devices_info method exercises its callback."""
234
# pylint: disable=E1101
235
self.be.wc.results[DEVICES_API] = SAMPLE_DEVICES_JSON
236
result = yield self.be.devices_info()
237
self.assertEqual(result, EXPECTED_DEVICES_INFO)
240
def test_devices_info_fails(self):
241
"""The devices_info method exercises its errback."""
242
# pylint: disable=E1101
243
self.be.wc.failure = 404
244
yield self.assertFailure(self.be.devices_info(), WebClientError)
247
def test_remove_device(self):
248
"""The remove_device method calls the right api."""
249
dtype, did = "Computer", "SAMPLE-TOKEN"
250
device_id = dtype + did
251
apiurl = DEVICE_REMOVE_API % (dtype, did)
252
# pylint: disable=E1101
253
self.be.wc.results[apiurl] = SAMPLE_DEVICES_JSON
254
result = yield self.be.remove_device(device_id)
255
self.assertEqual(result, device_id)
258
def test_remove_device_fails(self):
259
"""The remove_device method fails as expected."""
260
# pylint: disable=E1101
261
self.be.wc.failure = 404
262
yield self.assertFailure(self.be.devices_info(), WebClientError)
265
def test_change_limit_bandwidth(self):
266
"""The device settings are updated."""
267
backend.dbus_client.throttling = False
268
yield self.be.change_device_settings(self.local_token,
269
{"limit_bandwidth": "1"})
270
self.assertEqual(backend.dbus_client.throttling, True)
271
yield self.be.change_device_settings(self.local_token,
272
{"limit_bandwidth": "0"})
273
self.assertEqual(backend.dbus_client.throttling, False)
276
def test_change_upload_speed_limit(self):
277
"""The device settings are updated."""
278
backend.dbus_client.limits = {"download": -1, "upload": -1}
279
yield self.be.change_device_settings(self.local_token,
280
{"max_upload_speed": "1111"})
281
self.assertEqual(backend.dbus_client.limits["upload"], 1111)
282
self.assertEqual(backend.dbus_client.limits["download"], -1)
285
def test_change_download_speed_limit(self):
286
"""The device settings are updated."""
287
backend.dbus_client.limits = {"download": -1, "upload": -1}
288
yield self.be.change_device_settings(self.local_token,
289
{"max_download_speed": "99"})
290
self.assertEqual(backend.dbus_client.limits["upload"], -1)
291
self.assertEqual(backend.dbus_client.limits["download"], 99)
294
def test_changing_settings_for_wrong_id_has_no_effect(self):
295
"""If the id is wrong, no settings are changed."""
296
backend.dbus_client.throttling = False
297
backend.dbus_client.limits = {"download": -1, "upload": -1}
299
"max_download_speed": "99",
300
"max_upload_speed": "99",
301
"limit_bandwidth": "1",
303
yield self.be.change_device_settings("wrong token!", new_settings)
304
self.assertEqual(backend.dbus_client.throttling, False)
305
self.assertEqual(backend.dbus_client.limits["upload"], -1)
306
self.assertEqual(backend.dbus_client.limits["download"], -1)
309
class BackendVolumesTestCase(BackendBasicTestCase):
310
"""Volumes tests for the backend."""
313
def test_volumes_info(self):
314
"""The volumes_info method exercises its callback."""
315
result = yield self.be.volumes_info()
316
self.assertEqual(result, SAMPLE_VOLUMES)