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)
202
except Exception as e:
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())
374
except Exception as e:
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 as 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
def testExceptions(self):
512
#self.assertRaises(dbus.DBusException,
513
# lambda: self.iface.RaiseValueError)
514
#self.assertRaises(dbus.DBusException,
515
# lambda: self.iface.RaiseDBusExceptionNoTraceback)
516
#self.assertRaises(dbus.DBusException,
517
# lambda: self.iface.RaiseDBusExceptionWithTraceback)
520
self.iface.RaiseValueError()
521
except Exception as e:
522
self.assert_(isinstance(e, dbus.DBusException), e.__class__)
523
self.assert_('.ValueError: Traceback ' in str(e),
524
'Wanted a traceback but got:\n"""%s"""' % str(e))
526
raise AssertionError('Wanted an exception')
529
self.iface.RaiseDBusExceptionNoTraceback()
530
except Exception as e:
531
self.assert_(isinstance(e, dbus.DBusException), e.__class__)
532
self.assertEquals(str(e),
533
'com.example.Networking.ServerError: '
534
'Server not responding')
536
raise AssertionError('Wanted an exception')
539
self.iface.RaiseDBusExceptionWithTraceback()
540
except Exception as e:
541
self.assert_(isinstance(e, dbus.DBusException), e.__class__)
542
self.assert_(str(e).startswith('com.example.Misc.RealityFailure: '
544
'Wanted a traceback but got:\n%s' % str(e))
546
raise AssertionError('Wanted an exception')
548
""" Remove this for now
549
class TestDBusPythonToGLibBindings(unittest.TestCase):
551
self.bus = dbus.SessionBus()
552
self.remote_object = self.bus.get_object("org.freedesktop.DBus.TestSuiteGLibService", "/org/freedesktop/DBus/Tests/MyTestObject")
553
self.iface = dbus.Interface(self.remote_object, "org.freedesktop.DBus.Tests.MyObject")
555
def testIntrospection(self):
557
print "\n********* Introspection Test ************"
558
print self.remote_object.Introspect(dbus_interface="org.freedesktop.DBus.Introspectable")
559
print "Introspection test passed"
563
print "\n********* Call Test ************"
564
result = self.iface.ManyArgs(1000, 'Hello GLib', 2)
566
self.assert_(result == [2002.0, 'HELLO GLIB'])
568
arg0 = {"Dude": 1, "john": "palmieri", "python": 2.4}
569
result = self.iface.ManyStringify(arg0)
572
print "Call test passed"
575
def testPythonTypes(self):
576
print "\n********* Testing Python Types ***********"
578
for send_val in test_types_vals:
579
print "Testing %s"% str(send_val)
580
recv_val = self.iface.EchoVariant(send_val)
581
self.assertEquals(send_val, recv_val.object)
583
if __name__ == '__main__':
584
gobject.threads_init()
585
dbus.glib.init_threads()
added
removed
.pc/python3-support.patch/test/test-client.py
