~nataliabidart/ubuntu-sso-client/unify-signal-broadcaster

# -*- coding: utf-8 -*-
#
# Copyright 2012 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 main SSO client code."""

from twisted.internet import defer
from ubuntuone.devtools.testcases import skipIfOS

from ubuntu_sso import main
from ubuntu_sso.tests import (
    APP_NAME,
    CAPTCHA_ID,
    CAPTCHA_SOLUTION,
    EMAIL,
    EMAIL_TOKEN,
    NAME,
    PASSWORD,
    TOKEN,
)
from ubuntu_sso.main.tests import BaseTestCase, FakedCredentials

FILENAME = 'sample filename'


class FakedKeyring(object):
    """A faked Keyring object."""

    _keys = {}

    def get_credentials(self, app_name):
        """Return the credentials for app_name."""
        return defer.succeed(self._keys.get(app_name, {}))

    def delete_credentials(self, app_name):
        """Delete the credentials for app_name."""
        self._keys.pop(app_name, None)
        return defer.succeed(None)

    def set_credentials(self, app_name, token):
        """Store the credentials for app_name."""
        self._keys[app_name] = token
        return defer.succeed(None)


class AbstractTestCase(BaseTestCase):
    """The base test case with platform specific support."""

    timeout = 2
    method = None
    backend_method = method
    params = ()
    success_signal = None
    backend_result = None
    success_result = None
    error_signal = None

    @defer.inlineCallbacks
    def setUp(self):
        yield super(AbstractTestCase, self).setUp()
        self.keyring = FakedKeyring()
        self.patch(main, 'Keyring', lambda: self.keyring)

        # avoid putting stuff in the mainloops
        self.patch(main.source, 'timeout_func', lambda *a: None)
        self.patch(main.source, 'shutdown_func', lambda *a: None)

        self.sso_service = main.UbuntuSSOService()
        yield self.sso_service.start()
        self.addCleanup(self.sso_service.shutdown)

        self.sso_client = yield main.get_sso_client()
        self.addCleanup(self.sso_client.disconnect)

        self.client = self.patchable_backend = None

        if self.backend_method is None:
            self.backend_method = self.method

    def _backend_succeed(self, *args, **kwargs):
        """Make self.patchable_backend return self.backend_result."""
        return self.backend_result

    def _backend_fail(self, *args, **kwargs):
        """Make self.patchable_backend raise an exception."""
        raise ValueError((args, kwargs))

    @defer.inlineCallbacks
    def assert_method_correct(self, success_signal, error_signal,
                              patched_backend_method, expected_result=None):
        """Calling 'self.method' works ok.

        Check that 'success_signal' is emitted, and make the test fail if
        error_signal is received.

        The self.patchable_backend will be patched with
        'patched_backend_method'.

        """
        if self.method is None:
            defer.returnValue(None)

        d = defer.Deferred()

        cb = lambda *a: d.callback(a)
        match = self.client.connect_to_signal(success_signal, cb)
        self.addCleanup(self.client.disconnect_from_signal,
                        success_signal, match)

        eb = lambda *a: d.errback(AssertionError(a))
        match = self.client.connect_to_signal(error_signal, eb)
        self.addCleanup(self.client.disconnect_from_signal,
                        error_signal, match)

        self.patch(self.patchable_backend, self.backend_method,
                   patched_backend_method)

        yield self.client.call_method(self.method, *self.params)

        result = yield d
        self.assertEqual(expected_result, result)

    def test_success(self):
        """Test that the 'method' works ok."""
        success_signal = self.success_signal
        error_signal = self.error_signal
        patched_backend_method = self._backend_succeed
        expected_result = self.success_result

        return self.assert_method_correct(success_signal, error_signal,
                    patched_backend_method, expected_result)

    def test_error(self):
        """Test that the 'method' fails as expected."""
        success_signal = self.error_signal
        error_signal = self.success_signal
        patched_backend_method = self._backend_fail
        expected_result = (APP_NAME, dict(errtype='ValueError'))

        return self.assert_method_correct(success_signal, error_signal,
                    patched_backend_method, expected_result)


