Fix NEWS indentation
[dbus-python-phuang.git] / test / test-client.py
blob0d95d72209c16017d56e64c1ae796c072126dc64
1 #!/usr/bin/env python
3 # Copyright (C) 2004 Red Hat Inc. <http://www.redhat.com/>
4 # Copyright (C) 2005, 2006 Collabora Ltd. <http://www.collabora.co.uk/>
6 # Licensed under the Academic Free License version 2.1
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.
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.
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
23 import sys
24 import os
25 import unittest
26 import time
27 import logging
28 import weakref
30 builddir = os.path.normpath(os.environ["DBUS_TOP_BUILDDIR"])
31 pydir = os.path.normpath(os.environ["DBUS_TOP_SRCDIR"])
33 import dbus
34 import _dbus_bindings
35 import gobject
36 import dbus.glib
37 import dbus.service
40 logging.basicConfig()
43 pkg = dbus.__file__
44 if not pkg.startswith(pydir):
45 raise Exception("DBus modules (%s) are not being picked up from the package"%pkg)
47 if not _dbus_bindings.__file__.startswith(builddir):
48 raise Exception("DBus modules (%s) are not being picked up from the package"%_dbus_bindings.__file__)
50 test_types_vals = [1, 12323231, 3.14159265, 99999999.99,
51 "dude", "123", "What is all the fuss about?", "gob@gob.com",
52 u'\\u310c\\u310e\\u3114', u'\\u0413\\u0414\\u0415',
53 u'\\u2200software \\u2203crack', u'\\xf4\\xe5\\xe8',
54 [1,2,3], ["how", "are", "you"], [1.23,2.3], [1], ["Hello"],
55 (1,2,3), (1,), (1,"2",3), ("2", "what"), ("you", 1.2),
56 {1:"a", 2:"b"}, {"a":1, "b":2}, #{"a":(1,"B")},
57 {1:1.1, 2:2.2}, [[1,2,3],[2,3,4]], [["a","b"],["c","d"]],
58 True, False,
59 dbus.Int16(-10), dbus.UInt16(10), 'SENTINEL',
60 #([1,2,3],"c", 1.2, ["a","b","c"], {"a": (1,"v"), "b": (2,"d")})
63 NAME = "org.freedesktop.DBus.TestSuitePythonService"
64 IFACE = "org.freedesktop.DBus.TestSuiteInterface"
65 OBJECT = "/org/freedesktop/DBus/TestSuitePythonObject"
67 class TestDBusBindings(unittest.TestCase):
68 def setUp(self):
69 self.bus = dbus.SessionBus()
70 self.remote_object = self.bus.get_object(NAME, OBJECT)
71 self.remote_object_follow = self.bus.get_object(NAME, OBJECT,
72 follow_name_owner_changes=True)
73 self.iface = dbus.Interface(self.remote_object, IFACE)
75 def testGObject(self):
76 print "Testing ExportedGObject... ",
77 remote_gobject = self.bus.get_object(NAME, OBJECT + '/GObject')
78 iface = dbus.Interface(remote_gobject, IFACE)
79 print "introspection, ",
80 remote_gobject.Introspect(dbus_interface=dbus.INTROSPECTABLE_IFACE)
81 print "method call, ",
82 self.assertEquals(iface.Echo('123'), '123')
83 print "... OK"
85 def testWeakRefs(self):
86 # regression test for Sugar crash caused by smcv getting weak refs
87 # wrong - pre-bugfix, this would segfault
88 bus = dbus.SessionBus(private=True)
89 ref = weakref.ref(bus)
90 self.assert_(ref() is bus)
91 del bus
92 self.assert_(ref() is None)
94 def testInterfaceKeyword(self):
95 #test dbus_interface parameter
96 print self.remote_object.Echo("dbus_interface on Proxy test Passed", dbus_interface = IFACE)
97 print self.iface.Echo("dbus_interface on Interface test Passed", dbus_interface = IFACE)
98 self.assert_(True)
100 def testGetDBusMethod(self):
101 self.assertEquals(self.iface.get_dbus_method('AcceptListOfByte')('\1\2\3'), [1,2,3])
102 self.assertEquals(self.remote_object.get_dbus_method('AcceptListOfByte', dbus_interface=IFACE)('\1\2\3'), [1,2,3])
104 def testCallingConventionOptions(self):
105 self.assertEquals(self.iface.AcceptListOfByte('\1\2\3'), [1,2,3])
106 self.assertEquals(self.iface.AcceptListOfByte('\1\2\3', byte_arrays=True), '\1\2\3')
107 self.assertEquals(self.iface.AcceptByteArray('\1\2\3'), [1,2,3])
108 self.assertEquals(self.iface.AcceptByteArray('\1\2\3', byte_arrays=True), '\1\2\3')
109 self.assert_(isinstance(self.iface.AcceptUTF8String('abc'), unicode))
110 self.assert_(isinstance(self.iface.AcceptUTF8String('abc', utf8_strings=True), str))
111 self.assert_(isinstance(self.iface.AcceptUnicodeString('abc'), unicode))
112 self.assert_(isinstance(self.iface.AcceptUnicodeString('abc', utf8_strings=True), str))
114 def testIntrospection(self):
115 #test introspection
116 print "\n********* Introspection Test ************"
117 print self.remote_object.Introspect(dbus_interface="org.freedesktop.DBus.Introspectable")
118 print "Introspection test passed"
119 self.assert_(True)
121 def testMultiPathIntrospection(self):
122 # test introspection on an object exported in multiple places
123 # https://bugs.freedesktop.org/show_bug.cgi?id=11794
124 remote_object = self.bus.get_object(NAME, OBJECT + '/Multi1')
125 remote_object.Introspect(dbus_interface="org.freedesktop.DBus.Introspectable")
126 remote_object = self.bus.get_object(NAME, OBJECT + '/Multi2')
127 remote_object.Introspect(dbus_interface="org.freedesktop.DBus.Introspectable")
128 remote_object = self.bus.get_object(NAME, OBJECT + '/Multi2/3')
129 remote_object.Introspect(dbus_interface="org.freedesktop.DBus.Introspectable")
130 self.assert_(True)
132 def testPythonTypes(self):
133 #test sending python types and getting them back
134 print "\n********* Testing Python Types ***********"
136 for send_val in test_types_vals:
137 print "Testing %s"% str(send_val)
138 recv_val = self.iface.Echo(send_val)
139 self.assertEquals(send_val, recv_val)
140 self.assertEquals(recv_val.variant_level, 1)
142 def testMethodExtraInfoKeywords(self):
143 print "Testing MethodExtraInfoKeywords..."
144 sender, path, destination, message_cls = self.iface.MethodExtraInfoKeywords()
145 self.assert_(sender.startswith(':'))
146 self.assertEquals(path, '/org/freedesktop/DBus/TestSuitePythonObject')
147 # we're using the "early binding" form of get_object (without
148 # follow_name_owner_changes), so the destination we actually sent it
149 # to will be the unique name
150 self.assert_(destination.startswith(':'))
151 self.assertEquals(message_cls, 'dbus.lowlevel.MethodCallMessage')
153 def testUtf8StringsSync(self):
154 send_val = u'foo'
155 recv_val = self.iface.Echo(send_val, utf8_strings=True)
156 self.assert_(isinstance(recv_val, str))
157 self.assert_(isinstance(recv_val, dbus.UTF8String))
158 recv_val = self.iface.Echo(send_val, utf8_strings=False)
159 self.assert_(isinstance(recv_val, unicode))
160 self.assert_(isinstance(recv_val, dbus.String))
162 def testBenchmarkIntrospect(self):
163 print "\n********* Benchmark Introspect ************"
164 a = time.time()
165 print a
166 print self.iface.GetComplexArray()
167 b = time.time()
168 print b
169 print "Delta: %f" % (b - a)
170 self.assert_(True)
172 def testAsyncCalls(self):
173 #test sending python types and getting them back async
174 print "\n********* Testing Async Calls ***********"
176 failures = []
177 main_loop = gobject.MainLoop()
179 class async_check:
180 def __init__(self, test_controler, expected_result, do_exit, utf8):
181 self.expected_result = expected_result
182 self.do_exit = do_exit
183 self.utf8 = utf8
184 self.test_controler = test_controler
186 def callback(self, val):
187 try:
188 if self.do_exit:
189 main_loop.quit()
191 self.test_controler.assertEquals(val, self.expected_result)
192 self.test_controler.assertEquals(val.variant_level, 1)
193 if self.utf8 and not isinstance(val, dbus.UTF8String):
194 failures.append('%r should have been utf8 but was not' % val)
195 return
196 elif not self.utf8 and isinstance(val, dbus.UTF8String):
197 failures.append('%r should not have been utf8' % val)
198 return
199 except Exception, e:
200 failures.append("%s:\n%s" % (e.__class__, e))
202 def error_handler(self, error):
203 print error
204 if self.do_exit:
205 main_loop.quit()
207 failures.append('%s: %s' % (error.__class__, error))
209 last_type = test_types_vals[-1]
210 for send_val in test_types_vals:
211 print "Testing %s" % str(send_val)
212 utf8 = (send_val == 'gob@gob.com')
213 check = async_check(self, send_val, last_type == send_val,
214 utf8)
215 recv_val = self.iface.Echo(send_val,
216 reply_handler=check.callback,
217 error_handler=check.error_handler,
218 utf8_strings=utf8)
219 main_loop.run()
220 if failures:
221 self.assert_(False, failures)
223 def testStrictMarshalling(self):
224 print "\n********* Testing strict return & signal marshalling ***********"
226 # these values are the same as in the server, and the
227 # methods should only succeed when they are called with
228 # the right value number, because they have out_signature
229 # decorations, and return an unmatching type when called
230 # with a different number
231 values = ["", ("",""), ("","",""), [], {}, ["",""], ["","",""]]
232 methods = [
233 (self.iface.ReturnOneString, 'SignalOneString', set([0]), set([0])),
234 (self.iface.ReturnTwoStrings, 'SignalTwoStrings', set([1, 5]), set([1])),
235 (self.iface.ReturnStruct, 'SignalStruct', set([1, 5]), set([1])),
236 # all of our test values are sequences so will marshall correctly into an array :P
237 (self.iface.ReturnArray, 'SignalArray', set(range(len(values))), set([3, 5, 6])),
238 (self.iface.ReturnDict, 'SignalDict', set([0, 3, 4]), set([4]))
241 for (method, signal, success_values, return_values) in methods:
242 print "\nTrying correct behaviour of", method._method_name
243 for value in range(len(values)):
244 try:
245 ret = method(value)
246 except Exception, e:
247 print "%s(%r) raised %s: %s" % (method._method_name, values[value], e.__class__, e)
249 # should fail if it tried to marshal the wrong type
250 self.assert_(value not in success_values, "%s should succeed when we ask it to return %r\n%s\n%s" % (method._method_name, values[value], e.__class__, e))
251 else:
252 print "%s(%r) returned %r" % (method._method_name, values[value], ret)
254 # should only succeed if it's the right return type
255 self.assert_(value in success_values, "%s should fail when we ask it to return %r" % (method._method_name, values[value]))
257 # check the value is right too :D
258 returns = map(lambda n: values[n], return_values)
259 self.assert_(ret in returns, "%s should return one of %r but it returned %r instead" % (method._method_name, returns, ret))
261 print "\nTrying correct emission of", signal
262 for value in range(len(values)):
263 try:
264 self.iface.EmitSignal(signal, value)
265 except Exception, e:
266 print "EmitSignal(%s, %r) raised %s" % (signal, values[value], e.__class__)
268 # should fail if it tried to marshal the wrong type
269 self.assert_(value not in success_values, "EmitSignal(%s) should succeed when we ask it to return %r\n%s\n%s" % (signal, values[value], e.__class__, e))
270 else:
271 print "EmitSignal(%s, %r) appeared to succeed" % (signal, values[value])
273 # should only succeed if it's the right return type
274 self.assert_(value in success_values, "EmitSignal(%s) should fail when we ask it to return %r" % (signal, values[value]))
276 # FIXME: wait for the signal here
278 print
280 def testInheritance(self):
281 print "\n********* Testing inheritance from dbus.method.Interface ***********"
282 ret = self.iface.CheckInheritance()
283 print "CheckInheritance returned %s" % ret
284 self.assert_(ret, "overriding CheckInheritance from TestInterface failed")
286 def testAsyncMethods(self):
287 print "\n********* Testing asynchronous method implementation *******"
288 for async in (True, False):
289 for fail in (True, False):
290 try:
291 val = ('a', 1, False, [1,2], {1:2})
292 print "calling AsynchronousMethod with %s %s %s" % (async, fail, val)
293 ret = self.iface.AsynchronousMethod(async, fail, val)
294 except Exception, e:
295 self.assert_(fail, '%s: %s' % (e.__class__, e))
296 print "Expected failure: %s: %s" % (e.__class__, e)
297 else:
298 self.assert_(not fail, 'Expected failure but succeeded?!')
299 self.assertEquals(val, ret)
300 self.assertEquals(1, ret.variant_level)
302 def testBusInstanceCaching(self):
303 print "\n********* Testing dbus.Bus instance sharing *********"
305 # unfortunately we can't test the system bus here
306 # but the codepaths are the same
307 for (cls, type, func) in ((dbus.SessionBus, dbus.Bus.TYPE_SESSION, dbus.Bus.get_session), (dbus.StarterBus, dbus.Bus.TYPE_STARTER, dbus.Bus.get_starter)):
308 print "\nTesting %s:" % cls.__name__
310 share_cls = cls()
311 share_type = dbus.Bus(bus_type=type)
312 share_func = func()
314 private_cls = cls(private=True)
315 private_type = dbus.Bus(bus_type=type, private=True)
316 private_func = func(private=True)
318 print " - checking shared instances are the same..."
319 self.assert_(share_cls == share_type, '%s should equal %s' % (share_cls, share_type))
320 self.assert_(share_type == share_func, '%s should equal %s' % (share_type, share_func))
322 print " - checking private instances are distinct from the shared instance..."
323 self.assert_(share_cls != private_cls, '%s should not equal %s' % (share_cls, private_cls))
324 self.assert_(share_type != private_type, '%s should not equal %s' % (share_type, private_type))
325 self.assert_(share_func != private_func, '%s should not equal %s' % (share_func, private_func))
327 print " - checking private instances are distinct from each other..."
328 self.assert_(private_cls != private_type, '%s should not equal %s' % (private_cls, private_type))
329 self.assert_(private_type != private_func, '%s should not equal %s' % (private_type, private_func))
330 self.assert_(private_func != private_cls, '%s should not equal %s' % (private_func, private_cls))
332 def testSenderName(self):
333 print '\n******** Testing sender name keyword ********'
334 myself = self.iface.WhoAmI()
335 print "I am", myself
337 def testBusGetNameOwner(self):
338 ret = self.bus.get_name_owner(NAME)
339 self.assert_(ret.startswith(':'), ret)
341 def testBusListNames(self):
342 ret = self.bus.list_names()
343 self.assert_(NAME in ret, ret)
345 def testBusListActivatableNames(self):
346 ret = self.bus.list_activatable_names()
347 self.assert_(NAME in ret, ret)
349 def testBusNameHasOwner(self):
350 self.assert_(self.bus.name_has_owner(NAME))
351 self.assert_(not self.bus.name_has_owner('badger.mushroom.snake'))
353 def testBusNameCreation(self):
354 print '\n******** Testing BusName creation ********'
355 test = [('org.freedesktop.DBus.Python.TestName', True),
356 ('org.freedesktop.DBus.Python.TestName', True),
357 ('org.freedesktop.DBus.Python.InvalidName&^*%$', False)]
358 # Do some more intelligent handling/testing of queueing vs success?
359 # ('org.freedesktop.DBus.TestSuitePythonService', False)]
360 # For some reason this actually succeeds
361 # ('org.freedesktop.DBus', False)]
363 # make a method call to ensure the test service is active
364 self.iface.Echo("foo")
366 names = {}
367 for (name, succeed) in test:
368 try:
369 print "requesting %s" % name
370 busname = dbus.service.BusName(name, dbus.SessionBus())
371 except Exception, e:
372 print "%s:\n%s" % (e.__class__, e)
373 self.assert_(not succeed, 'did not expect registering bus name %s to fail' % name)
374 else:
375 print busname
376 self.assert_(succeed, 'expected registering bus name %s to fail'% name)
377 if name in names:
378 self.assert_(names[name] == busname, 'got a new instance for same name %s' % name)
379 print "instance of %s re-used, good!" % name
380 else:
381 names[name] = busname
383 del busname
385 print
387 del names
389 bus = dbus.Bus()
390 ret = bus.name_has_owner('org.freedesktop.DBus.Python.TestName')
391 self.assert_(not ret, 'deleting reference failed to release BusName org.freedesktop.DBus.Python.TestName')
393 def testMultipleReturnWithoutSignature(self):
394 # https://bugs.freedesktop.org/show_bug.cgi?id=10174
395 ret = self.iface.MultipleReturnWithoutSignature()
396 self.assert_(not isinstance(ret, dbus.Struct), repr(ret))
397 self.assertEquals(ret, ('abc', 123))
399 def testListExportedChildObjects(self):
400 self.assert_(self.iface.TestListExportedChildObjects())
402 def testRemoveFromConnection(self):
403 # https://bugs.freedesktop.org/show_bug.cgi?id=10457
404 self.assert_(not self.iface.HasRemovableObject())
405 self.assert_(self.iface.AddRemovableObject())
406 self.assert_(self.iface.HasRemovableObject())
408 removable = self.bus.get_object(NAME, OBJECT + '/RemovableObject')
409 iface = dbus.Interface(removable, IFACE)
410 self.assert_(iface.IsThere())
411 self.assert_(iface.RemoveSelf())
413 self.assert_(not self.iface.HasRemovableObject())
415 # and again...
416 self.assert_(self.iface.AddRemovableObject())
417 self.assert_(self.iface.HasRemovableObject())
418 self.assert_(iface.IsThere())
419 self.assert_(iface.RemoveSelf())
420 self.assert_(not self.iface.HasRemovableObject())
422 def testFallbackObjectTrivial(self):
423 obj = self.bus.get_object(NAME, OBJECT + '/Fallback')
424 iface = dbus.Interface(obj, IFACE)
425 path, rel, unique_name = iface.TestPathAndConnKeywords()
426 self.assertEquals(path, OBJECT + '/Fallback')
427 self.assertEquals(rel, '/')
428 self.assertEquals(unique_name, obj.bus_name)
430 def testFallbackObjectNested(self):
431 obj = self.bus.get_object(NAME, OBJECT + '/Fallback/Nested')
432 iface = dbus.Interface(obj, IFACE)
433 path, rel, unique_name = iface.TestPathAndConnKeywords()
434 self.assertEquals(path, OBJECT + '/Fallback/Nested')
435 self.assertEquals(rel, '/')
436 self.assertEquals(unique_name, obj.bus_name)
438 obj = self.bus.get_object(NAME, OBJECT + '/Fallback/Nested/Badger/Mushroom')
439 iface = dbus.Interface(obj, IFACE)
440 path, rel, unique_name = iface.TestPathAndConnKeywords()
441 self.assertEquals(path, OBJECT + '/Fallback/Nested/Badger/Mushroom')
442 self.assertEquals(rel, '/Badger/Mushroom')
443 self.assertEquals(unique_name, obj.bus_name)
445 def testFallbackObject(self):
446 obj = self.bus.get_object(NAME, OBJECT + '/Fallback/Badger/Mushroom')
447 iface = dbus.Interface(obj, IFACE)
448 path, rel, unique_name = iface.TestPathAndConnKeywords()
449 self.assertEquals(path, OBJECT + '/Fallback/Badger/Mushroom')
450 self.assertEquals(rel, '/Badger/Mushroom')
451 self.assertEquals(unique_name, obj.bus_name)
453 def testTimeoutSync(self):
454 self.assert_(self.iface.BlockFor500ms(timeout=1.0) is None)
455 self.assertRaises(dbus.DBusException,
456 lambda: self.iface.BlockFor500ms(timeout=0.25))
458 def testAsyncRaise(self):
459 self.assertRaises(dbus.DBusException, self.iface.AsyncRaise)
460 try:
461 self.iface.AsyncRaise()
462 except dbus.DBusException, e:
463 self.assert_(e.get_dbus_name() ==
464 'org.freedesktop.bugzilla.bug12403',
465 e.get_dbus_name())
466 else:
467 self.assert_(False)
469 def testClosePrivateBus(self):
470 # fd.o #12096
471 dbus.Bus(private=True).close()
473 def testTimeoutAsyncClient(self):
474 loop = gobject.MainLoop()
475 passes = []
476 fails = []
477 def correctly_returned():
478 passes.append('1000')
479 if len(passes) + len(fails) >= 2:
480 loop.quit()
481 def correctly_failed(exc):
482 passes.append('250')
483 if len(passes) + len(fails) >= 2:
484 loop.quit()
485 def incorrectly_returned():
486 fails.append('250')
487 if len(passes) + len(fails) >= 2:
488 loop.quit()
489 def incorrectly_failed(exc):
490 fails.append('1000')
491 if len(passes) + len(fails) >= 2:
492 loop.quit()
493 self.iface.BlockFor500ms(timeout=1.0,
494 reply_handler=correctly_returned,
495 error_handler=incorrectly_failed)
496 self.iface.BlockFor500ms(timeout=0.25,
497 reply_handler=incorrectly_returned,
498 error_handler=correctly_failed)
499 loop.run()
500 self.assertEquals(passes, ['250', '1000'])
501 self.assertEquals(fails, [])
503 def testTimeoutAsyncService(self):
504 self.assert_(self.iface.AsyncWait500ms(timeout=1.0) is None)
505 self.assertRaises(dbus.DBusException,
506 lambda: self.iface.AsyncWait500ms(timeout=0.25))
508 """ Remove this for now
509 class TestDBusPythonToGLibBindings(unittest.TestCase):
510 def setUp(self):
511 self.bus = dbus.SessionBus()
512 self.remote_object = self.bus.get_object("org.freedesktop.DBus.TestSuiteGLibService", "/org/freedesktop/DBus/Tests/MyTestObject")
513 self.iface = dbus.Interface(self.remote_object, "org.freedesktop.DBus.Tests.MyObject")
515 def testIntrospection(self):
516 #test introspection
517 print "\n********* Introspection Test ************"
518 print self.remote_object.Introspect(dbus_interface="org.freedesktop.DBus.Introspectable")
519 print "Introspection test passed"
520 self.assert_(True)
522 def testCalls(self):
523 print "\n********* Call Test ************"
524 result = self.iface.ManyArgs(1000, 'Hello GLib', 2)
525 print result
526 self.assert_(result == [2002.0, 'HELLO GLIB'])
528 arg0 = {"Dude": 1, "john": "palmieri", "python": 2.4}
529 result = self.iface.ManyStringify(arg0)
530 print result
532 print "Call test passed"
533 self.assert_(True)
535 def testPythonTypes(self):
536 print "\n********* Testing Python Types ***********"
538 for send_val in test_types_vals:
539 print "Testing %s"% str(send_val)
540 recv_val = self.iface.EchoVariant(send_val)
541 self.assertEquals(send_val, recv_val.object)
543 if __name__ == '__main__':
544 gobject.threads_init()
545 dbus.glib.init_threads()
547 unittest.main()