~mvo/ubuntu-sso-client/strawman-lp711413

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

# Author: Natalia Bidart <natalia.bidart@canonical.com>
# Author: Alejandro J. Cura <alecu@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/>.
"""Tests for the Credentials module."""

import logging
import urllib

import gobject

from twisted.internet import defer
from twisted.internet.defer import inlineCallbacks
from twisted.trial.unittest import TestCase, FailTest

from contrib.testing.testcase import MementoHandler
from ubuntu_sso import credentials
from ubuntu_sso.credentials import (APP_NAME_KEY, HELP_TEXT_KEY, NO_OP,
    PING_URL_KEY, TC_URL_KEY, UI_CLASS_KEY, UI_MODULE_KEY, WINDOW_ID_KEY,
    ERROR_KEY, ERROR_DETAIL_KEY)
from ubuntu_sso.tests import (APP_NAME, EMAIL, GTK_GUI_CLASS, GTK_GUI_MODULE,
    HELP_TEXT, PING_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


KWARGS = {
    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,
    UI_MODULE_KEY: 'ubuntu_sso.tests.test_credentials',
    UI_CLASS_KEY: 'FakedClientGUI',
}

UI_KWARGS = {
    APP_NAME_KEY: APP_NAME,
    TC_URL_KEY: TC_URL,
    HELP_TEXT_KEY: HELP_TEXT,
    WINDOW_ID_KEY: WINDOW_ID,
}

SIG_LOGIN_SUCCEEDED = '1'
SIG_REGISTRATION_SUCCEEDED = '2'
SIG_USER_CANCELATION = '3'


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


class FailingClientGUI(object):
    """Fake a failing SSO GUI."""

    def __init__(self, *args, **kwargs):
        raise SampleMiscException('A failing GUI class.')


class FakedClientGUI(object):
    """Fake a SSO GUI."""

    def __init__(self, *args, **kwargs):
        self.args = args
        self.kwargs = kwargs
        self.signals = []
        self.methods = []
        self.connect = lambda *a: self.signals.append(a)

    def finish_success(self):
        """Fake the success finish."""
        self.methods.append('finish_success')

    def finish_error(self, error):
        """Fake the error finish."""
        self.methods.append(('finish_error', error))


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

    def setUp(self):
        """Init."""
        self.patch(gobject, 'idle_add', lambda f, *a, **kw: f(*a, **kw))
        self._called = False  # helper

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

    def tearDown(self):
        """Clean up."""
        self._called = False

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


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

    def setUp(self):
        """Init."""
        super(CredentialsTestCase, self).setUp()
        self.obj = credentials.Credentials(success_cb=self.success,
                                           error_cb=self.error,
                                           denial_cb=self.denial,
                                           **KWARGS)

    def tearDown(self):
        """Clean up."""
        super(CredentialsTestCase, self).tearDown()

    def success(self, *args, **kwargs):
        """To be called on success."""
        self._set_called('success', *args, **kwargs)

    def error(self, *args, **kwargs):
        """To be called on error."""
        self._set_called('error', *args, **kwargs)

    def denial(self, *args, **kwargs):
        """To be called on credentials denial."""
        self._set_called('denial', *args, **kwargs)

    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)
        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], {})


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

    def test_callbacks_are_stored(self):
        """Creation callbacks are stored."""
        self.assertEqual(self.obj._success_cb, self.success)
        self.assertEqual(self.obj._error_cb, self.error)
        self.assertEqual(self.obj.denial_cb, self.denial)

    def test_callbacks_default_to_no_op(self):
        """The callbacks are a no-operation if not given."""
        self.obj = credentials.Credentials(**KWARGS)
        self.assertEqual(self.obj._success_cb, NO_OP)
        self.assertEqual(self.obj._error_cb, NO_OP)
        self.assertEqual(self.obj.denial_cb, NO_OP)

    def test_creation_parameters_are_stored(self):
        """Creation parameters are stored."""
        for key, value in KWARGS.iteritems():
            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_ui_class_defaults_to_gtk(self):
        """The ui class defaults to gtk."""
        newkw = KWARGS.copy()
        newkw.pop(UI_CLASS_KEY)
        self.obj = credentials.Credentials(**newkw)

        self.assertEqual(getattr(self.obj, UI_CLASS_KEY), GTK_GUI_CLASS)

    def test_ui_module_defaults_to_gtk(self):
        """The ui module defaults to gtk."""
        newkw = KWARGS.copy()
        newkw.pop(UI_MODULE_KEY)
        self.obj = credentials.Credentials(**newkw)

        self.assertEqual(getattr(self.obj, UI_MODULE_KEY), GTK_GUI_MODULE)

    def test_success_cb_gui_none(self):
        """Success callback calls the caller but not the GUI."""
        self.obj.gui = None
        self.obj.success_cb(creds=TOKEN)

        self.assertEqual(self._called, (('success', APP_NAME, TOKEN), {}),
                         'caller _success_cb was called.')

    def test_success_cb_gui_not_none(self):
        """Success callback calls the caller and finish the GUI."""
        self.obj.gui = ui = FakedClientGUI(APP_NAME)
        self.obj.success_cb(creds=TOKEN)

        self.assertEqual(self._called, (('success', APP_NAME, TOKEN), {}),
                         'caller _success_cb was called.')
        self.assertEqual(ui.methods, ['finish_success'],
                         'GUI finish_success was called.')
        self.assertTrue(self.obj.gui is None, 'gui is reset to None.')

    def test_error_cb_gui_none(self):
        """Error callback calls the caller but not the GUI."""
        self.obj.gui = None
        error_dict = {'foo': 'bar'}
        self.obj.error_cb(error_dict=error_dict)

        self.assertEqual(self._called, (('error', APP_NAME, error_dict), {}),
                         'caller _error_cb was called.')

    def test_error_cb_gui_not_none(self):
        """Error callback calls the caller and finish the GUI."""
        self.obj.gui = ui = FakedClientGUI(APP_NAME)
        error_dict = {'foo': 'bar'}
        self.obj.error_cb(error_dict=error_dict)

        self.assertEqual(self._called, (('error', APP_NAME, error_dict), {}),
                         'caller _error_cb was called.')
        self.assertEqual(ui.methods, [('finish_error', error_dict)],
                         'GUI finish_error was called.')
        self.assertTrue(self.obj.gui is None, 'gui is reset to None.')


