[pygobject] [gi] Add Pythonic gdbus method invocation

[Martin Pitt <martinpitt src gnome org>]

commit 63a60bcc20e724f96ea8d565ee0cf13a228b72b9
Author: Martin Pitt <martin pitt ubuntu com>
Date:   Tue Feb 8 15:38:21 2011 +0100

    [gi] Add Pythonic gdbus method invocation
    
    Provide a wrapper for Gio.DBusProxy for calling D-Bus methods like on a normal
    Python object. This will handle the Python object <-> GVariant conversion, and
    optional keyword arguments for flags, timeout, and a result handler for
    asynchronous calls.
    
    Require specifying the input argument signature as the first argument of each
    method call. This ensures that the types of e. g. integers are always correct,
    and avoids having to do expensive D-Bus introspection for each call.
    
    https://bugzilla.gnome.org/show_bug.cgi?id=640181

 gi/overrides/Gio.py |   99 +++++++++++++++++++++++++++++++++++++++++++++++++
 tests/test_gdbus.py |  102 ++++++++++++++++++++++++++++++++++++++++++++++++++-
 2 files changed, 199 insertions(+), 2 deletions(-)
---
diff --git a/gi/overrides/Gio.py b/gi/overrides/Gio.py
index 78affa2..20343a2 100644
--- a/gi/overrides/Gio.py
+++ b/gi/overrides/Gio.py
@@ -97,3 +97,102 @@ class Settings(Gio.Settings):
 
 Settings = override(Settings)
 __all__.append('Settings')
+
+class _DBusProxyMethodCall:
+    '''Helper class to implement DBusProxy method calls.'''
+
+    def __init__(self, dbus_proxy, method_name):
+        self.dbus_proxy = dbus_proxy
+        self.method_name = method_name
+
+    def __async_result_handler(self, obj, result, user_data):
+        (result_callback, error_callback, real_user_data) = user_data
+        try:
+            ret = obj.call_finish(result)
+        except Exception as e:
+            # return exception as value
+            if error_callback:
+                error_callback(obj, e, real_user_data)
+            else:
+                result_callback(obj, e, real_user_data)
+            return
+
+        result_callback(obj, self._unpack_result(ret), real_user_data)
+
+    def __call__(self, signature, *args, **kwargs):
+        arg_variant = GLib.Variant(signature, tuple(args))
+
+        if 'result_handler' in kwargs:
+            # asynchronous call
+            user_data = (kwargs['result_handler'],
+                    kwargs.get('error_handler'), kwargs.get('user_data'))
+            self.dbus_proxy.call(self.method_name, arg_variant,
+                    kwargs.get('flags', 0), kwargs.get('timeout', -1), None,
+                    self.__async_result_handler, user_data)
+        else:
+            # synchronous call
+            result = self.dbus_proxy.call_sync(self.method_name, arg_variant,
+                    kwargs.get('flags', 0), kwargs.get('timeout', -1), None)
+            return self._unpack_result(result)
+
+    @classmethod
+    def _unpack_result(klass, result):
+        '''Convert a D-BUS return variant into an appropriate return value'''
+
+        result = result.unpack()
+
+        # to be compatible with standard Python behaviour, unbox
+        # single-element tuples and return None for empty result tuples
+        if len(result) == 1:
+            result = result[0]
+        elif len(result) == 0:
+            result = None
+
+        return result
+
+class DBusProxy(Gio.DBusProxy):
+    '''Provide comfortable and pythonic method calls.
+    
+    This marshalls the method arguments into a GVariant, invokes the
+    call_sync() method on the DBusProxy object, and unmarshalls the result
+    GVariant back into a Python tuple.
+
+    The first argument always needs to be the D-Bus signature tuple of the
+    method call. Example:
+
+      proxy = Gio.DBusProxy.new_sync(...)
+      result = proxy.MyMethod('(is)', 42, 'hello')
+
+    Optional keyword arguments:
+
+    - timeout: timeout for the call in milliseconds (default to D-Bus timeout)
+
+    - flags: Combination of Gio.DBusCallFlags.*
+
+    - result_handler: Do an asynchronous method call and invoke
+         result_handler(proxy_object, result, user_data) when it finishes.
+
+    - error_handler: If the asynchronous call raises an exception,
+      error_handler(proxy_object, exception, user_data) is called when it
+      finishes. If error_handler is not given, result_handler is called with
+      the exception object as result instead.
+    
+    - user_data: Optional user data to pass to result_handler for
+      asynchronous calls.
+
+    Example for asynchronous calls:
+
+      def mymethod_done(proxy, result, user_data):
+          if isinstance(result, Exception):
+              # handle error
+          else:
+              # do something with result
+
+      proxy.MyMethod('(is)', 42, 'hello',
+          result_handler=mymethod_done, user_data='data')
+    '''
+    def __getattr__(self, name):
+        return _DBusProxyMethodCall(self, name)
+
+DBusProxy = override(DBusProxy)
+__all__.append('DBusProxy')
diff --git a/tests/test_gdbus.py b/tests/test_gdbus.py
index 567433c..ade62d1 100644
--- a/tests/test_gdbus.py
+++ b/tests/test_gdbus.py
@@ -62,7 +62,7 @@ class TestGDBusClient(unittest.TestCase):
 
     def test_native_calls_async(self):
         def call_done(obj, result, user_data):
