test/cross-test-server.py

   1 # Copyright (C) 2006 Collabora Ltd. <http://www.collabora.co.uk/>
   2 #
   3 # Licensed under the Academic Free License version 2.1
   4 #
   5 # This program is free software; you can redistribute it and/or modify
   6 # it under the terms of the GNU Lesser General Public License as published by
   7 # the Free Software Foundation; either version 2.1 of the License, or
   8 # (at your option) any later version.
   9 #
  10 # This program is distributed in the hope that it will be useful,
  11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 # GNU Lesser General Public License for more details.
  14 #
  15 # You should have received a copy of the GNU Lesser General Public License
  16 # along with this program; if not, write to the Free Software
  17 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  18
  19 from sets import Set
  20 import logging
  21
  22 import gobject
  23
  24 import dbus.glib
  25 from dbus import SessionBus
  26 from dbus.service import BusName
  27
  28 from crosstest import CROSS_TEST_PATH, CROSS_TEST_BUS_NAME, \
  29                       INTERFACE_SINGLE_TESTS, INTERFACE_TESTS,\
  30                       INTERFACE_CALLBACK_TESTS, INTERFACE_SIGNAL_TESTS,\
  31                       SignalTestsImpl
  32
  33
  34 logging.basicConfig()
  35 logging.getLogger().setLevel(1)
  36 logger = logging.getLogger('cross-test-server')
  37
  38
  39 class VerboseSet(Set):
  40     def add(self, thing):
  41         print '%s ok' % thing
  42         Set.add(self, thing)
  43
  44
  45 objects = {}
  46
  47
  48 tested_things = VerboseSet()
  49 testable_things = [
  50         INTERFACE_SINGLE_TESTS + '.Sum',
  51         INTERFACE_TESTS + '.Identity',
  52         INTERFACE_TESTS + '.IdentityByte',
  53         INTERFACE_TESTS + '.IdentityBool',
  54         INTERFACE_TESTS + '.IdentityInt16',
  55         INTERFACE_TESTS + '.IdentityUInt16',
  56         INTERFACE_TESTS + '.IdentityInt32',
  57         INTERFACE_TESTS + '.IdentityUInt32',
  58         INTERFACE_TESTS + '.IdentityInt64',
  59         INTERFACE_TESTS + '.IdentityUInt64',
  60         INTERFACE_TESTS + '.IdentityDouble',
  61         INTERFACE_TESTS + '.IdentityString',
  62         INTERFACE_TESTS + '.IdentityArray',
  63         INTERFACE_TESTS + '.IdentityByteArray',
  64         INTERFACE_TESTS + '.IdentityBoolArray',
  65         INTERFACE_TESTS + '.IdentityInt16Array',
  66         INTERFACE_TESTS + '.IdentityUInt16Array',
  67         INTERFACE_TESTS + '.IdentityInt32Array',
  68         INTERFACE_TESTS + '.IdentityUInt32Array',
  69         INTERFACE_TESTS + '.IdentityInt64Array',
  70         INTERFACE_TESTS + '.IdentityUInt64Array',
  71         INTERFACE_TESTS + '.IdentityDoubleArray',
  72         INTERFACE_TESTS + '.IdentityStringArray',
  73         INTERFACE_TESTS + '.Sum',
  74         INTERFACE_TESTS + '.InvertMapping',
  75         INTERFACE_TESTS + '.DeStruct',
  76         INTERFACE_TESTS + '.Primitize',
  77         INTERFACE_TESTS + '.Trigger',
  78         INTERFACE_TESTS + '.Exit',
  79         INTERFACE_TESTS + '.Invert',
  80         INTERFACE_SIGNAL_TESTS + '.Trigger',
  81 ]
  82
  83
  84 class SingleTestsImpl(dbus.service.Object):
  85
  86     @dbus.service.method(INTERFACE_SINGLE_TESTS, 'ay', 'u')
  87     def Sum(self, input):
  88         tested_things.add(INTERFACE_SINGLE_TESTS + '.Sum')
  89         u = sum(input)
  90         logger.info('Sum of %r is %r', input, u)
  91         return u
  92
  93
  94 class TestsImpl(dbus.service.Object):
  95
  96     def __init__(self, bus_name, service_path, exit_fn):
  97         self._exit_fn = exit_fn
  98         dbus.service.Object.__init__(self, bus_name, service_path)
  99
 100     @dbus.service.method(INTERFACE_TESTS, 'v', 'v')
 101     def Identity(self, input):
 102         tested_things.add(INTERFACE_TESTS + '.Identity')
 103         return input
 104
 105     @dbus.service.method(INTERFACE_TESTS, 'y', 'y')
 106     def IdentityByte(self, input):
 107         tested_things.add(INTERFACE_TESTS + '.IdentityByte')
 108         return input
 109
 110     @dbus.service.method(INTERFACE_TESTS, 'b', 'b')
 111     def IdentityBool(self, input):
 112         tested_things.add(INTERFACE_TESTS + '.IdentityBool')
 113         return input
 114
 115     @dbus.service.method(INTERFACE_TESTS, 'n', 'n')
 116     def IdentityInt16(self, input):
 117         tested_things.add(INTERFACE_TESTS + '.IdentityInt16')
 118         return input
 119
 120     @dbus.service.method(INTERFACE_TESTS, 'q', 'q')
 121     def IdentityUInt16(self, input):
 122         tested_things.add(INTERFACE_TESTS + '.IdentityUInt16')
 123         return input
 124
 125     @dbus.service.method(INTERFACE_TESTS, 'i', 'i')
 126     def IdentityInt32(self, input):
 127         tested_things.add(INTERFACE_TESTS + '.IdentityInt32')
 128         return input
 129
 130     @dbus.service.method(INTERFACE_TESTS, 'u', 'u')
 131     def IdentityUInt32(self, input):
 132         tested_things.add(INTERFACE_TESTS + '.IdentityUInt32')
 133         return input
 134
 135     @dbus.service.method(INTERFACE_TESTS, 'x', 'x')
 136     def IdentityInt64(self, input):
 137         tested_things.add(INTERFACE_TESTS + '.IdentityInt64')
 138         return input
 139
 140     @dbus.service.method(INTERFACE_TESTS, 't', 't')
 141     def IdentityUInt64(self, input):
 142         tested_things.add(INTERFACE_TESTS + '.IdentityUInt64')
 143         return input
 144
 145     @dbus.service.method(INTERFACE_TESTS, 'd', 'd')
 146     def IdentityDouble(self, input):
 147         tested_things.add(INTERFACE_TESTS + '.IdentityDouble')
 148         return input
 149
 150     @dbus.service.method(INTERFACE_TESTS, 's', 's')
 151     def IdentityString(self, input):
 152         tested_things.add(INTERFACE_TESTS + '.IdentityString')
 153         return input
 154
 155     @dbus.service.method(INTERFACE_TESTS, 'av', 'av')
 156     def IdentityArray(self, input):
 157         tested_things.add(INTERFACE_TESTS + '.IdentityArray')
 158         return input
 159
 160     @dbus.service.method(INTERFACE_TESTS, 'ay', 'ay')
 161     def IdentityByteArray(self, input):
 162         tested_things.add(INTERFACE_TESTS + '.IdentityByteArray')
 163         return input
 164
 165     @dbus.service.method(INTERFACE_TESTS, 'ab', 'ab')
 166     def IdentityBoolArray(self, input):
 167         tested_things.add(INTERFACE_TESTS + '.IdentityBoolArray')
 168         return input
 169
 170     @dbus.service.method(INTERFACE_TESTS, 'an', 'an')
 171     def IdentityInt16Array(self, input):
 172         tested_things.add(INTERFACE_TESTS + '.IdentityInt16Array')
 173         return input
 174
 175     @dbus.service.method(INTERFACE_TESTS, 'aq', 'aq')
 176     def IdentityUInt16Array(self, input):
 177         tested_things.add(INTERFACE_TESTS + '.IdentityUInt16Array')
 178         return input
 179
 180     @dbus.service.method(INTERFACE_TESTS, 'ai', 'ai')
 181     def IdentityInt32Array(self, input):
 182         tested_things.add(INTERFACE_TESTS + '.IdentityInt32Array')
 183         return input
 184
 185     @dbus.service.method(INTERFACE_TESTS, 'au', 'au')
 186     def IdentityUInt32Array(self, input):
 187         tested_things.add(INTERFACE_TESTS + '.IdentityUInt32Array')
 188         return input
 189
 190     @dbus.service.method(INTERFACE_TESTS, 'ax', 'ax')
 191     def IdentityInt64Array(self, input):
 192         tested_things.add(INTERFACE_TESTS + '.IdentityInt64Array')
 193         return input
 194
 195     @dbus.service.method(INTERFACE_TESTS, 'at', 'at')
 196     def IdentityUInt64Array(self, input):
 197         tested_things.add(INTERFACE_TESTS + '.IdentityUInt64Array')
 198         return input
 199
 200     @dbus.service.method(INTERFACE_TESTS, 'ad', 'ad')
 201     def IdentityDoubleArray(self, input):
 202         tested_things.add(INTERFACE_TESTS + '.IdentityDoubleArray')
 203         return input
 204
 205     @dbus.service.method(INTERFACE_TESTS, 'as', 'as')
 206     def IdentityStringArray(self, input):
 207         tested_things.add(INTERFACE_TESTS + '.IdentityStringArray')
 208         return input
 209
 210     @dbus.service.method(INTERFACE_TESTS, 'ai', 'x')
 211     def Sum(self, input):
 212         tested_things.add(INTERFACE_TESTS + '.Sum')
 213         x = sum(input)
 214         logger.info('Sum of %r is %r', input, x)
 215         return x
 216
 217
 218     @dbus.service.method(INTERFACE_TESTS, 'a{ss}', 'a{sas}', utf8_strings=True)
 219     def InvertMapping(self, input):
 220         tested_things.add(INTERFACE_TESTS + '.InvertMapping')
 221         output = dbus.Dictionary({})
 222         for k, v in input.iteritems():
 223             output.setdefault(v, []).append(k)
 224         return output
 225
 226     @dbus.service.method(INTERFACE_TESTS, '(sun)', 'sun')
 227     def DeStruct(self, input):
 228         tested_things.add(INTERFACE_TESTS + '.DeStruct')
 229         return input
 230
 231     @dbus.service.method(INTERFACE_TESTS, 'v', 'av')
 232     def Primitize(self, input):
 233         tested_things.add(INTERFACE_TESTS + '.Primitize')
 234         return list(self.primitivize_helper(input))
 235
 236     def primitivize_helper(self, input):
 237         if (isinstance(input, tuple) or isinstance(input, dbus.Struct)
 238             or isinstance(input, list) or isinstance(input, dbus.Array)):
 239             for x in input:
 240                 for y in self.primitivize_helper(x):
 241                     yield y
 242         elif isinstance(input, dbus.ByteArray):
 243             for x in input:
 244                 yield dbus.Byte(ord(x))
 245         elif isinstance(input, dict) or isinstance(input, dbus.Dictionary):
 246             for x in input:
 247                 for y in self.primitivize_helper(x):
 248                     yield y
 249                 for y in self.primitivize_helper(input[x]):
 250                     yield y
 251         elif input.variant_level > 0:
 252             yield input.__class__(input)
 253         else:
 254             yield input
 255
 256     @dbus.service.method(INTERFACE_TESTS, 'b', 'b')
 257     def Invert(self, input):
 258         tested_things.add(INTERFACE_TESTS + '.Invert')
 259         return not input
 260
 261     @dbus.service.method(INTERFACE_TESTS, 'st', '', utf8_strings=True,
 262                          connection_keyword='conn')
 263     def Trigger(self, object, parameter, conn=None):
 264         assert isinstance(object, str)
 265         logger.info('method/signal: client wants me to emit Triggered(%r) from %r', parameter, object)
 266         tested_things.add(INTERFACE_TESTS + '.Trigger')
 267         gobject.idle_add(lambda: self.emit_Triggered_from(conn, object,
 268                                                           parameter))
 269
 270     def emit_Triggered_from(self, conn, object, parameter):
 271         assert isinstance(object, str)
 272         logger.info('method/signal: Emitting Triggered(%r) from %r', parameter, object)
 273         obj = objects.get(object, None)
 274         if obj is None:
 275             obj = SignalTestsImpl(conn, object)
 276             objects[object] = obj
 277         obj.Triggered(parameter)
 278         logger.info('method/signal: Emitted Triggered')
 279
 280     @dbus.service.method(INTERFACE_TESTS, '', '')
 281     def Exit(self):
 282         logger.info('client wants me to Exit')
 283         tested_things.add(INTERFACE_TESTS + '.Exit')
 284         for x in testable_things:
 285             if x not in tested_things:
 286                 print '%s untested' % x
 287         logger.info('will quit when idle')
 288         gobject.idle_add(self._exit_fn)
 289
 290
 291 class Server(SingleTestsImpl, TestsImpl, SignalTestsImpl):
 292
 293     def triggered_by_client(self, parameter1, parameter2, sender, sender_path):
 294         # Called when the client emits TestSignals.Trigger from any object.
 295         logger.info('signal/callback: Triggered by client (%s:%s): (%r,%r)', sender, sender_path, parameter1, parameter2)
 296         tested_things.add(INTERFACE_SIGNAL_TESTS + '.Trigger')
 297         dbus.Interface(dbus.SessionBus().get_object(sender, sender_path),
 298                        INTERFACE_CALLBACK_TESTS).Response(parameter1, parameter2)
 299         logger.info('signal/callback: Sent Response')
 300
 301
 302
 303 if __name__ == '__main__':
 304     bus = SessionBus()
 305     bus_name = BusName(CROSS_TEST_BUS_NAME, bus=bus)
 306     loop = gobject.MainLoop()
 307     obj = Server(bus_name, CROSS_TEST_PATH, loop.quit)
 308     objects[CROSS_TEST_PATH] = obj
 309     bus.add_signal_receiver(obj.triggered_by_client,
 310                             signal_name='Trigger',
 311                             dbus_interface=INTERFACE_SIGNAL_TESTS,
 312                             named_service=None,
 313                             path=None,
 314                             sender_keyword='sender',
 315                             path_keyword='sender_path',
 316                             utf8_strings=True)
 317
 318     logger.info("running...")
 319     loop.run()
 320     logger.info("main loop exited.")