class CredentialsLoginSuccessTestCase(CredentialsTestCase):
    """Test suite for the Credentials login success callback."""

    @inlineCallbacks
    def test_cred_not_found(self):
        """On auth success, if cred not found, self.error_cb is called."""
        self.patch(credentials.Keyring, 'get_credentials',
                   lambda kr, app: defer.succeed(None))

        yield self.obj._login_success_cb(dialog=None, app_name=APP_NAME,
                                         email=EMAIL)

        msg = 'Creds are empty! This should not happen'
        self.assert_error_cb_called(msg='Problem while retrieving credentials',
                                    detailed_error=AssertionError(msg))

    @inlineCallbacks
    def test_cred_error(self):
        """On auth success, if cred failed, self.error_cb is called."""
        expected_error = SampleMiscException()
        self.patch(credentials.Keyring, 'get_credentials',
                   lambda kr, app: defer.fail(expected_error))

        yield self.obj._login_success_cb(dialog=None, app_name=APP_NAME,
                                         email=EMAIL)

        msg = 'Problem while retrieving credentials'
        self.assert_error_cb_called(msg=msg, detailed_error=expected_error)
        self.assertTrue(self.memento.check_exception(SampleMiscException))

    @inlineCallbacks
    def test_ping_success(self):
        """Auth success + cred found + ping success, success_cb is called."""
        self.patch(credentials.Keyring, 'get_credentials',
                   lambda kr, app: defer.succeed(TOKEN))
        self.patch(self.obj, '_ping_url', lambda *a, **kw: 200)

        yield self.obj._login_success_cb(dialog=None, app_name=APP_NAME,
                                         email=EMAIL)

        self.assertEqual(self._called, (('success', APP_NAME, TOKEN), {}))

    @inlineCallbacks
    def test_ping_error(self):
        """Auth success + cred found + ping error, error_cb is called.

        Credentials are removed. The exception is logged.
        """
        self.patch(credentials.Keyring, 'get_credentials',
                   lambda kr, app: defer.succeed(TOKEN))
        error = 'Bla'
        self.patch(credentials.urllib2, 'urlopen',
                   lambda *a, **kw: self.fail(error))
        self._cred_cleared = False
        self.patch(self.obj, 'clear_credentials',
                   lambda: setattr(self, '_cred_cleared', True))

        yield self.obj._login_success_cb(dialog=None, app_name=APP_NAME,
                                         email=EMAIL)

        # error cb called correctly
        msg = 'Problem opening the ping_url'
        self.assert_error_cb_called(msg=msg, detailed_error=FailTest(error))

        # credentials cleared
        self.assertTrue(self._cred_cleared)

        # exception logged
        self.assertTrue(self.memento.check_exception(FailTest, error))

    @inlineCallbacks
    def test_pings_url(self):
        """On auth success, self.ping_url is opened."""
        self.patch(credentials.Keyring, 'get_credentials',
                   lambda kr, app: defer.succeed(TOKEN))
        self.patch(credentials.Keyring, 'delete_credentials',
                   lambda kr, app: defer.succeed(None))

        self._url_pinged = False
        self.patch(self.obj, '_ping_url',
                   lambda *a, **kw: setattr(self, '_url_pinged', (a, kw)))

        yield self.obj._login_success_cb(dialog=None, app_name=APP_NAME,
                                         email=EMAIL)

        self.assertEqual(self._url_pinged, ((APP_NAME, EMAIL, TOKEN), {}))

    @inlineCallbacks
    def test_no_ping_url_is_success(self):
        """Auth success + cred found + no ping url, success_cb is called.

        Credentials are NOT removed.

        """
        self.patch(credentials.Keyring, 'get_credentials',
                   lambda kr, app: defer.succeed(TOKEN))  # auth success
        self.patch(self.obj, 'clear_credentials', self._set_called)
        self.obj.ping_url = None

        yield self.obj._login_success_cb(dialog=None, app_name=APP_NAME,
                                         email=EMAIL)

        self.assertEqual(self._called, (('success', APP_NAME, TOKEN), {}))


