tests/test_gdbus.py

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