Refactor GLib.child_watch_add to make it more testable

Break the argument munging code into a separate function which
can be tested in isolation of adding a child watch. Update tests
to reflect this. Add additional failing test which specify
all args as keywords which we eventually need to support for
consistency with the rest of PyGObject.

https://bugzilla.gnome.org/show_bug.cgi?id=640812

2 files changed, 56 insertions, 57 deletions

diff --git a/gi/overrides/GLib.py b/gi/overrides/GLib.py
index f4b1ef57..b2b22e7e 100644
--- a/gi/overrides/GLib.py
+++ b/gi/overrides/GLib.py
@@ -808,13 +808,13 @@ PollFD = override(PollFD)
 __all__.append('PollFD')
 
 
-# The real GLib API is child_watch_add(priority, pid, callback, data).
-# The old static bindings had the following API which we still need to support
-# for a while:
-#   child_watch_add(pid, callback, data=None, priority=GLib.PRIORITY_DEFAULT)
-# and the usual "call without user_data", in which case the callback does not
-# get an user_data either.
-def child_watch_add(priority_or_pid, pid_or_callback, *args, **kwargs):
+# The GI GLib API uses g_child_watch_add_full renamed to g_child_watch_add with
+# a signature of (priority, pid, callback, data).
+# Prior to PyGObject 3.8, this function was statically bound with an API closer to the
+# non-full version with a signature of: (pid, callback, data=None, priority=GLib.PRIORITY_DEFAULT)
+# We need to support this until we are okay with breaking API in a way which is
+# not backwards compatible.
+def _child_watch_add_get_args(priority_or_pid, pid_or_callback, *args, **kwargs):
     _unspecified = object()
 
     if callable(pid_or_callback):
@@ -851,7 +851,16 @@ def child_watch_add(priority_or_pid, pid_or_callback, *args, **kwargs):
     else:
         func = callback
 
-    return GLib.child_watch_add(priority, pid, func, user_data)
+    return priority, pid, func, user_data
+
+# we need this to be accessible for unit testing
+__all__.append('_child_watch_add_get_args')
+
+
+def child_watch_add(*args, **kwargs):
+    """child_watch_add(priority, pid, function, *data)"""
+    priority, pid, function, data = _child_watch_add_get_args(*args, **kwargs)
+    return GLib.child_watch_add(priority, pid, function, data)
 
 __all__.append('child_watch_add')
 
diff --git a/tests/test_subprocess.py b/tests/test_subprocess.py
index ef4c35ee..cf5ef041 100644
--- a/tests/test_subprocess.py
+++ b/tests/test_subprocess.py
@@ -12,68 +12,58 @@ from gi import PyGIDeprecationWarning
 class TestProcess(unittest.TestCase):
 
     def test_deprecated_child_watch_no_data(self):
-        def cb(pid, status):
-            self.status = status
-            self.loop.quit()
-
-        self.status = None
-        self.loop = GLib.MainLoop()
-        argv = [sys.executable, '-c', 'import sys']
-        pid, stdin, stdout, stderr = GLib.spawn_async(
-            argv, flags=GLib.SpawnFlags.DO_NOT_REAP_CHILD)
-        pid.close()
+        cb = lambda pid, status: None
+        pid = object()
         with warnings.catch_warnings(record=True) as w:
             warnings.simplefilter('always')
-            GLib.child_watch_add(pid, cb)
+            res = GLib._child_watch_add_get_args(pid, cb)
             self.assertTrue(issubclass(w[0].category, PyGIDeprecationWarning))
-        self.loop.run()
-        self.assertEqual(self.status, 0)
 
-    def test_deprecated_child_watch_data_priority(self):
-        def cb(pid, status, data):
-            self.data = data
-            self.status = status
-            self.loop.quit()
+        self.assertEqual(len(res), 4)
+        self.assertEqual(res[0], GLib.PRIORITY_DEFAULT)
+        self.assertEqual(res[1], pid)
+        self.assertTrue(callable(cb))
+        self.assertEqual(res[3], None)
 
-        self.status = None
-        self.data = None
-        self.loop = GLib.MainLoop()
-        argv = [sys.executable, '-c', 'import sys']
-        pid, stdin, stdout, stderr = GLib.spawn_async(
-            argv, flags=GLib.SpawnFlags.DO_NOT_REAP_CHILD)
-        pid.close()
+    def test_deprecated_child_watch_data_priority(self):
+        cb = lambda pid, status: None
+        pid = object()
         with warnings.catch_warnings(record=True) as w:
             warnings.simplefilter('always')
-            id = GLib.child_watch_add(pid, cb, 12345, GLib.PRIORITY_HIGH)
+            res = GLib._child_watch_add_get_args(pid, cb, 12345, GLib.PRIORITY_HIGH)
             self.assertTrue(issubclass(w[0].category, PyGIDeprecationWarning))
-        self.assertEqual(self.loop.get_context().find_source_by_id(id).priority,
-                         GLib.PRIORITY_HIGH)
-        self.loop.run()
-        self.assertEqual(self.data, 12345)
-        self.assertEqual(self.status, 0)
 
-    def test_deprecated_child_watch_data_priority_kwargs(self):
-        def cb(pid, status, data):
-            self.data = data
-            self.status = status
-            self.loop.quit()
+        self.assertEqual(len(res), 4)
+        self.assertEqual(res[0], GLib.PRIORITY_HIGH)
+        self.assertEqual(res[1], pid)
+        self.assertEqual(res[2], cb)
+        self.assertEqual(res[3], 12345)
 
-        self.status = None
-        self.data = None
-        self.loop = GLib.MainLoop()
-        argv = [sys.executable, '-c', 'import sys']
-        pid, stdin, stdout, stderr = GLib.spawn_async(
-            argv, flags=GLib.SpawnFlags.DO_NOT_REAP_CHILD)
-        pid.close()
+    def test_deprecated_child_watch_data_priority_kwargs(self):
+        cb = lambda pid, status: None
+        pid = object()
         with warnings.catch_warnings(record=True) as w:
             warnings.simplefilter('always')
-            id = GLib.child_watch_add(pid, cb, priority=GLib.PRIORITY_HIGH, data=12345)
+            res = GLib._child_watch_add_get_args(pid, cb, priority=GLib.PRIORITY_HIGH, data=12345)
             self.assertTrue(issubclass(w[0].category, PyGIDeprecationWarning))
-        self.assertEqual(self.loop.get_context().find_source_by_id(id).priority,
-                         GLib.PRIORITY_HIGH)
-        self.loop.run()
-        self.assertEqual(self.data, 12345)
-        self.assertEqual(self.status, 0)
+
+        self.assertEqual(len(res), 4)
+        self.assertEqual(res[0], GLib.PRIORITY_HIGH)
+        self.assertEqual(res[1], pid)
+        self.assertEqual(res[2], cb)
+        self.assertEqual(res[3], 12345)
+
+    @unittest.expectedFailure  # using keyword args is fully supported by PyGObject machinery
+    def test_child_watch_all_kwargs(self):
+        cb = lambda pid, status: None
+        pid = object()
+
+        res = GLib._child_watch_add_get_args(priority=GLib.PRIORITY_HIGH, pid=pid, function=cb, data=12345)
+        self.assertEqual(len(res), 4)
+        self.assertEqual(res[0], GLib.PRIORITY_HIGH)
+        self.assertEqual(res[1], pid)
+        self.assertEqual(res[2], cb)
+        self.assertEqual(res[3], 12345)
 
     def test_child_watch_no_data(self):
         def cb(pid, status):