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, bus_name, object_path=OBJECT + '/Fallback'):
  78         super(Fallback, self).__init__(bus_name, object_path)
  79
  80     @dbus.service.method(IFACE, in_signature='', out_signature='os',
  81                          path_keyword='path', # rel_path_keyword='rel',
  82                          connection_keyword='conn')
  83     def TestPathAndConnKeywords(self, path=None, conn=None):
  84         return path, conn.get_unique_name()
  85
  86     @dbus.service.signal(IFACE, signature='s', rel_path_keyword='rel_path')
  87     def SignalOneString(self, test, rel_path=None):
  88         logger.info('SignalOneString(%r) @ %r', test, rel_path)
  89
  90     # Deprecated
  91     @dbus.service.signal(IFACE, signature='ss', path_keyword='path')
  92     def SignalTwoStrings(self, test, test2, path=None):
  93         logger.info('SignalTwoStrings(%r, %r) @ %r', test, test2, path)
  94
  95     @dbus.service.method(IFACE, in_signature='su', out_signature='',
  96                          path_keyword='path')
  97     def EmitSignal(self, signal, value, path=None):
  98         sig = getattr(self, str(signal), None)
  99         assert sig is not None
 100
 101         assert path.startswith(OBJECT + '/Fallback')
 102         rel_path = path[len(OBJECT + '/Fallback'):]
 103         if rel_path == '':
 104             rel_path = '/'
 105
 106         if signal == 'SignalOneString':
 107             logger.info('Emitting %s from rel %r', signal, rel_path)
 108             sig('I am a fallback', rel_path=rel_path)
 109         else:
 110             val = ('I am', 'a fallback')
 111             logger.info('Emitting %s from abs %r', signal, path)
 112             sig('I am', 'a deprecated fallback', path=path)
 113
 114 class TestObject(dbus.service.Object, TestInterface):
 115     def __init__(self, bus_name, object_path=OBJECT):
 116         dbus.service.Object.__init__(self, bus_name, object_path)
 117         self._removable = RemovableObject()
 118
 119     """ Echo whatever is sent
 120     """
 121     @dbus.service.method(IFACE)
 122     def Echo(self, arg):
 123         return arg
 124
 125     @dbus.service.method(IFACE, in_signature='s', out_signature='s')
 126     def AcceptUnicodeString(self, foo):
 127         assert isinstance(foo, unicode), (foo, foo.__class__.__mro__)
 128         return foo
 129
 130     @dbus.service.method(IFACE, in_signature='s', out_signature='s', utf8_strings=True)
 131     def AcceptUTF8String(self, foo):
 132         assert isinstance(foo, str), (foo, foo.__class__.__mro__)
 133         return foo
 134
 135     @dbus.service.method(IFACE, in_signature='', out_signature='soss',
 136             sender_keyword='sender', path_keyword='path',
 137             destination_keyword='dest', message_keyword='msg')
 138     def MethodExtraInfoKeywords(self, sender=None, path=None, dest=None,
 139             msg=None):
 140         return (sender, path, dest,
 141                 msg.__class__.__module__ + '.' + msg.__class__.__name__)
 142
 143     @dbus.service.method(IFACE, in_signature='ay', out_signature='ay')
 144     def AcceptListOfByte(self, foo):
 145         assert isinstance(foo, list), (foo, foo.__class__.__mro__)
 146         return foo
 147
 148     @dbus.service.method(IFACE, in_signature='ay', out_signature='ay', byte_arrays=True)
 149     def AcceptByteArray(self, foo):
 150         assert isinstance(foo, str), (foo, foo.__class__.__mro__)
 151         return foo
 152
 153     @dbus.service.method(IFACE)
 154     def GetComplexArray(self):
 155         ret = []
 156         for i in range(0,100):
 157             ret.append((random.randint(0,100), random.randint(0,100), str(random.randint(0,100))))
 158
 159         return dbus.Array(ret, signature="(uus)")
 160
 161     def returnValue(self, test):
 162         if test == 0:
 163             return ""
 164         elif test == 1:
 165             return "",""
 166         elif test == 2:
 167             return "","",""
 168         elif test == 3:
 169             return []
 170         elif test == 4:
 171             return {}
 172         elif test == 5:
 173             return ["",""]
 174         elif test == 6:
 175             return ["","",""]
 176
 177     @dbus.service.method(IFACE, in_signature='u', out_signature='s')
 178     def ReturnOneString(self, test):
 179         return self.returnValue(test)
 180
 181     @dbus.service.method(IFACE, in_signature='u', out_signature='ss')
 182     def ReturnTwoStrings(self, test):
 183         return self.returnValue(test)
 184
 185     @dbus.service.method(IFACE, in_signature='u', out_signature='(ss)')
 186     def ReturnStruct(self, test):
 187         logger.info('ReturnStruct(%r) -> %r', test, self.returnValue(test))
 188         return self.returnValue(test)
 189
 190     @dbus.service.method(IFACE, in_signature='u', out_signature='as')
 191     def ReturnArray(self, test):
 192         return self.returnValue(test)
 193
 194     @dbus.service.method(IFACE, in_signature='u', out_signature='a{ss}')
 195     def ReturnDict(self, test):
 196         return self.returnValue(test)
 197
 198     @dbus.service.signal(IFACE, signature='s')
 199     def SignalOneString(self, test):
 200         logger.info('SignalOneString(%r)', test)
 201
 202     @dbus.service.signal(IFACE, signature='ss')
 203     def SignalTwoStrings(self, test, test2):
 204         logger.info('SignalTwoStrings(%r, %r)', test, test2)
 205
 206     @dbus.service.signal(IFACE, signature='(ss)')
 207     def SignalStruct(self, test):
 208         logger.info('SignalStruct(%r)', test)
 209
 210     @dbus.service.signal(IFACE, signature='as')
 211     def SignalArray(self, test):
 212         pass
 213
 214     @dbus.service.signal(IFACE, signature='a{ss}')
 215     def SignalDict(self, test):
 216         pass
 217
 218     @dbus.service.method(IFACE, in_signature='su', out_signature='')
 219     def EmitSignal(self, signal, value):
 220         sig = getattr(self, str(signal), None)
 221         assert(sig != None)
 222
 223         val = self.returnValue(value)
 224         # make two string case work by passing arguments in by tuple
 225         if (signal == 'SignalTwoStrings' and (value == 1 or value == 5)):
 226             val = tuple(val)
 227         else:
 228             val = tuple([val])
 229
 230         logger.info('Emitting %s with %r', signal, val)
 231         sig(*val)
 232
 233     def CheckInheritance(self):
 234         return True
 235
 236     @dbus.service.method(IFACE, in_signature='', out_signature='b')
 237     def TestListExportedChildObjects(self):
 238         objs_root = session_bus.list_exported_child_objects('/')
 239         assert objs_root == ['org'], objs_root
 240         objs_org = session_bus.list_exported_child_objects('/org')
 241         assert objs_org == ['freedesktop'], objs_org
 242         return True
 243
 244     @dbus.service.method(IFACE, in_signature='bbv', out_signature='v', async_callbacks=('return_cb', 'error_cb'))
 245     def AsynchronousMethod(self, async, fail, variant, return_cb, error_cb):
 246         try:
 247             if async:
 248                 gobject.timeout_add(500, self.AsynchronousMethod, False, fail, variant, return_cb, error_cb)
 249                 return
 250             else:
 251                 if fail:
 252                     raise RuntimeError
 253                 else:
 254                     return_cb(variant)
 255
 256                 return False # do not run again
 257         except Exception, e:
 258             error_cb(e)
 259
 260     @dbus.service.method(IFACE, in_signature='', out_signature='s', sender_keyword='sender')
 261     def WhoAmI(self, sender):
 262         return sender
 263
 264     @dbus.service.method(IFACE, in_signature='', out_signature='b')
 265     def AddRemovableObject(self):
 266         # Part of test for https://bugs.freedesktop.org/show_bug.cgi?id=10457
 267         # Keep the removable object reffed, since that's the use case for this
 268         self._removable.add_to_connection(self._connection,
 269                                           OBJECT + '/RemovableObject')
 270         return True
 271
 272     @dbus.service.method(IFACE, in_signature='', out_signature='b')
 273     def HasRemovableObject(self):
 274         # Part of test for https://bugs.freedesktop.org/show_bug.cgi?id=10457
 275         objs = session_bus.list_exported_child_objects(OBJECT)
 276         return ('RemovableObject' in objs)
 277
 278     @dbus.service.method(IFACE)
 279     def MultipleReturnWithoutSignature(self):
 280         # https://bugs.freedesktop.org/show_bug.cgi?id=10174
 281         return dbus.String('abc'), dbus.Int32(123)
 282
 283     @dbus.service.method(IFACE, in_signature='', out_signature='')
 284     def BlockFor500ms(self):
 285         sleep(0.5)
 286
 287     @dbus.service.method(IFACE, in_signature='', out_signature='',
 288                          async_callbacks=('return_cb', 'raise_cb'))
 289     def AsyncWait500ms(self, return_cb, raise_cb):
 290         def return_from_async_wait():
 291             return_cb()
 292             return False
 293         gobject.timeout_add(500, return_from_async_wait)
 294
 295 session_bus = dbus.SessionBus()
 296 global_name = dbus.service.BusName(NAME, bus=session_bus)
 297 object = TestObject(global_name)
 298 g_object = TestGObject(global_name)
 299 fallback_object = Fallback(session_bus)
 300 loop = gobject.MainLoop()
 301 loop.run()