3 # Copyright (C) 2004 Red Hat Inc. <http://www.redhat.com/>
4 # Copyright (C) 2005-2007 Collabora Ltd. <http://www.collabora.co.uk/>
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:
14 # The above copyright notice and this permission notice shall be
15 # included in all copies or substantial portions of the Software.
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.
29 from time import sleep
31 builddir = os.path.normpath(os.environ["DBUS_TOP_BUILDDIR"])
32 pydir = os.path.normpath(os.environ["DBUS_TOP_SRCDIR"])
36 if not dbus.__file__.startswith(pydir):
37 raise Exception("DBus modules are not being picked up from the package")
44 from dbus.gobject_service import ExportedGObject
47 logging.basicConfig(filename=builddir + '/test/test-service.log', filemode='w')
48 logging.getLogger().setLevel(1)
49 logger = logging.getLogger('test-service')
52 NAME = "org.freedesktop.DBus.TestSuitePythonService"
53 IFACE = "org.freedesktop.DBus.TestSuiteInterface"
54 OBJECT = "/org/freedesktop/DBus/TestSuitePythonObject"
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')
62 @dbus.service.method(IFACE, in_signature='', out_signature='b')
64 self.remove_from_connection()
67 class TestGObject(ExportedGObject):
68 def __init__(self, bus_name, object_path=OBJECT + '/GObject'):
69 super(TestGObject, self).__init__(bus_name, object_path)
71 @dbus.service.method(IFACE)
75 class TestInterface(dbus.service.Interface):
76 @dbus.service.method(IFACE, in_signature='', out_signature='b')
77 def CheckInheritance(self):
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')
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()
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)
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)
100 @dbus.service.method(IFACE, in_signature='su', out_signature='',
102 def EmitSignal(self, signal, value, path=None):
103 sig = getattr(self, str(signal), None)
104 assert sig is not None
106 assert path.startswith(OBJECT + '/Fallback')
107 rel_path = path[len(OBJECT + '/Fallback'):]
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)
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)
119 class MultiPathObject(dbus.service.Object):
120 SUPPORTS_MULTIPLE_OBJECT_PATHS = True
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')
132 """ Echo whatever is sent
134 @dbus.service.method(IFACE)
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__)
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__)
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,
153 return (sender, path, dest,
154 msg.__class__.__module__ + '.' + msg.__class__.__name__)
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__)
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__)
166 @dbus.service.method(IFACE)
167 def GetComplexArray(self):
169 for i in range(0,100):
170 ret.append((random.randint(0,100), random.randint(0,100), str(random.randint(0,100))))
172 return dbus.Array(ret, signature="(uus)")
174 def returnValue(self, test):
190 @dbus.service.method(IFACE, in_signature='u', out_signature='s')
191 def ReturnOneString(self, test):
192 return self.returnValue(test)
194 @dbus.service.method(IFACE, in_signature='u', out_signature='ss')
195 def ReturnTwoStrings(self, test):
196 return self.returnValue(test)
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)
203 @dbus.service.method(IFACE, in_signature='u', out_signature='as')
204 def ReturnArray(self, test):
205 return self.returnValue(test)
207 @dbus.service.method(IFACE, in_signature='u', out_signature='a{ss}')
208 def ReturnDict(self, test):
209 return self.returnValue(test)
211 @dbus.service.signal(IFACE, signature='s')
212 def SignalOneString(self, test):
213 logger.info('SignalOneString(%r)', test)
215 @dbus.service.signal(IFACE, signature='ss')
216 def SignalTwoStrings(self, test, test2):
217 logger.info('SignalTwoStrings(%r, %r)', test, test2)
219 @dbus.service.signal(IFACE, signature='(ss)')
220 def SignalStruct(self, test):
221 logger.info('SignalStruct(%r)', test)
223 @dbus.service.signal(IFACE, signature='as')
224 def SignalArray(self, test):
227 @dbus.service.signal(IFACE, signature='a{ss}')
228 def SignalDict(self, test):
231 @dbus.service.method(IFACE, in_signature='su', out_signature='')
232 def EmitSignal(self, signal, value):
233 sig = getattr(self, str(signal), None)
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)):
243 logger.info('Emitting %s with %r', signal, val)
246 def CheckInheritance(self):
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
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):
261 gobject.timeout_add(500, self.AsynchronousMethod, False, fail, variant, return_cb, error_cb)
269 return False # do not run again
273 @dbus.service.method(IFACE, in_signature='', out_signature='s', sender_keyword='sender')
274 def WhoAmI(self, sender):
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')
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)
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)
296 @dbus.service.method(IFACE, in_signature='', out_signature='')
297 def BlockFor500ms(self):
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():
306 gobject.timeout_add(500, return_from_async_wait)
308 @dbus.service.method(IFACE, in_signature='', out_signature='')
309 def RaiseValueError(self):
310 raise ValueError('Wrong!')
312 @dbus.service.method(IFACE, in_signature='', out_signature='')
313 def RaiseDBusExceptionNoTraceback(self):
314 class ServerError(dbus.DBusException):
315 """Exception representing a normal "environmental" error"""
316 include_traceback = False
317 _dbus_error_name = 'com.example.Networking.ServerError'
319 raise ServerError('Server not responding')
321 @dbus.service.method(IFACE, in_signature='', out_signature='')
322 def RaiseDBusExceptionWithTraceback(self):
323 class RealityFailure(dbus.DBusException):
324 """Exception representing a programming error"""
325 include_traceback = True
326 _dbus_error_name = 'com.example.Misc.RealityFailure'
328 raise RealityFailure('Botched invariant')
330 @dbus.service.method(IFACE, in_signature='', out_signature='',
331 async_callbacks=('return_cb', 'raise_cb'))
332 def AsyncRaise(self, return_cb, raise_cb):
333 class Fdo12403Error(dbus.DBusException):
334 _dbus_error_name = 'org.freedesktop.bugzilla.bug12403'
336 raise_cb(Fdo12403Error())
338 session_bus = dbus.SessionBus()
339 global_name = dbus.service.BusName(NAME, bus=session_bus)
340 object = TestObject(global_name)
341 g_object = TestGObject(global_name)
342 fallback_object = Fallback(session_bus)
343 loop = gobject.MainLoop()