python-ubuntu-sso-client.tests 13.10-0ubuntu11 / usr / lib / python2.7 / dist-packages / ubuntu-sso-client / ubuntu_sso / tests / test_credentials.py

# -*- coding: utf-8 -*-
#
# Copyright 2010-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/>.
#
# In addition, as a special exception, the copyright holders give
# permission to link the code of portions of this program with the
# OpenSSL library under certain conditions as described in each
# individual source file, and distribute linked combinations
# including the two.
# You must obey the GNU General Public License in all respects
# for all of the code used other than OpenSSL.  If you modify
# file(s) with this exception, you may extend this exception to your
# version of the file(s), but you are not obligated to do so.  If you
# do not wish to do so, delete this exception statement from your
# version.  If you delete this exception statement from all source
# files in the program, then also delete it here.
"""Tests for the Credentials module."""

import logging
import os

from twisted.internet import defer
from twisted.trial.unittest import TestCase
from ubuntuone.devtools.handlers import MementoHandler

import ubuntu_sso.main

from ubuntu_sso import credentials
from ubuntu_sso.credentials import (
    APP_NAME_KEY,
    HELP_TEXT_KEY,
    PING_URL_KEY,
    POLICY_URL_KEY,
    TC_URL_KEY,
    UI_EXECUTABLE_KEY,
    WINDOW_ID_KEY,
)
from ubuntu_sso.tests import (
    APP_NAME,
    EMAIL,
    HELP_TEXT,
    PASSWORD,
    PING_URL,
    POLICY_URL,
    TC_URL,
    TOKEN,
    WINDOW_ID,
)


# Access to a protected member of a client class
# pylint: disable=W0212

# Attribute defined outside __init__
# pylint: disable=W0201

# Instance of 'class' has no 'x' member (but some types could not be inferred)
# pylint: disable=E1103


KWARGS = {
    APP_NAME_KEY: APP_NAME,
    HELP_TEXT_KEY: HELP_TEXT,
    WINDOW_ID_KEY: WINDOW_ID,
    PING_URL_KEY: PING_URL,
    POLICY_URL_KEY: POLICY_URL,
    TC_URL_KEY: TC_URL,
    UI_EXECUTABLE_KEY: 'foo-bar-baz',
}

UI_KWARGS = {
    APP_NAME_KEY: APP_NAME,
    HELP_TEXT_KEY: HELP_TEXT,
    PING_URL_KEY: PING_URL,
    POLICY_URL_KEY: POLICY_URL,
    TC_URL_KEY: TC_URL,
    WINDOW_ID_KEY: WINDOW_ID,
}


class SampleMiscException(Exception):
    """An error to be used while testing."""


class FakedSSOLogin(object):
    """Fake a SSOLogin."""

    proxy = None

    def __init__(self, proxy):
        """Nothing."""
        FakedSSOLogin.proxy = proxy

    def login(self, *args, **kwargs):
        """Fake login."""


class BasicTestCase(TestCase):
    """Test case with a helper tracker."""

    bin_dir = 'some/bin/dir'

    @defer.inlineCallbacks
    def setUp(self):
        """Init."""
        yield super(BasicTestCase, self).setUp()
        self._called = False  # helper

        self.memento = MementoHandler()
        self.memento.setLevel(logging.DEBUG)
        credentials.logger.addHandler(self.memento)
        self.addCleanup(credentials.logger.removeHandler, self.memento)

        self.patch(credentials, 'get_bin_cmd',
                   lambda x: [os.path.join(self.bin_dir,
                                          KWARGS[UI_EXECUTABLE_KEY])])

    def _set_called(self, *args, **kwargs):
        """Set _called to True."""
        self._called = (args, kwargs)


class CredentialsTestCase(BasicTestCase):
    """Test suite for the Credentials class."""

    timeout = 5

    @defer.inlineCallbacks
    def setUp(self):
        """Init."""
        yield super(CredentialsTestCase, self).setUp()
        self.obj = credentials.Credentials(**KWARGS)


