Merge lp:~alecu/ubuntuone-control-panel/backend-born into lp:~nataliabidart/ubuntuone-control-panel/trunk

[Alejandro J. Cura]

The foundations of the control panel backend

The foundations of the control panel backend

[lp:~alecu/ubuntuone-control-panel/backend-born]

9. By Alejandro J. Cura on 2010-09-24

fix pylint complaints

10. By Alejandro J. Cura on 2010-09-24

style fixes

[Natalia Bidart (nataliabidart)]

Very nice!

review: Approve

=== added directory 'integration'
=== added file 'integration/__init__.py'
=== added file 'integration/run-tests'
--- integration/run-tests	1970-01-01 00:00:00 +0000
+++ integration/run-tests	2010-09-24 23:06:03 +0000
@@ -0,0 +1,28 @@
+#! /bin/bash
+#
+# Author: Natalia Bidart <[e-mail]>
+# Author: Alejandro J. Cura <[e-mail]>
+#
+# 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/>.
+
+if [ $# -ne 0 ]; then
+    MODULE="$@"
+else
+    MODULE="integration"
+fi
+
+echo "Running test suite for ""$MODULE"
+PYTHONPATH=. `which xvfb-run` u1trial "$MODULE"
+rm -rf _trial_temp

=== added file 'integration/test_dbus_client.py'
--- integration/test_dbus_client.py	1970-01-01 00:00:00 +0000
+++ integration/test_dbus_client.py	2010-09-24 23:06:03 +0000
@@ -0,0 +1,73 @@
+# -*- coding: utf-8 -*-
+
+# Authors: Alejandro J. Cura <[e-mail]>
+#
+# 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/>.
+
+"""Tests for the control panel backend DBus service."""
+
+import dbus
+
+from twisted.internet import defer
+from twisted.trial import unittest
+
+from ubuntuone.controlpanel import dbus_service
+from ubuntuone.controlpanel import dbus_client
+from ubuntuone.controlpanel import (DBUS_BUS_NAME, DBUS_PREFERENCES_PATH,
+                                    DBUS_PREFERENCES_IFACE)
+from ubuntuone.controlpanel.dbus_client import (DBUS_SSO_NAME, DBUS_SSO_PATH,
+                                                DBUS_SSO_IFACE)
+
+SAMPLE_CREDS = {
+    "token": "ABCDEF12345678",
+    "access_token": "DEADCAFE2010",
+}
+
+class MockDBusSSOClient(dbus.service.Object):
+    """A mock object that mimicks ussoc."""
+
+    @dbus.service.method(dbus_interface=DBUS_SSO_IFACE, in_signature="sssx")
+    def login_or_register_to_get_credentials(self, app_name, tcurl, help, wid):
+        """Get creds from the keyring, login/register if needed."""
+        self.CredentialsFound(app_name, SAMPLE_CREDS)
+
+    @dbus.service.signal(dbus_interface=DBUS_SSO_IFACE, signature="sa{ss}")
+    def CredentialsFound(self, app_name, credentials):
+        """Credentials were finally found."""
+
+class DBusClientTestCase(unittest.TestCase):
+    """test for the dbus client."""
+    timeout = 5
+
+    def setUp(self):
+        super(DBusClientTestCase, self).setUp()
+        dbus_service.init_mainloop()
+
+    def register_mockserver(self, bus_name, object_path, object_class):
+        """The mock service is registered on the DBus."""
+        session_bus = dbus.SessionBus()
+        name = session_bus.request_name(bus_name,
+                                        dbus.bus.NAME_FLAG_DO_NOT_QUEUE)
+        self.assertNotEqual(name, dbus.bus.REQUEST_NAME_REPLY_EXISTS)
+        bus_name = dbus.service.BusName(bus_name, bus=dbus.SessionBus())
+        object_class(object_path=DBUS_SSO_PATH, bus_name=bus_name)
+
+    @defer.inlineCallbacks
+    def test_get_credentials_ok(self):
+        """Test the success case for get_credentials."""
+        self.register_mockserver(DBUS_SSO_NAME, DBUS_SSO_PATH,
+                                 MockDBusSSOClient)
+        creds = yield dbus_client.get_credentials()
+        self.assertEqual(creds, SAMPLE_CREDS)

=== added file 'integration/test_dbus_service.py'
--- integration/test_dbus_service.py	1970-01-01 00:00:00 +0000
+++ integration/test_dbus_service.py	2010-09-24 23:06:03 +0000
@@ -0,0 +1,90 @@
+# -*- coding: utf-8 -*-
+
+# Authors: Alejandro J. Cura <[e-mail]>
+#
+# 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/>.
+
+"""Tests for the control panel backend DBus service."""
+
+import dbus
+
+from twisted.internet import defer
+from twisted.trial import unittest
+
+from ubuntuone.controlpanel import dbus_service
+from ubuntuone.controlpanel import (DBUS_BUS_NAME, DBUS_PREFERENCES_PATH,
+                                    DBUS_PREFERENCES_IFACE)
+
+
+class DBusServiceTestCase(unittest.TestCase):
+    """Test for the DBus service."""
+    timeout = 5
+
+    def setUp(self):
+        """Initialize each test run."""
+        super(DBusServiceTestCase, self).setUp()
+        dbus_service.init_mainloop()
+
+    def test_register_service(self):
+        """The DBus service is successfully registered."""
+        ret = dbus_service.register_service()
+        self.assertTrue(ret)
+
+    def test_cant_register_twice(self):
+        """The DBus service can't register if it already did."""
+        ret = dbus_service.register_service()
+        self.assertTrue(ret)
+        ret = dbus_service.register_service()
+        self.assertFalse(ret)
+    test_cant_register_twice.skip = "Must run 2nd check in another process."
+
+    def test_dbus_busname_created(self):
+        """The DBus BusName is created."""
+        busname = dbus_service.get_busname()
+        self.assertEqual(busname.get_name(), DBUS_BUS_NAME)
+
+    def test_account_info_returned(self):
+        """The account info is successfully returned."""
+        dbus_service.publish_backend()
+
+        bus = dbus.SessionBus()
+        obj = bus.get_object(bus_name=DBUS_BUS_NAME,
+                             object_path=DBUS_PREFERENCES_PATH,
+                             follow_name_owner_changes=True)
+        self.iface_name = DBUS_PREFERENCES_IFACE
+        backend = dbus.Interface(object=obj,
+                                 dbus_interface=DBUS_PREFERENCES_IFACE)
+        d = defer.Deferred()
+
+        def got_signal(account_info):
+            """The correct signal was fired."""
+            self.assertIn("quota_used", account_info)
+            self.assertIn("quota_total", account_info)
+            self.assertIn("type", account_info)
+            self.assertIn("name", account_info)
+            self.assertIn("email", account_info)
+            d.callback("success")
+
+        def got_error(*a):
+            """Some error happened in the DBus call."""
+            d.errback(*a)
+
+        def ignore(*a):
+            """Do nothing with the returned value."""
+
+        backend.connect_to_signal("AccountInfoReady", got_signal)
+
+        backend.account_info(reply_handler=ignore, error_handler=got_error)
+        return d

=== added file 'integration/test_webclient.py'
--- integration/test_webclient.py	1970-01-01 00:00:00 +0000
+++ integration/test_webclient.py	2010-09-24 23:06:03 +0000
@@ -0,0 +1,90 @@
+# -*- coding: utf-8 -*-
+
+# Authors: Alejandro J. Cura <[e-mail]>
+#
+# 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/>.
+
+"""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.internet.threads import deferToThread
+from twisted.trial import unittest
+from twisted.web import vhost, static, server, resource
+
+from ubuntuone.controlpanel import webclient
+
+SAMPLE_RESOURCE = "sample resource"
+
+
+class MockResource(resource.Resource):
+    """A simple web resource."""
+    isLeaf = True
+    contents = ""
+
+    def getChild(self, name, request):
+        """Get a given child resource."""
+        if name == '':
+            return self
+        return resource.Resource.getChild(self, name, request)
+
+    def render_GET(self, request):
+        """Make a bit of html out of these resource's content."""
+        return self.contents
+
+
+class MockWebService(object):
+    """A mock webservice for testing"""
+
+    def __init__(self):
+        """Start up this instance."""
+        root = MockResource()
+        root.contents = SAMPLE_RESOURCE
+        site = server.Site(root)
+        application = service.Application('web')
+        self.serviceCollection = service.IServiceCollection(application)
+        self.tcpserver = internet.TCPServer(0, site)
+        self.tcpserver.setServiceParent(self.serviceCollection)
+        self.serviceCollection.startService()
+
+    def get_url(self):
+        """Build the url for this mock server."""
+        portNum = self.tcpserver._port.getHost().port
+        return "http://localhost:%d/" % portNum
+
+    def stop(self):
+        """Shut it down."""
+        self.serviceCollection.stopService()
+
+
+class WebClientTestCase(unittest.TestCase):
+    """Test for the webservice client."""
+    timeout = 5
+
+    def setUp(self):
+        super(WebClientTestCase, self).setUp()
+        self.ws = MockWebService()
+
+    def tearDown(self):
+        super(WebClientTestCase, self).tearDown()
+        self.ws.stop()
+
+    @inlineCallbacks
+    def test_connect(self):
+        """Connect to the mock webservice."""
+        webclient.BASE_URL = self.ws.get_url()
+        ret = yield deferToThread(webclient.getinfo)
+        self.assertEqual(ret, SAMPLE_RESOURCE)

=== modified file 'ubuntuone/controlpanel/__init__.py'
--- ubuntuone/controlpanel/__init__.py	2010-09-15 21:43:48 +0000
+++ ubuntuone/controlpanel/__init__.py	2010-09-24 23:06:03 +0000
@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 
 # Authors: Natalia B Bidart <[e-mail]>
+# Authors: Alejandro J. Cura <[e-mail]>
 #
 # Copyright 2010 Canonical Ltd.
 #
@@ -22,3 +23,8 @@
 their Ubuntu One subscription and preferences.
 
 """
+
+# constants
+DBUS_BUS_NAME = "com.ubuntuone.controlpanel"
+DBUS_PREFERENCES_PATH = "/preferences"
+DBUS_PREFERENCES_IFACE = "com.ubuntuone.controlpanel.Preferences"

=== added file 'ubuntuone/controlpanel/backend.py'
--- ubuntuone/controlpanel/backend.py	1970-01-01 00:00:00 +0000
+++ ubuntuone/controlpanel/backend.py	2010-09-24 23:06:03 +0000
@@ -0,0 +1,22 @@
+# -*- coding: utf-8 -*-
+
+# Authors: Alejandro J. Cura <[e-mail]>
+#
+# 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/>.
+
+"""A backend for the Ubuntu One Control Panel."""
+
+class ControlBackend(object):
+    """A backend."""

=== added file 'ubuntuone/controlpanel/dbus_client.py'
--- ubuntuone/controlpanel/dbus_client.py	1970-01-01 00:00:00 +0000
+++ ubuntuone/controlpanel/dbus_client.py	2010-09-24 23:06:03 +0000
@@ -0,0 +1,54 @@
+# -*- coding: utf-8 -*-
+
+# Authors: Alejandro J. Cura <[e-mail]>
+#
+# 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/>.
+
+"""Client to use other DBus services."""
+
+import dbus.service
+
+from twisted.internet import defer
+
+DBUS_SSO_NAME = "com.ubuntu.sso"
+DBUS_SSO_PATH = "/credentials"
+DBUS_SSO_IFACE = "com.ubuntu.sso.ApplicationCredentials"
+
+
+def get_credentials():
+    """Get the credentials from the ubuntu-sso-client."""
+    bus = dbus.SessionBus()
+    obj = bus.get_object(bus_name=DBUS_SSO_NAME,
+                         object_path=DBUS_SSO_PATH,
+                         follow_name_owner_changes=True)
+    proxy = dbus.Interface(object=obj, dbus_interface=DBUS_SSO_IFACE)
+    d = defer.Deferred()
+
+    def ignore_return(*a):
+        """The return value of the call to the proxy can be safely ignored."""
+
+    def flag_error(*a):
+        """Make a twisted error from the error returned by the DBus call."""
+        d.errback(*a)
+
+    def found_credentials(app_name, creds):
+        """Credentials have been found."""
+        d.callback(creds)
+
+    proxy.connect_to_signal("CredentialsFound", found_credentials)
+    proxy.login_or_register_to_get_credentials("app", "tcurl", "help", 0,
+                                               reply_handler=ignore_return,
+                                               error_handler=flag_error)
+    return d

=== added file 'ubuntuone/controlpanel/dbus_service.py'
--- ubuntuone/controlpanel/dbus_service.py	1970-01-01 00:00:00 +0000
+++ ubuntuone/controlpanel/dbus_service.py	2010-09-24 23:06:03 +0000
@@ -0,0 +1,74 @@
+# -*- coding: utf-8 -*-
+
+# Authors: Alejandro J. Cura <[e-mail]>
+#
+# 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/>.
+
+"""Export the control backend thru DBus."""
+
+import dbus.service
+
+from dbus.mainloop.glib import DBusGMainLoop
+from dbus.service import method, signal
+
+from ubuntuone.controlpanel import (DBUS_BUS_NAME, DBUS_PREFERENCES_PATH,
+                                    DBUS_PREFERENCES_IFACE)
+
+# XXX: should get this info from the web service
+SAMPLE_DATA = {
+    "quota_used": "12345",
+    "quota_total": "54321",
+    "type": "paying customer",
+    "name": "john carlos",
+    "email": "[e-mail]",
+}
+
+class ControlPanelBackend(dbus.service.Object):
+    """Export the Control Panel backend thru DBus."""
+    # pylint: disable=C0103
+
+    @method(dbus_interface=DBUS_PREFERENCES_IFACE, in_signature="")
+    def account_info(self):
+        """Find out the account info for the current logged in user."""
+        self.AccountInfoReady(SAMPLE_DATA)
+
+    @signal(dbus_interface=DBUS_PREFERENCES_IFACE, signature="a{ss}")
+    def AccountInfoReady(self, info):
+        """The info for the current user is available now."""
+
+def init_mainloop():
+    """Start the DBus mainloop."""
+    DBusGMainLoop(set_as_default=True)
+
+def register_service():
+    """Try to register DBus service for making sure we run only one instance.
+
+    Return True if succesfully registered, False if already running.
+    """
+    session_bus = dbus.SessionBus()
+    name = session_bus.request_name(DBUS_BUS_NAME,
+                                    dbus.bus.NAME_FLAG_DO_NOT_QUEUE)
+    return name != dbus.bus.REQUEST_NAME_REPLY_EXISTS
+
+
+def get_busname():
+    """Build the DBus BusName."""
+    return dbus.service.BusName(DBUS_BUS_NAME, bus=dbus.SessionBus())
+
+
+def publish_backend():
+    """Publish the backend on the DBus."""
+    ControlPanelBackend(object_path=DBUS_PREFERENCES_PATH,
+                        bus_name=get_busname())

=== added file 'ubuntuone/controlpanel/tests/test_backend.py'
--- ubuntuone/controlpanel/tests/test_backend.py	1970-01-01 00:00:00 +0000
+++ ubuntuone/controlpanel/tests/test_backend.py	2010-09-24 23:06:03 +0000
@@ -0,0 +1,31 @@
+# -*- coding: utf-8 -*-
+
+# Authors: Alejandro J. Cura <[e-mail]>
+#
+# 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/>.
+
+"""Tests for the control panel backend."""
+
+from twisted.trial.unittest import TestCase
+
+from ubuntuone.controlpanel import backend
+
+class BackendBasicTestCase(TestCase):
+    """Simple tests for the backend."""
+
+    def test_backend_creation(self):
+        """The backend instance is created."""
+        be = backend.ControlBackend()
+        self.assertNotEqual(be, None)

=== added file 'ubuntuone/controlpanel/webclient.py'
--- ubuntuone/controlpanel/webclient.py	1970-01-01 00:00:00 +0000
+++ ubuntuone/controlpanel/webclient.py	2010-09-24 23:06:03 +0000
@@ -0,0 +1,26 @@
+# Authors: Alejandro J. Cura <[e-mail]>
+#
+# 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/>.
+
+"""The control panel backend webservice client."""
+
+import urllib
+
+BASE_URL = "https://one.ubuntu.com/api/1.0/"
+
+def getinfo():
+    """Get the info from the url."""
+    f = urllib.urlopen(BASE_URL + "devices/")
+    return f.read()