Blame test/cross-test-server.py

Packit 130fc8
#!/usr/bin/env python
Packit 130fc8
Packit 130fc8
# Copyright (C) 2006 Collabora Ltd. <http://www.collabora.co.uk/>
Packit 130fc8
#
Packit 130fc8
# Permission is hereby granted, free of charge, to any person
Packit 130fc8
# obtaining a copy of this software and associated documentation
Packit 130fc8
# files (the "Software"), to deal in the Software without
Packit 130fc8
# restriction, including without limitation the rights to use, copy,
Packit 130fc8
# modify, merge, publish, distribute, sublicense, and/or sell copies
Packit 130fc8
# of the Software, and to permit persons to whom the Software is
Packit 130fc8
# furnished to do so, subject to the following conditions:
Packit 130fc8
#
Packit 130fc8
# The above copyright notice and this permission notice shall be
Packit 130fc8
# included in all copies or substantial portions of the Software.
Packit 130fc8
#
Packit 130fc8
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
Packit 130fc8
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
Packit 130fc8
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
Packit 130fc8
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
Packit 130fc8
# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
Packit 130fc8
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
Packit 130fc8
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
Packit 130fc8
# DEALINGS IN THE SOFTWARE.
Packit 130fc8
Packit 130fc8
from __future__ import print_function
Packit 130fc8
import logging
Packit 130fc8
Packit 130fc8
try:
Packit 130fc8
    from gi.repository import GObject as gobject
Packit 130fc8
except ImportError:
Packit 130fc8
    raise SystemExit(77)
Packit 130fc8
Packit 130fc8
import dbus.glib
Packit 130fc8
from dbus import SessionBus
Packit 130fc8
from dbus.service import BusName
Packit 130fc8
from dbus._compat import is_py2
Packit 130fc8
Packit 130fc8
from crosstest import (
Packit 130fc8
    CROSS_TEST_BUS_NAME, CROSS_TEST_PATH, INTERFACE_CALLBACK_TESTS,
Packit 130fc8
    INTERFACE_SIGNAL_TESTS, INTERFACE_SINGLE_TESTS, INTERFACE_TESTS,
Packit 130fc8
    SignalTestsImpl)
Packit 130fc8
Packit 130fc8
Packit 130fc8
logging.basicConfig()
Packit 130fc8
logging.getLogger().setLevel(1)
Packit 130fc8
logger = logging.getLogger('cross-test-server')
Packit 130fc8
Packit 130fc8
Packit 130fc8
class VerboseSet(set):
Packit 130fc8
    def add(self, thing):
Packit 130fc8
        print('%s ok' % thing)
Packit 130fc8
        set.add(self, thing)
Packit 130fc8
Packit 130fc8
Packit 130fc8
objects = {}
Packit 130fc8
Packit 130fc8
Packit 130fc8
tested_things = VerboseSet()
Packit 130fc8
testable_things = [
Packit 130fc8
        INTERFACE_SINGLE_TESTS + '.Sum',
Packit 130fc8
        INTERFACE_TESTS + '.Identity',
Packit 130fc8
        INTERFACE_TESTS + '.IdentityByte',
Packit 130fc8
        INTERFACE_TESTS + '.IdentityBool',
Packit 130fc8
        INTERFACE_TESTS + '.IdentityInt16',
Packit 130fc8
        INTERFACE_TESTS + '.IdentityUInt16',
Packit 130fc8
        INTERFACE_TESTS + '.IdentityInt32',
Packit 130fc8
        INTERFACE_TESTS + '.IdentityUInt32',
Packit 130fc8
        INTERFACE_TESTS + '.IdentityInt64',
Packit 130fc8
        INTERFACE_TESTS + '.IdentityUInt64',
Packit 130fc8
        INTERFACE_TESTS + '.IdentityDouble',
Packit 130fc8
        INTERFACE_TESTS + '.IdentityString',
Packit 130fc8
        INTERFACE_TESTS + '.IdentityArray',
Packit 130fc8
        INTERFACE_TESTS + '.IdentityByteArray',
Packit 130fc8
        INTERFACE_TESTS + '.IdentityBoolArray',
Packit 130fc8
        INTERFACE_TESTS + '.IdentityInt16Array',
Packit 130fc8
        INTERFACE_TESTS + '.IdentityUInt16Array',
Packit 130fc8
        INTERFACE_TESTS + '.IdentityInt32Array',
Packit 130fc8
        INTERFACE_TESTS + '.IdentityUInt32Array',
Packit 130fc8
        INTERFACE_TESTS + '.IdentityInt64Array',
Packit 130fc8
        INTERFACE_TESTS + '.IdentityUInt64Array',
Packit 130fc8
        INTERFACE_TESTS + '.IdentityDoubleArray',
Packit 130fc8
        INTERFACE_TESTS + '.IdentityStringArray',
Packit 130fc8
        INTERFACE_TESTS + '.Sum',
Packit 130fc8
        INTERFACE_TESTS + '.InvertMapping',
Packit 130fc8
        INTERFACE_TESTS + '.DeStruct',
Packit 130fc8
        INTERFACE_TESTS + '.Primitize',
Packit 130fc8
        INTERFACE_TESTS + '.Trigger',
Packit 130fc8
        INTERFACE_TESTS + '.Exit',
Packit 130fc8
        INTERFACE_TESTS + '.Invert',
Packit 130fc8
        INTERFACE_SIGNAL_TESTS + '.Trigger',
Packit 130fc8
]
Packit 130fc8
Packit 130fc8
Packit 130fc8
class SingleTestsImpl(dbus.service.Object):
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_SINGLE_TESTS, 'ay', 'u')
Packit 130fc8
    def Sum(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_SINGLE_TESTS + '.Sum')
