1
# -*- coding: utf-8 -*-
3
# Authors: Alejandro J. Cura <alecu@canonical.com>
5
# Copyright 2011 Canonical Ltd.
7
# This program is free software: you can redistribute it and/or modify it
8
# under the terms of the GNU General Public License version 3, as published
9
# by the Free Software Foundation.
11
# This program is distributed in the hope that it will be useful, but
12
# WITHOUT ANY WARRANTY; without even the implied warranties of
13
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
14
# PURPOSE. See the GNU General Public License for more details.
16
# You should have received a copy of the GNU General Public License along
17
# with this program. If not, see <http://www.gnu.org/licenses/>.
19
"""Unit tests for the control panel backend twisted webservice client."""
23
from twisted.application import internet, service
24
from twisted.internet import defer, reactor
25
from twisted.internet.defer import inlineCallbacks
26
from twisted.web import server, resource
27
from ubuntuone.devtools.testcases import skipIfOS
29
from ubuntuone.controlpanel.web_client import txwebclient
30
from ubuntuone.controlpanel.tests import TestCase
34
SAMPLE_VALUE = "sample result"
35
SAMPLE_RESOURCE = '{"%s": "%s"}' % (SAMPLE_KEY, SAMPLE_VALUE)
36
SAMPLE_CREDENTIALS = dict(
37
consumer_key="consumer key",
38
consumer_secret="consumer secret",
39
token="the real token",
40
token_secret="the token secret",
44
def sample_get_credentials():
45
"""Will return the sample credentials right now."""
46
return defer.succeed(SAMPLE_CREDENTIALS)
49
class MockResource(resource.Resource):
50
"""A simple web resource."""
54
# pylint: disable=C0103
55
# t.w.resource methods have freeform cased names
57
def getChild(self, name, request):
58
"""Get a given child resource."""
61
return resource.Resource.getChild(self, name, request)
63
def render_GET(self, request):
64
"""Make a bit of html out of these resource's content."""
68
class MockWebService(object):
69
"""A mock webservice for testing"""
72
"""Start up this instance."""
73
root = resource.Resource()
74
devices_resource = MockResource()
75
devices_resource.contents = SAMPLE_RESOURCE
76
root.putChild("devices", devices_resource)
77
root.putChild("throwerror", resource.NoResource())
78
unauthorized = resource.ErrorPage(resource.http.UNAUTHORIZED,
79
"Unauthrorized", "Unauthrorized")
80
root.putChild("unauthorized", unauthorized)
82
site = server.Site(root)
83
application = service.Application('web')
84
self.service_collection = service.IServiceCollection(application)
85
#pylint: disable=E1101
86
self.tcpserver = internet.TCPServer(0, site)
87
self.tcpserver.setServiceParent(self.service_collection)
88
self.service_collection.startService()
91
"""Build the url for this mock server."""
92
#pylint: disable=W0212
93
port_num = self.tcpserver._port.getHost().port
94
return "http://localhost:%d/" % port_num
98
#pylint: disable=E1101
99
return self.service_collection.stopService()
102
class FakeAsyncTimestamper(object):
103
"""A fake timestamp."""
106
"""Initialize this instance."""
109
def get_faithful_time(self):
110
"""Return the server checked timestamp."""
112
return defer.succeed(time.time())
115
class WebClientTestCase(TestCase):
116
"""Test for the webservice client."""
120
@defer.inlineCallbacks
122
yield super(WebClientTestCase, self).setUp()
123
self.ws = MockWebService()
124
test_base_url = self.ws.get_url()
125
self.wc = txwebclient.WebClient(sample_get_credentials, test_base_url)
126
self.timestamper = FakeAsyncTimestamper()
127
self.patch(txwebclient, "timestamp_checker", self.timestamper)
128
self.addCleanup(self.wc.shutdown)
129
self.addCleanup(self.ws.stop)
132
def test_get_url(self):
133
"""A method is successfully called in the mock webservice."""
134
result = yield self.wc.call_api("devices")
135
self.assertIn(SAMPLE_KEY, result)
136
self.assertEqual(SAMPLE_VALUE, result[SAMPLE_KEY])
139
def test_call_api_uses_timestamp(self):
140
"""Check that call_api uses the timestamp."""
141
yield self.wc.call_api("devices")
142
self.assertTrue(self.timestamper.called,
143
"The timestamper must be used.")
146
def test_get_url_error(self):
147
"""The errback is called when there's some error."""
148
yield self.assertFailure(self.wc.call_api("throwerror"),
149
txwebclient.WebClientError)
152
def test_unauthorized(self):
153
"""Detect when a request failed with UNAUTHORIZED."""
154
yield self.assertFailure(self.wc.call_api("unauthorized"),
155
txwebclient.UnauthorizedError)
158
class WebClientShutdownTestCase(TestCase):
159
"""The webclient behaviour during shutdown."""
161
@skipIfOS('win32', 'Failing on windows, see LP: #851158.')
163
def test_shutdown(self):
164
"""The webclient behaves well during shutdown."""
165
self.patch(txwebclient, "timestamp_checker", FakeAsyncTimestamper())
166
d3 = defer.Deferred()
167
# pylint: disable=E1101
168
reactor.callLater(1, d3.callback, None)
169
ws = MockWebService()
170
test_base_url = ws.get_url()
171
wc = txwebclient.WebClient(sample_get_credentials, test_base_url)
172
d1 = wc.call_api("throwerror")
176
yield defer.DeferredList([d1, d3], fireOnOneCallback=True,
177
fireOnOneErrback=True)
added
removed
ubuntuone/controlpanel/web_client/tests/test_txwebclient.py
