[pygobject] Add Python implementation of Object.connect_data()

[Simon Feltman <sfeltman src gnome org>]

commit 581acc4c56be127b3a724df504bb46a40959fdd9
Author: Simon Feltman <sfeltman src gnome org>
Date:   Mon Aug 11 21:21:42 2014 -0700

    Add Python implementation of Object.connect_data()
    
    Add GObject.Object.connect_data() which takes an optional "connect_flags"
    keyword argument accepting GObject.ConnectFlags enum values. This is
    for supporting user data swapping (ConnectFlags.SWAPPED).
    
    https://bugzilla.gnome.org/show_bug.cgi?id=701843

 gi/overrides/GObject.py |   37 ++++++++++++++++
 tests/test_signal.py    |  109 +++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 146 insertions(+), 0 deletions(-)
---
diff --git a/gi/overrides/GObject.py b/gi/overrides/GObject.py
index 33f5007..f884af5 100644
--- a/gi/overrides/GObject.py
+++ b/gi/overrides/GObject.py
@@ -609,6 +609,43 @@ class Object(GObjectModule.Object):
         super(Object, self).freeze_notify()
         return _FreezeNotifyManager(self)
 
+    def connect_data(self, detailed_signal, handler, *data, **kwargs):
+        """Connect a callback to the given signal with optional user data.
+
+        :param str detailed_signal:
+            A detailed signal to connect to.
+        :param callable handler:
+            Callback handler to connect to the signal.
+        :param *data:
+            Variable data which is passed through to the signal handler.
+        :param GObject.ConnectFlags connect_flags:
+            Flags used for connection options.
+        :returns:
+            A signal id which can be used with disconnect.
+        """
+        flags = kwargs.get('connect_flags', 0)
+        if flags & GObjectModule.ConnectFlags.AFTER:
+            connect_func = _gobject.GObject.connect_after
+        else:
+            connect_func = _gobject.GObject.connect
+
+        if flags & GObjectModule.ConnectFlags.SWAPPED:
+            if len(data) != 1:
+                raise ValueError('Using GObject.ConnectFlags.SWAPPED requires exactly '
+                                 'one argument for user data, got: %s' % [data])
+
+            def new_handler(obj, *args):
+                # Swap obj with the last element in args which will be the user
+                # data passed to the connect function.
+                args = list(args)
+                swap = args.pop()
+                args = args + [obj]
+                return handler(swap, *args)
+        else:
+            new_handler = handler
+
+        return connect_func(self, detailed_signal, new_handler, *data)
+
     #
     # Aliases
     #
diff --git a/tests/test_signal.py b/tests/test_signal.py
index c18c05e..7b718cb 100644
--- a/tests/test_signal.py
+++ b/tests/test_signal.py
@@ -850,6 +850,115 @@ class TestSignalConnectors(unittest.TestCase):
         self.assertEqual(self.value, 3)
 
 
