Merge lp:~nataliabidart/ubuntu-sso-client/new-dbus-iface into lp:ubuntu-sso-client

[Natalia Bidart]

Adding a new DBus iface to manage credentials with flexible API (LP: #653113).

Adding a new DBus iface to manage credentials. The methods in this interface have a flexible API in the sense that they accept a dictionary (string, string) as parameter so we can easily expand as necessary.

Also, {find, clear}_credentials are now not blocking operations, and the result of those are returned via DBus signals.

[Rodrigo Moya (rodrigo-moya)]

review: Approve

[Alejandro J. Cura (alecu)]

Looks fine!

The documentation at https://wiki.ubuntu.com/SingleSignOn/UbuntuSsoClient should be updated as well.
Perhaps have a /SingleSignOn/UbuntuSsoClient/BetaInterface while Natty is in development, then move it as the main page and move the current to /SingleSignOn/UbuntuSsoClient/1.0-Deprecated

review: Approve

=== modified file 'ubuntu_sso/__init__.py'
--- ubuntu_sso/__init__.py	2010-10-01 18:26:56 +0000
+++ ubuntu_sso/__init__.py	2010-10-07 12:44:44 +0000
@@ -22,11 +22,6 @@
 DBUS_IFACE_AUTH_NAME = "com.ubuntu.sso"
 DBUS_IFACE_USER_NAME = "com.ubuntu.sso.UserManagement"
 DBUS_IFACE_CRED_NAME = "com.ubuntu.sso.ApplicationCredentials"
+DBUS_IFACE_CREDENTIALS_NAME = "com.ubuntu.sso.CredentialsManagement"
 
 NO_OP = lambda *args, **kwargs: None
-
-APP_NAME_KEY = 'app_name'
-TC_URL_KEY = 'tc_url'
-HELP_TEXT_KEY = 'help_text'
-WINDOW_ID_KEY = 'window_id'
-PING_URL_KEY = 'ping_url'

=== modified file 'ubuntu_sso/credentials.py'
--- ubuntu_sso/credentials.py	2010-10-04 23:34:38 +0000
+++ ubuntu_sso/credentials.py	2010-10-07 12:44:44 +0000
@@ -50,6 +50,18 @@
 logger = setup_logging('ubuntu_sso.credentials')
 
 
+APP_NAME_KEY = 'app_name'
+TC_URL_KEY = 'tc_url'
+HELP_TEXT_KEY = 'help_text'
+WINDOW_ID_KEY = 'window_id'
+PING_URL_KEY = 'ping_url'
+SUCCESS_CB_KEY = 'success_cb'
+ERROR_CB_KEY = 'error_cb'
+DENIAL_CB_KEY = 'denial_cb'
+ERROR_KEY = 'error_message'
+ERROR_DETAIL_KEY = 'detailed_error'
+
+
 def keyring_store_credentials(app_name, credentials):
     """Store the credentials in the keyring."""
     logger.info('keyring_store_credentials: app_name "%s".', app_name)
@@ -91,7 +103,9 @@
                                  f.__name__, self.app_name, msg)
                 logger.error('%s (app_name: %s): Calling error_cb at %r.',
                                  f.__name__, self.app_name, self.error_cb)
-                self.error_cb(self.app_name, msg, traceback.format_exc())
+                error_dict = {ERROR_KEY: msg,
+                              ERROR_DETAIL_KEY: traceback.format_exc()}
+                self.error_cb(self.app_name, error_dict)
             return result
 
         return inner
@@ -173,7 +187,7 @@
         """Handle UI error when login/registration failed."""
         logger.warning('Login/registration failed app %r, error %r',
                        app_name, error)
-        self.error_cb(app_name, error, 'No detailed error from the GUI.')
+        self.error_cb(app_name, {ERROR_KEY: error})
 
     def _auth_denial_cb(self, dialog, app_name):
         """The user decided not to allow the registration or login."""

=== modified file 'ubuntu_sso/main.py'
--- ubuntu_sso/main.py	2010-10-04 18:05:06 +0000
+++ ubuntu_sso/main.py	2010-10-07 12:44:44 +0000
@@ -33,9 +33,12 @@
 
 import dbus.service
 
-from ubuntu_sso import DBUS_IFACE_USER_NAME, DBUS_IFACE_CRED_NAME
+from ubuntu_sso import (DBUS_IFACE_USER_NAME, DBUS_IFACE_CRED_NAME,
+    DBUS_IFACE_CREDENTIALS_NAME)
 from ubuntu_sso.account import Account
