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