bpo-40094: Add os.waitstatus_to_exitcode() (GH-19201)

Add os.waitstatus_to_exitcode() function to convert a wait status to an
exitcode.

Suggest waitstatus_to_exitcode() usage in the documentation when
appropriate.

Use waitstatus_to_exitcode() in:

* multiprocessing, os, subprocess and _bootsubprocess modules;
* test.support.wait_process();
* setup.py: run_command();
* and many tests.

11 files changed, 50 insertions, 58 deletions

diff --git a/Lib/_bootsubprocess.py b/Lib/_bootsubprocess.py
index 9c1912f..014782f 100644
--- a/Lib/_bootsubprocess.py
+++ b/Lib/_bootsubprocess.py
@@ -6,15 +6,6 @@ subprocess is unavailable. setup.py is not used on Windows.
 import os
 
 
-def _waitstatus_to_exitcode(status):
-    if os.WIFEXITED(status):
-        return os.WEXITSTATUS(status)
-    elif os.WIFSIGNALED(status):
-        return -os.WTERMSIG(status)
-    else:
-        raise ValueError(f"invalid wait status: {status!r}")
-
-
 # distutils.spawn used by distutils.command.build_ext
 # calls subprocess.Popen().wait()
 class Popen:
@@ -37,7 +28,7 @@ class Popen:
         else:
             # Parent process
             _, status = os.waitpid(pid, 0)
-            self.returncode = _waitstatus_to_exitcode(status)
+            self.returncode = os.waitstatus_to_exitcode(status)
 
         return self.returncode
 
@@ -87,7 +78,7 @@ def check_output(cmd, **kwargs):
     try:
         # system() spawns a shell
         status = os.system(cmd)
-        exitcode = _waitstatus_to_exitcode(status)
+        exitcode = os.waitstatus_to_exitcode(status)
         if exitcode:
             raise ValueError(f"Command {cmd!r} returned non-zero "
                              f"exit status {exitcode!r}")
diff --git a/Lib/multiprocessing/forkserver.py b/Lib/multiprocessing/forkserver.py
index 215ac39..22a911a 100644
--- a/Lib/multiprocessing/forkserver.py
+++ b/Lib/multiprocessing/forkserver.py
@@ -237,14 +237,8 @@ def main(listener_fd, alive_r, preload, main_path=None, sys_path=None):
                             break
                         child_w = pid_to_fd.pop(pid, None)
                         if child_w is not None:
-                            if os.WIFSIGNALED(sts):
-                                returncode = -os.WTERMSIG(sts)
-                            else:
-                                if not os.WIFEXITED(sts):
-                                    raise AssertionError(
-                                        "Child {0:n} status is {1:n}".format(
-                                            pid,sts))
-                                returncode = os.WEXITSTATUS(sts)
+                            returncode = os.waitstatus_to_exitcode(sts)
+
                             # Send exit code to client process
                             try:
                                 write_signed(child_w, returncode)
diff --git a/Lib/multiprocessing/popen_fork.py b/Lib/multiprocessing/popen_fork.py
index a65b06f..625981c 100644
--- a/Lib/multiprocessing/popen_fork.py
+++ b/Lib/multiprocessing/popen_fork.py
@@ -30,11 +30,7 @@ class Popen(object):
                 # e.errno == errno.ECHILD == 10
                 return None
             if pid == self.pid:
-                if os.WIFSIGNALED(sts):
-                    self.returncode = -os.WTERMSIG(sts)
-                else:
-                    assert os.WIFEXITED(sts), "Status is {:n}".format(sts)
-                    self.returncode = os.WEXITSTATUS(sts)
+                self.returncode = os.waitstatus_to_exitcode(sts)
         return self.returncode
 
     def wait(self, timeout=None):