-from ubuntu_sso.credentials import Credentials, keyring_store_credentials
+from ubuntu_sso.credentials import (Credentials, keyring_store_credentials,
+    HELP_TEXT_KEY, PING_URL_KEY, TC_URL_KEY, WINDOW_ID_KEY,
+    SUCCESS_CB_KEY, ERROR_CB_KEY, DENIAL_CB_KEY, ERROR_KEY, ERROR_DETAIL_KEY)
 from ubuntu_sso.keyring import get_token_name, U1_APP_NAME
 from ubuntu_sso.logger import setup_logging
 
@@ -45,7 +48,7 @@
 
 
 logger = setup_logging("ubuntu_sso.main")
-PING_URL = "https://one.ubuntu.com/oauth/sso-finished-so-get-tokens/"
+U1_PING_URL = "https://one.ubuntu.com/oauth/sso-finished-so-get-tokens/"
 
 
 class SSOLoginProcessor(Account):
@@ -267,7 +270,13 @@
 
     def __init__(self, *args, **kwargs):
         dbus.service.Object.__init__(self, *args, **kwargs)
-        self.ping_url = os.environ.get('USSOC_PING_URL', PING_URL)
+        self.ping_url = os.environ.get('USSOC_PING_URL', U1_PING_URL)
+
+    def _process_error(self, app_name, error_dict):
+        """Process the 'error_dict' and emit CredentialsError."""
+        msg = error_dict.get(ERROR_KEY, 'No error message given.')
+        detail = error_dict.get(ERROR_DETAIL_KEY, 'No detailed error given.')
+        self.CredentialsError(app_name, msg, detail)
 
     @dbus.service.signal(DBUS_IFACE_CRED_NAME, signature="s")
     def AuthorizationDenied(self, app_name):
@@ -317,7 +326,7 @@
                           tc_url=terms_and_conditions_url,
                           help_text=help_text, window_id=window_id,
                           success_cb=self.CredentialsFound,
-                          error_cb=self.CredentialsError,
+                          error_cb=self._process_error,
                           denial_cb=self.AuthorizationDenied)
         obj.register()
 
@@ -337,7 +346,7 @@
         obj = Credentials(app_name=app_name, ping_url=ping_url, tc_url=None,
                           help_text=help_text, window_id=window_id,
                           success_cb=self.CredentialsFound,
-                          error_cb=self.CredentialsError,
+                          error_cb=self._process_error,
                           denial_cb=self.AuthorizationDenied)
         obj.login()
 
