test/test-service.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (C) 2004 Red Hat Inc. <http://www.redhat.com/>
   4 # Copyright (C) 2005-2007 Collabora Ltd. <http://www.collabora.co.uk/>
   5 #
   6 # Permission is hereby granted, free of charge, to any person
   7 # obtaining a copy of this software and associated documentation
   8 # files (the "Software"), to deal in the Software without
   9 # restriction, including without limitation the rights to use, copy,
  10 # modify, merge, publish, distribute, sublicense, and/or sell copies
  11 # of the Software, and to permit persons to whom the Software is
  12 # furnished to do so, subject to the following conditions:
  13 #
  14 # The above copyright notice and this permission notice shall be
  15 # included in all copies or substantial portions of the Software.
  16 #
  17 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
  18 # EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
  19 # MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
  20 # NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
  21 # HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
  22 # WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  23 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
  24 # DEALINGS IN THE SOFTWARE.
  25
  26 import sys
  27 import os
  28 import logging
  29 from time import sleep
  30
  31 builddir = os.path.normpath(os.environ["DBUS_TOP_BUILDDIR"])
  32 pydir = os.path.normpath(os.environ["DBUS_TOP_SRCDIR"])
  33
  34 import dbus
  35
  36 if not dbus.__file__.startswith(pydir):
  37     raise Exception("DBus modules are not being picked up from the package")
  38
  39 import dbus.service
  40 import dbus.glib
  41 import gobject
  42 import random
  43
  44 from dbus.gobject_service import ExportedGObject
  45
  46
  47 logging.basicConfig(filename=builddir + '/test/test-service.log', filemode='w')
  48 logging.getLogger().setLevel(1)
  49 logger = logging.getLogger('test-service')
  50
  51
  52 NAME = "org.freedesktop.DBus.TestSuitePythonService"
  53 IFACE = "org.freedesktop.DBus.TestSuiteInterface"
  54 OBJECT = "/org/freedesktop/DBus/TestSuitePythonObject"
  55
  56 class RemovableObject(dbus.service.Object):
  57     # Part of test for https://bugs.freedesktop.org/show_bug.cgi?id=10457
  58     @dbus.service.method(IFACE, in_signature='', out_signature='b')
  59     def IsThere(self):
  60         return True
  61
  62     @dbus.service.method(IFACE, in_signature='', out_signature='b')
  63     def RemoveSelf(self):
  64         self.remove_from_connection()
  65         return True
  66
  67 class TestGObject(ExportedGObject):
  68     def __init__(self, bus_name, object_path=OBJECT + '/GObject'):
  69         super(TestGObject, self).__init__(bus_name, object_path)
  70
  71     @dbus.service.method(IFACE)
  72     def Echo(self, arg):
  73         return arg
  74
  75 class TestInterface(dbus.service.Interface):
  76     @dbus.service.method(IFACE, in_signature='', out_signature='b')
  77     def CheckInheritance(self):
  78         return False
  79
  80 class Fallback(dbus.service.FallbackObject):
  81     def __init__(self, conn, object_path=OBJECT + '/Fallback'):
  82         super(Fallback, self).__init__(conn, object_path)
  83         self.add_to_connection(conn, object_path + '/Nested')
  84
  85     @dbus.service.method(IFACE, in_signature='', out_signature='oos',
  86                          path_keyword='path', rel_path_keyword='rel',
  87                          connection_keyword='conn')
  88     def TestPathAndConnKeywords(self, path=None, conn=None, rel=None):
  89         return path, rel, conn.get_unique_name()
  90
  91     @dbus.service.signal(IFACE, signature='s', rel_path_keyword='rel_path')
  92     def SignalOneString(self, test, rel_path=None):
  93         logger.info('SignalOneString(%r) @ %r', test, rel_path)
  94
  95     # Deprecated usage
  96     @dbus.service.signal(IFACE, signature='ss', path_keyword='path')
  97     def SignalTwoStrings(self, test, test2, path=None):
  98         logger.info('SignalTwoStrings(%r, %r) @ %r', test, test2, path)
  99
 100     @dbus.service.method(IFACE, in_signature='su', out_signature='',
 101                          path_keyword='path')
 102     def EmitSignal(self, signal, value, path=None):
 103         sig = getattr(self, str(signal), None)
 104         assert sig is not None
 105
 106         assert path.startswith(OBJECT + '/Fallback')
 107         rel_path = path[len(OBJECT + '/Fallback'):]
 108         if rel_path == '':
 109             rel_path = '/'
 110
 111         if signal == 'SignalOneString':
 112             logger.info('Emitting %s from rel %r', signal, rel_path)
 113             sig('I am a fallback', rel_path=rel_path)
 114         else:
 115             val = ('I am', 'a fallback')
 116             logger.info('Emitting %s from abs %r', signal, path)
 117             sig('I am', 'a deprecated fallback', path=path)
 118
 119 class MultiPathObject(dbus.service.Object):
 120     SUPPORTS_MULTIPLE_OBJECT_PATHS = True
 121
 122 class TestObject(dbus.service.Object, TestInterface):
 123     def __init__(self, bus_name, object_path=OBJECT):
 124         dbus.service.Object.__init__(self, bus_name, object_path)
 125         self._removable = RemovableObject()
 126         self._multi = MultiPathObject(bus_name, object_path + '/Multi1')
 127         self._multi.add_to_connection(bus_name.get_bus(),
 128                                       object_path + '/Multi2')
 129         self._multi.add_to_connection(bus_name.get_bus(),
 130                                       object_path + '/Multi2/3')
 131
 132     """ Echo whatever is sent
 133     """
 134     @dbus.service.method(IFACE)
 135     def Echo(self, arg):
 136         return arg
 137
 138     @dbus.service.method(IFACE, in_signature='s', out_signature='s')
 139     def AcceptUnicodeString(self, foo):
 140         assert isinstance(foo, unicode), (foo, foo.__class__.__mro__)
 141         return foo
 142
 143     @dbus.service.method(IFACE, in_signature='s', out_signature='s', utf8_strings=True)
 144     def AcceptUTF8String(self, foo):
 145         assert isinstance(foo, str), (foo, foo.__class__.__mro__)
 146         return foo
 147
 148     @dbus.service.method(IFACE, in_signature='', out_signature='soss',
 149             sender_keyword='sender', path_keyword='path',
 150             destination_keyword='dest', message_keyword='msg')
 151     def MethodExtraInfoKeywords(self, sender=None, path=None, dest=None,
 152             msg=None):
 153         return (sender, path, dest,
 154                 msg.__class__.__module__ + '.' + msg.__class__.__name__)
 155
 156     @dbus.service.method(IFACE, in_signature='ay', out_signature='ay')
 157     def AcceptListOfByte(self, foo):
 158         assert isinstance(foo, list), (foo, foo.__class__.__mro__)
 159         return foo
 160
 161     @dbus.service.method(IFACE, in_signature='ay', out_signature='ay', byte_arrays=True)
 162     def AcceptByteArray(self, foo):
 163         assert isinstance(foo, str), (foo, foo.__class__.__mro__)
 164         return foo
 165
 166     @dbus.service.method(IFACE)
 167     def GetComplexArray(self):
 168         ret = []
 169         for i in range(0,100):
 170             ret.append((random.randint(0,100), random.randint(0,100), str(random.randint(0,100))))
 171
 172         return dbus.Array(ret, signature="(uus)")
 173
 174     def returnValue(self, test):
 175         if test == 0:
 176             return ""
 177         elif test == 1:
 178             return "",""
 179         elif test == 2:
 180             return "","",""
 181         elif test == 3:
 182             return []
 183         elif test == 4:
 184             return {}
 185         elif test == 5:
 186             return ["",""]
 187         elif test == 6:
 188             return ["","",""]
 189
 190     @dbus.service.method(IFACE, in_signature='u', out_signature='s')
 191     def ReturnOneString(self, test):
 192         return self.returnValue(test)
 193
 194     @dbus.service.method(IFACE, in_signature='u', out_signature='ss')
 195     def ReturnTwoStrings(self, test):
 196         return self.returnValue(test)
 197
 198     @dbus.service.method(IFACE, in_signature='u', out_signature='(ss)')
 199     def ReturnStruct(self, test):
 200         logger.info('ReturnStruct(%r) -> %r', test, self.returnValue(test))
 201         return self.returnValue(test)
 202
 203     @dbus.service.method(IFACE, in_signature='u', out_signature='as')
 204     def ReturnArray(self, test):
 205         return self.returnValue(test)
 206
 207     @dbus.service.method(IFACE, in_signature='u', out_signature='a{ss}')
 208     def ReturnDict(self, test):
 209         return self.returnValue(test)
 210
 211     @dbus.service.signal(IFACE, signature='s')
 212     def SignalOneString(self, test):
 213         logger.info('SignalOneString(%r)', test)
 214
 215     @dbus.service.signal(IFACE, signature='ss')
 216     def SignalTwoStrings(self, test, test2):
 217         logger.info('SignalTwoStrings(%r, %r)', test, test2)
 218
 219     @dbus.service.signal(IFACE, signature='(ss)')
 220     def SignalStruct(self, test):
 221         logger.info('SignalStruct(%r)', test)
 222
 223     @dbus.service.signal(IFACE, signature='as')
 224     def SignalArray(self, test):
 225         pass
 226
 227     @dbus.service.signal(IFACE, signature='a{ss}')
 228     def SignalDict(self, test):
 229         pass
 230
 231     @dbus.service.method(IFACE, in_signature='su', out_signature='')
 232     def EmitSignal(self, signal, value):
 233         sig = getattr(self, str(signal), None)
 234         assert(sig != None)
 235
 236         val = self.returnValue(value)
 237         # make two string case work by passing arguments in by tuple
 238         if (signal == 'SignalTwoStrings' and (value == 1 or value == 5)):
 239             val = tuple(val)
 240         else:
 241             val = tuple([val])
 242
 243         logger.info('Emitting %s with %r', signal, val)
 244         sig(*val)
 245
 246     def CheckInheritance(self):
 247         return True
 248
 249     @dbus.service.method(IFACE, in_signature='', out_signature='b')
 250     def TestListExportedChildObjects(self):
 251         objs_root = session_bus.list_exported_child_objects('/')
 252         assert objs_root == ['org'], objs_root
 253         objs_org = session_bus.list_exported_child_objects('/org')
 254         assert objs_org == ['freedesktop'], objs_org
 255         return True
 256
 257     @dbus.service.method(IFACE, in_signature='bbv', out_signature='v', async_callbacks=('return_cb', 'error_cb'))
 258     def AsynchronousMethod(self, async, fail, variant, return_cb, error_cb):
 259         try:
 260             if async:
 261                 gobject.timeout_add(500, self.AsynchronousMethod, False, fail, variant, return_cb, error_cb)
 262                 return
 263             else:
 264                 if fail:
 265                     raise RuntimeError
 266                 else:
 267                     return_cb(variant)
 268
 269                 return False # do not run again
 270         except Exception, e:
 271             error_cb(e)
 272
 273     @dbus.service.method(IFACE, in_signature='', out_signature='s', sender_keyword='sender')
 274     def WhoAmI(self, sender):
 275         return sender
 276
 277     @dbus.service.method(IFACE, in_signature='', out_signature='b')
 278     def AddRemovableObject(self):
 279         # Part of test for https://bugs.freedesktop.org/show_bug.cgi?id=10457
 280         # Keep the removable object reffed, since that's the use case for this
 281         self._removable.add_to_connection(self._connection,
 282                                           OBJECT + '/RemovableObject')
 283         return True
 284
 285     @dbus.service.method(IFACE, in_signature='', out_signature='b')
 286     def HasRemovableObject(self):
 287         # Part of test for https://bugs.freedesktop.org/show_bug.cgi?id=10457
 288         objs = session_bus.list_exported_child_objects(OBJECT)
 289         return ('RemovableObject' in objs)
 290
 291     @dbus.service.method(IFACE)
 292     def MultipleReturnWithoutSignature(self):
 293         # https://bugs.freedesktop.org/show_bug.cgi?id=10174
 294         return dbus.String('abc'), dbus.Int32(123)
 295
 296     @dbus.service.method(IFACE, in_signature='', out_signature='')
 297     def BlockFor500ms(self):
 298         sleep(0.5)
 299
 300     @dbus.service.method(IFACE, in_signature='', out_signature='',
 301                          async_callbacks=('return_cb', 'raise_cb'))
 302     def AsyncWait500ms(self, return_cb, raise_cb):
 303         def return_from_async_wait():
 304             return_cb()
 305             return False
 306         gobject.timeout_add(500, return_from_async_wait)
 307
 308     @dbus.service.method(IFACE, in_signature='', out_signature='',
 309                          async_callbacks=('return_cb', 'raise_cb'))
 310     def AsyncRaise(self, return_cb, raise_cb):
 311         class Fdo12403Error(dbus.DBusException):
 312             _dbus_error_name = 'org.freedesktop.bugzilla.bug12403'
 313
 314         raise_cb(Fdo12403Error())
 315
 316 session_bus = dbus.SessionBus()
 317 global_name = dbus.service.BusName(NAME, bus=session_bus)
 318 object = TestObject(global_name)
 319 g_object = TestGObject(global_name)
 320 fallback_object = Fallback(session_bus)
 321 loop = gobject.MainLoop()
 322 loop.run()