python-ubuntu-sso-client.tests 3.0.0-0ubuntu1 / usr / share / pyshared / ubuntu-sso-client / ubuntu_sso / utils / tests / test_ipc.py

# -*- coding: utf-8 -*-
#
# Copyright 2011-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.
"""Perspective Broker IPC test cases."""

import logging

from collections import namedtuple

from twisted.internet import defer
from twisted.spread.pb import (
    DeadReferenceError,
    NoSuchMethod,
)
from ubuntuone.devtools.handlers import MementoHandler

from ubuntu_sso.tests import TestCase
from ubuntu_sso.utils import ipc


TEST_SERVICE = 'foo-service'
TEST_CMDLINE = 'foo.bin'
TEST_PORT = 40404


class RandomException(Exception):
    """A random exception."""


class FakeActivationClient(object):
    """A fake ActivationClient."""

    def __init__(self, config):
        """Initialize this fake instance."""
        self.config = config

    def get_active_port(self):
        """Return the port where the pb server is running."""
        return defer.succeed(self.config.port)


class FakeActivationInstance(FakeActivationClient):
    """A fake ActivationInstance."""

    get_port = FakeActivationClient.get_active_port


# Invalid name "callRemote", etc.
# pylint: disable=C0102, C0103


class FakeReactor(object):
    """A fake reactor."""

    def __init__(self):
        """Initialize this faker."""
        self.connection_class = namedtuple("Connection", "host port factory")
        self.connections = []

    def connectTCP(self, host, port, factory):
        """Store the connected factory."""
        connection = self.connection_class(host, port, factory)
        self.connections.append(connection)

    def listenTCP(self, port, factory, interface):
        """Store the listenning factory."""
        connection = self.connection_class(interface, port, factory)
        self.connections.append(connection)


class FakeRemoteClient(object):
    """A fake RemoteClient."""

    missing_signal = "missing"
    failing_signal = "failing"
    dead_remote = False
    random_exception = RandomException()

    def __init__(self, dead_remote=False):
        self.called = False
        self.dead_remote = dead_remote

    def callRemote(self, signal_name, *a, **kw):
        """Fake a call to a given remote method."""
        if self.dead_remote:
            raise DeadReferenceError("Calling Stale Broker")

        if signal_name == self.missing_signal:
            return defer.fail(NoSuchMethod())
        if signal_name == self.failing_signal:
            return defer.fail(self.random_exception)

        self.called = (a, kw)
        return defer.succeed(None)


class DummyRemoteService(ipc.RemoteService):
    """Represent a dummy IPC object."""

    remote_calls = ['foo', 'bar']
    next_result = None

    def foo(self):
        """Dummy foo."""
        self.Success(self.next_result)

    def bar(self, error):
        """Dummy bar."""
        self.NotSuccess(error)

    @ipc.signal
    def Success(self, param):
        """Fire a signal to notify a success."""

    @ipc.signal
    def NotSuccess(self, error):
        """Fire a signal to notify a not-success."""

    @ipc.signal
    def NoArgs(self):
        """Get no args passed."""

    @ipc.signal
    def JustArgs(self, *args):
        """Just get args."""

    @ipc.signal
    def JustKwargs(self, **kwargs):
        """Just get kwargs."""

    @ipc.signal
    def BothArgsAndKwargs(self, *args, **kwargs):
        """Both args and kwargs."""

# pylint: enable=C0102, C0103


class DummyService(ipc.BaseService):
    """Represent a dummy root service."""

    services = {'dummy': DummyRemoteService}
    name = 'Dummy Service'
    port = TEST_PORT
    cmdline = 'yadda yo'


class DummyRemoteClient(ipc.RemoteClient):
    """Represent a dummy remote client."""

    call_remote_functions = DummyRemoteService.remote_calls
    signal_handlers = ['Success', 'NotSuccess']


class DummyClient(ipc.BaseClient):
    """Represent a dummy base client."""

    clients = {'dummy': DummyRemoteClient}
    service_name = DummyService.name
    service_port = DummyService.port
    service_cmdline = DummyService.cmdline


