test/python/test-client.py

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