class CredentialsCallbacksTestCase(CredentialsTestCase):
    """Test suite for the Credentials callbacks."""

    def test_creation_parameters_are_stored(self):
        """Creation parameters are stored."""
        for key, value in KWARGS.items():
            self.assertEqual(getattr(self.obj, key), value)

    def test_tc_url_defaults_to_none(self):
        """The T&C url defaults to None."""
        newkw = KWARGS.copy()
        newkw.pop(TC_URL_KEY)
        self.obj = credentials.Credentials(**newkw)

        self.assertEqual(getattr(self.obj, TC_URL_KEY), None)

    def test_help_text_defaults_to_empty_string(self):
        """The T&C url defaults to the emtpy string."""
        newkw = KWARGS.copy()
        newkw.pop(HELP_TEXT_KEY)
        self.obj = credentials.Credentials(**newkw)

        self.assertEqual(getattr(self.obj, HELP_TEXT_KEY), '')

    def test_window_id_defaults_to_zero(self):
        """The T&C url defaults to 0."""
        newkw = KWARGS.copy()
        newkw.pop(WINDOW_ID_KEY)
        self.obj = credentials.Credentials(**newkw)

        self.assertEqual(getattr(self.obj, WINDOW_ID_KEY), 0)

    def test_ping_url_defaults_to_none(self):
        """The ping url defaults to None."""
        newkw = KWARGS.copy()
        newkw.pop(PING_URL_KEY)
        self.obj = credentials.Credentials(**newkw)

        self.assertEqual(getattr(self.obj, PING_URL_KEY), None)

    def test_policy_url_defaults_to_none(self):
        """The policy url defaults to None."""
        newkw = KWARGS.copy()
        newkw.pop(POLICY_URL_KEY)
        self.obj = credentials.Credentials(**newkw)

        self.assertEqual(getattr(self.obj, POLICY_URL_KEY), None)

    def test_ui_executable_default(self):
        """The ui class defaults to Qt."""
        newkw = KWARGS.copy()
        newkw.pop(UI_EXECUTABLE_KEY)
        self.obj = credentials.Credentials(**newkw)

        self.assertEqual(getattr(self.obj, UI_EXECUTABLE_KEY),
                         credentials.UI_EXECUTABLE_QT)


class FindCredentialsTestCase(CredentialsTestCase):
    """Test suite for the find_credentials method."""

    @defer.inlineCallbacks
    def test_find_credentials(self):
        """A deferred with credentials is returned when found."""
        self.patch(credentials.Keyring, 'get_credentials',
                   lambda kr, app: defer.succeed(TOKEN))

        token = yield self.obj.find_credentials()
        self.assertEqual(token, TOKEN)

    @defer.inlineCallbacks
    def test_credentials_not_found(self):
        """find_credentials returns {} when no credentials are found."""
        self.patch(credentials.Keyring, 'get_credentials',
                   lambda kr, app: defer.succeed(None))

        token = yield self.obj.find_credentials()
        self.assertEqual(token, {})

    @defer.inlineCallbacks
    def test_keyring_failure(self):
        """Failures from the keyring are handled."""
        expected_error = SampleMiscException()
        self.patch(credentials.Keyring, 'get_credentials',
                   lambda kr, app: defer.fail(expected_error))

        yield self.assertFailure(self.obj.find_credentials(),
                                 SampleMiscException)


class ClearCredentialsTestCase(CredentialsTestCase):
    """Test suite for the clear_credentials method."""

    @defer.inlineCallbacks
    def test_clear_credentials(self):
        """The credentials are cleared."""
        self.patch(credentials.Keyring, 'delete_credentials',
                   lambda kr, app: defer.succeed(self._set_called(app)))

        yield self.obj.clear_credentials()
        self.assertEqual(self._called, ((APP_NAME,), {}))

    @defer.inlineCallbacks
    def test_keyring_failure(self):
        """Failures from the keyring are handled."""
        expected_error = SampleMiscException()
        self.patch(credentials.Keyring, 'delete_credentials',
                   lambda kr, app: defer.fail(expected_error))

        yield self.assertFailure(self.obj.clear_credentials(),
                                 SampleMiscException)


class StoreCredentialsTestCase(CredentialsTestCase):
    """Test suite for the store_credentials method."""

    @defer.inlineCallbacks
    def test_store_credentials(self):
        """The credentials are stored."""
        self.patch(credentials.Keyring, 'set_credentials',
                   lambda kr, *a: defer.succeed(self._set_called(*a)))

        yield self.obj.store_credentials(TOKEN)
        self.assertEqual(self._called, ((APP_NAME, TOKEN,), {}))

    @defer.inlineCallbacks
    def test_keyring_failure(self):
        """Failures from the keyring are handled."""
        expected_error = SampleMiscException()
        self.patch(credentials.Keyring, 'set_credentials',
                   lambda kr, app, token: defer.fail(expected_error))

        yield self.assertFailure(self.obj.store_credentials(TOKEN),
                                 SampleMiscException)