-            user_data['result'] = self.dbus_proxy.call_finish(result)
+            user_data['result'] = obj.call_finish(result)
             user_data['main_loop'].quit()
 
         main_loop = gobject.MainLoop()
@@ -80,7 +80,7 @@ class TestGDBusClient(unittest.TestCase):
     def test_native_calls_async_errors(self):
         def call_done(obj, result, user_data):
             try:
-                self.dbus_proxy.call_finish(result)
+                obj.call_finish(result)
                 self.fail('call_finish() for unknown method should raise an exception')
             except Exception as e:
                 self.assertTrue('UnknownMethod' in str(e))
@@ -92,3 +92,101 @@ class TestGDBusClient(unittest.TestCase):
         self.dbus_proxy.call('UnknownMethod', None,
                 Gio.DBusCallFlags.NO_AUTO_START, 500, None, call_done, data)
         main_loop.run()
+
+    def test_python_calls_sync(self):
+        # single value return tuples get unboxed to the one element
+        result = self.dbus_proxy.ListNames('()')
+        self.assertTrue(isinstance(result, list))
+        self.assertTrue(len(result) > 1)
+        self.assertTrue('org.freedesktop.DBus' in result)
+
+        result = self.dbus_proxy.GetNameOwner('(s)', 'org.freedesktop.DBus')
+        self.assertEqual(type(result), type(''))
+
+        # empty return tuples get unboxed to None
+        self.assertEqual(self.dbus_proxy.ReloadConfig('()'), None)
+
+        # multiple return values remain a tuple; unfortunately D-BUS itself
+        # does not have any method returning multiple results, so try talking
+        # to notification-daemon (and don't fail the test if it does not exist)
+        try:
+            notification_daemon = Gio.DBusProxy.new_sync(self.bus,
+                    Gio.DBusProxyFlags.NONE, None,
+                    'org.freedesktop.Notifications',
+                    '/org/freedesktop/Notifications',
+                    'org.freedesktop.Notifications', None)
+            result = notification_daemon.GetServerInformation('()')
+            self.assertTrue(isinstance(result, tuple))
+            self.assertEqual(len(result), 4)
+            for i in result:
+                self.assertEqual(type(i), type(''))
+        except Exception as e:
+            if 'Error.ServiceUnknown' not in str(e):
+                raise
+
+        # test keyword argument; timeout=0 will fail immediately
+        try:
+            self.dbus_proxy.GetConnectionUnixProcessID('()', timeout=0)
+            self.fail('call with timeout=0 should raise an exception')
+        except Exception as e:
+            self.assertTrue('Timeout' in str(e), str(e))
+
+    def test_python_calls_sync_errors(self):
+        # error case: invalid argument types
+        try:
+            self.dbus_proxy.GetConnectionUnixProcessID('()')
+            self.fail('call with invalid arguments should raise an exception')
+        except Exception as e:
+            self.assertTrue('InvalidArgs' in str(e), str(e))
+
+    def test_python_calls_async(self):
+        def call_done(obj, result, user_data):
+            user_data['result'] = result
+            user_data['main_loop'].quit()
+
+        main_loop = gobject.MainLoop()
+        data = {'main_loop': main_loop}
+        self.dbus_proxy.ListNames('()', result_handler=call_done,
+                user_data=data)
+        main_loop.run()
+
+        result = data['result']
+        self.assertEqual(type(result), type([]))
+        self.assertTrue(len(result) > 1)
+        self.assertTrue('org.freedesktop.DBus' in result)
+
+    def test_python_calls_async_error_result(self):
+        # when only specifying a result handler, this will get the error
+        def call_done(obj, result, user_data):
+            user_data['result'] = result
+            user_data['main_loop'].quit()
+
+        main_loop = gobject.MainLoop()
+        data = {'main_loop': main_loop}
+        self.dbus_proxy.ListNames('(s)', 'invalid_argument',
+                result_handler=call_done, user_data=data)
+        main_loop.run()
+
+        self.assertTrue(isinstance(data['result'], Exception))
+        self.assertTrue('InvalidArgs' in str(data['result']), str(data['result']))
+
+    def test_python_calls_async_error(self):
+        # when specifying an explicit error handler, this will get the error
+        def call_done(obj, result, user_data):
+            user_data['main_loop'].quit()
+            self.fail('result handler should not be called')
+
+        def call_error(obj, error, user_data):
+            user_data['error'] = error
+            user_data['main_loop'].quit()
+
+        main_loop = gobject.MainLoop()
+        data = {'main_loop': main_loop}
+        self.dbus_proxy.ListNames('(s)', 'invalid_argument',
+                result_handler=call_done, error_handler=call_error,
+                user_data=data)
+        main_loop.run()
+
+        self.assertTrue(isinstance(data['error'], Exception))
+        self.assertTrue('InvalidArgs' in str(data['error']), str(data['error']))
+