test/test-client.py: untabify
[dbus-python-phuang.git] / test / test-client.py
blob49578f7e89641174efd1ec8095266e60729b7114
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 class TestDBusBindings(unittest.TestCase):
64 def setUp(self):
65 self.bus = dbus.SessionBus()
66 self.remote_object = self.bus.get_object("org.freedesktop.DBus.TestSuitePythonService", "/org/freedesktop/DBus/TestSuitePythonObject")
67 self.iface = dbus.Interface(self.remote_object, "org.freedesktop.DBus.TestSuiteInterface")
69 def testWeakRefs(self):
70 # regression test for Sugar crash caused by smcv getting weak refs
71 # wrong - pre-bugfix, this would segfault
72 bus = dbus.SessionBus(private=True)
73 ref = weakref.ref(bus)
74 self.assert_(ref() is bus)
75 del bus
76 self.assert_(ref() is None)
78 def testInterfaceKeyword(self):
79 #test dbus_interface parameter
80 print self.remote_object.Echo("dbus_interface on Proxy test Passed", dbus_interface = "org.freedesktop.DBus.TestSuiteInterface")
81 print self.iface.Echo("dbus_interface on Interface test Passed", dbus_interface = "org.freedesktop.DBus.TestSuiteInterface")
82 self.assert_(True)
84 def testGetDBusMethod(self):
85 self.assertEquals(self.iface.get_dbus_method('AcceptListOfByte')('\1\2\3'), [1,2,3])
86 self.assertEquals(self.remote_object.get_dbus_method('AcceptListOfByte', dbus_interface='org.freedesktop.DBus.TestSuiteInterface')('\1\2\3'), [1,2,3])
88 def testCallingConventionOptions(self):
89 self.assertEquals(self.iface.AcceptListOfByte('\1\2\3'), [1,2,3])
90 self.assertEquals(self.iface.AcceptListOfByte('\1\2\3', byte_arrays=True), '\1\2\3')
91 self.assertEquals(self.iface.AcceptByteArray('\1\2\3'), [1,2,3])
92 self.assertEquals(self.iface.AcceptByteArray('\1\2\3', byte_arrays=True), '\1\2\3')
93 self.assert_(isinstance(self.iface.AcceptUTF8String('abc'), unicode))
94 self.assert_(isinstance(self.iface.AcceptUTF8String('abc', utf8_strings=True), str))
95 self.assert_(isinstance(self.iface.AcceptUnicodeString('abc'), unicode))
96 self.assert_(isinstance(self.iface.AcceptUnicodeString('abc', utf8_strings=True), str))
98 def testIntrospection(self):
99 #test introspection
100 print "\n********* Introspection Test ************"
101 print self.remote_object.Introspect(dbus_interface="org.freedesktop.DBus.Introspectable")
102 print "Introspection test passed"
103 self.assert_(True)
105 def testPythonTypes(self):
106 #test sending python types and getting them back
107 print "\n********* Testing Python Types ***********"
109 for send_val in test_types_vals:
110 print "Testing %s"% str(send_val)
111 recv_val = self.iface.Echo(send_val)
112 self.assertEquals(send_val, recv_val)
113 self.assertEquals(recv_val.variant_level, 1)
115 def testUtf8StringsSync(self):
116 send_val = u'foo'
117 recv_val = self.iface.Echo(send_val, utf8_strings=True)
118 self.assert_(isinstance(recv_val, str))
119 self.assert_(isinstance(recv_val, dbus.UTF8String))
120 recv_val = self.iface.Echo(send_val, utf8_strings=False)
121 self.assert_(isinstance(recv_val, unicode))
122 self.assert_(isinstance(recv_val, dbus.String))
124 def testBenchmarkIntrospect(self):
125 print "\n********* Benchmark Introspect ************"
126 a = time.time()
127 print a
128 print self.iface.GetComplexArray()
129 b = time.time()
130 print b
131 print "Delta: %f" % (b - a)
132 self.assert_(True)
134 def testAsyncCalls(self):
135 #test sending python types and getting them back async
136 print "\n********* Testing Async Calls ***********"
138 failures = []
139 main_loop = gobject.MainLoop()
141 class async_check:
142 def __init__(self, test_controler, expected_result, do_exit, utf8):
143 self.expected_result = expected_result
144 self.do_exit = do_exit
145 self.utf8 = utf8
146 self.test_controler = test_controler
148 def callback(self, val):
149 try:
150 if self.do_exit:
151 main_loop.quit()
153 self.test_controler.assertEquals(val, self.expected_result)
154 self.test_controler.assertEquals(val.variant_level, 1)
155 if self.utf8 and not isinstance(val, dbus.UTF8String):
156 failures.append('%r should have been utf8 but was not' % val)
157 return
158 elif not self.utf8 and isinstance(val, dbus.UTF8String):
159 failures.append('%r should not have been utf8' % val)
160 return
161 except Exception, e:
162 failures.append("%s:\n%s" % (e.__class__, e))
164 def error_handler(self, error):
165 print error
166 if self.do_exit:
167 main_loop.quit()
169 failures.append('%s: %s' % (error.__class__, error))
171 last_type = test_types_vals[-1]
172 for send_val in test_types_vals:
173 print "Testing %s" % str(send_val)
174 utf8 = (send_val == 'gob@gob.com')
175 check = async_check(self, send_val, last_type == send_val,
176 utf8)
177 recv_val = self.iface.Echo(send_val,
178 reply_handler=check.callback,
179 error_handler=check.error_handler,
180 utf8_strings=utf8)
181 main_loop.run()
182 if failures:
183 self.assert_(False, failures)
185 def testStrictMarshalling(self):
186 print "\n********* Testing strict return & signal marshalling ***********"
188 # these values are the same as in the server, and the
189 # methods should only succeed when they are called with
190 # the right value number, because they have out_signature
191 # decorations, and return an unmatching type when called
192 # with a different number
193 values = ["", ("",""), ("","",""), [], {}, ["",""], ["","",""]]
194 methods = [
195 (self.iface.ReturnOneString, 'SignalOneString', set([0]), set([0])),
196 (self.iface.ReturnTwoStrings, 'SignalTwoStrings', set([1, 5]), set([1])),
197 (self.iface.ReturnStruct, 'SignalStruct', set([1, 5]), set([1])),
198 # all of our test values are sequences so will marshall correctly into an array :P
199 (self.iface.ReturnArray, 'SignalArray', set(range(len(values))), set([3, 5, 6])),
200 (self.iface.ReturnDict, 'SignalDict', set([0, 3, 4]), set([4]))
203 for (method, signal, success_values, return_values) in methods:
204 print "\nTrying correct behaviour of", method._method_name
205 for value in range(len(values)):
206 try:
207 ret = method(value)
208 except Exception, e:
209 print "%s(%r) raised %s: %s" % (method._method_name, values[value], e.__class__, e)
211 # should fail if it tried to marshal the wrong type
212 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))
213 else:
214 print "%s(%r) returned %r" % (method._method_name, values[value], ret)
216 # should only succeed if it's the right return type
217 self.assert_(value in success_values, "%s should fail when we ask it to return %r" % (method._method_name, values[value]))
219 # check the value is right too :D
220 returns = map(lambda n: values[n], return_values)
221 self.assert_(ret in returns, "%s should return one of %r but it returned %r instead" % (method._method_name, returns, ret))
223 print "\nTrying correct emission of", signal
224 for value in range(len(values)):
225 try:
226 self.iface.EmitSignal(signal, value)
227 except Exception, e:
228 print "EmitSignal(%s, %r) raised %s" % (signal, values[value], e.__class__)
230 # should fail if it tried to marshal the wrong type
231 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))
232 else:
233 print "EmitSignal(%s, %r) appeared to succeed" % (signal, values[value])
235 # should only succeed if it's the right return type
236 self.assert_(value in success_values, "EmitSignal(%s) should fail when we ask it to return %r" % (signal, values[value]))
238 # FIXME: wait for the signal here
240 print
242 def testInheritance(self):
243 print "\n********* Testing inheritance from dbus.method.Interface ***********"
244 ret = self.iface.CheckInheritance()
245 print "CheckInheritance returned %s" % ret
246 self.assert_(ret, "overriding CheckInheritance from TestInterface failed")
248 def testAsyncMethods(self):
249 print "\n********* Testing asynchronous method implementation *******"
250 for async in (True, False):
251 for fail in (True, False):
252 try:
253 val = ('a', 1, False, [1,2], {1:2})
254 print "calling AsynchronousMethod with %s %s %s" % (async, fail, val)
255 ret = self.iface.AsynchronousMethod(async, fail, val)
256 except Exception, e:
257 self.assert_(fail, '%s: %s' % (e.__class__, e))
258 print "Expected failure: %s: %s" % (e.__class__, e)
259 else:
260 self.assert_(not fail, 'Expected failure but succeeded?!')
261 self.assertEquals(val, ret)
262 self.assertEquals(1, ret.variant_level)
264 def testBusInstanceCaching(self):
265 print "\n********* Testing dbus.Bus instance sharing *********"
267 # unfortunately we can't test the system bus here
268 # but the codepaths are the same
269 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)):
270 print "\nTesting %s:" % cls.__name__
272 share_cls = cls()
273 share_type = dbus.Bus(bus_type=type)
274 share_func = func()
276 private_cls = cls(private=True)
277 private_type = dbus.Bus(bus_type=type, private=True)
278 private_func = func(private=True)
280 print " - checking shared instances are the same..."
281 self.assert_(share_cls == share_type, '%s should equal %s' % (share_cls, share_type))
282 self.assert_(share_type == share_func, '%s should equal %s' % (share_type, share_func))
284 print " - checking private instances are distinct from the shared instance..."
285 self.assert_(share_cls != private_cls, '%s should not equal %s' % (share_cls, private_cls))
286 self.assert_(share_type != private_type, '%s should not equal %s' % (share_type, private_type))
287 self.assert_(share_func != private_func, '%s should not equal %s' % (share_func, private_func))
289 print " - checking private instances are distinct from each other..."
290 self.assert_(private_cls != private_type, '%s should not equal %s' % (private_cls, private_type))
291 self.assert_(private_type != private_func, '%s should not equal %s' % (private_type, private_func))
292 self.assert_(private_func != private_cls, '%s should not equal %s' % (private_func, private_cls))
294 def testSenderName(self):
295 print '\n******** Testing sender name keyword ********'
296 myself = self.iface.WhoAmI()
297 print "I am", myself
299 def testBusNameCreation(self):
300 print '\n******** Testing BusName creation ********'
301 test = [('org.freedesktop.DBus.Python.TestName', True),
302 ('org.freedesktop.DBus.Python.TestName', True),
303 ('org.freedesktop.DBus.Python.InvalidName&^*%$', False)]
304 # Do some more intelligent handling/testing of queueing vs success?
305 # ('org.freedesktop.DBus.TestSuitePythonService', False)]
306 # For some reason this actually succeeds
307 # ('org.freedesktop.DBus', False)]
309 # make a method call to ensure the test service is active
310 self.iface.Echo("foo")
312 names = {}
313 for (name, succeed) in test:
314 try:
315 print "requesting %s" % name
316 busname = dbus.service.BusName(name)
317 except Exception, e:
318 print "%s:\n%s" % (e.__class__, e)
319 self.assert_(not succeed, 'did not expect registering bus name %s to fail' % name)
320 else:
321 print busname
322 self.assert_(succeed, 'expected registering bus name %s to fail'% name)
323 if name in names:
324 self.assert_(names[name] == busname, 'got a new instance for same name %s' % name)
325 print "instance of %s re-used, good!" % name
326 else:
327 names[name] = busname
329 del busname
331 print
333 del names
335 bus = dbus.Bus()
336 ret = bus.name_has_owner('org.freedesktop.DBus.Python.TestName')
337 self.assert_(not ret, 'deleting reference failed to release BusName org.freedesktop.DBus.Python.TestName')
339 """ Remove this for now
340 class TestDBusPythonToGLibBindings(unittest.TestCase):
341 def setUp(self):
342 self.bus = dbus.SessionBus()
343 self.remote_object = self.bus.get_object("org.freedesktop.DBus.TestSuiteGLibService", "/org/freedesktop/DBus/Tests/MyTestObject")
344 self.iface = dbus.Interface(self.remote_object, "org.freedesktop.DBus.Tests.MyObject")
346 def testIntrospection(self):
347 #test introspection
348 print "\n********* Introspection Test ************"
349 print self.remote_object.Introspect(dbus_interface="org.freedesktop.DBus.Introspectable")
350 print "Introspection test passed"
351 self.assert_(True)
353 def testCalls(self):
354 print "\n********* Call Test ************"
355 result = self.iface.ManyArgs(1000, 'Hello GLib', 2)
356 print result
357 self.assert_(result == [2002.0, 'HELLO GLIB'])
359 arg0 = {"Dude": 1, "john": "palmieri", "python": 2.4}
360 result = self.iface.ManyStringify(arg0)
361 print result
363 print "Call test passed"
364 self.assert_(True)
366 def testPythonTypes(self):
367 print "\n********* Testing Python Types ***********"
369 for send_val in test_types_vals:
370 print "Testing %s"% str(send_val)
371 recv_val = self.iface.EchoVariant(send_val)
372 self.assertEquals(send_val, recv_val.object)
374 if __name__ == '__main__':
375 gobject.threads_init()
376 dbus.glib.init_threads()
378 unittest.main()