Logo Search packages:      
Sourcecode: ubuntuone-control-panel version File versions  Download package

test_webclient.py

# -*- coding: utf-8 -*-

# Authors: Alejandro J. Cura <alecu@canonical.com>
# Authors: Natalia B. Bidart <nataliabidart@canonical.com>
#
# Copyright 2010 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/>.

"""Integration tests for the control panel backend webservice client."""

from twisted.application import internet, service
from twisted.internet import defer
from twisted.internet.defer import inlineCallbacks
from twisted.web import server, resource

from ubuntuone.controlpanel import webclient
from ubuntuone.controlpanel.integrationtests import TestCase


SAMPLE_KEY = "result"
SAMPLE_VALUE = "sample result"
SAMPLE_RESOURCE = '{"%s": "%s"}' % (SAMPLE_KEY, SAMPLE_VALUE)
SAMPLE_CREDENTIALS = dict(
    consumer_key="consumer key",
    consumer_secret="consumer secret",
    token="the real token",
    token_secret="the token secret",
)


def sample_get_credentials():
    """Will return the sample credentials right now."""
    return defer.succeed(SAMPLE_CREDENTIALS)


00047 class MockResource(resource.Resource):
    """A simple web resource."""
    isLeaf = True
    contents = ""

    # pylint: disable=C0103
    # t.w.resource methods have freeform cased names

00055     def getChild(self, name, request):
        """Get a given child resource."""
        if name == '':
            return self
        return resource.Resource.getChild(self, name, request)

00061     def render_GET(self, request):
        """Make a bit of html out of these resource's content."""
        return self.contents


00066 class MockWebService(object):
    """A mock webservice for testing"""

00069     def __init__(self):
        """Start up this instance."""
        root = resource.Resource()
        devices_resource = MockResource()
        devices_resource.contents = SAMPLE_RESOURCE
        root.putChild("devices", devices_resource)
        root.putChild("throwerror", resource.NoResource())
        unauthorized = resource.ErrorPage(resource.http.UNAUTHORIZED,
                                          "Unauthrorized", "Unauthrorized")
        root.putChild("unauthorized", unauthorized)

        site = server.Site(root)
        application = service.Application('web')
        self.service_collection = service.IServiceCollection(application)
        #pylint: disable=E1101
        self.tcpserver = internet.TCPServer(0, site)
        self.tcpserver.setServiceParent(self.service_collection)
        self.service_collection.startService()

00088     def get_url(self):
        """Build the url for this mock server."""
        #pylint: disable=W0212
        port_num = self.tcpserver._port.getHost().port
        return "http://localhost:%d/" % port_num

00094     def stop(self):
        """Shut it down."""
        #pylint: disable=E1101
        self.service_collection.stopService()


00100 class WebClientTestCase(TestCase):
    """Test for the webservice client."""

    timeout = 8

    def setUp(self):
        super(WebClientTestCase, self).setUp()
        self.ws = MockWebService()
        test_base_url = self.ws.get_url()
        self.wc = webclient.WebClient(sample_get_credentials, test_base_url)

    @inlineCallbacks
    def tearDown(self):
        super(WebClientTestCase, self).tearDown()
        yield self.ws.stop()
        self.wc.shutdown()

    @inlineCallbacks
00118     def test_get_url(self):
        """A method is successfully called in the mock webservice."""
        result = yield self.wc.call_api("devices")
        self.assertIn(SAMPLE_KEY, result)
        self.assertEqual(SAMPLE_VALUE, result[SAMPLE_KEY])

    @inlineCallbacks
00125     def test_get_url_error(self):
        """The errback is called when there's some error."""
        yield self.assertFailure(self.wc.call_api("throwerror"),
                                 webclient.WebClientError)

    @inlineCallbacks
00131     def test_unauthorized(self):
        """Detect when a request failed with UNAUTHORIZED."""
        yield self.assertFailure(self.wc.call_api("unauthorized"),
                                 webclient.UnauthorizedError)


00137 class OAuthTestCase(TestCase):
    """Test for the oauth signing code."""

00140     def test_build_oauth_headers(self):
        """Build the oauth headers for a sample request."""

        sample_method = "GET"
        sample_url = "http://one.ubuntu.com/"
        headers = webclient.build_oauth_headers(sample_method, sample_url,
                                                SAMPLE_CREDENTIALS)
        self.assertIn("Authorization", headers)

00149     def test_add_oauth_headers(self):
        """Add the OAuth headers to a libsoup request."""

        def sample_build_headers(*a):
            """Build some sample headers."""
            return {"header1": "h1", "header2": "h2"}

        self.patch(webclient, "build_oauth_headers", sample_build_headers)
        test_request_headers = {}
        append_method = test_request_headers.__setitem__
        webclient.add_oauth_headers(append_method, "GET", "http://this", {})
        self.assertIn("header1", test_request_headers)
        self.assertIn("header2", test_request_headers)

Generated by  Doxygen 1.6.0   Back to index