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