bpo-40094: Add test.support.wait_process() (GH-19254)

Moreover, the following tests now check the child process exit code:

* test_os.PtyTests
* test_mailbox.test_lock_conflict()
* test_tempfile.test_process_awareness()
* test_uuid.testIssue8621()
* multiprocessing resource tracker tests

Author: vstinner

Doc/library/test.rst

@@ -825,6 +825,21 @@ The :mod:`test.support` module defines the following functions:
    target of the "as" clause, if there is one.
 
 
+.. function:: wait_process(pid, *, exitcode, timeout=None)
+
+   Wait until process *pid* completes and check that the process exit code is
+   *exitcode*.
+
+   Raise an :exc:`AssertionError` if the process exit code is not equal to
+   *exitcode*.
+
+   If the process runs longer than *timeout* seconds (:data:`SHORT_TIMEOUT` by
+   default), kill the process and raise an :exc:`AssertionError`. The timeout
+   feature is not available on Windows.
+
+   .. versionadded:: 3.9
+
+
 .. function:: wait_threads_exit(timeout=60.0)
 
    Context manager to wait until all threads created in the ``with`` statement

Lib/test/_test_multiprocessing.py

@@ -5124,7 +5124,7 @@ def check_resource_tracker_death(self, signum, should_die):
         pid = _resource_tracker._pid
         if pid is not None:
             os.kill(pid, signal.SIGKILL)
-            os.waitpid(pid, 0)
+            support.wait_process(pid, exitcode=-signal.SIGKILL)
         with warnings.catch_warnings():
             warnings.simplefilter("ignore")
             _resource_tracker.ensure_running()

Lib/test/fork_wait.py

@@ -44,16 +44,7 @@ def f(self, id):
                 pass
 
     def wait_impl(self, cpid):
-        for i in range(10):
-            # waitpid() shouldn't hang, but some of the buildbots seem to hang
-            # in the forking tests.  This is an attempt to fix the problem.
-            spid, status = os.waitpid(cpid, os.WNOHANG)
-            if spid == cpid:
-                break
-            time.sleep(2 * SHORTSLEEP)
-
-        self.assertEqual(spid, cpid)
-        self.assertEqual(status, 0, "cause = %d, exit = %d" % (status&0xff, status>>8))
+        support.wait_process(cpid, exitcode=0)
 
     def test_wait(self):
         for i in range(NUM_THREADS):

Lib/test/support/__init__.py

@@ -3400,3 +3400,62 @@ def __exit__(self, *exc_info):
         del self.exc_value
         del self.exc_traceback
         del self.thread
+
+
+def wait_process(pid, *, exitcode, timeout=None):
+    """
+    Wait until process pid completes and check that the process exit code is
+    exitcode.
+
+    Raise an AssertionError if the process exit code is not equal to exitcode.
+
+    If the process runs longer than timeout seconds (SHORT_TIMEOUT by default),
+    kill the process (if signal.SIGKILL is available) and raise an
+    AssertionError. The timeout feature is not available on Windows.
+    """
+    if os.name != "nt":
+        if timeout is None:
+            timeout = SHORT_TIMEOUT
+        t0 = time.monotonic()
+        deadline = t0 + timeout
+        sleep = 0.001
+        max_sleep = 0.1
+        while True:
+            pid2, status = os.waitpid(pid, os.WNOHANG)
+            if pid2 != 0:
+                break
+            # process is still running
+
+            dt = time.monotonic() - t0
+            if dt > SHORT_TIMEOUT:
+                try:
+                    os.kill(pid, signal.SIGKILL)
+                    os.waitpid(pid, 0)
+                except OSError:
+                    # Ignore errors like ChildProcessError or PermissionError
+                    pass
+
+                raise AssertionError(f"process {pid} is still running "
+                                     f"after {dt:.1f} seconds")
+
+            sleep = min(sleep * 2, max_sleep)
+            time.sleep(sleep)
+
+        if os.WIFEXITED(status):
+            exitcode2 = os.WEXITSTATUS(status)
+        elif os.WIFSIGNALED(status):
+            exitcode2 = -os.WTERMSIG(status)
+        else:
+            raise ValueError(f"invalid wait status: {status!r}")
+    else:
+        # Windows implementation
+        pid2, status = os.waitpid(pid, 0)
+        exitcode2 = (status >> 8)
+
+    if exitcode2 != exitcode:
+        raise AssertionError(f"process {pid} exited with code {exitcode2}, "
+                             f"but exit code {exitcode} is expected")
+
+    # sanity check: it should not fail in practice
+    if pid2 != pid:
+        raise AssertionError(f"pid {pid2} != pid {pid}")

