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."""
21
from twisted.application import internet, service
22
from twisted.internet import defer, reactor
23
from twisted.internet.defer import inlineCallbacks
24
from twisted.web import server, resource
25
from ubuntuone.devtools.testcase import skipIfOS
27
from ubuntuone.controlpanel.web_client import txwebclient
28
from ubuntuone.controlpanel.tests import TestCase
32
SAMPLE_VALUE = "sample result"
33
SAMPLE_RESOURCE = '{"%s": "%s"}' % (SAMPLE_KEY, SAMPLE_VALUE)
34
SAMPLE_CREDENTIALS = dict(
35
consumer_key="consumer key",
36
consumer_secret="consumer secret",
37
token="the real token",
38
token_secret="the token secret",
42
def sample_get_credentials():
43
"""Will return the sample credentials right now."""
44
return defer.succeed(SAMPLE_CREDENTIALS)
47
class MockResource(resource.Resource):
48
"""A simple web resource."""
52
# pylint: disable=C0103
53
# t.w.resource methods have freeform cased names
55
def getChild(self, name, request):
56
"""Get a given child resource."""
59
return resource.Resource.getChild(self, name, request)
61
def render_GET(self, request):
62
"""Make a bit of html out of these resource's content."""
66
class MockWebService(object):
67
"""A mock webservice for testing"""
70
"""Start up this instance."""
71
root = resource.Resource()
72
devices_resource = MockResource()
73
devices_resource.contents = SAMPLE_RESOURCE
74
root.putChild("devices", devices_resource)
75
root.putChild("throwerror", resource.NoResource())
76
unauthorized = resource.ErrorPage(resource.http.UNAUTHORIZED,
77
"Unauthrorized", "Unauthrorized")
78
root.putChild("unauthorized", unauthorized)
80
site = server.Site(root)
81
application = service.Application('web')
82
self.service_collection = service.IServiceCollection(application)
83
#pylint: disable=E1101
84
self.tcpserver = internet.TCPServer(0, site)
85
self.tcpserver.setServiceParent(self.service_collection)
86
self.service_collection.startService()
89
"""Build the url for this mock server."""
90
#pylint: disable=W0212
91
port_num = self.tcpserver._port.getHost().port
92
return "http://localhost:%d/" % port_num
96
#pylint: disable=E1101
97
self.service_collection.stopService()
100
class WebClientTestCase(TestCase):
101
"""Test for the webservice client."""
106
super(WebClientTestCase, self).setUp()
107
self.ws = MockWebService()
108
test_base_url = self.ws.get_url()
109
self.wc = txwebclient.WebClient(sample_get_credentials, test_base_url)
113
super(WebClientTestCase, self).tearDown()
118
def test_get_url(self):
119
"""A method is successfully called in the mock webservice."""
120
result = yield self.wc.call_api("devices")
121
self.assertIn(SAMPLE_KEY, result)
122
self.assertEqual(SAMPLE_VALUE, result[SAMPLE_KEY])
125
def test_get_url_error(self):
126
"""The errback is called when there's some error."""
127
yield self.assertFailure(self.wc.call_api("throwerror"),
128
txwebclient.WebClientError)
131
def test_unauthorized(self):
132
"""Detect when a request failed with UNAUTHORIZED."""
133
yield self.assertFailure(self.wc.call_api("unauthorized"),
134
txwebclient.UnauthorizedError)
137
class WebClientShutdownTestCase(TestCase):
138
"""The webclient behaviour during shutdown."""
140
@skipIfOS('win32', 'Failing on windows, see LP: #851158.')
142
def test_shutdown(self):
143
"""The webclient behaves well during shutdown."""
144
d3 = defer.Deferred()
145
# pylint: disable=E1101
146
reactor.callLater(1, d3.callback, None)
147
ws = MockWebService()
148
test_base_url = ws.get_url()
149
wc = txwebclient.WebClient(sample_get_credentials, test_base_url)
150
d1 = wc.call_api("throwerror")
154
yield defer.DeferredList([d1, d3], fireOnOneCallback=True,
155
fireOnOneErrback=True)