test/cross-test-client.py

   1 import sys
   2 from sets import Set
   3 from time import sleep
   4 import logging
   5
   6 import gobject
   7
   8 from dbus import SessionBus, Interface, Array, Byte, Double, Variant, Boolean, ByteArray, Int32
   9 from dbus.service import BusName
  10 import dbus.glib
  11
  12 from crosstest import CROSS_TEST_PATH, CROSS_TEST_BUS_NAME,\
  13                       INTERFACE_SINGLE_TESTS, INTERFACE_TESTS,\
  14                       INTERFACE_SIGNAL_TESTS, INTERFACE_CALLBACK_TESTS,\
  15                       SignalTestsImpl
  16
  17
  18 logging.basicConfig()
  19
  20
  21 class Client(SignalTestsImpl):
  22     fail_id = 0
  23     expected = Set()
  24
  25     def quit(self):
  26         for x in self.expected:
  27             self.fail_id += 1
  28             print "<%s> fail %d" % (x, self.fail_id)
  29             print "report %d: reply to %s didn't arrive" % (self.fail_id, x)
  30         sys.stderr.write("CLIENT: asking server to Exit\n")
  31         Interface(self.obj, INTERFACE_TESTS).Exit(reply_handler=self.quit_reply_handler, error_handler=self.quit_error_handler)
  32         # if the server doesn't reply we'll just exit anyway
  33         gobject.timeout_add(1000, lambda: (loop.quit(), False)[1])
  34
  35     def quit_reply_handler(self):
  36         sys.stderr.write("CLIENT: server says it will exit\n")
  37         loop.quit()
  38
  39     def quit_error_handler(self, e):
  40         sys.stderr.write("CLIENT: error telling server to quit: %s %s\n"
  41                          % (e.__class__, e))
  42         loop.quit()
  43
  44     @dbus.service.method(INTERFACE_CALLBACK_TESTS, 'qd')
  45     def Response(self, input1, input2):
  46         sys.stderr.write("CLIENT: signal/callback: Response received (%r,%r)\n" % (input1, input2))
  47         self.expected.discard('%s.Trigger' % INTERFACE_SIGNAL_TESTS)
  48         if (input1, input2) != (42, 23):
  49             self.fail_id += 1
  50             print "<%s.Trigger> fail %d" % (INTERFACE_SIGNAL_TESTS, self.fail_id)
  51             print ("report %d: expected (42,23), got %r"
  52                    % (self.fail_id, (input1, input2)))
  53         else:
  54             print "<%s.Trigger> pass" % INTERFACE_SIGNAL_TESTS
  55         self.quit()
  56
  57     def assert_method_eq(self, interface, ret, member, *args):
  58         if_obj = Interface(self.obj, interface)
  59         method = getattr(if_obj, member)
  60         try:
  61             real_ret = method(*args)
  62         except Exception, e:
  63             self.fail_id += 1
  64             print "<%s.%s> fail %d" % (interface, member, self.fail_id)
  65             print ("report %d: %s.%s%r: expected %r, raised %r \"%s\""
  66                    % (self.fail_id, interface, member, args, ret, e, e))
  67             __import__('traceback').print_exc()
  68             return
  69         if real_ret != ret:
  70             self.fail_id += 1
  71             print "<%s.%s> fail %d" % (interface, member, self.fail_id)
  72             print ("report %d: %s.%s%r: expected %r, got %r"
  73                    % (self.fail_id, interface, member, args, ret, real_ret))
  74             return
  75         print "<%s.%s> pass" % (interface, member)
  76
  77
  78     def triggered_cb(self, param, sender_path):
  79         sys.stderr.write("CLIENT: method/signal: Triggered(%r) by %r\n"
  80                          % (param, sender_path))
  81         self.expected.discard('%s.Trigger' % INTERFACE_TESTS)
  82         if sender_path != '/Where/Ever':
  83             self.fail_id += 1
  84             print "<%s.Trigger> fail %d" % (INTERFACE_TESTS, self.fail_id)
  85             print ("report %d: expected signal from /Where/Ever, got %r"
  86                    % (self.fail_id, sender_path))
  87         elif param != 42:
  88             self.fail_id += 1
  89             print "<%s.Trigger> fail %d" % (INTERFACE_TESTS, self.fail_id)
  90             print ("report %d: expected signal param 42, got %r"
  91                    % (self.fail_id, parameter))
  92         else:
  93             print "<%s.Trigger> pass" % INTERFACE_TESTS
  94
  95     def trigger_returned_cb(self):
  96         sys.stderr.write('CLIENT: method/signal: Trigger() returned\n')
  97         # Callback tests
  98         sys.stderr.write("CLIENT: signal/callback: Emitting signal to trigger callback\n")
  99         self.expected.add('%s.Trigger' % INTERFACE_SIGNAL_TESTS)
 100         self.Trigger(42, 23)
 101         sys.stderr.write("CLIENT: signal/callback: Emitting signal returned\n")
 102
 103     def run_client(self):
 104         bus = SessionBus()
 105         obj = bus.get_object(CROSS_TEST_BUS_NAME, CROSS_TEST_PATH)
 106         self.obj = obj
 107
 108         self.run_synchronous_tests(obj)
 109
 110         # Signal tests
 111         sys.stderr.write("CLIENT: Binding signal handler for Triggered\n")
 112         # FIXME: doesn't seem to work when going via the Interface method
 113         # FIXME: should be possible to ask the proxy object for its
 114         # bus name
 115         bus.add_signal_receiver(self.triggered_cb, 'Triggered',
 116                                 INTERFACE_SIGNAL_TESTS,
 117                                 CROSS_TEST_BUS_NAME,
 118                                 path_keyword='sender_path')
 119         sys.stderr.write("CLIENT: method/signal: Triggering signal\n")
 120         self.expected.add('%s.Trigger' % INTERFACE_TESTS)
 121         Interface(obj, INTERFACE_TESTS).Trigger('/Where/Ever', 42, reply_handler=self.trigger_returned_cb, error_handler=self.trigger_error_handler)
 122
 123     def trigger_error_handler(self, e):
 124         sys.stderr.write("CLIENT: method/signal: %s %s\n", e.__class__, e)
 125         Interface(self.obj, INTERFACE_TESTS).Exit()
 126         self.quit()
 127
 128     def run_synchronous_tests(self, obj):
 129         # "Single tests"
 130         self.assert_method_eq(INTERFACE_SINGLE_TESTS, 6, 'Sum', [1, 2, 3])
 131         self.assert_method_eq(INTERFACE_SINGLE_TESTS, 6, 'Sum', ['\x01', '\x02', '\x03'])
 132         self.assert_method_eq(INTERFACE_SINGLE_TESTS, 6, 'Sum', [Byte(1), Byte(2), Byte(3)])
 133
 134         # Main tests
 135         self.assert_method_eq(INTERFACE_TESTS, Variant(u'foo', 's'), 'Identity', 'foo')
 136         self.assert_method_eq(INTERFACE_TESTS, Variant(Byte(42)), 'Identity', Byte(42))
 137         self.assert_method_eq(INTERFACE_TESTS, Variant(Byte(42)), 'Identity', Variant(Byte(42)))
 138         self.assert_method_eq(INTERFACE_TESTS, Variant(Variant(Variant(Variant(Variant(Variant(Variant(Byte(42)))))))), 'Identity', Variant(Variant(Variant(Variant(Variant(Variant(Variant(Byte(42)))))))))
 139         self.assert_method_eq(INTERFACE_TESTS, Variant(42.5), 'Identity', Double(42.5))
 140         self.assert_method_eq(INTERFACE_TESTS, Variant(-42.5), 'Identity', -42.5)
 141         self.assert_method_eq(INTERFACE_TESTS, Byte(0x42), 'IdentityByte', '\x42')
 142         self.assert_method_eq(INTERFACE_TESTS, Byte(42), 'IdentityByte', Byte(42))
 143         self.assert_method_eq(INTERFACE_TESTS, True, 'IdentityBool', 42)
 144         self.assert_method_eq(INTERFACE_TESTS, True, 'IdentityBool', Boolean(42))
 145         self.assert_method_eq(INTERFACE_TESTS, 42, 'IdentityInt16', 42)
 146         self.assert_method_eq(INTERFACE_TESTS, 42, 'IdentityUInt16', 42)
 147         self.assert_method_eq(INTERFACE_TESTS, 42, 'IdentityInt32', 42)
 148         self.assert_method_eq(INTERFACE_TESTS, 42, 'IdentityUInt32', 42)
 149         self.assert_method_eq(INTERFACE_TESTS, 42, 'IdentityInt64', 42)
 150         self.assert_method_eq(INTERFACE_TESTS, 42, 'IdentityUInt64', 42)
 151         self.assert_method_eq(INTERFACE_TESTS, 42.3, 'IdentityDouble', 42.3)
 152         self.assert_method_eq(INTERFACE_TESTS, u'\xa9', 'IdentityString', u'\xa9')
 153         self.assert_method_eq(INTERFACE_TESTS, [Variant(Byte('\x01')),Variant(Byte('\x02')),Variant(Byte('\x03'))], 'IdentityArray', ByteArray('\x01\x02\x03'))
 154         self.assert_method_eq(INTERFACE_TESTS, [Variant(Int32(1)),Variant(Int32(2)),Variant(Int32(3))], 'IdentityArray', [1,2,3])
 155         self.assert_method_eq(INTERFACE_TESTS, [Variant(u'a'),Variant(u'b'),Variant(u'c')], 'IdentityArray', ['a','b','c'])
 156         self.assert_method_eq(INTERFACE_TESTS, ['\x01','\x02','\x03'], 'IdentityByteArray', ByteArray('\x01\x02\x03'))
 157         self.assert_method_eq(INTERFACE_TESTS, ['\x01','\x02','\x03'], 'IdentityByteArray', ['\x01', '\x02', '\x03'])
 158         self.assert_method_eq(INTERFACE_TESTS, [False,True,True], 'IdentityBoolArray', [0,1,2])
 159         self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityInt16Array', [1,2,3])
 160         self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityUInt16Array', [1,2,3])
 161         self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityInt32Array', [1,2,3])
 162         self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityUInt32Array', [1,2,3])
 163         self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityInt64Array', [1,2,3])
 164         self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityUInt64Array', [1,2,3])
 165         self.assert_method_eq(INTERFACE_TESTS, [1.0,2.5,3.1], 'IdentityDoubleArray', [1,2.5,3.1])
 166         self.assert_method_eq(INTERFACE_TESTS, ['a','b','c'], 'IdentityStringArray', ['a','b','c'])
 167         self.assert_method_eq(INTERFACE_TESTS, 6, 'Sum', [1,2,3])
 168         self.assert_method_eq(INTERFACE_TESTS, {'fps': ['unreal', 'quake'], 'rts': ['warcraft']}, 'InvertMapping', {'unreal': 'fps', 'quake': 'fps', 'warcraft': 'rts'})
 169
 170         self.assert_method_eq(INTERFACE_TESTS, ('a', 1, 2), 'DeStruct', ('a', 1, 2))
 171         self.assert_method_eq(INTERFACE_TESTS, [Variant('x')], 'Primitize', [Variant(Variant(['x']))])
 172         self.assert_method_eq(INTERFACE_TESTS, [Variant('x'), Variant(Byte(1)), Variant(Byte(2))], 'Primitize', [Variant([Variant(['x']), Array([Byte(1), Byte(2)])])])
 173         self.assert_method_eq(INTERFACE_TESTS, False, 'Invert', 42)
 174         self.assert_method_eq(INTERFACE_TESTS, True, 'Invert', 0)
 175
 176
 177 if __name__ == '__main__':
 178     # FIXME: should be possible to export objects without a bus name
 179     bus_name = BusName('com.example.Argh')
 180     client = Client(bus_name, '/Client')
 181     gobject.idle_add(client.run_client)
 182
 183     loop = gobject.MainLoop()
 184     sys.stderr.write("CLIENT: running...\n")
 185     loop.run()
 186     sys.stderr.write("CLIENT: main loop exited.\n")