mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-04 00:48:58 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

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

Browse source 
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.
This commit is contained in:
Victor Stinner 2020-04-01 18:49:29 +02:00 • committed by GitHub
parent 5dd836030e
commit 65a796e527
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
18 changed files with 258 additions and 61 deletions
Download patch file Download diff file Expand all files Collapse all files

13
Lib/_bootsubprocess.py
View file

@ -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}")

10
Lib/multiprocessing/forkserver.py
View file

@ -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)

6
Lib/multiprocessing/popen_fork.py
View file

@ -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):

8
Lib/os.py
View file

@ -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

18
Lib/subprocess.py
View file

@ -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):

9
Lib/test/support/__init__.py
View file

@ -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")

29
Lib/test/test_os.py
View file

@ -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):

5
Lib/test/test_popen.py
View file

@ -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:

4
Lib/test/test_pty.py
View file

@ -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:

3
Lib/test/test_wait3.py
View file

@ -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):

3
Lib/test/test_wait4.py
View file

@ -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():