[pygobject] Add Python implementation of Object.connect_data()
- From: Simon Feltman <sfeltman src gnome org>
- To: commits-list gnome org
- Cc:
- Subject: [pygobject] Add Python implementation of Object.connect_data()
- Date: Tue, 12 Aug 2014 05:53:56 +0000 (UTC)
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
[
Date Prev][
Date Next] [
Thread Prev][
Thread Next]
[
Thread Index]
[
Date Index]
[
Author Index]