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     def Trigger(self, object, parameter):
 263         assert isinstance(object, str)
 264         logger.info('method/signal: client wants me to emit Triggered(%r) from %r', parameter, object)
 265         tested_things.add(INTERFACE_TESTS + '.Trigger')
 266         gobject.idle_add(lambda: self.emit_Triggered_from(object, parameter))
 267
 268     def emit_Triggered_from(self, object, parameter):
 269         assert isinstance(object, str)
 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                             utf8_strings=True)
 315
 316     logger.info("running...")
 317     loop.run()
 318     logger.info("main loop exited.")