class RegisterTestCase(CredentialsTestCase):
    """Test suite for the register method."""

    operation = 'register'
    login_only = False
    kwargs = {}

    @defer.inlineCallbacks
    def setUp(self):
        yield super(RegisterTestCase, self).setUp()
        self.deferred = defer.Deferred()
        self.patch_inner(self.fake_inner)
        self.exe_path = os.path.join(self.bin_dir, KWARGS[UI_EXECUTABLE_KEY])
        self.inner_args = [
            self.exe_path,
            '--app_name', APP_NAME,
            '--help_text', HELP_TEXT,
            '--ping_url', PING_URL,
            '--policy_url', POLICY_URL,
            '--tc_url', TC_URL,
            '--window_id', str(WINDOW_ID),
        ]
        if self.login_only:
            self.inner_args.append('--login_only')

        self._next_inner_result = 0

        self.method_call = getattr(self.obj, self.operation)

    def patch_inner(self, f):
        """Patch the inner call."""
        self.patch(credentials.runner, 'spawn_program', f)

    def fake_inner(self, args):
        """Fake the runner.spawn_program."""
        self.deferred.callback(args)

        if self._next_inner_result == credentials.USER_SUCCESS:
            # fake that tokens were retrieved from the UI
            self.patch(credentials.Keyring, 'get_credentials',
                       lambda kr, app: defer.succeed(TOKEN))

        return defer.succeed(self._next_inner_result)

    def fail_inner(self, args):
        """Make the inner call fail."""
        return defer.fail(SampleMiscException(args))

    def assert_exc_msg_logged(self, exception_class, msg):
        """Check that 'msg' was logged as part as the logger.exception call."""
        for rec in self.memento.records:
            if rec.exc_info and rec.exc_info[0] == exception_class:
                self.assertIn(msg, rec.getMessage())
                break

    @defer.inlineCallbacks
    def test_with_existent_token(self):
        """The operation returns the credentials if already in keyring."""
        self.patch(credentials.Keyring, 'get_credentials',
                   lambda kr, app: defer.succeed(TOKEN))

        result = yield self.method_call(**self.kwargs)

        self.assertEqual(result, TOKEN)
        self.assertFalse(self.deferred.called, 'No program was spawnned.')

    @defer.inlineCallbacks
    def test_without_existent_token_and_return_code_success(self):
        """The operation returns the credentials gathered by the inner call."""
        self.patch(credentials.Keyring, 'get_credentials',
                   lambda kr, app: defer.succeed(None))
        self._next_inner_result = credentials.USER_SUCCESS

        result = yield self.method_call(**self.kwargs)
        self.assertEqual(result, TOKEN)

        # the ui was opened and proper params were passed
        args = yield self.deferred
        self.assertEqual(self.inner_args, args)

    @defer.inlineCallbacks
    def test_ui_executable_fall_back(self):
        """The executable falls back to the Qt UI if given does not exist."""
        self.patch(credentials.Keyring, 'get_credentials',
                   lambda kr, app: defer.succeed(None))
        self._next_inner_result = credentials.USER_SUCCESS

        def get_bin_cmd_raises_if_not_qt(path):
            """Simulates nonexistence of exes that aren't qt."""
            if not path.endswith(credentials.UI_EXECUTABLE_QT):
                raise OSError()
            return [os.path.join(self.bin_dir,
                                credentials.UI_EXECUTABLE_QT)]

        self.patch(credentials, 'get_bin_cmd',
                   get_bin_cmd_raises_if_not_qt)

        result = yield self.method_call(**self.kwargs)
        self.assertEqual(result, TOKEN)

        # the ui was opened and proper params were passed
        args = yield self.deferred
        self.inner_args[0] = os.path.join(self.bin_dir,
                                          credentials.UI_EXECUTABLE_QT)
        self.assertEqual(self.inner_args, args)

    @defer.inlineCallbacks
    def test_raises_exception_if_no_ui_available(self):
        """If no GUI is available, raise GUINotAvailableError."""
        self.patch(credentials.Keyring, 'get_credentials',
                   lambda kr, app: defer.succeed(None))

        def get_bin_cmd_just_raises(*args, **kwargs):
            """Workaround for lambda: raise."""
            raise OSError()

        self.patch(credentials, 'get_bin_cmd', get_bin_cmd_just_raises)

        f = self.method_call(**self.kwargs)
        yield self.assertFailure(f, credentials.GUINotAvailableError)

    @defer.inlineCallbacks
    def test_without_existent_token_and_return_code_cancel(self, exc=None):
        """The operation returns exc if defined, else UserCancellationError."""
        self.patch(credentials.Keyring, 'get_credentials',
                   lambda kr, app: defer.succeed(None))
        self._next_inner_result = credentials.USER_CANCELLATION

        if exc is None:
            exc = credentials.UserCancellationError
        yield self.assertFailure(self.method_call(**self.kwargs), exc)

    @defer.inlineCallbacks
    def test_without_existent_token_and_return_other_code(self, result=None):
        """The operation returns CredentialsError with 'result'."""
        self.patch(credentials.Keyring, 'get_credentials',
                   lambda kr, app: defer.succeed(None))
        self._next_inner_result = credentials.USER_CANCELLATION * 2  # other

        exc = yield self.assertFailure(self.method_call(**self.kwargs),
                                       credentials.CredentialsError)
        if result is None:
            result = self._next_inner_result
        self.assertEqual(exc.args, (result,))

    @defer.inlineCallbacks
    def test_with_exception_on_credentials(self):
        """The operation calls the error callback if a exception occurs."""
        self.patch(credentials.Keyring, 'get_credentials',
                   lambda kr, app: defer.fail(SampleMiscException()))

        yield self.assertFailure(self.method_call(**self.kwargs),
                                 SampleMiscException)

        self.assertTrue(self.memento.check_exception(SampleMiscException))
        msg = 'Problem while getting credentials from the keyring'
        self.assert_exc_msg_logged(SampleMiscException, msg)

    @defer.inlineCallbacks
    def test_with_exception_on_inner_call(self):
        """The operation calls the error callback if a exception occurs."""
        self.patch(credentials.Keyring, 'get_credentials',
                   lambda kr, app: defer.succeed(None))
        self.patch_inner(self.fail_inner)

        yield self.assertFailure(self.method_call(**self.kwargs),
                                 SampleMiscException)

        self.assertTrue(self.memento.check_exception(SampleMiscException))
        msg = 'Problem while performing %s' % self.operation
        self.assert_exc_msg_logged(SampleMiscException, msg)


