test/cross-test-server.py

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