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.
33 builddir
= os
.path
.normpath(os
.environ
["DBUS_TOP_BUILDDIR"])
34 pydir
= os
.path
.normpath(os
.environ
["DBUS_TOP_SRCDIR"])
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"]],
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
):
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')
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
)
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
)
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
):
119 print "\n********* Introspection Test ************"
120 print self
.remote_object
.Introspect(dbus_interface
="org.freedesktop.DBus.Introspectable")
121 print "Introspection test passed"
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")
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
):
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 ************"
169 print self
.iface
.GetComplexArray()
172 print "Delta: %f" % (b
- a
)
175 def testAsyncCalls(self
):
176 #test sending python types and getting them back async
177 print "\n********* Testing Async Calls ***********"
180 main_loop
= gobject
.MainLoop()
183 def __init__(self
, test_controler
, expected_result
, do_exit
, utf8
):
184 self
.expected_result
= expected_result
185 self
.do_exit
= do_exit
187 self
.test_controler
= test_controler
189 def callback(self
, val
):
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
)
199 elif not self
.utf8
and isinstance(val
, dbus
.UTF8String
):
200 failures
.append('%r should not have been utf8' % val
)
203 failures
.append("%s:\n%s" % (e
.__class
__, e
))
205 def error_handler(self
, error
):
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
,
218 recv_val
= self
.iface
.Echo(send_val
,
219 reply_handler
=check
.callback
,
220 error_handler
=check
.error_handler
,
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
= ["", ("",""), ("","",""), [], {}, ["",""], ["","",""]]
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
)):
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
))
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
)):
267 self
.iface
.EmitSignal(signal
, value
)
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
))
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
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):
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
)
298 self
.assert_(fail
, '%s: %s' % (e
.__class
__, e
))
299 print "Expected failure: %s: %s" % (e
.__class
__, e
)
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
__
314 share_type
= dbus
.Bus(bus_type
=type)
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()
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")
370 for (name
, succeed
) in test
:
372 print "requesting %s" % name
373 busname
= dbus
.service
.BusName(name
, dbus
.SessionBus())
375 print "%s:\n%s" % (e
.__class
__, e
)
376 self
.assert_(not succeed
, 'did not expect registering bus name %s to fail' % name
)
379 self
.assert_(succeed
, 'expected registering bus name %s to fail'% name
)
381 self
.assert_(names
[name
] == busname
, 'got a new instance for same name %s' % name
)
382 print "instance of %s re-used, good!" % name
384 names
[name
] = busname
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())
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
)
464 self
.iface
.AsyncRaise()
465 except dbus
.DBusException
, e
:
466 self
.assert_(e
.get_dbus_name() ==
467 'org.freedesktop.bugzilla.bug12403',
472 def testClosePrivateBus(self
):
474 dbus
.Bus(private
=True).close()
476 def testTimeoutAsyncClient(self
):
477 loop
= gobject
.MainLoop()
480 def correctly_returned():
481 passes
.append('1000')
482 if len(passes
) + len(fails
) >= 2:
484 def correctly_failed(exc
):
486 if len(passes
) + len(fails
) >= 2:
488 def incorrectly_returned():
490 if len(passes
) + len(fails
) >= 2:
492 def incorrectly_failed(exc
):
494 if len(passes
) + len(fails
) >= 2:
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
)
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):
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):
520 print "\n********* Introspection Test ************"
521 print self.remote_object.Introspect(dbus_interface="org.freedesktop.DBus.Introspectable")
522 print "Introspection test passed"
526 print "\n********* Call Test ************"
527 result = self.iface.ManyArgs(1000, 'Hello GLib', 2)
529 self.assert_(result == [2002.0, 'HELLO GLIB'])
531 arg0 = {"Dude": 1, "john": "palmieri", "python": 2.4}
532 result = self.iface.ManyStringify(arg0)
535 print "Call test passed"
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()