class SSOLoginProxyTestCase(AbstractTestCase):
    """Test the SSOLoginProxy interface."""

    @defer.inlineCallbacks
    def setUp(self):
        yield super(SSOLoginProxyTestCase, self).setUp()
        self.client = self.sso_client.sso_login
        self.patchable_backend = self.sso_service.sso_login.processor


class GenerateCaptchaTestCase(SSOLoginProxyTestCase):
    """Test the generate_captcha method."""

    method = 'generate_captcha'
    params = (APP_NAME, FILENAME)
    success_signal = 'CaptchaGenerated'
    error_signal = 'CaptchaGenerationError'
    backend_result = 'a captcha id'
    success_result = (APP_NAME, backend_result)


class RegisterUserTestCase(SSOLoginProxyTestCase):
    """Test the register_user method."""

    method = 'register_user'
    params = (APP_NAME, EMAIL, PASSWORD, NAME, CAPTCHA_ID, CAPTCHA_SOLUTION)
    success_signal = 'UserRegistered'
    error_signal = 'UserRegistrationError'
    backend_result = EMAIL
    success_result = (APP_NAME, backend_result)


class LoginTestCase(SSOLoginProxyTestCase):
    """Test the login method."""

    method = 'login'
    params = (APP_NAME, EMAIL, PASSWORD)
    success_signal = 'LoggedIn'
    error_signal = 'LoginError'
    backend_result = TOKEN
    success_result = (APP_NAME, EMAIL)

    @defer.inlineCallbacks
    def test_success(self):
        """Test that the 'method' works ok when the user is validated."""
        self.patch(self.patchable_backend, 'is_validated', lambda _: True)

        yield super(LoginTestCase, self).test_success()

        expected_credentials = yield self.keyring.get_credentials(APP_NAME)
        self.assertEqual(expected_credentials, TOKEN)

    def test_not_validated(self):
        """Test that the 'method' works ok when the user is not validated."""
        self.patch(self.patchable_backend, 'is_validated', lambda _: False)
        self.patch(self, 'success_signal', 'UserNotValidated')

        return super(LoginTestCase, self).test_success()

    def test_error_when_setting_credentials(self):
        """The 'error_signal' is emitted when credentials can't be set."""
        self.patch(self.patchable_backend, 'is_validated', lambda _: True)
        exc = TypeError('foo')
        self.patch(self.keyring, 'set_credentials', lambda *a: defer.fail(exc))
        patched_backend_method = lambda *a, **kw: self.backend_result
        expected_result = (APP_NAME, dict(errtype='TypeError', message='foo'))

        return self.assert_method_correct('LoginError', 'LoggedIn',
                    patched_backend_method, expected_result)


class ValidateEmailTestCase(SSOLoginProxyTestCase):
    """Test the validate_email method."""

    method = 'validate_email'
    params = (APP_NAME, EMAIL, PASSWORD, EMAIL_TOKEN)
    success_signal = 'EmailValidated'
    error_signal = 'EmailValidationError'
    backend_result = TOKEN
    success_result = (APP_NAME, EMAIL)


class RequestPasswordResetTokenTestCase(SSOLoginProxyTestCase):
    """Test the request_password_reset_token method."""

    method = 'request_password_reset_token'
    params = (APP_NAME, EMAIL)
    success_signal = 'PasswordResetTokenSent'
    error_signal = 'PasswordResetError'
    backend_result = EMAIL
    success_result = (APP_NAME, backend_result)


class SetNewPasswordTestCase(SSOLoginProxyTestCase):
    """Test the set_new_password method."""

    method = 'set_new_password'
    params = (APP_NAME, EMAIL, EMAIL_TOKEN, PASSWORD)
    success_signal = 'PasswordChanged'
    error_signal = 'PasswordChangeError'
    backend_result = EMAIL
    success_result = (APP_NAME, backend_result)


