test/test-client.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (C) 2004 Red Hat Inc. <http://www.redhat.com/>
   4 # Copyright (C) 2005, 2006 Collabora Ltd. <http://www.collabora.co.uk/>
   5 #
   6 # Licensed under the Academic Free License version 2.1
   7 #
   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 # This program is free software; you can redistribute it and/or modify
  11 # it under the terms of the GNU General Public License as published by
  12 # the Free Software Foundation; either version 2 of the License, or
  13 # (at your option) any later version.
  14 #
  15 # This program is distributed in the hope that it will be useful,
  16 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  17 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  18 # GNU General Public License for more details.
  19 #
  20 # You should have received a copy of the GNU General Public License
  21 # along with this program; if not, write to the Free Software
  22 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  23
  24
  25 import sys
  26 import os
  27 import unittest
  28 import time
  29 import logging
  30 import weakref
  31
  32 builddir = os.path.normpath(os.environ["DBUS_TOP_BUILDDIR"])
  33 pydir = os.path.normpath(os.environ["DBUS_TOP_SRCDIR"])
  34
  35 import dbus
  36 import _dbus_bindings
  37 import gobject
  38 import dbus.glib
  39 import dbus.service
  40
  41
  42 logging.basicConfig()
  43
  44
  45 pkg = dbus.__file__
  46 if not pkg.startswith(pydir):
  47     raise Exception("DBus modules (%s) are not being picked up from the package"%pkg)
  48
  49 if not _dbus_bindings.__file__.startswith(builddir):
  50     raise Exception("DBus modules (%s) are not being picked up from the package"%_dbus_bindings.__file__)
  51
  52 test_types_vals = [1, 12323231, 3.14159265, 99999999.99,
  53                  "dude", "123", "What is all the fuss about?", "gob@gob.com",
  54                  u'\\u310c\\u310e\\u3114', u'\\u0413\\u0414\\u0415',
  55                  u'\\u2200software \\u2203crack', u'\\xf4\\xe5\\xe8',
  56                  [1,2,3], ["how", "are", "you"], [1.23,2.3], [1], ["Hello"],
  57                  (1,2,3), (1,), (1,"2",3), ("2", "what"), ("you", 1.2),
  58                  {1:"a", 2:"b"}, {"a":1, "b":2}, #{"a":(1,"B")},
  59                  {1:1.1, 2:2.2}, [[1,2,3],[2,3,4]], [["a","b"],["c","d"]],
  60                  True, False,
  61                  dbus.Int16(-10), dbus.UInt16(10), 'SENTINEL',
  62                  #([1,2,3],"c", 1.2, ["a","b","c"], {"a": (1,"v"), "b": (2,"d")})
  63                  ]
  64
  65 class TestDBusBindings(unittest.TestCase):
  66     def setUp(self):
  67         self.bus = dbus.SessionBus()
  68         self.remote_object = self.bus.get_object("org.freedesktop.DBus.TestSuitePythonService", "/org/freedesktop/DBus/TestSuitePythonObject")
  69         self.iface = dbus.Interface(self.remote_object, "org.freedesktop.DBus.TestSuiteInterface")
  70
  71     def testWeakRefs(self):
  72         # regression test for Sugar crash caused by smcv getting weak refs
  73         # wrong - pre-bugfix, this would segfault
  74         bus = dbus.SessionBus(private=True)
  75         ref = weakref.ref(bus)
  76         self.assert_(ref() is bus)
  77         del bus
  78         self.assert_(ref() is None)
  79
  80     def testInterfaceKeyword(self):
  81         #test dbus_interface parameter
  82         print self.remote_object.Echo("dbus_interface on Proxy test Passed", dbus_interface = "org.freedesktop.DBus.TestSuiteInterface")
  83         print self.iface.Echo("dbus_interface on Interface test Passed", dbus_interface = "org.freedesktop.DBus.TestSuiteInterface")
  84         self.assert_(True)
  85
  86     def testGetDBusMethod(self):
  87         self.assertEquals(self.iface.get_dbus_method('AcceptListOfByte')('\1\2\3'), [1,2,3])
  88         self.assertEquals(self.remote_object.get_dbus_method('AcceptListOfByte', dbus_interface='org.freedesktop.DBus.TestSuiteInterface')('\1\2\3'), [1,2,3])
  89
  90     def testCallingConventionOptions(self):
  91         self.assertEquals(self.iface.AcceptListOfByte('\1\2\3'), [1,2,3])
  92         self.assertEquals(self.iface.AcceptListOfByte('\1\2\3', byte_arrays=True), '\1\2\3')
  93         self.assertEquals(self.iface.AcceptByteArray('\1\2\3'), [1,2,3])
  94         self.assertEquals(self.iface.AcceptByteArray('\1\2\3', byte_arrays=True), '\1\2\3')
  95         self.assert_(isinstance(self.iface.AcceptUTF8String('abc'), unicode))
  96         self.assert_(isinstance(self.iface.AcceptUTF8String('abc', utf8_strings=True), str))
  97         self.assert_(isinstance(self.iface.AcceptUnicodeString('abc'), unicode))
  98         self.assert_(isinstance(self.iface.AcceptUnicodeString('abc', utf8_strings=True), str))
  99
 100     def testIntrospection(self):
 101         #test introspection
 102         print "\n********* Introspection Test ************"
 103         print self.remote_object.Introspect(dbus_interface="org.freedesktop.DBus.Introspectable")
 104         print "Introspection test passed"
 105         self.assert_(True)
 106
 107     def testPythonTypes(self):
 108         #test sending python types and getting them back
 109         print "\n********* Testing Python Types ***********"
 110
 111         for send_val in test_types_vals:
 112             print "Testing %s"% str(send_val)
 113             recv_val = self.iface.Echo(send_val)
 114             self.assertEquals(send_val, recv_val)
 115             self.assertEquals(recv_val.variant_level, 1)
 116
 117     def testUtf8StringsSync(self):
 118         send_val = u'foo'
 119         recv_val = self.iface.Echo(send_val, utf8_strings=True)
 120         self.assert_(isinstance(recv_val, str))
 121         self.assert_(isinstance(recv_val, dbus.UTF8String))
 122         recv_val = self.iface.Echo(send_val, utf8_strings=False)
 123         self.assert_(isinstance(recv_val, unicode))
 124         self.assert_(isinstance(recv_val, dbus.String))
 125
 126     def testBenchmarkIntrospect(self):
 127         print "\n********* Benchmark Introspect ************"
 128         a = time.time()
 129         print a
 130         print self.iface.GetComplexArray()
 131         b = time.time()
 132         print b
 133         print "Delta: %f" % (b - a)
 134         self.assert_(True)
 135
 136     def testAsyncCalls(self):
 137         #test sending python types and getting them back async
 138         print "\n********* Testing Async Calls ***********"
 139
 140         failures = []
 141         main_loop = gobject.MainLoop()
 142
 143         class async_check:
 144             def __init__(self, test_controler, expected_result, do_exit, utf8):
 145                 self.expected_result = expected_result
 146                 self.do_exit = do_exit
 147                 self.utf8 = utf8
 148                 self.test_controler = test_controler
 149
 150             def callback(self, val):
 151                 try:
 152                     if self.do_exit:
 153                         main_loop.quit()
 154
 155                     self.test_controler.assertEquals(val, self.expected_result)
 156                     self.test_controler.assertEquals(val.variant_level, 1)
 157                     if self.utf8 and not isinstance(val, dbus.UTF8String):
 158                         failures.append('%r should have been utf8 but was not' % val)
 159                         return
 160                     elif not self.utf8 and isinstance(val, dbus.UTF8String):
 161                         failures.append('%r should not have been utf8' % val)
 162                         return
 163                 except Exception, e:
 164                     failures.append("%s:\n%s" % (e.__class__, e))
 165
 166             def error_handler(self, error):
 167                 print error
 168                 if self.do_exit:
 169                     main_loop.quit()
 170
 171                 failures.append('%s: %s' % (error.__class__, error))
 172
 173         last_type = test_types_vals[-1]
 174         for send_val in test_types_vals:
 175             print "Testing %s" % str(send_val)
 176             utf8 = (send_val == 'gob@gob.com')
 177             check = async_check(self, send_val, last_type == send_val,
 178                                 utf8)
 179             recv_val = self.iface.Echo(send_val,
 180                                        reply_handler=check.callback,
 181                                        error_handler=check.error_handler,
 182                                        utf8_strings=utf8)
 183         main_loop.run()
 184         if failures:
 185             self.assert_(False, failures)
 186
 187     def testStrictMarshalling(self):
 188         print "\n********* Testing strict return & signal marshalling ***********"
 189
 190         # these values are the same as in the server, and the
 191         # methods should only succeed when they are called with
 192         # the right value number, because they have out_signature
 193         # decorations, and return an unmatching type when called
 194         # with a different number
 195         values = ["", ("",""), ("","",""), [], {}, ["",""], ["","",""]]
 196         methods = [
 197                     (self.iface.ReturnOneString, 'SignalOneString', set([0]), set([0])),
 198                     (self.iface.ReturnTwoStrings, 'SignalTwoStrings', set([1, 5]), set([1])),
 199                     (self.iface.ReturnStruct, 'SignalStruct', set([1, 5]), set([1])),
 200                     # all of our test values are sequences so will marshall correctly into an array :P
 201                     (self.iface.ReturnArray, 'SignalArray', set(range(len(values))), set([3, 5, 6])),
 202                     (self.iface.ReturnDict, 'SignalDict', set([0, 3, 4]), set([4]))
 203                 ]
 204
 205         for (method, signal, success_values, return_values) in methods:
 206             print "\nTrying correct behaviour of", method._method_name
 207             for value in range(len(values)):
 208                 try:
 209                     ret = method(value)
 210                 except Exception, e:
 211                     print "%s(%r) raised %s: %s" % (method._method_name, values[value], e.__class__, e)
 212
 213                     # should fail if it tried to marshal the wrong type
 214                     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))
 215                 else:
 216                     print "%s(%r) returned %r" % (method._method_name, values[value], ret)
 217
 218                     # should only succeed if it's the right return type
 219                     self.assert_(value in success_values, "%s should fail when we ask it to return %r" % (method._method_name, values[value]))
 220
 221                     # check the value is right too :D
 222                     returns = map(lambda n: values[n], return_values)
 223                     self.assert_(ret in returns, "%s should return one of %r but it returned %r instead" % (method._method_name, returns, ret))
 224
 225             print "\nTrying correct emission of", signal
 226             for value in range(len(values)):
 227                 try:
 228                     self.iface.EmitSignal(signal, value)
 229                 except Exception, e:
 230                     print "EmitSignal(%s, %r) raised %s" % (signal, values[value], e.__class__)
 231
 232                     # should fail if it tried to marshal the wrong type
 233                     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))
 234                 else:
 235                     print "EmitSignal(%s, %r) appeared to succeed" % (signal, values[value])
 236
 237                     # should only succeed if it's the right return type
 238                     self.assert_(value in success_values, "EmitSignal(%s) should fail when we ask it to return %r" % (signal, values[value]))
 239
 240                     # FIXME: wait for the signal here
 241
 242         print
 243
 244     def testInheritance(self):
 245         print "\n********* Testing inheritance from dbus.method.Interface ***********"
 246         ret = self.iface.CheckInheritance()
 247         print "CheckInheritance returned %s" % ret
 248         self.assert_(ret, "overriding CheckInheritance from TestInterface failed")
 249
 250     def testAsyncMethods(self):
 251         print "\n********* Testing asynchronous method implementation *******"
 252         for async in (True, False):
 253             for fail in (True, False):
 254                 try:
 255                     val = ('a', 1, False, [1,2], {1:2})
 256                     print "calling AsynchronousMethod with %s %s %s" % (async, fail, val)
 257                     ret = self.iface.AsynchronousMethod(async, fail, val)
 258                 except Exception, e:
 259                     self.assert_(fail, '%s: %s' % (e.__class__, e))
 260                     print "Expected failure: %s: %s" % (e.__class__, e)
 261                 else:
 262                     self.assert_(not fail, 'Expected failure but succeeded?!')
 263                     self.assertEquals(val, ret)
 264                     self.assertEquals(1, ret.variant_level)
 265
 266     def testBusInstanceCaching(self):
 267         print "\n********* Testing dbus.Bus instance sharing *********"
 268
 269         # unfortunately we can't test the system bus here
 270         # but the codepaths are the same
 271         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)):
 272             print "\nTesting %s:" % cls.__name__
 273
 274             share_cls = cls()
 275             share_type = dbus.Bus(bus_type=type)
 276             share_func = func()
 277
 278             private_cls = cls(private=True)
 279             private_type = dbus.Bus(bus_type=type, private=True)
 280             private_func = func(private=True)
 281
 282             print " - checking shared instances are the same..."
 283             self.assert_(share_cls == share_type, '%s should equal %s' % (share_cls, share_type))
 284             self.assert_(share_type == share_func, '%s should equal %s' % (share_type, share_func))
 285
 286             print " - checking private instances are distinct from the shared instance..."
 287             self.assert_(share_cls != private_cls, '%s should not equal %s' % (share_cls, private_cls))
 288             self.assert_(share_type != private_type, '%s should not equal %s' % (share_type, private_type))
 289             self.assert_(share_func != private_func, '%s should not equal %s' % (share_func, private_func))
 290
 291             print " - checking private instances are distinct from each other..."
 292             self.assert_(private_cls != private_type, '%s should not equal %s' % (private_cls, private_type))
 293             self.assert_(private_type != private_func, '%s should not equal %s' % (private_type, private_func))
 294             self.assert_(private_func != private_cls, '%s should not equal %s' % (private_func, private_cls))
 295
 296     def testSenderName(self):
 297         print '\n******** Testing sender name keyword ********'
 298         myself = self.iface.WhoAmI()
 299         print "I am", myself
 300
 301     def testBusNameCreation(self):
 302         print '\n******** Testing BusName creation ********'
 303         test = [('org.freedesktop.DBus.Python.TestName', True),
 304                 ('org.freedesktop.DBus.Python.TestName', True),
 305                 ('org.freedesktop.DBus.Python.InvalidName&^*%$', False)]
 306         # Do some more intelligent handling/testing of queueing vs success?
 307         # ('org.freedesktop.DBus.TestSuitePythonService', False)]
 308         # For some reason this actually succeeds
 309         # ('org.freedesktop.DBus', False)]
 310
 311         # make a method call to ensure the test service is active
 312         self.iface.Echo("foo")
 313
 314         names = {}
 315         for (name, succeed) in test:
 316             try:
 317                 print "requesting %s" % name
 318                 busname = dbus.service.BusName(name)
 319             except Exception, e:
 320                 print "%s:\n%s" % (e.__class__, e)
 321                 self.assert_(not succeed, 'did not expect registering bus name %s to fail' % name)
 322             else:
 323                 print busname
 324                 self.assert_(succeed, 'expected registering bus name %s to fail'% name)
 325                 if name in names:
 326                     self.assert_(names[name] == busname, 'got a new instance for same name %s' % name)
 327                     print "instance of %s re-used, good!" % name
 328                 else:
 329                     names[name] = busname
 330
 331                 del busname
 332
 333             print
 334
 335         del names
 336
 337         bus = dbus.Bus()
 338         ret = bus.name_has_owner('org.freedesktop.DBus.Python.TestName')
 339         self.assert_(not ret, 'deleting reference failed to release BusName org.freedesktop.DBus.Python.TestName')
 340
 341 """ Remove this for now
 342 class TestDBusPythonToGLibBindings(unittest.TestCase):
 343     def setUp(self):
 344         self.bus = dbus.SessionBus()
 345         self.remote_object = self.bus.get_object("org.freedesktop.DBus.TestSuiteGLibService", "/org/freedesktop/DBus/Tests/MyTestObject")
 346         self.iface = dbus.Interface(self.remote_object, "org.freedesktop.DBus.Tests.MyObject")
 347
 348     def testIntrospection(self):
 349         #test introspection
 350         print "\n********* Introspection Test ************"
 351         print self.remote_object.Introspect(dbus_interface="org.freedesktop.DBus.Introspectable")
 352         print "Introspection test passed"
 353         self.assert_(True)
 354
 355     def testCalls(self):
 356         print "\n********* Call Test ************"
 357         result =  self.iface.ManyArgs(1000, 'Hello GLib', 2)
 358         print result
 359         self.assert_(result == [2002.0, 'HELLO GLIB'])
 360
 361         arg0 = {"Dude": 1, "john": "palmieri", "python": 2.4}
 362         result = self.iface.ManyStringify(arg0)
 363         print result
 364
 365         print "Call test passed"
 366         self.assert_(True)
 367
 368     def testPythonTypes(self):
 369         print "\n********* Testing Python Types ***********"
 370
 371         for send_val in test_types_vals:
 372             print "Testing %s"% str(send_val)
 373             recv_val = self.iface.EchoVariant(send_val)
 374             self.assertEquals(send_val, recv_val.object)
 375 """
 376 if __name__ == '__main__':
 377     gobject.threads_init()
 378     dbus.glib.init_threads()
 379
 380     unittest.main()