tests/test_gdbus.py

   1 # -*- Mode: Python; py-indent-offset: 4 -*-
   2 # vim: tabstop=4 shiftwidth=4 expandtab
   3
   4 import unittest
   5
   6 import sys
   7 sys.path.insert(0, "../")
   8
   9 import gobject
  10 from gi.repository import GLib
  11 from gi.repository import Gio
  12
  13 class TestGDBusClient(unittest.TestCase):
  14     def setUp(self):
  15         self.bus = Gio.bus_get_sync(Gio.BusType.SESSION, None)
  16
  17         self.dbus_proxy = Gio.DBusProxy.new_sync(self.bus,
  18                 Gio.DBusProxyFlags.NONE, None,
  19                 'org.freedesktop.DBus',
  20                 '/org/freedesktop/DBus',
  21                 'org.freedesktop.DBus', None)
  22
  23     def test_native_calls_sync(self):
  24         result = self.dbus_proxy.call_sync('ListNames', None,
  25                 Gio.DBusCallFlags.NO_AUTO_START, 500, None)
  26         self.assertTrue(isinstance(result, GLib.Variant))
  27         result = result.unpack()[0] # result is always a tuple
  28         self.assertTrue(len(result) > 1)
  29         self.assertTrue('org.freedesktop.DBus' in result)
  30
  31         result = self.dbus_proxy.call_sync('GetNameOwner',
  32                 GLib.Variant('(s)', ('org.freedesktop.DBus',)),
  33                 Gio.DBusCallFlags.NO_AUTO_START, 500, None)
  34         self.assertTrue(isinstance(result, GLib.Variant))
  35         self.assertEqual(type(result.unpack()[0]), type(''))
  36
  37     def test_native_calls_sync_errors(self):
  38         # error case: invalid argument types
  39         try:
  40             self.dbus_proxy.call_sync('GetConnectionUnixProcessID', None,
  41                     Gio.DBusCallFlags.NO_AUTO_START, 500, None)
  42             self.fail('call with invalid arguments should raise an exception')
  43         except Exception:
  44             etype, e = sys.exc_info()[:2]
  45             self.assertTrue('InvalidArgs' in str(e))
  46
  47         # error case: invalid argument
  48         try:
  49             self.dbus_proxy.call_sync('GetConnectionUnixProcessID',
  50                     GLib.Variant('(s)', (' unknown',)),
  51                     Gio.DBusCallFlags.NO_AUTO_START, 500, None)
  52             self.fail('call with invalid arguments should raise an exception')
  53         except Exception:
  54             etype, e = sys.exc_info()[:2]
  55
  56             self.assertTrue('NameHasNoOwner' in str(e))
  57
  58         # error case: unknown method
  59         try:
  60             self.dbus_proxy.call_sync('UnknownMethod', None,
  61                     Gio.DBusCallFlags.NO_AUTO_START, 500, None)
  62             self.fail('call for unknown method should raise an exception')
  63         except Exception:
  64             etype, e = sys.exc_info()[:2]
  65
  66             self.assertTrue('UnknownMethod' in str(e))
  67
  68     def test_native_calls_async(self):
  69         def call_done(obj, result, user_data):
  70             user_data['result'] = obj.call_finish(result)
  71             user_data['main_loop'].quit()
  72
  73         main_loop = gobject.MainLoop()
  74         data = {'main_loop': main_loop}
  75         self.dbus_proxy.call('ListNames', None,
  76                 Gio.DBusCallFlags.NO_AUTO_START, 500, None,
  77                 call_done, data)
  78         main_loop.run()
  79
  80         self.assertTrue(isinstance(data['result'], GLib.Variant))
  81         result = data['result'].unpack()[0] # result is always a tuple
  82         self.assertTrue(len(result) > 1)
  83         self.assertTrue('org.freedesktop.DBus' in result)
  84
  85     def test_native_calls_async_errors(self):
  86         def call_done(obj, result, user_data):
  87             try:
  88                 obj.call_finish(result)
  89                 self.fail('call_finish() for unknown method should raise an exception')
  90             except Exception:
  91                 etype, e = sys.exc_info()[:2]
  92
  93                 self.assertTrue('UnknownMethod' in str(e))
  94             finally:
  95                 user_data['main_loop'].quit()
  96
  97         main_loop = gobject.MainLoop()
  98         data = {'main_loop': main_loop}
  99         self.dbus_proxy.call('UnknownMethod', None,
 100                 Gio.DBusCallFlags.NO_AUTO_START, 500, None, call_done, data)
 101         main_loop.run()
 102
 103     def test_python_calls_sync(self):
 104         # single value return tuples get unboxed to the one element
 105         result = self.dbus_proxy.ListNames('()')
 106         self.assertTrue(isinstance(result, list))
 107         self.assertTrue(len(result) > 1)
 108         self.assertTrue('org.freedesktop.DBus' in result)
 109
 110         result = self.dbus_proxy.GetNameOwner('(s)', 'org.freedesktop.DBus')
 111         self.assertEqual(type(result), type(''))
 112
 113         # empty return tuples get unboxed to None
 114         self.assertEqual(self.dbus_proxy.ReloadConfig('()'), None)
 115
 116         # multiple return values remain a tuple; unfortunately D-BUS itself
 117         # does not have any method returning multiple results, so try talking
 118         # to notification-daemon (and don't fail the test if it does not exist)
 119         try:
 120             notification_daemon = Gio.DBusProxy.new_sync(self.bus,
 121                     Gio.DBusProxyFlags.NONE, None,
 122                     'org.freedesktop.Notifications',
 123                     '/org/freedesktop/Notifications',
 124                     'org.freedesktop.Notifications', None)
 125             result = notification_daemon.GetServerInformation('()')
 126             self.assertTrue(isinstance(result, tuple))
 127             self.assertEqual(len(result), 4)
 128             for i in result:
 129                 self.assertEqual(type(i), type(''))
 130         except Exception:
 131             etype, e = sys.exc_info()[:2]
 132
 133             if 'Error.ServiceUnknown' not in str(e):
 134                 raise
 135
 136         # test keyword argument; timeout=0 will fail immediately
 137         try:
 138             self.dbus_proxy.GetConnectionUnixProcessID('()', timeout=0)
 139             self.fail('call with timeout=0 should raise an exception')
 140         except Exception:
 141             etype, e = sys.exc_info()[:2]
 142
 143             self.assertTrue('Timeout' in str(e), str(e))
 144
 145     def test_python_calls_sync_noargs(self):
 146         # methods without arguments don't need an explicit signature
 147         result = self.dbus_proxy.ListNames()
 148         self.assertTrue(isinstance(result, list))
 149         self.assertTrue(len(result) > 1)
 150         self.assertTrue('org.freedesktop.DBus' in result)
 151
 152     def test_python_calls_sync_errors(self):
 153         # error case: invalid argument types
 154         try:
 155             self.dbus_proxy.GetConnectionUnixProcessID('()')
 156             self.fail('call with invalid arguments should raise an exception')
 157         except Exception:
 158             etype, e = sys.exc_info()[:2]
 159
 160             self.assertTrue('InvalidArgs' in str(e), str(e))
 161
 162         try:
 163             self.dbus_proxy.GetConnectionUnixProcessID(None, 'foo')
 164             self.fail('call with None signature should raise an exception')
 165         except TypeError as e:
 166             self.assertTrue('signature' in str(e), str(e))
 167
 168     def test_python_calls_async(self):
 169         def call_done(obj, result, user_data):
 170             user_data['result'] = result
 171             user_data['main_loop'].quit()
 172
 173         main_loop = gobject.MainLoop()
 174         data = {'main_loop': main_loop}
 175         self.dbus_proxy.ListNames('()', result_handler=call_done,
 176                 user_data=data)
 177         main_loop.run()
 178
 179         result = data['result']
 180         self.assertEqual(type(result), type([]))
 181         self.assertTrue(len(result) > 1)
 182         self.assertTrue('org.freedesktop.DBus' in result)
 183
 184     def test_python_calls_async_error_result(self):
 185         # when only specifying a result handler, this will get the error
 186         def call_done(obj, result, user_data):
 187             user_data['result'] = result
 188             user_data['main_loop'].quit()
 189
 190         main_loop = gobject.MainLoop()
 191         data = {'main_loop': main_loop}
 192         self.dbus_proxy.ListNames('(s)', 'invalid_argument',
 193                 result_handler=call_done, user_data=data)
 194         main_loop.run()
 195
 196         self.assertTrue(isinstance(data['result'], Exception))
 197         self.assertTrue('InvalidArgs' in str(data['result']), str(data['result']))
 198
 199     def test_python_calls_async_error(self):
 200         # when specifying an explicit error handler, this will get the error
 201         def call_done(obj, result, user_data):
 202             user_data['main_loop'].quit()
 203             self.fail('result handler should not be called')
 204
 205         def call_error(obj, error, user_data):
 206             user_data['error'] = error
 207             user_data['main_loop'].quit()
 208
 209         main_loop = gobject.MainLoop()
 210         data = {'main_loop': main_loop}
 211         self.dbus_proxy.ListNames('(s)', 'invalid_argument',
 212                 result_handler=call_done, error_handler=call_error,
 213                 user_data=data)
 214         main_loop.run()
 215
 216         self.assertTrue(isinstance(data['error'], Exception))
 217         self.assertTrue('InvalidArgs' in str(data['error']), str(data['error']))
 218