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 # the Free Software Foundation; either version 2 of the License, or
  11 # (at your option) any later version.
  12 #
  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.
  17 #
  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
  21
  22
  23 import sys
  24 import os
  25 import unittest
  26 import time
  27 import logging
  28 import weakref
  29
  30 builddir = os.path.normpath(os.environ["DBUS_TOP_BUILDDIR"])
  31 pydir = os.path.normpath(os.environ["DBUS_TOP_SRCDIR"])
  32
  33 import dbus
  34 import _dbus_bindings
  35 import gobject
  36 import dbus.glib
  37 import dbus.service
  38
  39
  40 logging.basicConfig()
  41
  42
  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)
  46
  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__)
  49
  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")})
  61                  ]
  62
  63 NAME = "org.freedesktop.DBus.TestSuitePythonService"
  64 IFACE = "org.freedesktop.DBus.TestSuiteInterface"
  65 OBJECT = "/org/freedesktop/DBus/TestSuitePythonObject"
  66
  67 class TestDBusBindings(unittest.TestCase):
  68     def setUp(self):
  69         self.bus = dbus.SessionBus()
  70         self.remote_object = self.bus.get_object(NAME, OBJECT)
  71         self.remote_object_follow = self.bus.get_object(NAME, OBJECT,
  72                 follow_name_owner_changes=True)
  73         self.iface = dbus.Interface(self.remote_object, IFACE)
  74
  75     def testGObject(self):
  76         print "Testing ExportedGObject... ",
  77         remote_gobject = self.bus.get_object(NAME, OBJECT + '/GObject')
  78         iface = dbus.Interface(remote_gobject, IFACE)
  79         print "introspection, ",
  80         remote_gobject.Introspect(dbus_interface=dbus.INTROSPECTABLE_IFACE)
  81         print "method call, ",
  82         self.assertEquals(iface.Echo('123'), '123')
  83         print "... OK"
  84
  85     def testWeakRefs(self):
  86         # regression test for Sugar crash caused by smcv getting weak refs
  87         # wrong - pre-bugfix, this would segfault
  88         bus = dbus.SessionBus(private=True)
  89         ref = weakref.ref(bus)
  90         self.assert_(ref() is bus)
  91         del bus
  92         self.assert_(ref() is None)
  93
  94     def testInterfaceKeyword(self):
  95         #test dbus_interface parameter
  96         print self.remote_object.Echo("dbus_interface on Proxy test Passed", dbus_interface = IFACE)
  97         print self.iface.Echo("dbus_interface on Interface test Passed", dbus_interface = IFACE)
  98         self.assert_(True)
  99
 100     def testGetDBusMethod(self):
 101         self.assertEquals(self.iface.get_dbus_method('AcceptListOfByte')('\1\2\3'), [1,2,3])
 102         self.assertEquals(self.remote_object.get_dbus_method('AcceptListOfByte', dbus_interface=IFACE)('\1\2\3'), [1,2,3])
 103
 104     def testCallingConventionOptions(self):
 105         self.assertEquals(self.iface.AcceptListOfByte('\1\2\3'), [1,2,3])
 106         self.assertEquals(self.iface.AcceptListOfByte('\1\2\3', byte_arrays=True), '\1\2\3')
 107         self.assertEquals(self.iface.AcceptByteArray('\1\2\3'), [1,2,3])
 108         self.assertEquals(self.iface.AcceptByteArray('\1\2\3', byte_arrays=True), '\1\2\3')
 109         self.assert_(isinstance(self.iface.AcceptUTF8String('abc'), unicode))
 110         self.assert_(isinstance(self.iface.AcceptUTF8String('abc', utf8_strings=True), str))
 111         self.assert_(isinstance(self.iface.AcceptUnicodeString('abc'), unicode))
 112         self.assert_(isinstance(self.iface.AcceptUnicodeString('abc', utf8_strings=True), str))
 113
 114     def testIntrospection(self):
 115         #test introspection
 116         print "\n********* Introspection Test ************"
 117         print self.remote_object.Introspect(dbus_interface="org.freedesktop.DBus.Introspectable")
 118         print "Introspection test passed"
 119         self.assert_(True)
 120
 121     def testPythonTypes(self):
 122         #test sending python types and getting them back
 123         print "\n********* Testing Python Types ***********"
 124
 125         for send_val in test_types_vals:
 126             print "Testing %s"% str(send_val)
 127             recv_val = self.iface.Echo(send_val)
 128             self.assertEquals(send_val, recv_val)
 129             self.assertEquals(recv_val.variant_level, 1)
 130
 131     def testMethodExtraInfoKeywords(self):
 132         print "Testing MethodExtraInfoKeywords..."
 133         sender, path, destination, message_cls = self.iface.MethodExtraInfoKeywords()
 134         self.assert_(sender.startswith(':'))
 135         self.assertEquals(path, '/org/freedesktop/DBus/TestSuitePythonObject')
 136         # we're using the "early binding" form of get_object (without
 137         # follow_name_owner_changes), so the destination we actually sent it
 138         # to will be the unique name
 139         self.assert_(destination.startswith(':'))
 140         self.assertEquals(message_cls, 'dbus.lowlevel.MethodCallMessage')
 141
 142     def testUtf8StringsSync(self):
 143         send_val = u'foo'
 144         recv_val = self.iface.Echo(send_val, utf8_strings=True)
 145         self.assert_(isinstance(recv_val, str))
 146         self.assert_(isinstance(recv_val, dbus.UTF8String))
 147         recv_val = self.iface.Echo(send_val, utf8_strings=False)
 148         self.assert_(isinstance(recv_val, unicode))
 149         self.assert_(isinstance(recv_val, dbus.String))
 150
 151     def testBenchmarkIntrospect(self):
 152         print "\n********* Benchmark Introspect ************"
 153         a = time.time()
 154         print a
 155         print self.iface.GetComplexArray()
 156         b = time.time()
 157         print b
 158         print "Delta: %f" % (b - a)
 159         self.assert_(True)
 160
 161     def testAsyncCalls(self):
 162         #test sending python types and getting them back async
 163         print "\n********* Testing Async Calls ***********"
 164
 165         failures = []
 166         main_loop = gobject.MainLoop()
 167
 168         class async_check:
 169             def __init__(self, test_controler, expected_result, do_exit, utf8):
 170                 self.expected_result = expected_result
 171                 self.do_exit = do_exit
 172                 self.utf8 = utf8
 173                 self.test_controler = test_controler
 174
 175             def callback(self, val):
 176                 try:
 177                     if self.do_exit:
 178                         main_loop.quit()
 179
 180                     self.test_controler.assertEquals(val, self.expected_result)
 181                     self.test_controler.assertEquals(val.variant_level, 1)
 182                     if self.utf8 and not isinstance(val, dbus.UTF8String):
 183                         failures.append('%r should have been utf8 but was not' % val)
 184                         return
 185                     elif not self.utf8 and isinstance(val, dbus.UTF8String):
 186                         failures.append('%r should not have been utf8' % val)
 187                         return
 188                 except Exception, e:
 189                     failures.append("%s:\n%s" % (e.__class__, e))
 190
 191             def error_handler(self, error):
 192                 print error
 193                 if self.do_exit:
 194                     main_loop.quit()
 195
 196                 failures.append('%s: %s' % (error.__class__, error))
 197
 198         last_type = test_types_vals[-1]
 199         for send_val in test_types_vals:
 200             print "Testing %s" % str(send_val)
 201             utf8 = (send_val == 'gob@gob.com')
 202             check = async_check(self, send_val, last_type == send_val,
 203                                 utf8)
 204             recv_val = self.iface.Echo(send_val,
 205                                        reply_handler=check.callback,
 206                                        error_handler=check.error_handler,
 207                                        utf8_strings=utf8)
 208         main_loop.run()
 209         if failures:
 210             self.assert_(False, failures)
 211
 212     def testStrictMarshalling(self):
 213         print "\n********* Testing strict return & signal marshalling ***********"
 214
 215         # these values are the same as in the server, and the
 216         # methods should only succeed when they are called with
 217         # the right value number, because they have out_signature
 218         # decorations, and return an unmatching type when called
 219         # with a different number
 220         values = ["", ("",""), ("","",""), [], {}, ["",""], ["","",""]]
 221         methods = [
 222                     (self.iface.ReturnOneString, 'SignalOneString', set([0]), set([0])),
 223                     (self.iface.ReturnTwoStrings, 'SignalTwoStrings', set([1, 5]), set([1])),
 224                     (self.iface.ReturnStruct, 'SignalStruct', set([1, 5]), set([1])),
 225                     # all of our test values are sequences so will marshall correctly into an array :P
 226                     (self.iface.ReturnArray, 'SignalArray', set(range(len(values))), set([3, 5, 6])),
 227                     (self.iface.ReturnDict, 'SignalDict', set([0, 3, 4]), set([4]))
 228                 ]
 229
 230         for (method, signal, success_values, return_values) in methods:
 231             print "\nTrying correct behaviour of", method._method_name
 232             for value in range(len(values)):
 233                 try:
 234                     ret = method(value)
 235                 except Exception, e:
 236                     print "%s(%r) raised %s: %s" % (method._method_name, values[value], e.__class__, e)
 237
 238                     # should fail if it tried to marshal the wrong type
 239                     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))
 240                 else:
 241                     print "%s(%r) returned %r" % (method._method_name, values[value], ret)
 242
 243                     # should only succeed if it's the right return type
 244                     self.assert_(value in success_values, "%s should fail when we ask it to return %r" % (method._method_name, values[value]))
 245
 246                     # check the value is right too :D
 247                     returns = map(lambda n: values[n], return_values)
 248                     self.assert_(ret in returns, "%s should return one of %r but it returned %r instead" % (method._method_name, returns, ret))
 249
 250             print "\nTrying correct emission of", signal
 251             for value in range(len(values)):
 252                 try:
 253                     self.iface.EmitSignal(signal, value)
 254                 except Exception, e:
 255                     print "EmitSignal(%s, %r) raised %s" % (signal, values[value], e.__class__)
 256
 257                     # should fail if it tried to marshal the wrong type
 258                     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))
 259                 else:
 260                     print "EmitSignal(%s, %r) appeared to succeed" % (signal, values[value])
 261
 262                     # should only succeed if it's the right return type
 263                     self.assert_(value in success_values, "EmitSignal(%s) should fail when we ask it to return %r" % (signal, values[value]))
 264
 265                     # FIXME: wait for the signal here
 266
 267         print
 268
 269     def testInheritance(self):
 270         print "\n********* Testing inheritance from dbus.method.Interface ***********"
 271         ret = self.iface.CheckInheritance()
 272         print "CheckInheritance returned %s" % ret
 273         self.assert_(ret, "overriding CheckInheritance from TestInterface failed")
 274
 275     def testAsyncMethods(self):
 276         print "\n********* Testing asynchronous method implementation *******"
 277         for async in (True, False):
 278             for fail in (True, False):
 279                 try:
 280                     val = ('a', 1, False, [1,2], {1:2})
 281                     print "calling AsynchronousMethod with %s %s %s" % (async, fail, val)
 282                     ret = self.iface.AsynchronousMethod(async, fail, val)
 283                 except Exception, e:
 284                     self.assert_(fail, '%s: %s' % (e.__class__, e))
 285                     print "Expected failure: %s: %s" % (e.__class__, e)
 286                 else:
 287                     self.assert_(not fail, 'Expected failure but succeeded?!')
 288                     self.assertEquals(val, ret)
 289                     self.assertEquals(1, ret.variant_level)
 290
 291     def testBusInstanceCaching(self):
 292         print "\n********* Testing dbus.Bus instance sharing *********"
 293
 294         # unfortunately we can't test the system bus here
 295         # but the codepaths are the same
 296         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)):
 297             print "\nTesting %s:" % cls.__name__
 298
 299             share_cls = cls()
 300             share_type = dbus.Bus(bus_type=type)
 301             share_func = func()
 302
 303             private_cls = cls(private=True)
 304             private_type = dbus.Bus(bus_type=type, private=True)
 305             private_func = func(private=True)
 306
 307             print " - checking shared instances are the same..."
 308             self.assert_(share_cls == share_type, '%s should equal %s' % (share_cls, share_type))
 309             self.assert_(share_type == share_func, '%s should equal %s' % (share_type, share_func))
 310
 311             print " - checking private instances are distinct from the shared instance..."
 312             self.assert_(share_cls != private_cls, '%s should not equal %s' % (share_cls, private_cls))
 313             self.assert_(share_type != private_type, '%s should not equal %s' % (share_type, private_type))
 314             self.assert_(share_func != private_func, '%s should not equal %s' % (share_func, private_func))
 315
 316             print " - checking private instances are distinct from each other..."
 317             self.assert_(private_cls != private_type, '%s should not equal %s' % (private_cls, private_type))
 318             self.assert_(private_type != private_func, '%s should not equal %s' % (private_type, private_func))
 319             self.assert_(private_func != private_cls, '%s should not equal %s' % (private_func, private_cls))
 320
 321     def testSenderName(self):
 322         print '\n******** Testing sender name keyword ********'
 323         myself = self.iface.WhoAmI()
 324         print "I am", myself
 325
 326     def testBusGetNameOwner(self):
 327         ret = self.bus.get_name_owner(NAME)
 328         self.assert_(ret.startswith(':'), ret)
 329
 330     def testBusListNames(self):
 331         ret = self.bus.list_names()
 332         self.assert_(NAME in ret, ret)
 333
 334     def testBusListActivatableNames(self):
 335         ret = self.bus.list_activatable_names()
 336         self.assert_(NAME in ret, ret)
 337
 338     def testBusNameHasOwner(self):
 339         self.assert_(self.bus.name_has_owner(NAME))
 340         self.assert_(not self.bus.name_has_owner('badger.mushroom.snake'))
 341
 342     def testBusNameCreation(self):
 343         print '\n******** Testing BusName creation ********'
 344         test = [('org.freedesktop.DBus.Python.TestName', True),
 345                 ('org.freedesktop.DBus.Python.TestName', True),
 346                 ('org.freedesktop.DBus.Python.InvalidName&^*%$', False)]
 347         # Do some more intelligent handling/testing of queueing vs success?
 348         # ('org.freedesktop.DBus.TestSuitePythonService', False)]
 349         # For some reason this actually succeeds
 350         # ('org.freedesktop.DBus', False)]
 351
 352         # make a method call to ensure the test service is active
 353         self.iface.Echo("foo")
 354
 355         names = {}
 356         for (name, succeed) in test:
 357             try:
 358                 print "requesting %s" % name
 359                 busname = dbus.service.BusName(name, dbus.SessionBus())
 360             except Exception, e:
 361                 print "%s:\n%s" % (e.__class__, e)
 362                 self.assert_(not succeed, 'did not expect registering bus name %s to fail' % name)
 363             else:
 364                 print busname
 365                 self.assert_(succeed, 'expected registering bus name %s to fail'% name)
 366                 if name in names:
 367                     self.assert_(names[name] == busname, 'got a new instance for same name %s' % name)
 368                     print "instance of %s re-used, good!" % name
 369                 else:
 370                     names[name] = busname
 371
 372                 del busname
 373
 374             print
 375
 376         del names
 377
 378         bus = dbus.Bus()
 379         ret = bus.name_has_owner('org.freedesktop.DBus.Python.TestName')
 380         self.assert_(not ret, 'deleting reference failed to release BusName org.freedesktop.DBus.Python.TestName')
 381
 382     def testMultipleReturnWithoutSignature(self):
 383         # https://bugs.freedesktop.org/show_bug.cgi?id=10174
 384         ret = self.iface.MultipleReturnWithoutSignature()
 385         self.assert_(not isinstance(ret, dbus.Struct), repr(ret))
 386         self.assertEquals(ret, ('abc', 123))
 387
 388     def testListExportedChildObjects(self):
 389         self.assert_(self.iface.TestListExportedChildObjects())
 390
 391     def testRemoveFromConnection(self):
 392         # https://bugs.freedesktop.org/show_bug.cgi?id=10457
 393         self.assert_(not self.iface.HasRemovableObject())
 394         self.assert_(self.iface.AddRemovableObject())
 395         self.assert_(self.iface.HasRemovableObject())
 396
 397         removable = self.bus.get_object(NAME, OBJECT + '/RemovableObject')
 398         iface = dbus.Interface(removable, IFACE)
 399         self.assert_(iface.IsThere())
 400         self.assert_(iface.RemoveSelf())
 401
 402         self.assert_(not self.iface.HasRemovableObject())
 403
 404         # and again...
 405         self.assert_(self.iface.AddRemovableObject())
 406         self.assert_(self.iface.HasRemovableObject())
 407         self.assert_(iface.IsThere())
 408         self.assert_(iface.RemoveSelf())
 409         self.assert_(not self.iface.HasRemovableObject())
 410
 411     def testFallbackObjectTrivial(self):
 412         obj = self.bus.get_object(NAME, OBJECT + '/Fallback')
 413         iface = dbus.Interface(obj, IFACE)
 414         path, unique_name = iface.TestPathAndConnKeywords()
 415         self.assertEquals(path, OBJECT + '/Fallback')
 416         #self.assertEquals(rel, '/Badger/Mushroom')
 417         self.assertEquals(unique_name, obj.bus_name)
 418
 419     def testFallbackObject(self):
 420         obj = self.bus.get_object(NAME, OBJECT + '/Fallback/Badger/Mushroom')
 421         iface = dbus.Interface(obj, IFACE)
 422         path, unique_name = iface.TestPathAndConnKeywords()
 423         self.assertEquals(path, OBJECT + '/Fallback/Badger/Mushroom')
 424         #self.assertEquals(rel, '/Badger/Mushroom')
 425         self.assertEquals(unique_name, obj.bus_name)
 426
 427     def testTimeoutSync(self):
 428         self.assert_(self.iface.BlockFor500ms(timeout=1.0) is None)
 429         self.assertRaises(dbus.DBusException,
 430                           lambda: self.iface.BlockFor500ms(timeout=0.25))
 431
 432     def testTimeoutAsyncClient(self):
 433         loop = gobject.MainLoop()
 434         passes = []
 435         fails = []
 436         def correctly_returned():
 437             passes.append('1000')
 438             if len(passes) + len(fails) >= 2:
 439                 loop.quit()
 440         def correctly_failed(exc):
 441             passes.append('250')
 442             if len(passes) + len(fails) >= 2:
 443                 loop.quit()
 444         def incorrectly_returned():
 445             fails.append('250')
 446             if len(passes) + len(fails) >= 2:
 447                 loop.quit()
 448         def incorrectly_failed(exc):
 449             fails.append('1000')
 450             if len(passes) + len(fails) >= 2:
 451                 loop.quit()
 452         self.iface.BlockFor500ms(timeout=1.0,
 453                                  reply_handler=correctly_returned,
 454                                  error_handler=incorrectly_failed)
 455         self.iface.BlockFor500ms(timeout=0.25,
 456                                  reply_handler=incorrectly_returned,
 457                                  error_handler=correctly_failed)
 458         loop.run()
 459         self.assertEquals(passes, ['250', '1000'])
 460         self.assertEquals(fails, [])
 461
 462     def testTimeoutAsyncService(self):
 463         self.assert_(self.iface.AsyncWait500ms(timeout=1.0) is None)
 464         self.assertRaises(dbus.DBusException,
 465                           lambda: self.iface.AsyncWait500ms(timeout=0.25))
 466
 467 """ Remove this for now
 468 class TestDBusPythonToGLibBindings(unittest.TestCase):
 469     def setUp(self):
 470         self.bus = dbus.SessionBus()
 471         self.remote_object = self.bus.get_object("org.freedesktop.DBus.TestSuiteGLibService", "/org/freedesktop/DBus/Tests/MyTestObject")
 472         self.iface = dbus.Interface(self.remote_object, "org.freedesktop.DBus.Tests.MyObject")
 473
 474     def testIntrospection(self):
 475         #test introspection
 476         print "\n********* Introspection Test ************"
 477         print self.remote_object.Introspect(dbus_interface="org.freedesktop.DBus.Introspectable")
 478         print "Introspection test passed"
 479         self.assert_(True)
 480
 481     def testCalls(self):
 482         print "\n********* Call Test ************"
 483         result =  self.iface.ManyArgs(1000, 'Hello GLib', 2)
 484         print result
 485         self.assert_(result == [2002.0, 'HELLO GLIB'])
 486
 487         arg0 = {"Dude": 1, "john": "palmieri", "python": 2.4}
 488         result = self.iface.ManyStringify(arg0)
 489         print result
 490
 491         print "Call test passed"
 492         self.assert_(True)
 493
 494     def testPythonTypes(self):
 495         print "\n********* Testing Python Types ***********"
 496
 497         for send_val in test_types_vals:
 498             print "Testing %s"% str(send_val)
 499             recv_val = self.iface.EchoVariant(send_val)
 500             self.assertEquals(send_val, recv_val.object)
 501 """
 502 if __name__ == '__main__':
 503     gobject.threads_init()
 504     dbus.glib.init_threads()
 505
 506     unittest.main()