test/test-client.py

   1 #!/usr/bin/env python
   2 import sys
   3 import os
   4 import unittest
   5 import time
   6 import logging
   7
   8 builddir = os.environ["DBUS_TOP_BUILDDIR"]
   9 pydir = builddir
  10
  11 sys.path.insert(0, pydir)
  12 sys.path.insert(0, pydir + 'dbus')
  13
  14 import dbus
  15 import _dbus_bindings
  16 import gobject
  17 import dbus.glib
  18 import dbus.service
  19
  20
  21 logging.basicConfig()
  22
  23
  24 pkg = dbus.__file__
  25 if not pkg.startswith(pydir):
  26     raise Exception("DBus modules (%s) are not being picked up from the package"%pkg)
  27
  28 if not _dbus_bindings.__file__.startswith(pydir):
  29     raise Exception("DBus modules (%s) are not being picked up from the package"%_dbus_bindings.__file__)
  30
  31 test_types_vals = [1, 12323231, 3.14159265, 99999999.99,
  32                  "dude", "123", "What is all the fuss about?", "gob@gob.com",
  33                  u'\\u310c\\u310e\\u3114', u'\\u0413\\u0414\\u0415',
  34                  u'\\u2200software \\u2203crack', u'\\xf4\\xe5\\xe8',
  35                  [1,2,3], ["how", "are", "you"], [1.23,2.3], [1], ["Hello"],
  36                  (1,2,3), (1,), (1,"2",3), ("2", "what"), ("you", 1.2),
  37                  {1:"a", 2:"b"}, {"a":1, "b":2}, #{"a":(1,"B")},
  38                  {1:1.1, 2:2.2}, [[1,2,3],[2,3,4]], [["a","b"],["c","d"]],
  39                  True, False,
  40                  dbus.Int16(-10), dbus.UInt16(10), 'SENTINEL',
  41                  #([1,2,3],"c", 1.2, ["a","b","c"], {"a": (1,"v"), "b": (2,"d")})
  42                  ]
  43
  44 class TestDBusBindings(unittest.TestCase):
  45     def setUp(self):
  46         self.bus = dbus.SessionBus()
  47         self.remote_object = self.bus.get_object("org.freedesktop.DBus.TestSuitePythonService", "/org/freedesktop/DBus/TestSuitePythonObject")
  48         self.iface = dbus.Interface(self.remote_object, "org.freedesktop.DBus.TestSuiteInterface")
  49
  50     def testInterfaceKeyword(self):
  51         #test dbus_interface parameter
  52         print self.remote_object.Echo("dbus_interface on Proxy test Passed", dbus_interface = "org.freedesktop.DBus.TestSuiteInterface")
  53         print self.iface.Echo("dbus_interface on Interface test Passed", dbus_interface = "org.freedesktop.DBus.TestSuiteInterface")
  54         self.assert_(True)
  55
  56     def testIntrospection(self):
  57         #test introspection
  58         print "\n********* Introspection Test ************"
  59         print self.remote_object.Introspect(dbus_interface="org.freedesktop.DBus.Introspectable")
  60         print "Introspection test passed"
  61         self.assert_(True)
  62
  63     def testPythonTypes(self):
  64         #test sending python types and getting them back
  65         print "\n********* Testing Python Types ***********"
  66
  67         for send_val in test_types_vals:
  68             print "Testing %s"% str(send_val)
  69             recv_val = self.iface.Echo(send_val)
  70             self.assertEquals(send_val, recv_val.object)
  71
  72     def testBenchmarkIntrospect(self):
  73         print "\n********* Benchmark Introspect ************"
  74         a = time.time()
  75         print a
  76         print self.iface.GetComplexArray()
  77         b = time.time()
  78         print b
  79         print "Delta: %f" % (b - a)
  80         self.assert_(True)
  81
  82     def testAsyncCalls(self):
  83         #test sending python types and getting them back async
  84         print "\n********* Testing Async Calls ***********"
  85
  86
  87         main_loop = gobject.MainLoop()
  88         class async_check:
  89             def __init__(self, test_controler, expected_result, do_exit):
  90                 self.expected_result = expected_result
  91                 self.do_exit = do_exit
  92                 self.test_controler = test_controler
  93
  94             def callback(self, val):
  95                 try:
  96                     if self.do_exit:
  97                         main_loop.quit()
  98
  99                     self.test_controler.assertEquals(val.object, self.expected_result)
 100                 except Exception, e:
 101                     print "%s:\n%s" % (e.__class__, e)
 102
 103             def error_handler(self, error):
 104                 print error
 105                 if self.do_exit:
 106                     main_loop.quit()
 107
 108                 self.test_controler.assert_(False, '%s: %s' % (error.__class__,
 109                                                                error))
 110
 111         last_type = test_types_vals[-1]
 112         for send_val in test_types_vals:
 113             print "Testing %s"% str(send_val)
 114             check = async_check(self, send_val, last_type == send_val)
 115             recv_val = self.iface.Echo(send_val,
 116                                        reply_handler = check.callback,
 117                                        error_handler = check.error_handler)
 118
 119         main_loop.run()
 120
 121     def testStrictMarshalling(self):
 122         print "\n********* Testing strict return & signal marshalling ***********"
 123
 124         # these values are the same as in the server, and the
 125         # methods should only succeed when they are called with
 126         # the right value number, because they have out_signature
 127         # decorations, and return an unmatching type when called
 128         # with a different number
 129         values = ["", ("",""), ("","",""), [], {}, ["",""], ["","",""]]
 130         methods = [
 131                     (self.iface.ReturnOneString, 'SignalOneString', set([0]), set([0])),
 132                     (self.iface.ReturnTwoStrings, 'SignalTwoStrings', set([1, 5]), set([1])),
 133                     (self.iface.ReturnStruct, 'SignalStruct', set([1, 5]), set([1])),
 134                     # all of our test values are sequences so will marshall correctly into an array :P
 135                     (self.iface.ReturnArray, 'SignalArray', set(range(len(values))), set([3, 5, 6])),
 136                     (self.iface.ReturnDict, 'SignalDict', set([0, 3, 4]), set([4]))
 137                 ]
 138
 139         for (method, signal, success_values, return_values) in methods:
 140             print "\nTrying correct behaviour of", method._method_name
 141             for value in range(len(values)):
 142                 try:
 143                     ret = method(value)
 144                 except Exception, e:
 145                     print "%s(%r) raised %s: %s" % (method._method_name, values[value], e.__class__, e)
 146
 147                     # should fail if it tried to marshal the wrong type
 148                     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))
 149                 else:
 150                     print "%s(%r) returned %r" % (method._method_name, values[value], ret)
 151
 152                     # should only succeed if it's the right return type
 153                     self.assert_(value in success_values, "%s should fail when we ask it to return %r" % (method._method_name, values[value]))
 154
 155                     # check the value is right too :D
 156                     returns = map(lambda n: values[n], return_values)
 157                     self.assert_(ret in returns, "%s should return one of %r but it returned %r instead" % (method._method_name, returns, ret))
 158
 159             print "\nTrying correct emission of", signal
 160             for value in range(len(values)):
 161                 try:
 162                     self.iface.EmitSignal(signal, value)
 163                 except Exception, e:
 164                     print "EmitSignal(%s, %r) raised %s" % (signal, values[value], e.__class__)
 165
 166                     # should fail if it tried to marshal the wrong type
 167                     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))
 168                 else:
 169                     print "EmitSignal(%s, %r) appeared to succeed" % (signal, values[value])
 170
 171                     # should only succeed if it's the right return type
 172                     self.assert_(value in success_values, "EmitSignal(%s) should fail when we ask it to return %r" % (signal, values[value]))
 173
 174                     # FIXME: wait for the signal here
 175
 176         print
 177
 178     def testInheritance(self):
 179         print "\n********* Testing inheritance from dbus.method.Interface ***********"
 180         ret = self.iface.CheckInheritance()
 181         print "CheckInheritance returned %s" % ret
 182         self.assert_(ret, "overriding CheckInheritance from TestInterface failed")
 183
 184     def testAsyncMethods(self):
 185         print "\n********* Testing asynchronous method implementation *******"
 186         for (async, fail) in ((False, False), (False, True), (True, False), (True, True)):
 187             try:
 188                 val = ('a', 1, False, [1,2], {1:2})
 189                 print "calling AsynchronousMethod with %s %s %s" % (async, fail, val)
 190                 ret = self.iface.AsynchronousMethod(async, fail, val)
 191             except Exception, e:
 192                 self.assert_(fail, '%s: %s' % (e.__class__, e))
 193                 print "Expected failure: %s: %s" % (e.__class__, e)
 194             else:
 195                 self.assert_(not fail, 'Expected failure but succeeded?!')
 196                 self.assertEquals(val, ret.object)
 197
 198     def testBusInstanceCaching(self):
 199         print "\n********* Testing dbus.Bus instance sharing *********"
 200
 201         # unfortunately we can't test the system bus here
 202         # but the codepaths are the same
 203         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)):
 204             print "\nTesting %s:" % cls.__name__
 205
 206             share_cls = cls()
 207             share_type = dbus.Bus(bus_type=type)
 208             share_func = func()
 209
 210             private_cls = cls(private=True)
 211             private_type = dbus.Bus(bus_type=type, private=True)
 212             private_func = func(private=True)
 213
 214             print " - checking shared instances are the same..."
 215             self.assert_(share_cls == share_type, '%s should equal %s' % (share_cls, share_type))
 216             self.assert_(share_type == share_func, '%s should equal %s' % (share_type, share_func))
 217
 218             print " - checking private instances are distinct from the shared instance..."
 219             self.assert_(share_cls != private_cls, '%s should not equal %s' % (share_cls, private_cls))
 220             self.assert_(share_type != private_type, '%s should not equal %s' % (share_type, private_type))
 221             self.assert_(share_func != private_func, '%s should not equal %s' % (share_func, private_func))
 222
 223             print " - checking private instances are distinct from each other..."
 224             self.assert_(private_cls != private_type, '%s should not equal %s' % (private_cls, private_type))
 225             self.assert_(private_type != private_func, '%s should not equal %s' % (private_type, private_func))
 226             self.assert_(private_func != private_cls, '%s should not equal %s' % (private_func, private_cls))
 227
 228     def testSenderName(self):
 229         print '\n******** Testing sender name keyword ********'
 230         myself = self.iface.WhoAmI()
 231         print "I am", myself
 232
 233     def testBusNameCreation(self):
 234         print '\n******** Testing BusName creation ********'
 235         test = [('org.freedesktop.DBus.Python.TestName', True),
 236                 ('org.freedesktop.DBus.Python.TestName', True),
 237                 ('org.freedesktop.DBus.Python.InvalidName&^*%$', False)]
 238         # Do some more intelligent handling/testing of queueing vs success?
 239         # ('org.freedesktop.DBus.TestSuitePythonService', False)]
 240         # For some reason this actually succeeds
 241         # ('org.freedesktop.DBus', False)]
 242
 243         # make a method call to ensure the test service is active
 244         self.iface.Echo("foo")
 245
 246         names = {}
 247         for (name, succeed) in test:
 248             try:
 249                 print "requesting %s" % name
 250                 busname = dbus.service.BusName(name)
 251             except Exception, e:
 252                 print "%s:\n%s" % (e.__class__, e)
 253                 self.assert_(not succeed, 'did not expect registering bus name %s to fail' % name)
 254             else:
 255                 print busname
 256                 self.assert_(succeed, 'expected registering bus name %s to fail'% name)
 257                 if name in names:
 258                     self.assert_(names[name] == busname, 'got a new instance for same name %s' % name)
 259                     print "instance of %s re-used, good!" % name
 260                 else:
 261                     names[name] = busname
 262
 263                 del busname
 264
 265             print
 266
 267         del names
 268
 269         bus = dbus.Bus()
 270         ret = bus.name_has_owner('org.freedesktop.DBus.Python.TestName')
 271         self.assert_(not ret, 'deleting reference failed to release BusName org.freedesktop.DBus.Python.TestName')
 272
 273 """ Remove this for now
 274 class TestDBusPythonToGLibBindings(unittest.TestCase):
 275     def setUp(self):
 276         self.bus = dbus.SessionBus()
 277         self.remote_object = self.bus.get_object("org.freedesktop.DBus.TestSuiteGLibService", "/org/freedesktop/DBus/Tests/MyTestObject")
 278         self.iface = dbus.Interface(self.remote_object, "org.freedesktop.DBus.Tests.MyObject")
 279
 280     def testIntrospection(self):
 281         #test introspection
 282         print "\n********* Introspection Test ************"
 283         print self.remote_object.Introspect(dbus_interface="org.freedesktop.DBus.Introspectable")
 284         print "Introspection test passed"
 285         self.assert_(True)
 286
 287     def testCalls(self):
 288         print "\n********* Call Test ************"
 289         result =  self.iface.ManyArgs(1000, 'Hello GLib', 2)
 290         print result
 291         self.assert_(result == [2002.0, 'HELLO GLIB'])
 292
 293         arg0 = {"Dude": 1, "john": "palmieri", "python": 2.4}
 294         result = self.iface.ManyStringify(arg0)
 295         print result
 296
 297         print "Call test passed"
 298         self.assert_(True)
 299
 300     def testPythonTypes(self):
 301         print "\n********* Testing Python Types ***********"
 302
 303         for send_val in test_types_vals:
 304             print "Testing %s"% str(send_val)
 305             recv_val = self.iface.EchoVariant(send_val)
 306             self.assertEquals(send_val, recv_val.object)
 307 """
 308 if __name__ == '__main__':
 309     gobject.threads_init()
 310     dbus.glib.init_threads()
 311
 312     unittest.main()