diff --git a/Lib/os.py b/Lib/os.py
index 8459baa..8acd6f1 100644
--- a/Lib/os.py
+++ b/Lib/os.py
@@ -864,12 +864,8 @@ if _exists("fork") and not _exists("spawnv") and _exists("execv"):
                 wpid, sts = waitpid(pid, 0)
                 if WIFSTOPPED(sts):
                     continue
-                elif WIFSIGNALED(sts):
-                    return -WTERMSIG(sts)
-                elif WIFEXITED(sts):
-                    return WEXITSTATUS(sts)
-                else:
-                    raise OSError("Not stopped, signaled or exited???")
+
+                return waitstatus_to_exitcode(sts)
 
     def spawnv(mode, file, args):
         """spawnv(mode, file, args) -> integer
diff --git a/Lib/subprocess.py b/Lib/subprocess.py
index c8db387..1eeccea 100644
--- a/Lib/subprocess.py
+++ b/Lib/subprocess.py
@@ -1838,23 +1838,17 @@ class Popen(object):
                 raise child_exception_type(err_msg)
 
 
-        def _handle_exitstatus(self, sts, _WIFSIGNALED=os.WIFSIGNALED,
-                _WTERMSIG=os.WTERMSIG, _WIFEXITED=os.WIFEXITED,
-                _WEXITSTATUS=os.WEXITSTATUS, _WIFSTOPPED=os.WIFSTOPPED,
-                _WSTOPSIG=os.WSTOPSIG):
+        def _handle_exitstatus(self, sts,
+                               waitstatus_to_exitcode=os.waitstatus_to_exitcode,
+                               _WIFSTOPPED=os.WIFSTOPPED,
+                               _WSTOPSIG=os.WSTOPSIG):
             """All callers to this function MUST hold self._waitpid_lock."""
             # This method is called (indirectly) by __del__, so it cannot
             # refer to anything outside of its local scope.
-            if _WIFSIGNALED(sts):
-                self.returncode = -_WTERMSIG(sts)
-            elif _WIFEXITED(sts):
-                self.returncode = _WEXITSTATUS(sts)
-            elif _WIFSTOPPED(sts):
+            if _WIFSTOPPED(sts):
                 self.returncode = -_WSTOPSIG(sts)
             else:
-                # Should never happen
-                raise SubprocessError("Unknown child exit status!")
-
+                self.returncode = waitstatus_to_exitcode(sts)
 
         def _internal_poll(self, _deadstate=None, _waitpid=os.waitpid,
                 _WNOHANG=os.WNOHANG, _ECHILD=errno.ECHILD):
diff --git a/Lib/test/support/__init__.py b/Lib/test/support/__init__.py
index 7272d47..1f792d8 100644
--- a/Lib/test/support/__init__.py
+++ b/Lib/test/support/__init__.py
@@ -3442,18 +3442,11 @@ def wait_process(pid, *, exitcode, timeout=None):
 
             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)
 
+    exitcode2 = os.waitstatus_to_exitcode(status)
     if exitcode2 != exitcode:
         raise AssertionError(f"process {pid} exited with code {exitcode2}, "
                              f"but exit code {exitcode} is expected")
diff --git a/Lib/test/test_os.py b/Lib/test/test_os.py
index be85616..142cfea 100644
--- a/Lib/test/test_os.py
+++ b/Lib/test/test_os.py
@@ -2794,6 +2794,35 @@ class PidTests(unittest.TestCase):
         pid = os.spawnv(os.P_NOWAIT, FakePath(args[0]), args)
         support.wait_process(pid, exitcode=0)
 
+    def test_waitstatus_to_exitcode(self):
+        exitcode = 23
+        filename = support.TESTFN
+        self.addCleanup(support.unlink, filename)
+
+        with open(filename, "w") as fp:
+            print(f'import sys; sys.exit({exitcode})', file=fp)
+            fp.flush()
+        args = [sys.executable, filename]
+        pid = os.spawnv(os.P_NOWAIT, args[0], args)
+
+        pid2, status = os.waitpid(pid, 0)
+        self.assertEqual(os.waitstatus_to_exitcode(status), exitcode)
+        self.assertEqual(pid2, pid)
+
+    # Skip the test on Windows
+    @unittest.skipUnless(hasattr(signal, 'SIGKILL'), 'need signal.SIGKILL')
+    def test_waitstatus_to_exitcode_kill(self):
+        signum = signal.SIGKILL
+        args = [sys.executable, '-c',
+                f'import time; time.sleep({support.LONG_TIMEOUT})']
+        pid = os.spawnv(os.P_NOWAIT, args[0], args)
+
+        os.kill(pid, signum)
+
+        pid2, status = os.waitpid(pid, 0)
+        self.assertEqual(os.waitstatus_to_exitcode(status), -signum)
+        self.assertEqual(pid2, pid)
+
 
 class SpawnTests(unittest.TestCase):
     def create_args(self, *, with_env=False, use_bytes=False):
diff --git a/Lib/test/test_popen.py b/Lib/test/test_popen.py
index da01a87..ab1bc77 100644
--- a/Lib/test/test_popen.py
+++ b/Lib/test/test_popen.py
@@ -44,10 +44,11 @@ class PopenTest(unittest.TestCase):
 
     def test_return_code(self):
         self.assertEqual(os.popen("exit 0").close(), None)
+        status = os.popen("exit 42").close()
         if os.name == 'nt':
-            self.assertEqual(os.popen("exit 42").close(), 42)
+            self.assertEqual(status, 42)
         else:
-            self.assertEqual(os.popen("exit 42").close(), 42 << 8)
+            self.assertEqual(os.waitstatus_to_exitcode(status), 42)
 
     def test_contextmanager(self):
         with os.popen("echo hello") as f:
diff --git a/Lib/test/test_pty.py b/Lib/test/test_pty.py
index ce85f57..aa5c687 100644
--- a/Lib/test/test_pty.py
+++ b/Lib/test/test_pty.py
@@ -200,8 +200,8 @@ class PtyTest(unittest.TestCase):
             ##    raise TestFailed("Unexpected output from child: %r" % line)
 
             (pid, status) = os.waitpid(pid, 0)
-            res = status >> 8
-            debug("Child (%d) exited with status %d (%d)." % (pid, res, status))
+            res = os.waitstatus_to_exitcode(status)
+            debug("Child (%d) exited with code %d (status %d)." % (pid, res, status))
             if res == 1:
                 self.fail("Child raised an unexpected exception in os.setsid()")
             elif res == 2:
diff --git a/Lib/test/test_wait3.py b/Lib/test/test_wait3.py
index 6e06049..aa166ba 100644
--- a/Lib/test/test_wait3.py
+++ b/Lib/test/test_wait3.py
@@ -30,8 +30,7 @@ class Wait3Test(ForkWait):
             time.sleep(0.1)
 
         self.assertEqual(spid, cpid)
-        self.assertEqual(status, exitcode << 8,
-                         "cause = %d, exit = %d" % (status&0xff, status>>8))
+        self.assertEqual(os.waitstatus_to_exitcode(status), exitcode)
         self.assertTrue(rusage)
 
     def test_wait3_rusage_initialized(self):
diff --git a/Lib/test/test_wait4.py b/Lib/test/test_wait4.py
index 6c7ebcb..f8d5e13 100644
--- a/Lib/test/test_wait4.py
+++ b/Lib/test/test_wait4.py
@@ -29,8 +29,7 @@ class Wait4Test(ForkWait):
                 break
             time.sleep(0.1)
         self.assertEqual(spid, cpid)
-        self.assertEqual(status, exitcode << 8,
-                         "cause = %d, exit = %d" % (status&0xff, status>>8))
+        self.assertEqual(os.waitstatus_to_exitcode(status), exitcode)
         self.assertTrue(rusage)
 
 def tearDownModule():