class LoginTestCase(RegisterTestCase):
    """Test suite for the login method."""

    operation = 'login'
    login_only = True


class LoginEmailPasswordTestCase(RegisterTestCase):
    """Test suite for the login_email_password method."""

    operation = 'login'
    login_only = True
    kwargs = {'email': EMAIL, 'password': PASSWORD}

    @defer.inlineCallbacks
    def setUp(self):
        yield super(LoginEmailPasswordTestCase, self).setUp()
        self.patch(ubuntu_sso.main, 'SSOLogin', FakedSSOLogin)
        self.patch_inner(self.fake_inner)
        self.inner_args = dict(app_name=APP_NAME,
                               email=EMAIL, password=PASSWORD,
                               ping_url=PING_URL)

    def patch_inner(self, f):
        """Patch the inner call."""
        self.patch(FakedSSOLogin, 'login', f)

    def fake_inner(self, **kwargs):
        """Fake the runner.spawn_program."""
        self.deferred.callback(kwargs)

        if self._next_inner_result == credentials.USER_SUCCESS:
            # fake that tokens were retrieved from the UI
            self.patch(credentials.Keyring, 'get_credentials',
                       lambda kr, app: defer.succeed(TOKEN))
            FakedSSOLogin.proxy.LoggedIn(APP_NAME, EMAIL)
        elif self._next_inner_result == credentials.USER_CANCELLATION:
            FakedSSOLogin.proxy.UserNotValidated(APP_NAME, EMAIL)
        else:
            FakedSSOLogin.proxy.LoginError(APP_NAME, {'errtype': 'foo'})

    def fail_inner(self, **kwargs):
        """Make the inner call fail."""
        raise SampleMiscException(kwargs)

    def test_ui_executable_fall_back(self):
        """This check does not apply for this test case."""

    def test_ui_executable_uses_qt_ui_if_none_available(self):
        """This check does not apply for this test case."""

    def test_raises_exception_if_no_ui_available(self):
        """This check does not apply for this test case."""

    def test_without_existent_token_and_return_code_cancel(self, exc=None):
        """The operation returns UserNotValidatedError."""
        exc = credentials.UserNotValidatedError
        return super(LoginEmailPasswordTestCase, self).\
               test_without_existent_token_and_return_code_cancel(exc=exc)

    def test_without_existent_token_and_return_other_code(self, result=None):
        """The operation returns CredentialsError with 'result'."""
        result = 'foo'
        return super(LoginEmailPasswordTestCase, self).\
               test_without_existent_token_and_return_other_code(result=result)