class CredentialsAuthDeniedTestCase(CredentialsTestCase):
    """Test suite for the Credentials auth denied callback."""

    def test_auth_denial_cb(self):
        """On auth denied, self.denial_cb is called."""
        self.obj.gui = ui = FakedClientGUI(APP_NAME)
        self.obj._auth_denial_cb(dialog=None, app_name=APP_NAME)
        self.assertEqual(self._called, (('denial', APP_NAME), {}))
        self.assertEqual(ui.methods, [], 'no GUI method was called.')


class PingUrlTestCase(CredentialsTestCase):
    """Test suite for the URL pinging."""

    def setUp(self):
        super(PingUrlTestCase, self).setUp()
        self._request = None

        def faked_urlopen(request):
            """Fake urlopener."""
            self._request = request
            response = urllib.addinfourl(fp=open('/dev/null'),
                                         headers=request.headers,
                                         url=request.get_full_url(),
                                         code=200)
            return response

        self.patch(credentials.urllib2, 'urlopen', faked_urlopen)

    def test_ping_url_if_url_is_none(self):
        """self.ping_url is opened."""
        self.patch(credentials.urllib2, 'urlopen', self.fail)
        self.obj.ping_url = None
        self.obj._ping_url(app_name=APP_NAME, email=EMAIL, credentials=TOKEN)
        # no failure

    def test_ping_url(self):
        """On auth success, self.ping_url is opened."""
        self.obj._ping_url(app_name=APP_NAME, email=EMAIL, credentials=TOKEN)

        self.assertIsInstance(self._request, credentials.urllib2.Request)
        self.assertEqual(self._request.get_full_url(),
                         self.obj.ping_url + EMAIL)

    def test_request_is_signed_with_credentials(self):
        """The http request to self.ping_url is signed with the credentials."""
        result = self.obj._ping_url(APP_NAME, EMAIL, TOKEN)
        headers = self._request.headers

        self.assertIn('Authorization', headers)
        oauth_stuff = headers['Authorization']

        expected = 'oauth_consumer_key="xQ7xDAz", ' \
            'oauth_signature_method="HMAC-SHA1", oauth_version="1.0", ' \
            'oauth_token="GkInOfSMGwTXAUoVQwLUoPxElEEUdhsLVNTPhxHJDUIeHCPNEo' \
            '", oauth_signature="'
        self.assertIn(expected, oauth_stuff)
        self.assertEqual(result, 200)

    def test_ping_url_error(self):
        """Exception is handled if ping fails."""
        error = 'Blu'
        self.patch(credentials.urllib2, 'urlopen', lambda r: self.fail(error))

        self.obj._ping_url(APP_NAME, EMAIL, TOKEN)

        msg = 'Problem opening the ping_url'
        self.assert_error_cb_called(msg=msg, detailed_error=FailTest(error))
        self.assertTrue(self.memento.check_exception(FailTest, error))


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

    @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)

    @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, {})

    @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.obj.find_credentials()

        msg = 'Problem while retrieving credentials'
        self.assert_error_cb_called(msg=msg, detailed_error=expected_error)
        self.assertTrue(self.memento.check_exception(SampleMiscException))


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

    @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,), {}))

    @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.obj.clear_credentials()

        msg = 'Problem while deleting credentials'
        self.assert_error_cb_called(msg=msg, detailed_error=expected_error)
        self.assertTrue(self.memento.check_exception(SampleMiscException))


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

    operation = 'register'
    login_only = False

    def setUp(self):
        super(RegisterTestCase, self).setUp()
        self.ui_kwargs = UI_KWARGS.copy()
        self.ui_kwargs['login_only'] = self.login_only

    @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))

        yield getattr(self.obj, self.operation)()

        self.assertEqual(self._called, (('success', APP_NAME, TOKEN), {}))

    @inlineCallbacks
    def test_without_existent_token(self):
        """The operation returns the credentials gathered by the GUI."""
        self.patch(credentials.Keyring, 'get_credentials',
                   lambda kr, app: defer.succeed(None))

        yield getattr(self.obj, self.operation)()

        self.assertEqual(self.obj.gui.kwargs, self.ui_kwargs)

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

        yield getattr(self.obj, self.operation)()

        msg = 'Problem while retrieving credentials'
        self.assert_error_cb_called(msg=msg, detailed_error=expected_error)
        self.assertTrue(self.memento.check_exception(SampleMiscException))

    @inlineCallbacks
    def test_with_exception_on_gui(self):
        """The operation calls the error callback if a GUI exception occurs."""
        self.patch(credentials.Keyring, 'get_credentials',
                   lambda kr, app: defer.succeed(None))
        err = 'A failing GUI class.'
        self.obj.ui_class = 'FailingClientGUI'

        yield getattr(self.obj, self.operation)()

        msg = 'Problem opening the Ubuntu SSO user interface'
        self.assert_error_cb_called(msg=msg,
                                    detailed_error=SampleMiscException(err))
        self.assertTrue(self.memento.check_exception(SampleMiscException, err))

    @inlineCallbacks
    def test_connects_gui_signals(self):
        """Outcome signals from GUI are connected to callbacks."""
        self.patch(credentials.Keyring, 'get_credentials',
                   lambda kr, app: defer.succeed(None))

        expected = [
            (SIG_LOGIN_SUCCEEDED, self.obj._login_success_cb),
            (SIG_REGISTRATION_SUCCEEDED, self.obj._login_success_cb),
            (SIG_USER_CANCELATION, self.obj._auth_denial_cb),
        ]
        yield getattr(self.obj, self.operation)()
        self.assertEqual(self.obj.gui.signals, expected)

    @inlineCallbacks
    def test_gui_is_created(self):
        """The GUI is created and stored."""
        self.patch(credentials.Keyring, 'get_credentials',
                   lambda kr, app: defer.succeed(None))

        yield getattr(self.obj, self.operation)()

        self.assertIsInstance(self.obj.gui, FakedClientGUI)
        self.assertEqual(self.obj.gui.args, ())
        self.assertEqual(self.obj.gui.kwargs, self.ui_kwargs)


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

    operation = 'login'
    login_only = True