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, 2006 Collabora Ltd. <http://www.collabora.co.uk/>
   5 #
   6 # Licensed under the Academic Free License version 2.1
   7 #
   8 # This program is free software; you can redistribute it and/or modify
   9 # it under the terms of the GNU General Public License as published by
  10 # the Free Software Foundation; either version 2 of the License, or
  11 # (at your option) any later version.
  12 #
  13 # This program is distributed in the hope that it will be useful,
  14 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  15 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16 # GNU General Public License for more details.
  17 #
  18 # You should have received a copy of the GNU General Public License
  19 # along with this program; if not, write to the Free Software
  20 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  21
  22 import sys
  23 import os
  24 import logging
  25 from time import sleep
  26
  27 builddir = os.path.normpath(os.environ["DBUS_TOP_BUILDDIR"])
  28 pydir = os.path.normpath(os.environ["DBUS_TOP_SRCDIR"])
  29
  30 import dbus
  31
  32 if not dbus.__file__.startswith(pydir):
  33     raise Exception("DBus modules are not being picked up from the package")
  34
  35 import dbus.service
  36 import dbus.glib
  37 import gobject
  38 import random
  39
  40 from dbus.gobject_service import ExportedGObject
  41
  42
  43 logging.basicConfig(filename=builddir + '/test/test-service.log', filemode='w')
  44 logging.getLogger().setLevel(1)
  45 logger = logging.getLogger('test-service')
  46
  47
  48 NAME = "org.freedesktop.DBus.TestSuitePythonService"
  49 IFACE = "org.freedesktop.DBus.TestSuiteInterface"
  50 OBJECT = "/org/freedesktop/DBus/TestSuitePythonObject"
  51
  52 class RemovableObject(dbus.service.Object):
  53     # Part of test for https://bugs.freedesktop.org/show_bug.cgi?id=10457
  54     @dbus.service.method(IFACE, in_signature='', out_signature='b')
  55     def IsThere(self):
  56         return True
  57
  58     @dbus.service.method(IFACE, in_signature='', out_signature='b')
  59     def RemoveSelf(self):
  60         self.remove_from_connection()
  61         return True
  62
  63 class TestGObject(ExportedGObject):
  64     def __init__(self, bus_name, object_path=OBJECT + '/GObject'):
  65         super(TestGObject, self).__init__(bus_name, object_path)
  66
  67     @dbus.service.method(IFACE)
  68     def Echo(self, arg):
  69         return arg
  70
  71 class TestInterface(dbus.service.Interface):
  72     @dbus.service.method(IFACE, in_signature='', out_signature='b')
  73     def CheckInheritance(self):
  74         return False
  75
  76 class Fallback(dbus.service.FallbackObject):
  77     def __init__(self, conn, object_path=OBJECT + '/Fallback'):
  78         super(Fallback, self).__init__(conn, object_path)
  79         self.add_to_connection(conn, object_path + '/Nested')
  80
  81     @dbus.service.method(IFACE, in_signature='', out_signature='oos',
  82                          path_keyword='path', rel_path_keyword='rel',
  83                          connection_keyword='conn')
  84     def TestPathAndConnKeywords(self, path=None, conn=None, rel=None):
  85         return path, rel, conn.get_unique_name()
  86
  87     @dbus.service.signal(IFACE, signature='s', rel_path_keyword='rel_path')
  88     def SignalOneString(self, test, rel_path=None):
  89         logger.info('SignalOneString(%r) @ %r', test, rel_path)
  90
  91     # Deprecated usage
  92     @dbus.service.signal(IFACE, signature='ss', path_keyword='path')
  93     def SignalTwoStrings(self, test, test2, path=None):
  94         logger.info('SignalTwoStrings(%r, %r) @ %r', test, test2, path)
  95
  96     @dbus.service.method(IFACE, in_signature='su', out_signature='',
  97                          path_keyword='path')
  98     def EmitSignal(self, signal, value, path=None):
  99         sig = getattr(self, str(signal), None)
 100         assert sig is not None
 101
 102         assert path.startswith(OBJECT + '/Fallback')
 103         rel_path = path[len(OBJECT + '/Fallback'):]
 104         if rel_path == '':
 105             rel_path = '/'
 106
 107         if signal == 'SignalOneString':
 108             logger.info('Emitting %s from rel %r', signal, rel_path)
 109             sig('I am a fallback', rel_path=rel_path)
 110         else:
 111             val = ('I am', 'a fallback')
 112             logger.info('Emitting %s from abs %r', signal, path)
 113             sig('I am', 'a deprecated fallback', path=path)
 114
 115 class MultiPathObject(dbus.service.Object):
 116     SUPPORTS_MULTIPLE_OBJECT_PATHS = True
 117
 118 class TestObject(dbus.service.Object, TestInterface):
 119     def __init__(self, bus_name, object_path=OBJECT):
 120         dbus.service.Object.__init__(self, bus_name, object_path)
 121         self._removable = RemovableObject()
 122         self._multi = MultiPathObject(bus_name, object_path + '/Multi1')
 123         self._multi.add_to_connection(bus_name.get_bus(),
 124                                       object_path + '/Multi2')
 125         self._multi.add_to_connection(bus_name.get_bus(),
 126                                       object_path + '/Multi2/3')
 127
 128     """ Echo whatever is sent
 129     """
 130     @dbus.service.method(IFACE)
 131     def Echo(self, arg):
 132         return arg
 133
 134     @dbus.service.method(IFACE, in_signature='s', out_signature='s')
 135     def AcceptUnicodeString(self, foo):
 136         assert isinstance(foo, unicode), (foo, foo.__class__.__mro__)
 137         return foo
 138
 139     @dbus.service.method(IFACE, in_signature='s', out_signature='s', utf8_strings=True)
 140     def AcceptUTF8String(self, foo):
 141         assert isinstance(foo, str), (foo, foo.__class__.__mro__)
 142         return foo
 143
 144     @dbus.service.method(IFACE, in_signature='', out_signature='soss',
 145             sender_keyword='sender', path_keyword='path',
 146             destination_keyword='dest', message_keyword='msg')
 147     def MethodExtraInfoKeywords(self, sender=None, path=None, dest=None,
 148             msg=None):
 149         return (sender, path, dest,
 150                 msg.__class__.__module__ + '.' + msg.__class__.__name__)
 151
 152     @dbus.service.method(IFACE, in_signature='ay', out_signature='ay')
 153     def AcceptListOfByte(self, foo):
 154         assert isinstance(foo, list), (foo, foo.__class__.__mro__)
 155         return foo
 156
 157     @dbus.service.method(IFACE, in_signature='ay', out_signature='ay', byte_arrays=True)
 158     def AcceptByteArray(self, foo):
 159         assert isinstance(foo, str), (foo, foo.__class__.__mro__)
 160         return foo
 161
 162     @dbus.service.method(IFACE)
 163     def GetComplexArray(self):
 164         ret = []
 165         for i in range(0,100):
 166             ret.append((random.randint(0,100), random.randint(0,100), str(random.randint(0,100))))
 167
 168         return dbus.Array(ret, signature="(uus)")
 169
 170     def returnValue(self, test):
 171         if test == 0:
 172             return ""
 173         elif test == 1:
 174             return "",""
 175         elif test == 2:
 176             return "","",""
 177         elif test == 3:
 178             return []
 179         elif test == 4:
 180             return {}
 181         elif test == 5:
 182             return ["",""]
 183         elif test == 6:
 184             return ["","",""]
 185
 186     @dbus.service.method(IFACE, in_signature='u', out_signature='s')
 187     def ReturnOneString(self, test):
 188         return self.returnValue(test)
 189
 190     @dbus.service.method(IFACE, in_signature='u', out_signature='ss')
 191     def ReturnTwoStrings(self, test):
 192         return self.returnValue(test)
 193
 194     @dbus.service.method(IFACE, in_signature='u', out_signature='(ss)')
 195     def ReturnStruct(self, test):
 196         logger.info('ReturnStruct(%r) -> %r', test, self.returnValue(test))
 197         return self.returnValue(test)
 198
 199     @dbus.service.method(IFACE, in_signature='u', out_signature='as')
 200     def ReturnArray(self, test):
 201         return self.returnValue(test)
 202
 203     @dbus.service.method(IFACE, in_signature='u', out_signature='a{ss}')
 204     def ReturnDict(self, test):
 205         return self.returnValue(test)
 206
 207     @dbus.service.signal(IFACE, signature='s')
 208     def SignalOneString(self, test):
 209         logger.info('SignalOneString(%r)', test)
 210
 211     @dbus.service.signal(IFACE, signature='ss')
 212     def SignalTwoStrings(self, test, test2):
 213         logger.info('SignalTwoStrings(%r, %r)', test, test2)
 214
 215     @dbus.service.signal(IFACE, signature='(ss)')
 216     def SignalStruct(self, test):
 217         logger.info('SignalStruct(%r)', test)
 218
 219     @dbus.service.signal(IFACE, signature='as')
 220     def SignalArray(self, test):
 221         pass
 222
 223     @dbus.service.signal(IFACE, signature='a{ss}')
 224     def SignalDict(self, test):
 225         pass
 226
 227     @dbus.service.method(IFACE, in_signature='su', out_signature='')
 228     def EmitSignal(self, signal, value):
 229         sig = getattr(self, str(signal), None)
 230         assert(sig != None)
 231
 232         val = self.returnValue(value)
 233         # make two string case work by passing arguments in by tuple
 234         if (signal == 'SignalTwoStrings' and (value == 1 or value == 5)):
 235             val = tuple(val)
 236         else:
 237             val = tuple([val])
 238
 239         logger.info('Emitting %s with %r', signal, val)
 240         sig(*val)
 241
 242     def CheckInheritance(self):
 243         return True
 244
 245     @dbus.service.method(IFACE, in_signature='', out_signature='b')
 246     def TestListExportedChildObjects(self):
 247         objs_root = session_bus.list_exported_child_objects('/')
 248         assert objs_root == ['org'], objs_root
 249         objs_org = session_bus.list_exported_child_objects('/org')
 250         assert objs_org == ['freedesktop'], objs_org
 251         return True
 252
 253     @dbus.service.method(IFACE, in_signature='bbv', out_signature='v', async_callbacks=('return_cb', 'error_cb'))
 254     def AsynchronousMethod(self, async, fail, variant, return_cb, error_cb):
 255         try:
 256             if async:
 257                 gobject.timeout_add(500, self.AsynchronousMethod, False, fail, variant, return_cb, error_cb)
 258                 return
 259             else:
 260                 if fail:
 261                     raise RuntimeError
 262                 else:
 263                     return_cb(variant)
 264
 265                 return False # do not run again
 266         except Exception, e:
 267             error_cb(e)
 268
 269     @dbus.service.method(IFACE, in_signature='', out_signature='s', sender_keyword='sender')
 270     def WhoAmI(self, sender):
 271         return sender
 272
 273     @dbus.service.method(IFACE, in_signature='', out_signature='b')
 274     def AddRemovableObject(self):
 275         # Part of test for https://bugs.freedesktop.org/show_bug.cgi?id=10457
 276         # Keep the removable object reffed, since that's the use case for this
 277         self._removable.add_to_connection(self._connection,
 278                                           OBJECT + '/RemovableObject')
 279         return True
 280
 281     @dbus.service.method(IFACE, in_signature='', out_signature='b')
 282     def HasRemovableObject(self):
 283         # Part of test for https://bugs.freedesktop.org/show_bug.cgi?id=10457
 284         objs = session_bus.list_exported_child_objects(OBJECT)
 285         return ('RemovableObject' in objs)
 286
 287     @dbus.service.method(IFACE)
 288     def MultipleReturnWithoutSignature(self):
 289         # https://bugs.freedesktop.org/show_bug.cgi?id=10174
 290         return dbus.String('abc'), dbus.Int32(123)
 291
 292     @dbus.service.method(IFACE, in_signature='', out_signature='')
 293     def BlockFor500ms(self):
 294         sleep(0.5)
 295
 296     @dbus.service.method(IFACE, in_signature='', out_signature='',
 297                          async_callbacks=('return_cb', 'raise_cb'))
 298     def AsyncWait500ms(self, return_cb, raise_cb):
 299         def return_from_async_wait():
 300             return_cb()
 301             return False
 302         gobject.timeout_add(500, return_from_async_wait)
 303
 304 session_bus = dbus.SessionBus()
 305 global_name = dbus.service.BusName(NAME, bus=session_bus)
 306 object = TestObject(global_name)
 307 g_object = TestGObject(global_name)
 308 fallback_object = Fallback(session_bus)
 309 loop = gobject.MainLoop()
 310 loop.run()