@@ -350,3 +359,131 @@
         """
         obj = Credentials(app_name=app_name)
         obj.clear_credentials()
+
+
+class CredentialsManagement(dbus.service.Object):
+    """DBus object that manages credentials.
+
+    Every exposed method in this class requires one mandatory argument:
+
+        - 'app_name': the name of the application. Will be displayed in the
+        GUI header, plus it will be used to find/build/clear tokens.
+
+    And accepts another parameter named 'args', which is a dictionary that
+    can contain the following:
+
+        - 'help_text': an explanatory text for the end-users, will be
+        shown below the header. This is an optional free text field.
+
+        - 'ping_url': the url to open after successful token retrieval. If
+        defined, the email will be attached to the url and will be pinged
+        with a OAuth-signed request.
+
+        - 'tc_url': the link to the Terms and Conditions page. If defined,
+        the checkbox to agree to the terms will link to it.
+
+        - 'window_id': the id of the window which will be set as a parent
+        of the GUI. If not defined, no parent will be set.
+
+    """
+
+    # Operator not preceded by a space (fails with dbus decorators)
+    # pylint: disable=C0322
+
+    def _parse_args(self, args):
+        """Retrieve values from the generic param 'args'."""
+        result = dict((k, v) for k, v in args.iteritems() if k in
+                      (HELP_TEXT_KEY, PING_URL_KEY, TC_URL_KEY, WINDOW_ID_KEY))
+        result[WINDOW_ID_KEY] = int(args.get(WINDOW_ID_KEY, 0))
+        result[SUCCESS_CB_KEY] = self.CredentialsFound
+        result[ERROR_CB_KEY] = self.CredentialsError
+        result[DENIAL_CB_KEY] = self.AuthorizationDenied
+        return result
+
+    @dbus.service.signal(DBUS_IFACE_CREDENTIALS_NAME, signature='s')
+    def AuthorizationDenied(self, app_name):
+        """Signal thrown when the user denies the authorization."""
+        logger.info('%s: emitting AuthorizationDenied with app_name "%s".',
+                    self.__class__.__name__, app_name)
+
+    @dbus.service.signal(DBUS_IFACE_CREDENTIALS_NAME, signature='sa{ss}')
+    def CredentialsFound(self, app_name, credentials):
+        """Signal thrown when the credentials are found."""
+        logger.info('%s: emitting CredentialsFound with app_name "%s".',
+                    self.__class__.__name__, app_name)
+
+    @dbus.service.signal(DBUS_IFACE_CREDENTIALS_NAME, signature='s')
+    def CredentialsNotFound(self, app_name):
+        """Signal thrown when the credentials are not found."""
+        logger.info('%s: emitting CredentialsNotFound with app_name "%s".',
+                    self.__class__.__name__, app_name)
+
+    @dbus.service.signal(DBUS_IFACE_CREDENTIALS_NAME, signature='s')
+    def CredentialsCleared(self, app_name):
+        """Signal thrown when the credentials were cleared."""
+        logger.info('%s: emitting CredentialsCleared with app_name "%s".',
+                    self.__class__.__name__, app_name)
+
+    @dbus.service.signal(DBUS_IFACE_CREDENTIALS_NAME, signature='sa{ss}')
+    def CredentialsError(self, app_name, error_dict):
+        """Signal thrown when there is a problem getting the credentials."""
+        logger.error('%s: emitting CredentialsError with app_name "%s" and '
+                     'error_dict %r.', self.__class__.__name__, app_name,
+                     error_dict)
+
+    @dbus.service.method(dbus_interface=DBUS_IFACE_CREDENTIALS_NAME,
+                         in_signature='sa{ss}', out_signature='')
+    def find_credentials(self, app_name, args):
+        """Look for the credentials for an application.
+
+        - 'app_name': the name of the application which credentials are
+        going to be removed.
+
+        - 'args' is a dictionary, currently not used.
+
+        """
+        obj = Credentials(app_name)
+
+        def success_cb(app, token):
+            """Find credentials and notify using signals."""
+            if token is not None and len(token) > 0:
+                self.CredentialsFound(app, token)
+            else:
+                self.CredentialsNotFound(app)
+
+        blocking(obj.find_credentials, app_name,
+                 success_cb, self.CredentialsError)
+
+    @dbus.service.method(dbus_interface=DBUS_IFACE_CREDENTIALS_NAME,
+                         in_signature='sa{ss}', out_signature='')
+    def clear_credentials(self, app_name, args):
+        """Clear the token for an application.
+
+        - 'app_name': the name of the application which credentials are
+        going to be removed.
+
+        - 'args' is a dictionary, currently not used.
+
+        """
+        obj = Credentials(app_name)
+
+        def success_cb(app_name, result):
+            """Clear credentials and notify using signals."""
+            self.CredentialsCleared(app_name)
+
+        blocking(obj.clear_credentials, app_name,
+                 success_cb, self.CredentialsError)
+
+    @dbus.service.method(dbus_interface=DBUS_IFACE_CREDENTIALS_NAME,
+                         in_signature='sa{ss}', out_signature='')
+    def register(self, app_name, args):
+        """Get credentials if found else prompt GUI to register."""
+        obj = Credentials(app_name, **self._parse_args(args))
+        obj.register()
+
+    @dbus.service.method(dbus_interface=DBUS_IFACE_CREDENTIALS_NAME,
+                         in_signature='sa{ss}', out_signature='')
+    def login(self, app_name, args):
+        """Get credentials if found else prompt GUI to login."""
+        obj = Credentials(app_name, **self._parse_args(args))
+        obj.login()

=== modified file 'ubuntu_sso/tests/__init__.py'
--- ubuntu_sso/tests/__init__.py	2010-10-01 15:21:25 +0000
+++ ubuntu_sso/tests/__init__.py	2010-10-07 12:44:44 +0000
@@ -20,7 +20,7 @@
 
 from ubuntu_sso.keyring import get_token_name
 
-APP_NAME = 'The Super testing app!'
+APP_NAME = 'The Super App!'
 CAPTCHA_ID = 'test'
 CAPTCHA_PATH = os.path.abspath(os.path.join(os.curdir, 'ubuntu_sso', 'tests',
                                             'files', 'captcha.png'))

=== modified file 'ubuntu_sso/tests/test_credentials.py'
--- ubuntu_sso/tests/test_credentials.py	2010-10-04 18:13:52 +0000
+++ ubuntu_sso/tests/test_credentials.py	2010-10-07 12:44:44 +0000
@@ -25,8 +25,9 @@
 from twisted.trial.unittest import TestCase, FailTest
 
 from contrib.testing.testcase import MementoHandler
-from ubuntu_sso import (APP_NAME_KEY, HELP_TEXT_KEY, NO_OP,
-    PING_URL_KEY, TC_URL_KEY, WINDOW_ID_KEY, credentials, gui)
+from ubuntu_sso import credentials, gui
+from ubuntu_sso.credentials import (APP_NAME_KEY, HELP_TEXT_KEY, NO_OP,
+    PING_URL_KEY, TC_URL_KEY, WINDOW_ID_KEY, ERROR_KEY, ERROR_DETAIL_KEY)
 from ubuntu_sso.tests import (APP_NAME, EMAIL, HELP_TEXT, PING_URL, TC_URL,
     TOKEN, WINDOW_ID)
 
@@ -113,13 +114,17 @@
         """To be called on credentials denial."""
         self._set_called('denial', *args, **kwargs)
 
-    def assert_error_cb_called(self, msg, detailed_error):
+    def assert_error_cb_called(self, msg, detailed_error=None):
         """Check that self.error_cb was called with proper arguments."""
         self.assertEqual(len(self._called), 2)
         self.assertEqual(self._called[0][0], 'error')
         self.assertEqual(self._called[0][1], APP_NAME)
-        self.assertEqual(self._called[0][2], msg)
-        self.assertIn(str(detailed_error), self._called[0][3])
+        error_dict = self._called[0][2]
+        self.assertEqual(error_dict[ERROR_KEY], msg)
+        if detailed_error is not None:
+            self.assertIn(str(detailed_error), error_dict[ERROR_DETAIL_KEY])
+        else:
+            self.assertNotIn(ERROR_DETAIL_KEY, error_dict)
         self.assertEqual(self._called[1], {})
 
 
@@ -245,8 +250,7 @@
     def test_login_error_cb(self):
         """On login/register error, self.error_cb is called."""
         self.obj._login_error_cb(dialog=None, app_name=APP_NAME, error='Bla')
-        msg = 'No detailed error from the GUI.'
-        self.assert_error_cb_called(msg='Bla', detailed_error=msg)
+        self.assert_error_cb_called(msg='Bla')
 
     def test_auth_denial_cb(self):
         """On auth denied, self.denial_cb is called."""

=== modified file 'ubuntu_sso/tests/test_main.py'
--- ubuntu_sso/tests/test_main.py	2010-10-01 18:13:41 +0000
+++ ubuntu_sso/tests/test_main.py	2010-10-07 12:44:44 +0000
@@ -20,6 +20,7 @@
 # with this program.  If not, see <http://www.gnu.org/licenses/>.
 """Tests for the main SSO client code."""
 
+import logging
 import os
 
 from mocker import Mocker, MockerTestCase, ARGS, KWARGS
@@ -29,13 +30,16 @@
 import ubuntu_sso.keyring
 import ubuntu_sso.main
 
-from ubuntu_sso import credentials
+from contrib.testing.testcase import MementoHandler
+from ubuntu_sso import credentials, DBUS_IFACE_CREDENTIALS_NAME
 from ubuntu_sso.keyring import U1_APP_NAME
-from ubuntu_sso.main import (blocking, except_to_errdict,
-    keyring_store_credentials, PING_URL, SSOCredentials, SSOLogin)
+from ubuntu_sso.main import (U1_PING_URL, blocking, except_to_errdict,
+    CredentialsManagement, keyring_store_credentials, SSOCredentials, SSOLogin)
+from ubuntu_sso.main import (HELP_TEXT_KEY, PING_URL_KEY,
+    TC_URL_KEY, WINDOW_ID_KEY, SUCCESS_CB_KEY, ERROR_CB_KEY, DENIAL_CB_KEY)
 from ubuntu_sso.tests import (APP_NAME, TC_URL, HELP_TEXT, CAPTCHA_ID,
-    CAPTCHA_SOLUTION, EMAIL, EMAIL_TOKEN, PASSWORD, TOKEN, TOKEN_NAME,
-    WINDOW_ID)
+    CAPTCHA_SOLUTION, EMAIL, EMAIL_TOKEN, PASSWORD, PING_URL, TOKEN,
+    TOKEN_NAME, WINDOW_ID)
 
 
 # Access to a protected member 'yyy' of a client class
@@ -46,12 +50,26 @@
     """The exception that will be thrown by the fake blocking."""
 
 
+def fake_ok_blocking(f, app, cb, eb):
+    """A fake blocking function that succeeds."""
+    cb(app, f())
+
+
+def fake_err_blocking(f, app, cb, eb):
+    """A fake blocking function that fails."""
+    try:
+        f()
+    except Exception, e:  # pylint: disable=W0703
+        eb(app, except_to_errdict(e))
+    else:
+        eb(app, except_to_errdict(BlockingSampleException()))
+
+
 class SsoDbusTestCase(TestCase):
     """Test the SSOLogin DBus interface."""
 
     def setUp(self):
         """Create the mocking bus."""
-        self.real_blocking = ubuntu_sso.main.blocking
         self.mocker = Mocker()
         self.mockbusname = self.mocker.mock()
         mockbus = self.mocker.mock()
@@ -76,19 +94,6 @@
         self.mocker.verify()
         self.mocker.restore()
 
-    def fake_ok_blocking(self, f, app, cb, eb):
-        """A fake blocking function that succeeds."""
-        cb(app, f())
-
-    def fake_err_blocking(self, f, app, cb, eb):
-        """A fake blocking function that fails."""
-        try:
-            f()
-        except Exception, e:  # pylint: disable=W0703
-            eb(app, except_to_errdict(e))
-        else:
-            eb(app, except_to_errdict(BlockingSampleException()))
-
     def test_creation(self):
         """Test that the object creation is successful."""
         self.mocker.replay()
@@ -109,7 +114,7 @@
         expected_result = "expected result"
         self.create_mock_processor().generate_captcha(filename)
         self.mocker.result(expected_result)
-        self.patch(ubuntu_sso.main, "blocking", self.fake_ok_blocking)
+        self.patch(ubuntu_sso.main, "blocking", fake_ok_blocking)
         self.mocker.replay()
 
         def verify(app_name, result):
@@ -132,7 +137,7 @@
         expected_result = "expected result"
         self.create_mock_processor().generate_captcha(filename)
         self.mocker.result(expected_result)
-        self.patch(ubuntu_sso.main, "blocking", self.fake_err_blocking)
+        self.patch(ubuntu_sso.main, "blocking", fake_err_blocking)
         self.mocker.replay()
 
         def verify(app_name, errdict):
@@ -155,7 +160,7 @@
         self.create_mock_processor().register_user(EMAIL, PASSWORD, CAPTCHA_ID,
                                                    CAPTCHA_SOLUTION)
         self.mocker.result(expected_result)
-        self.patch(ubuntu_sso.main, "blocking", self.fake_ok_blocking)
+        self.patch(ubuntu_sso.main, "blocking", fake_ok_blocking)
         self.mocker.replay()
 
         def verify(app_name, result):
@@ -179,7 +184,7 @@
         self.create_mock_processor().register_user(EMAIL, PASSWORD, CAPTCHA_ID,
                                                    CAPTCHA_SOLUTION)
         self.mocker.result(expected_result)
-        self.patch(ubuntu_sso.main, "blocking", self.fake_err_blocking)
+        self.patch(ubuntu_sso.main, "blocking", fake_err_blocking)
         self.mocker.replay()
 
         def verify(app_name, errdict):
@@ -201,7 +206,7 @@
         d = Deferred()
         self.create_mock_processor().login(EMAIL, PASSWORD, TOKEN_NAME)
         self.mocker.result(TOKEN)
-        self.patch(ubuntu_sso.main, "blocking", self.fake_ok_blocking)
+        self.patch(ubuntu_sso.main, "blocking", fake_ok_blocking)
         self.mocker.replay()
 
         def verify(app_name, result):
@@ -222,7 +227,7 @@
         """Test that the login method fails as expected."""
         d = Deferred()
         self.mockprocessorclass = self.mocker.mock()
-        self.patch(ubuntu_sso.main, "blocking", self.fake_err_blocking)
+        self.patch(ubuntu_sso.main, "blocking", fake_err_blocking)
 
         def fake_gtn(*args):
             """A fake get_token_name that fails."""
@@ -251,7 +256,7 @@
         self.create_mock_processor().validate_email(EMAIL, PASSWORD,
                                                     EMAIL_TOKEN, TOKEN_NAME)
         self.mocker.result(TOKEN)
-        self.patch(ubuntu_sso.main, "blocking", self.fake_ok_blocking)
+        self.patch(ubuntu_sso.main, "blocking", fake_ok_blocking)
         self.mocker.replay()
 
         def verify(app_name, result):
@@ -272,7 +277,7 @@
         """Test that the validate_email method fails as expected."""
         d = Deferred()
         self.mockprocessorclass = self.mocker.mock()
-        self.patch(ubuntu_sso.main, "blocking", self.fake_err_blocking)
+        self.patch(ubuntu_sso.main, "blocking", fake_err_blocking)
 
         def fake_gtn(*args):
             """A fake get_token_name that fails."""
@@ -300,7 +305,7 @@
         d = Deferred()
         processor = self.create_mock_processor()
         processor.request_password_reset_token(EMAIL)
-        self.patch(ubuntu_sso.main, "blocking", self.fake_ok_blocking)
+        self.patch(ubuntu_sso.main, "blocking", fake_ok_blocking)
         self.mocker.result(EMAIL)
         self.mocker.replay()
 
@@ -323,7 +328,7 @@
 
         self.create_mock_processor().request_password_reset_token(EMAIL)
         self.mocker.result(EMAIL)
-        self.patch(ubuntu_sso.main, "blocking", self.fake_err_blocking)
+        self.patch(ubuntu_sso.main, "blocking", fake_err_blocking)
         self.mocker.replay()
 
         def verify(app_name, errdict):
@@ -345,7 +350,7 @@
         self.create_mock_processor().set_new_password(EMAIL, EMAIL_TOKEN,
                                                       PASSWORD)
         self.mocker.result(EMAIL)
-        self.patch(ubuntu_sso.main, "blocking", self.fake_ok_blocking)
+        self.patch(ubuntu_sso.main, "blocking", fake_ok_blocking)
         self.mocker.replay()
 
         def verify(app_name, result):
@@ -369,7 +374,7 @@
         self.create_mock_processor().set_new_password(EMAIL, EMAIL_TOKEN,
                                                       PASSWORD)
         self.mocker.result(expected_result)
-        self.patch(ubuntu_sso.main, "blocking", self.fake_err_blocking)
+        self.patch(ubuntu_sso.main, "blocking", fake_err_blocking)
         self.mocker.replay()
 
         def verify(app_name, errdict):
@@ -576,15 +581,17 @@
     def setUp(self):
         MockerTestCase.setUp(self)
         self.client = SSOCredentials(self.mocker.mock())
+        self.args = {PING_URL_KEY: self.ping_url,
+                     TC_URL_KEY: TC_URL, HELP_TEXT_KEY: HELP_TEXT,
+                     WINDOW_ID_KEY: WINDOW_ID,
+                     SUCCESS_CB_KEY: self.client.CredentialsFound,
+                     ERROR_CB_KEY: self.client._process_error,
+                     DENIAL_CB_KEY: self.client.AuthorizationDenied}
 
     def test_login_or_register(self):
         """login_or_register is correct."""
         mock_class = self.mocker.replace("ubuntu_sso.credentials.Credentials")
-        mock_class(app_name=self.app_name, ping_url=self.ping_url,
-                   tc_url=TC_URL, help_text=HELP_TEXT, window_id=WINDOW_ID,
-                   success_cb=self.client.CredentialsFound,
-                   error_cb=self.client.CredentialsError,
-                   denial_cb=self.client.AuthorizationDenied)
+        mock_class(app_name=self.app_name, **self.args)
         creds_obj = self.mocker.mock()
         self.mocker.result(creds_obj)
 
@@ -596,12 +603,9 @@
 
     def test_login_only(self):
         """login_or_register is correct."""
+        self.args[TC_URL_KEY] = None
         mock_class = self.mocker.replace("ubuntu_sso.credentials.Credentials")
-        mock_class(app_name=self.app_name, ping_url=self.ping_url,
-                   tc_url=None, help_text=HELP_TEXT, window_id=WINDOW_ID,
-                   success_cb=self.client.CredentialsFound,
-                   error_cb=self.client.CredentialsError,
-                   denial_cb=self.client.AuthorizationDenied)
+        mock_class(app_name=self.app_name, **self.args)
         creds_obj = self.mocker.mock()
         self.mocker.result(creds_obj)
 
@@ -620,7 +624,7 @@
     """
 
     app_name = U1_APP_NAME
