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}')
 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', '')
 264     def Trigger(self, object, parameter):
 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(object, parameter))
 268
 269     def emit_Triggered_from(self, object, parameter):
 270         logger.info('method/signal: Emitting Triggered(%r) from %r', parameter, object)
 271         obj = objects.get(object, None)
 272         if obj is None:
 273             obj = SignalTestsImpl(dbus.service.BusName(CROSS_TEST_BUS_NAME), object)
 274             objects[object] = obj
 275         obj.Triggered(parameter)
 276         logger.info('method/signal: Emitted Triggered')
 277
 278     @dbus.service.method(INTERFACE_TESTS, '', '')
 279     def Exit(self):
 280         logger.info('client wants me to Exit')
 281         tested_things.add(INTERFACE_TESTS + '.Exit')
 282         for x in testable_things:
 283             if x not in tested_things:
 284                 print '%s untested' % x
 285         logger.info('will quit when idle')
 286         gobject.idle_add(self._exit_fn)
 287
 288
 289 class Server(SingleTestsImpl, TestsImpl, SignalTestsImpl):
 290
 291     def triggered_by_client(self, parameter1, parameter2, sender, sender_path):
 292         # Called when the client emits TestSignals.Trigger from any object.
 293         logger.info('signal/callback: Triggered by client (%s:%s): (%r,%r)', sender, sender_path, parameter1, parameter2)
 294         tested_things.add(INTERFACE_SIGNAL_TESTS + '.Trigger')
 295         dbus.Interface(dbus.SessionBus().get_object(sender, sender_path),
 296                        INTERFACE_CALLBACK_TESTS).Response(parameter1, parameter2)
 297         logger.info('signal/callback: Sent Response')
 298
 299
 300
 301 if __name__ == '__main__':
 302     bus = SessionBus()
 303     bus_name = BusName(CROSS_TEST_BUS_NAME)
 304     loop = gobject.MainLoop()
 305     obj = Server(bus_name, CROSS_TEST_PATH, loop.quit)
 306     objects[CROSS_TEST_PATH] = obj
 307     bus.add_signal_receiver(obj.triggered_by_client,
 308                             signal_name='Trigger',
 309                             dbus_interface=INTERFACE_SIGNAL_TESTS,
 310                             named_service=None,
 311                             path=None,
 312                             sender_keyword='sender',
 313                             path_keyword='sender_path')
 314
 315     logger.info("running...")
 316     loop.run()
 317     logger.info("main loop exited.")