Lib/test/test_builtin.py

@@ -25,6 +25,7 @@
 from textwrap import dedent
 from types import AsyncGeneratorType, FunctionType
 from operator import neg
+from test import support
 from test.support import (
     EnvironmentVarGuard, TESTFN, check_warnings, swap_attr, unlink,
     maybe_get_event_loop_policy)
@@ -1890,7 +1891,7 @@ def run_child(self, child, terminal_input):
         os.close(fd)
 
         # Wait until the child process completes
-        os.waitpid(pid, 0)
+        support.wait_process(pid, exitcode=0)
 
         return lines
 

Lib/test/test_logging.py

@@ -727,30 +727,19 @@ def lock_holder_thread_fn():
 
         locks_held__ready_to_fork.wait()
         pid = os.fork()
-        if pid == 0:  # Child.
+        if pid == 0:
+            # Child process
             try:
                 test_logger.info(r'Child process did not deadlock. \o/')
             finally:
                 os._exit(0)
-        else:  # Parent.
+        else:
+            # Parent process
             test_logger.info(r'Parent process returned from fork. \o/')
             fork_happened__release_locks_and_end_thread.set()
             lock_holder_thread.join()
-            start_time = time.monotonic()
-            while True:
-                test_logger.debug('Waiting for child process.')
-                waited_pid, status = os.waitpid(pid, os.WNOHANG)
-                if waited_pid == pid:
-                    break  # child process exited.
-                if time.monotonic() - start_time > 7:
-                    break  # so long? implies child deadlock.
-                time.sleep(0.05)
-            test_logger.debug('Done waiting.')
-            if waited_pid != pid:
-                os.kill(pid, signal.SIGKILL)
-                waited_pid, status = os.waitpid(pid, 0)
-                self.fail("child process deadlocked.")
-            self.assertEqual(status, 0, msg="child process error")
+
+            support.wait_process(pid, exitcode=0)
 
 
 class BadStream(object):

Lib/test/test_mailbox.py

@@ -1092,7 +1092,7 @@ def test_lock_conflict(self):
             # Signal the child it can now release the lock and exit.
             p.send(b'p')
             # Wait for child to exit.  Locking should now succeed.
-            exited_pid, status = os.waitpid(pid, 0)
+            support.wait_process(pid, exitcode=0)
 
         self._box.lock()
         self._box.unlock()

Lib/test/test_os.py

@@ -2792,8 +2792,7 @@ def test_waitpid(self):
         args = [sys.executable, '-c', 'pass']
         # Add an implicit test for PyUnicode_FSConverter().
         pid = os.spawnv(os.P_NOWAIT, FakePath(args[0]), args)
-        status = os.waitpid(pid, 0)
-        self.assertEqual(status, (pid, 0))
+        support.wait_process(pid, exitcode=0)
 
 
 class SpawnTests(unittest.TestCase):
@@ -2877,14 +2876,7 @@ def test_spawnvpe(self):
     def test_nowait(self):
         args = self.create_args()
         pid = os.spawnv(os.P_NOWAIT, args[0], args)
-        result = os.waitpid(pid, 0)
-        self.assertEqual(result[0], pid)
-        status = result[1]
-        if hasattr(os, 'WIFEXITED'):
-            self.assertTrue(os.WIFEXITED(status))
-            self.assertEqual(os.WEXITSTATUS(status), self.exitcode)
-        else:
-            self.assertEqual(status, self.exitcode << 8)
+        support.wait_process(pid, exitcode=self.exitcode)
 
     @requires_os_func('spawnve')
     def test_spawnve_bytes(self):

Lib/test/test_platform.py

@@ -236,9 +236,7 @@ def test_mac_ver_with_fork(self):
 
         else:
             # parent
-            cpid, sts = os.waitpid(pid, 0)
-            self.assertEqual(cpid, pid)
-            self.assertEqual(sts, 0)
+            support.wait_process(pid, exitcode=0)
 
     def test_libc_ver(self):
         # check that libc_ver(executable) doesn't raise an exception