Don't try to distribute COPYING.AFL-2.1 and COPYING.GPL-2, which we no longer have...
[dbus-python-phuang.git] / test / test-service.py
blob74829d45d244b79ff4603f8f7f5b0d75682bfa6f
1 #!/usr/bin/env python
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.
26 import sys
27 import os
28 import logging
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"])
34 import dbus
36 if not dbus.__file__.startswith(pydir):
37 raise Exception("DBus modules are not being picked up from the package")
39 import dbus.service
40 import dbus.glib
41 import gobject
42 import random
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')
59 def IsThere(self):
60 return True
62 @dbus.service.method(IFACE, in_signature='', out_signature='b')
63 def RemoveSelf(self):
64 self.remove_from_connection()
65 return True
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)
72 def Echo(self, arg):
73 return arg
75 class TestInterface(dbus.service.Interface):
76 @dbus.service.method(IFACE, in_signature='', out_signature='b')
77 def CheckInheritance(self):
78 return False
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)
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)
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
106 assert path.startswith(OBJECT + '/Fallback')
107 rel_path = path[len(OBJECT + '/Fallback'):]
108 if rel_path == '':
109 rel_path = '/'
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)
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)
135 def Echo(self, arg):
136 return arg
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
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
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__)
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
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
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))))
172 return dbus.Array(ret, signature="(uus)")
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 ["","",""]
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):
225 pass
227 @dbus.service.signal(IFACE, signature='a{ss}')
228 def SignalDict(self, test):
229 pass
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)
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])
243 logger.info('Emitting %s with %r', signal, val)
244 sig(*val)
246 def CheckInheritance(self):
247 return True
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
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)
269 return False # do not run again
270 except Exception, e:
271 error_cb(e)
273 @dbus.service.method(IFACE, in_signature='', out_signature='s', sender_keyword='sender')
274 def WhoAmI(self, sender):
275 return 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')
283 return True
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):
298 sleep(0.5)
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)
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'
314 raise_cb(Fdo12403Error())
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()