-    ping_url = PING_URL
+    ping_url = U1_PING_URL
 
 
 class ApplicationCredentialsClearTokenTestCase(TestCase, MockerTestCase):
@@ -660,4 +664,246 @@
     def test_no_override_ping_url(self):
         """If the environ is unset, the default ping url is used."""
         creds = SSOCredentials(None)
-        self.assertEqual(creds.ping_url, PING_URL)
+        self.assertEqual(creds.ping_url, U1_PING_URL)
+
+
+class CredentialsManagementTestCase(TestCase):
+    """Tests for the CredentialsManagement DBus interface."""
+
+    base_args = {HELP_TEXT_KEY: HELP_TEXT, PING_URL_KEY: PING_URL,
+                 TC_URL_KEY: TC_URL, WINDOW_ID_KEY: WINDOW_ID}
+
+    def setUp(self):
+        self.mocker = Mocker()
+        self.client = CredentialsManagement()
+        self.args = {}
+        self.cred_args = {}
+
+    def tearDown(self):
+        """Verify the mocking stuff and shut it down."""
+        self.mocker.verify()
+        self.mocker.restore()
+
+    def assert_dbus_method_correct(self, method):
+        """Check that 'method' is a dbus method with proper signatures."""
+        self.assertTrue(method._dbus_is_method)
+        self.assertEqual(method._dbus_interface, DBUS_IFACE_CREDENTIALS_NAME)
+        self.assertEqual(method._dbus_in_signature, 'sa{ss}')
+        self.assertEqual(method._dbus_out_signature, '')
+
+    def create_mock_backend(self):
+        """Create a mock backend."""
+        mock_class = self.mocker.replace("ubuntu_sso.credentials.Credentials")
+        mock_class(APP_NAME, **self.cred_args)
+        creds_obj = self.mocker.mock()
+        self.mocker.result(creds_obj)
+
+        return creds_obj
+
+    def test_is_dbus_object(self):
+        """CredentialsManagement is a Dbus object."""
+        self.assertIsInstance(self.client, ubuntu_sso.main.dbus.service.Object)
+
+
+class CredentialsManagementFindClearTestCase(CredentialsManagementTestCase):
+    """Tests for the CredentialsManagement find/clear methods."""
+
+    timeout = 5
+
+    def test_find_credentials(self):
+        """The credentials are asked and returned in signals."""
+        self.create_mock_backend().find_credentials()
+        self.mocker.replay()
+
+        self.client.find_credentials(APP_NAME, self.args)
+        self.assert_dbus_method_correct(self.client.find_credentials)
+
+    def test_find_credentials_does_not_block_when_found(self):
+        """Calling find_credentials does not block but return thru signals.
+
+        If the creds are found, CredentialsFound is emitted.
+
+        """
+        d = Deferred()
+
+        def verify(app_name, creds):
+            """The actual test."""
+            try:
+                self.assertEqual(app_name, APP_NAME)
+                self.assertEqual(creds, TOKEN)
+            except Exception, e:  # pylint: disable=W0703
+                d.errback(e)
+            else:
+                d.callback(creds)
+
+        self.patch(ubuntu_sso.main, 'blocking', fake_ok_blocking)
+        self.patch(self.client, 'CredentialsFound', verify)
+        self.patch(self.client, 'CredentialsNotFound', d.errback)
+
+        self.create_mock_backend().find_credentials()
+        self.mocker.result(TOKEN)
+        self.mocker.replay()
+
+        self.client.find_credentials(APP_NAME, self.args)
+        return d
+
+    def test_find_credentials_does_not_block_when_not_found(self):
+        """Calling find_credentials does not block but return thru signals.
+
+        If the creds are not found, CredentialsNotFound is emitted.
+
+        """
+        d = Deferred()
+
+        def verify(app_name):
+            """The actual test."""
+            try:
+                self.assertEqual(app_name, APP_NAME)
+            except Exception, e:  # pylint: disable=W0703
+                d.errback(e)
+            else:
+                d.callback(app_name)
+
+        self.patch(ubuntu_sso.main, 'blocking', fake_ok_blocking)
+        self.patch(self.client, 'CredentialsFound', d.errback)
+        self.patch(self.client, 'CredentialsNotFound', verify)
+
+        self.create_mock_backend().find_credentials()
+        self.mocker.result({})
+        self.mocker.replay()
+
+        self.client.find_credentials(APP_NAME, self.args)
+        return d
+
+    def test_clear_credentials(self):
+        """The credentials are removed."""
+        self.create_mock_backend().clear_credentials()
+        self.mocker.replay()
+
+        self.client.clear_credentials(APP_NAME, self.args)
+        self.assert_dbus_method_correct(self.client.clear_credentials)
+
+    def test_clear_credentials_does_not_block(self):
+        """Calling clear_credentials does not block but return thru signals."""
+        d = Deferred()
+
+        def verify(app_name):
+            """The actual test."""
+            try:
+                self.assertEqual(app_name, APP_NAME)
+            except Exception, e:  # pylint: disable=W0703
+                d.errback(e)
+            else:
+                d.callback(app_name)
+
+        self.patch(ubuntu_sso.main, 'blocking', fake_ok_blocking)
+        self.patch(self.client, 'CredentialsCleared', verify)
+
+        self.create_mock_backend().clear_credentials()
+        self.mocker.replay()
+
+        self.client.clear_credentials(APP_NAME, self.args)
+        return d
+
+
+class CredentialsManagementOpsTestCase(CredentialsManagementTestCase):
+    """Tests for the CredentialsManagement login/register methods."""
+
+    def setUp(self):
+        super(CredentialsManagementOpsTestCase, self).setUp()
+        self.args = dict((k, str(v)) for k, v in self.base_args.iteritems())
+        self.cred_args = self.base_args.copy()
+        self.cred_args[SUCCESS_CB_KEY] = self.client.CredentialsFound
+        self.cred_args[ERROR_CB_KEY] = self.client.CredentialsError
+        self.cred_args[DENIAL_CB_KEY] = self.client.AuthorizationDenied
+
+    def test_register(self):
+        """The registration is correct."""
+        self.create_mock_backend().register()
+        self.mocker.replay()
+
+        self.client.register(APP_NAME, self.args)
+        self.assert_dbus_method_correct(self.client.register)
+
+    def test_login(self):
+        """The login is correct."""
+        self.create_mock_backend().login()
+        self.mocker.replay()
+
+        self.client.login(APP_NAME, self.args)
+        self.assert_dbus_method_correct(self.client.login)
+
+
+class CredentialsManagementParamsTestCase(CredentialsManagementOpsTestCase):
+    """Tests for the CredentialsManagement extra parameters handling."""
+
+    def setUp(self):
+        super(CredentialsManagementParamsTestCase, self).setUp()
+        self.args['dummy'] = 'nothing useful'
+
+
+class CredentialsManagementSignalsTestCase(TestCase):
+    """Tests for the CredentialsManagement DBus signals."""
+
+    def setUp(self):
+        self.client = CredentialsManagement()
+
+        self.memento = MementoHandler()
+        self.memento.setLevel(logging.DEBUG)
+        ubuntu_sso.main.logger.addHandler(self.memento)
+
+    def assert_dbus_signal_correct(self, signal, signature):
+        """Check that 'signal' is a dbus signal with proper 'signature'."""
+        self.assertTrue(signal._dbus_is_signal)
+        self.assertEqual(signal._dbus_interface, DBUS_IFACE_CREDENTIALS_NAME)
+        self.assertEqual(signal._dbus_signature, signature)
+
+    def test_credentials_found(self):
+        """The CredentialsFound signal."""
+        self.client.CredentialsFound(APP_NAME, TOKEN)
+        msgs = (self.client.__class__.__name__,
+                self.client.CredentialsFound.__name__, APP_NAME)
+        self.assertTrue(self.memento.check_info(*msgs))
+
+        msg = 'credentials must not be logged (found %r in log).'
+        for val in TOKEN.itervalues():
+            self.assertFalse(self.memento.check_info(val), msg % val)
+
+        self.assert_dbus_signal_correct(self.client.CredentialsFound, 'sa{ss}')
+
+    def test_credentials_not_found(self):
+        """The CredentialsNotFound signal."""
+        self.client.CredentialsNotFound(APP_NAME)
+        msgs = (self.client.__class__.__name__,
+                self.client.CredentialsNotFound.__name__, APP_NAME)
+        self.assertTrue(self.memento.check_info(*msgs))
+        self.assert_dbus_signal_correct(self.client.CredentialsNotFound, 's')
+
+    def test_credentials_cleared(self):
+        """The CredentialsCleared signal."""
+        self.client.CredentialsCleared(APP_NAME)
+        msgs = (self.client.__class__.__name__,
+                self.client.CredentialsCleared.__name__, APP_NAME)
+        self.assertTrue(self.memento.check_info(*msgs))
+
+        self.assert_dbus_signal_correct(self.client.CredentialsCleared, 's')
+
+    def test_credentials_error(self):
+        """The CredentialsError signal."""
+        error = {'error_message': 'failed!', 'detailed error': 'yadda yadda'}
+        self.client.CredentialsError(APP_NAME, error)
+        msgs = (self.client.__class__.__name__,
+                self.client.CredentialsError.__name__,
+                APP_NAME, str(error))
+        self.assertTrue(self.memento.check_error(*msgs))
+
+        self.assert_dbus_signal_correct(self.client.CredentialsError, 'sa{ss}')
+
+    def test_authorization_denied(self):
+        """The AuthorizationDenied signal."""
+        self.client.AuthorizationDenied(APP_NAME)
+        msgs = (self.client.__class__.__name__,
+                self.client.AuthorizationDenied.__name__, APP_NAME)
+        self.assertTrue(self.memento.check_info(*msgs))
+
+        self.assert_dbus_signal_correct(self.client.AuthorizationDenied, 's')