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