Packit 130fc8
        u = sum(input)
Packit 130fc8
        logger.info('Sum of %r is %r', input, u)
Packit 130fc8
        return u
Packit 130fc8
Packit 130fc8
Packit 130fc8
class TestsImpl(dbus.service.Object):
Packit 130fc8
Packit 130fc8
    def __init__(self, bus_name, service_path, exit_fn):
Packit 130fc8
        self._exit_fn = exit_fn
Packit 130fc8
        dbus.service.Object.__init__(self, bus_name, service_path)
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'v', 'v')
Packit 130fc8
    def Identity(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.Identity')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'y', 'y')
Packit 130fc8
    def IdentityByte(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.IdentityByte')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'b', 'b')
Packit 130fc8
    def IdentityBool(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.IdentityBool')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'n', 'n')
Packit 130fc8
    def IdentityInt16(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.IdentityInt16')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'q', 'q')
Packit 130fc8
    def IdentityUInt16(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.IdentityUInt16')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'i', 'i')
Packit 130fc8
    def IdentityInt32(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.IdentityInt32')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'u', 'u')
Packit 130fc8
    def IdentityUInt32(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.IdentityUInt32')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'x', 'x')
Packit 130fc8
    def IdentityInt64(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.IdentityInt64')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 't', 't')
Packit 130fc8
    def IdentityUInt64(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.IdentityUInt64')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'd', 'd')
Packit 130fc8
    def IdentityDouble(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.IdentityDouble')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 's', 's')
Packit 130fc8
    def IdentityString(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.IdentityString')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'av', 'av')
Packit 130fc8
    def IdentityArray(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.IdentityArray')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'ay', 'ay')
Packit 130fc8
    def IdentityByteArray(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.IdentityByteArray')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'ab', 'ab')
Packit 130fc8
    def IdentityBoolArray(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.IdentityBoolArray')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'an', 'an')
Packit 130fc8
    def IdentityInt16Array(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.IdentityInt16Array')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'aq', 'aq')
Packit 130fc8
    def IdentityUInt16Array(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.IdentityUInt16Array')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'ai', 'ai')
Packit 130fc8
    def IdentityInt32Array(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.IdentityInt32Array')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'au', 'au')
Packit 130fc8
    def IdentityUInt32Array(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.IdentityUInt32Array')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'ax', 'ax')
Packit 130fc8
    def IdentityInt64Array(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.IdentityInt64Array')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'at', 'at')
Packit 130fc8
    def IdentityUInt64Array(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.IdentityUInt64Array')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'ad', 'ad')
Packit 130fc8
    def IdentityDoubleArray(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.IdentityDoubleArray')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'as', 'as')
Packit 130fc8
    def IdentityStringArray(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.IdentityStringArray')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'ai', 'x')
Packit 130fc8
    def Sum(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.Sum')
Packit 130fc8
        x = sum(input)
Packit 130fc8
        logger.info('Sum of %r is %r', input, x)
Packit 130fc8
        return x
Packit 130fc8
Packit 130fc8
Packit 130fc8
    kwargs = {}
Packit 130fc8
    if is_py2:
Packit 130fc8
        kwargs['utf8_strings'] = True
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'a{ss}', 'a{sas}', **kwargs)
Packit 130fc8
    def InvertMapping(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.InvertMapping')
Packit 130fc8
        output = dbus.Dictionary({})
Packit 130fc8
        for k, v in input.items():
Packit 130fc8
            output.setdefault(v, []).append(k)
Packit 130fc8
        return output
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, '(sun)', 'sun')
Packit 130fc8
    def DeStruct(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.DeStruct')
Packit 130fc8
        return input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'v', 'av')
Packit 130fc8
    def Primitize(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.Primitize')
Packit 130fc8
        return list(self.primitivize_helper(input))
Packit 130fc8
Packit 130fc8
    def primitivize_helper(self, input):
Packit 130fc8
        if (isinstance(input, tuple) or isinstance(input, dbus.Struct)
Packit 130fc8
            or isinstance(input, list) or isinstance(input, dbus.Array)):
Packit 130fc8
            for x in input:
Packit 130fc8
                for y in self.primitivize_helper(x):
Packit 130fc8
                    yield y
Packit 130fc8
        elif isinstance(input, dbus.ByteArray):
Packit 130fc8
            for x in input:
Packit 130fc8
                yield dbus.Byte(ord(x))
Packit 130fc8
        elif isinstance(input, dict) or isinstance(input, dbus.Dictionary):
Packit 130fc8
            for x in input:
Packit 130fc8
                for y in self.primitivize_helper(x):
Packit 130fc8
                    yield y
Packit 130fc8
                for y in self.primitivize_helper(input[x]):
Packit 130fc8
                    yield y
Packit 130fc8
        elif input.variant_level > 0:
Packit 130fc8
            yield input.__class__(input)
Packit 130fc8
        else:
Packit 130fc8
            yield input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'b', 'b')
Packit 130fc8
    def Invert(self, input):
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.Invert')
Packit 130fc8
        return not input
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, 'st', '',
Packit 130fc8
                         connection_keyword='conn',
Packit 130fc8
                         **kwargs)
Packit 130fc8
    def Trigger(self, object, parameter, conn=None):
Packit 130fc8
        assert isinstance(object, str)
Packit 130fc8
        logger.info('method/signal: client wants me to emit Triggered(%r) from %r', parameter, object)
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.Trigger')
Packit 130fc8
        gobject.idle_add(lambda: self.emit_Triggered_from(conn, object,
Packit 130fc8
                                                          parameter))
Packit 130fc8
    
Packit 130fc8
    def emit_Triggered_from(self, conn, object, parameter):
Packit 130fc8
        assert isinstance(object, str)
Packit 130fc8
        logger.info('method/signal: Emitting Triggered(%r) from %r', parameter, object)
Packit 130fc8
        obj = objects.get(object, None)
Packit 130fc8
        if obj is None:
Packit 130fc8
            obj = SignalTestsImpl(conn, object)
Packit 130fc8
            objects[object] = obj
Packit 130fc8
        obj.Triggered(parameter)
Packit 130fc8
        logger.info('method/signal: Emitted Triggered')
Packit 130fc8
Packit 130fc8
    @dbus.service.method(INTERFACE_TESTS, '', '')
Packit 130fc8
    def Exit(self):
Packit 130fc8
        logger.info('client wants me to Exit')
Packit 130fc8
        tested_things.add(INTERFACE_TESTS + '.Exit')
Packit 130fc8
        for x in testable_things:
Packit 130fc8
            if x not in tested_things:
Packit 130fc8
                print('%s untested' % x)
Packit 130fc8
        logger.info('will quit when idle')
Packit 130fc8
        gobject.idle_add(self._exit_fn)
Packit 130fc8
Packit 130fc8
Packit 130fc8
class Server(SingleTestsImpl, TestsImpl, SignalTestsImpl):
Packit 130fc8
Packit 130fc8
    def triggered_by_client(self, parameter1, parameter2, sender, sender_path):
Packit 130fc8
        # Called when the client emits TestSignals.Trigger from any object.
Packit 130fc8
        logger.info('signal/callback: Triggered by client (%s:%s): (%r,%r)', sender, sender_path, parameter1, parameter2)
Packit 130fc8
        tested_things.add(INTERFACE_SIGNAL_TESTS + '.Trigger')
Packit 130fc8
        dbus.Interface(dbus.SessionBus().get_object(sender, sender_path),
Packit 130fc8
                       INTERFACE_CALLBACK_TESTS).Response(parameter1, parameter2)
Packit 130fc8
        logger.info('signal/callback: Sent Response')
Packit 130fc8
Packit 130fc8
Packit 130fc8
Packit 130fc8
if __name__ == '__main__':
Packit 130fc8
    bus = SessionBus()
Packit 130fc8
    bus_name = BusName(CROSS_TEST_BUS_NAME, bus=bus)
Packit 130fc8
    loop = gobject.MainLoop()
Packit 130fc8
    obj = Server(bus_name, CROSS_TEST_PATH, loop.quit)
Packit 130fc8
    objects[CROSS_TEST_PATH] = obj
Packit 130fc8
    kwargs = {}
Packit 130fc8
    if is_py2:
Packit 130fc8
        kwargs['utf8_strings'] = True
Packit 130fc8
    bus.add_signal_receiver(obj.triggered_by_client,
Packit 130fc8
                            signal_name='Trigger',
Packit 130fc8
                            dbus_interface=INTERFACE_SIGNAL_TESTS,
Packit 130fc8
                            named_service=None,
Packit 130fc8
                            path=None,
Packit 130fc8
                            sender_keyword='sender',
Packit 130fc8
                            path_keyword='sender_path',
Packit 130fc8
                            **kwargs)
Packit 130fc8
Packit 130fc8
    logger.info("running...")
Packit 130fc8
    loop.run()
Packit 130fc8
    logger.info("main loop exited.")