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