class BaseIPCTestCase(TestCase):
    """Set the ipc to a random port for this instance."""

    timeout = 5

    client_class = None  # the BaseClient instance
    service_class = None  # the BaseService instance

    remote_client_name = None  # the name of the remote client in the client
    remote_service_name = None  # the name of the remote service in the service

    method_mapping = []
    signal_mapping = []

    @defer.inlineCallbacks
    def setUp(self):
        yield super(BaseIPCTestCase, self).setUp()

        self.service = None
        self.client = None

        if self.service_class is not None:
            # pylint: disable=E1102
            self.service = self.service_class()
            yield self.service.start()
            self.addCleanup(self.service.shutdown)

            self.client = self.client_class()
            yield self.client.connect()
            self.addCleanup(self.client.disconnect)

    @property
    def remote_service(self):
        """Get the service named 'service_name'."""
        return getattr(self.service, self.remote_service_name)

    @property
    def remote_client(self):
        """Get the client named 'remote_client_name'."""
        return getattr(self.client, self.remote_client_name)

    @defer.inlineCallbacks
    def assert_method_called(self, service, method, result, *args, **kwargs):
        """Check that calling 'method(*args, **kwargs)' should query 'service'.

        The returned result from calling 'method(*args, **kwargs)' should be
        equal to the given parameter 'result'. If 'result' is a deferred, its
        result attribute will be used as expected result (ergo the deferred
        should be already called).

        """
        client = self.remote_client

        # hack to handle service methods returning a deferred with result
        if isinstance(result, defer.Deferred):
            real_result = result.result
        else:
            real_result = result

        self.patch(service, method, lambda *a, **kw: result)
        actual = yield client.call_method(method, *args, **kwargs)
        self.assertEqual(real_result, actual)
        self.assertEqual(service.called, {method: [(args, kwargs)]})

    def assert_remote_method(self, method_name, *args, **kwargs):
        """Assert that 'method_name' is a remote method.

        The parameters args and kwargs will be passed as such to the method
        itself, to exercise it.

        """
        self.assertIn(method_name, self.remote_service.remote_calls)
        method = getattr(self.remote_service, method_name)
        method(*args, **kwargs)

    def assert_remote_signal(self, signal_name, *args, **kwargs):
        """Assert that 'signal' is a remote signal.

        The parameters args and kwargs will be passed as such to the signal
        itself, to exercise it.

        """
        self.patch(self.remote_service, 'emit_signal', self._set_called)
        signal = getattr(self.remote_service, signal_name)
        signal(*args, **kwargs)

        expected = (signal_name,) + args
        self.assertEqual(self._called, (expected, kwargs))

    def test_remote_methods(self):
        """Check every method defined in self.method_mapping.

        Assert that every method is a remote method and that it has the
        expected signature.

        """
        for method, args, kwargs in self.method_mapping:
            self.assert_remote_method(method, *args, **kwargs)

    def test_remote_signals(self):
        """Check every signal defined in self.signal_mapping.

        Assert that every signal is a remote signal and that it has the
        expected signature.

        """
        for signal_name, args, kwargs in self.signal_mapping:
            self.assert_remote_signal(signal_name, *args, **kwargs)


class ListenConnectTestCase(BaseIPCTestCase):
    """Test suite for the server_listen and client_connect methods."""

    @defer.inlineCallbacks
    def setUp(self):
        yield super(ListenConnectTestCase, self).setUp()
        self.fake_reactor = FakeReactor()

    @defer.inlineCallbacks
    def test_server_listen(self):
        """Test the server_listen function."""
        self.patch(ipc, "ActivationInstance", FakeActivationInstance)

        fake_factory = object()
        yield ipc.server_listen(fake_factory, TEST_SERVICE,
                                TEST_CMDLINE, TEST_PORT,
                                reactor=self.fake_reactor)

        self.assertEqual(len(self.fake_reactor.connections), 1)
        conn = self.fake_reactor.connections[0]
        self.assertEqual(conn.port, TEST_PORT)
        self.assertEqual(conn.factory, fake_factory)
        self.assertEqual(conn.host, ipc.LOCALHOST)

    @defer.inlineCallbacks
    def test_client_connect(self):
        """Test the client_connect function."""
        self.patch(ipc, "ActivationClient", FakeActivationClient)

        fake_factory = object()
        yield ipc.client_connect(fake_factory, TEST_SERVICE,
                                 TEST_CMDLINE, TEST_PORT,
                                 reactor=self.fake_reactor)

        self.assertEqual(len(self.fake_reactor.connections), 1)
        conn = self.fake_reactor.connections[0]
        self.assertEqual(conn.port, TEST_PORT)
        self.assertEqual(conn.factory, fake_factory)
        self.assertEqual(conn.host, ipc.LOCALHOST)


class DummyClientTestCase(BaseIPCTestCase):
    """Test the status client class."""

    client_class = DummyClient
    service_class = DummyService

    remote_client_name = remote_service_name = 'dummy'

    method_mapping = [
        ('foo', (), {}),
        ('bar', (object(),), {}),
    ]
    signal_mapping = [
        ('Success', ('test',), {}),
        ('NotSuccess', ('yadda',), {}),
        ('NoArgs', (), {}),
        ('JustArgs', (object(), 'foo'), {}),
        ('JustKwargs', (), {'foo': 'bar'}),
        ('BothArgsAndKwargs', ('zaraza', 8), {'foo': -42}),
    ]

    @defer.inlineCallbacks
    def test_deprecated_siganl_is_also_sent(self):
        """Old-style, deprecated signals handler are also called."""
        d1 = defer.Deferred()
        d2 = defer.Deferred()

        self.remote_service.next_result = 'yadda'

        # old, deprecated way
        self.remote_client.connect_to_signal('Success', d1.callback)
        self.remote_client.on_success_cb = d2.callback

        self.remote_client.foo()

        result = yield defer.gatherResults([d1, d2])

        self.assertEqual(result, ['yadda', 'yadda'])

    @defer.inlineCallbacks
    def test_register_to_signals(self):
        """Test the register_to_signals method."""
        yield self.remote_client.register_to_signals()
        self.addCleanup(self.remote_client.unregister_to_signals)

        for signal in self.remote_client.signal_handlers:
            self.assertIn(signal, self.service.dummy.clients_per_signal)

    @defer.inlineCallbacks
    def test_unregister_to_signals(self):
        """Test the register_to_signals method."""
        yield self.remote_client.register_to_signals()
        yield self.remote_client.unregister_to_signals()

        for signal in self.remote_client.signal_handlers:
            actual = len(self.remote_service.clients_per_signal[signal])
            self.assertEqual(0, actual)