class CredentialsManagementProxyTestCase(AbstractTestCase):
    """Tests for the CredentialsManagementProxy DBus interface."""

    error_signal = 'CredentialsError'
    args = dict(foo='bar', fuh='baz')
    params = (APP_NAME, args)

    @defer.inlineCallbacks
    def setUp(self):
        yield super(CredentialsManagementProxyTestCase, self).setUp()
        self.credentials = FakedCredentials()
        self.patch(main, 'Credentials', lambda *a, **kw: self.credentials)

        self.client = self.sso_client.cred_manager
        self.patchable_backend = self.credentials

    def _backend_succeed(self, *args, **kwargs):
        """Make self.patchable_backend return self.backend_result."""
        return defer.succeed(self.backend_result)

    def _backend_fail(self, *args, **kwargs):
        """Make self.patchable_backend return a failed deferred."""
        return defer.fail(ValueError((args, kwargs)))


class FindCredentialsTestCase(CredentialsManagementProxyTestCase):
    """Test the find_credentials method."""

    method = 'find_credentials'
    success_signal = 'CredentialsFound'
    backend_result = TOKEN
    success_result = (APP_NAME, backend_result)

    @skipIfOS('win32', 'find_credentials_sync is only provided in Linux '
                       'due to compatibility issues with old clients.')
    @defer.inlineCallbacks
    def test_find_credentials_sync(self):
        """The credentials are asked and returned in a sync call."""
        d = defer.Deferred()

        self.client.call_method('find_credentials_sync',
                                APP_NAME, self.args,
                                reply_handler=d.callback,
                                error_handler=d.errback)
        creds = yield d
        self.assertEqual(creds, TOKEN)

    @skipIfOS('win32', 'find_credentials_sync is only provided in Linux '
                       'due to compatibility issues with old clients.')
    @defer.inlineCallbacks
    def test_find_credentials_sync_error(self):
        """If find_credentials_sync fails, error_handler is called."""
        self.patch(self.credentials, 'find_credentials', self._backend_fail)
        d = defer.Deferred()

        self.client.call_method('find_credentials_sync',
                                APP_NAME, self.args,
                                reply_handler=d.errback,
                                error_handler=d.callback)
        error = yield d
        error = error.args[0]
        self.assertEqual(error, 'ValueError')


class ClearCredentialsTestCase(CredentialsManagementProxyTestCase):
    """Test the clear_credentials method."""

    method = 'clear_credentials'
    success_signal = 'CredentialsCleared'
    success_result = (APP_NAME,)


class StoreCredentialsTestCase(CredentialsManagementProxyTestCase):
    """Test the store_credentials method."""

    method = 'store_credentials'
    success_signal = 'CredentialsStored'
    success_result = (APP_NAME,)


class CredentialsManagementOpsTestCase(CredentialsManagementProxyTestCase):
    """Tests for the CredentialsManagementProxy login/register methods."""

    success_signal = 'CredentialsFound'
    success_result = (APP_NAME, TOKEN)

    def _backend_succeed(self, *args, **kwargs):
        """Make self.patchable_backend return self.backend_result."""
        self.sso_service.cred_manager.CredentialsFound(APP_NAME, TOKEN)

    def _backend_fail(self, *args, **kwargs):
        """Make self.patchable_backend fail."""
        self.sso_service.cred_manager.CredentialsError(APP_NAME,
            dict(errtype='ValueError'))


class RegisterTestCase(CredentialsManagementOpsTestCase):
    """Test the register method."""

    method = 'register'


class LoginOnlyTestCase(CredentialsManagementOpsTestCase):
    """Test the login method."""

    method = 'login'


class LoginEmailPasswordTestCase(CredentialsManagementOpsTestCase):
    """Test the login method."""

    method = 'login_email_password'
    backend_method = 'login'
    args = dict(email=EMAIL, password=PASSWORD)
    params = (APP_NAME, args)