+class _ConnectDataTestBase(object):
+    # Notes:
+    #  - self.Object is overridden in sub-classes.
+    #  - Numeric suffixes indicate the number of user data args passed in.
+    Object = None
+
+    def run_connect_test(self, emit_args, user_data, flags=0):
+        obj = self.Object()
+        callback_args = []
+
+        def callback(*args):
+            callback_args.append(args)
+            return 0
+
+        obj.connect_data('sig-with-int64-prop', callback, connect_flags=flags, *user_data)
+        obj.emit('sig-with-int64-prop', *emit_args)
+        self.assertEqual(len(callback_args), 1)
+        return callback_args[0]
+
+    def test_0(self):
+        obj, value = self.run_connect_test([GObject.G_MAXINT64], user_data=[])
+        self.assertIsInstance(obj, self.Object)
+        self.assertEqual(value, GObject.G_MAXINT64)
+
+    def test_1(self):
+        obj, value, data = self.run_connect_test([GObject.G_MAXINT64],
+                                                 user_data=['mydata'])
+        self.assertIsInstance(obj, self.Object)
+        self.assertEqual(value, GObject.G_MAXINT64)
+        self.assertEqual(data, 'mydata')
+
+    def test_after_0(self):
+        obj, value = self.run_connect_test([GObject.G_MAXINT64],
+                                           user_data=[],
+                                           flags=GObject.ConnectFlags.AFTER)
+        self.assertIsInstance(obj, self.Object)
+        self.assertEqual(value, GObject.G_MAXINT64)
+
+    def test_after_1(self):
+        obj, value, data = self.run_connect_test([GObject.G_MAXINT64],
+                                                 user_data=['mydata'],
+                                                 flags=GObject.ConnectFlags.AFTER)
+        self.assertIsInstance(obj, self.Object)
+        self.assertEqual(value, GObject.G_MAXINT64)
+        self.assertEqual(data, 'mydata')
+
+    def test_swaped_0(self):
+        # Swapped only works with a single user data argument.
+        with self.assertRaises(ValueError):
+            self.run_connect_test([GObject.G_MAXINT64],
+                                  user_data=[],
+                                  flags=GObject.ConnectFlags.SWAPPED)
+
+    def test_swaped_1(self):
+        # Notice obj and data are reversed in the return.
+        data, value, obj = self.run_connect_test([GObject.G_MAXINT64],
+                                                 user_data=['mydata'],
+                                                 flags=GObject.ConnectFlags.SWAPPED)
+        self.assertIsInstance(obj, self.Object)
+        self.assertEqual(value, GObject.G_MAXINT64)
+        self.assertEqual(data, 'mydata')
+
+    def test_swaped_2(self):
+        # Swapped only works with a single user data argument.
+        with self.assertRaises(ValueError):
+            self.run_connect_test([GObject.G_MAXINT64],
+                                  user_data=[1, 2],
+                                  flags=GObject.ConnectFlags.SWAPPED)
+
+    def test_after_and_swapped_0(self):
+        # Swapped only works with a single user data argument.
+        with self.assertRaises(ValueError):
+            self.run_connect_test([GObject.G_MAXINT64],
+                                  user_data=[],
+                                  flags=GObject.ConnectFlags.AFTER | GObject.ConnectFlags.SWAPPED)
+
+    def test_after_and_swapped_1(self):
+        # Notice obj and data are reversed in the return.
+        data, value, obj = self.run_connect_test([GObject.G_MAXINT64],
+                                                 user_data=['mydata'],
+                                                 flags=GObject.ConnectFlags.AFTER | 
GObject.ConnectFlags.SWAPPED)
+        self.assertIsInstance(obj, self.Object)
+        self.assertEqual(value, GObject.G_MAXINT64)
+        self.assertEqual(data, 'mydata')
+
+    def test_after_and_swapped_2(self):
+        # Swapped only works with a single user data argument.
+        with self.assertRaises(ValueError):
+            self.run_connect_test([GObject.G_MAXINT64],
+                                  user_data=[],
+                                  flags=GObject.ConnectFlags.AFTER | GObject.ConnectFlags.SWAPPED)
+
+
+class TestConnectDataNonIntrospected(unittest.TestCase, _ConnectDataTestBase):
+    # This tests connect_data with non-introspected signals
+    # (created in Python in this case).
+    class Object(GObject.Object):
+        test = GObject.Signal()
+        sig_with_int64_prop = GObject.Signal(return_type=GObject.TYPE_INT64,
+                                             arg_types=[GObject.TYPE_INT64],
+                                             flags=GObject.SignalFlags.RUN_LAST)
+
+
+ unittest skipUnless(has_cairo, 'built without cairo support')
+class TestConnectDataIntrospected(unittest.TestCase, _ConnectDataTestBase):
+    # This tests connect_data with introspected signals brought in from Regress.
+    Object = Regress.TestObj
+
+
 class TestInstallSignals(unittest.TestCase):
     # These tests only test how signalhelper.install_signals works
     # with the __gsignals__ dict and therefore does not need to use