class RemoteMetaTestCase(TestCase):
    """Tests for the RemoteMeta metaclass."""

    def test_remote_calls_renamed(self):
        """The remote_calls are renamed."""
        test_token = object()

        class TestClass(object):
            """A class for testing."""

            __metaclass__ = ipc.RemoteMeta

            remote_calls = ['test_method']

            def test_method(self):
                """Fake call."""
                return test_token

        tc = TestClass()
        self.assertEquals(tc.test_method(), test_token)
        # pylint: disable=E1101
        self.assertEquals(tc.remote_test_method(), test_token)
        # pylint: enable=E1101


class SignalBroadcasterTestCase(TestCase):
    """Test the signal broadcaster code."""

    @defer.inlineCallbacks
    def setUp(self):
        yield super(SignalBroadcasterTestCase, self).setUp()
        self.client = FakeRemoteClient()
        self.sb = ipc.SignalBroadcaster()

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

    def test_remote_register_to_signals(self):
        """Assert that the client was added."""
        signals = ["demo_signal1", "demo_signal2"]
        self.sb.remote_register_to_signals(self.client, signals)
        for signal in signals:
            clients = self.sb.clients_per_signal[signal]
            self.assertTrue(self.client in clients)

    def test_emit_signal(self):
        """Assert that the client method was called."""
        first = 1
        second = 2
        word = 'word'
        signal_name = 'on_test'

        self.client.callRemote(signal_name, first, second, word=word)

        signals = [signal_name]
        self.sb.remote_register_to_signals(self.client, signals)
        self.sb.emit_signal(signal_name, first, second, foo=word)

        self.assertEqual(self.client.called, ((first, second), dict(foo=word)))

    def test_emit_signal_dead_reference(self):
        """Test dead reference while emitting a signal."""
        sample_signal = "sample_signal"
        fake_remote_client = FakeRemoteClient(dead_remote=True)

        self.sb.remote_register_to_signals(fake_remote_client, [sample_signal])
        self.assertIn(fake_remote_client,
                      self.sb.clients_per_signal[sample_signal])

        self.sb.emit_signal(sample_signal)
        self.assertNotIn(fake_remote_client,
                         self.sb.clients_per_signal[sample_signal])

    def test_emit_signal_some_dead_some_not(self):
        """Test a clean reference after a dead one."""
        sample_signal = "sample_signal"
        fake_dead_remote = FakeRemoteClient(dead_remote=True)
        fake_alive_remote = FakeRemoteClient()

        self.sb.remote_register_to_signals(fake_dead_remote, [sample_signal])
        self.sb.remote_register_to_signals(fake_alive_remote, [sample_signal])
        self.sb.emit_signal(sample_signal)

        self.assertTrue(fake_alive_remote.called, "The alive must be called.")

    def test_emit_signal_ignore_missing_handlers(self):
        """A missing signal handler should just log a debug line."""
        fake_remote_client = FakeRemoteClient()

        signal = fake_remote_client.missing_signal
        signals = [signal]
        self.sb.remote_register_to_signals(fake_remote_client, signals)
        sb_clients = self.sb.clients_per_signal[signal]
        self.assertIn(fake_remote_client, sb_clients)
        self.sb.emit_signal(signal)

        expected = ipc.SignalBroadcaster.MSG_NO_SIGNAL_HANDLER % (
            signal,
            fake_remote_client,
        )
        self.assertTrue(self.memento.check_debug(*expected))

    def test_emit_signal_log_other_errors(self):
        """Other errors should be logged as warnings."""
        fake_remote_client = FakeRemoteClient()

        signal = fake_remote_client.failing_signal
        signals = [signal]
        self.sb.remote_register_to_signals(fake_remote_client, signals)
        sb_clients = self.sb.clients_per_signal[signal]
        self.assertIn(fake_remote_client, sb_clients)
        self.sb.emit_signal(signal)

        expected = ipc.SignalBroadcaster.MSG_COULD_NOT_EMIT_SIGNAL % (
            signal,
            fake_remote_client,
            fake_remote_client.random_exception,
        )
        self.assertTrue(self.memento.check_warning(*expected))