subprocess was killed.) On Windows systems, the return value
contains the signed integer return code from the child process.
+ On Unix, :func:`waitstatus_to_exitcode` can be used to convert the ``close``
+ method result (exit status) into an exit code if it is not ``None``. On
+ Windows, the ``close`` method result is directly the exit code
+ (or ``None``).
+
This is implemented using :class:`subprocess.Popen`; see that class's
documentation for more powerful ways to manage and communicate with
subprocesses.
to using this function. See the :ref:`subprocess-replacements` section in
the :mod:`subprocess` documentation for some helpful recipes.
+ On Unix, :func:`waitstatus_to_exitcode` can be used to convert the result
+ (exit status) into an exit code. On Windows, the result is directly the exit
+ code.
+
.. audit-event:: os.system command os.system
.. availability:: Unix, Windows.
number is zero); the high bit of the low byte is set if a core file was
produced.
+ :func:`waitstatus_to_exitcode` can be used to convert the exit status into an
+ exit code.
+
.. availability:: Unix.
+ .. seealso::
+
+ :func:`waitpid` can be used to wait for the completion of a specific
+ child process and has more options.
+
.. function:: waitid(idtype, id, options)
Wait for the completion of one or more child processes.
id is known, not necessarily a child process. The :func:`spawn\* <spawnl>`
functions called with :const:`P_NOWAIT` return suitable process handles.
+ :func:`waitstatus_to_exitcode` can be used to convert the exit status into an
+ exit code.
+
.. versionchanged:: 3.5
If the system call is interrupted and the signal handler does not raise an
exception, the function now retries the system call instead of raising an
information. The option argument is the same as that provided to
:func:`waitpid` and :func:`wait4`.
+ :func:`waitstatus_to_exitcode` can be used to convert the exit status into an
+ exitcode.
+
.. availability:: Unix.
resource usage information. The arguments to :func:`wait4` are the same
as those provided to :func:`waitpid`.
+ :func:`waitstatus_to_exitcode` can be used to convert the exit status into an
+ exitcode.
+
.. availability:: Unix.
+.. function:: waitstatus_to_exitcode(status)
+
+ Convert a wait status to an exit code.
+
+ On Unix:
+
+ * If the process exited normally (if ``WIFEXITED(status)`` is true),
+ return the process exit status (return ``WEXITSTATUS(status)``):
+ result greater than or equal to 0.
+ * If the process was terminated by a signal (if ``WIFSIGNALED(status)`` is
+ true), return ``-signum`` where *signum* is the number of the signal that
+ caused the process to terminate (return ``-WTERMSIG(status)``):
+ result less than 0.
+ * Otherwise, raise a :exc:`ValueError`.
+
+ On Windows, return *status* shifted right by 8 bits.
+
+ On Unix, if the process is being traced or if :func:`waitpid` was called
+ with :data:`WUNTRACED` option, the caller must first check if
+ ``WIFSTOPPED(status)`` is true. This function must not be called if
+ ``WIFSTOPPED(status)`` is true.
+
+ .. seealso::
+
+ :func:`WIFEXITED`, :func:`WEXITSTATUS`, :func:`WIFSIGNALED`,
+ :func:`WTERMSIG`, :func:`WIFSTOPPED`, :func:`WSTOPSIG` functions.
+
+ .. versionadded:: 3.9
+
+
.. data:: WNOHANG
The option for :func:`waitpid` to return immediately if no child process status
*select* throws an error on your platform when passed three empty lists. This
is a bug, documented in `issue 26228 <https://bugs.python.org/issue26228>`_.
+ Return the exit status value from :func:`os.waitpid` on the child process.
+
+ :func:`waitstatus_to_exitcode` can be used to convert the exit status into
+ an exit code.
+
.. audit-event:: pty.spawn argv pty.spawn
.. versionchanged:: 3.4
available.
(Contributed by Victor Stinner in :issue:`39395`.)
+Add :func:`os.waitstatus_to_exitcode` function:
+convert a wait status to an exit code.
+(Contributed by Victor Stinner in :issue:`40094`.)
+
pathlib
-------
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:
else:
# Parent process
_, status = os.waitpid(pid, 0)
- self.returncode = _waitstatus_to_exitcode(status)
+ self.returncode = os.waitstatus_to_exitcode(status)
return self.returncode
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}")
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)
# 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):
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
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):
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")
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):
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:
## 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:
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):
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():
--- /dev/null
+Add :func:`os.waitstatus_to_exitcode` function:
+convert a wait status to an exit code.
#endif /* defined(MS_WINDOWS) */
+#if (defined(WIFEXITED) || defined(MS_WINDOWS))
+
+PyDoc_STRVAR(os_waitstatus_to_exitcode__doc__,
+"waitstatus_to_exitcode($module, /, status)\n"
+"--\n"
+"\n"
+"Convert a wait status to an exit code.\n"
+"\n"
+"On Unix:\n"
+"\n"
+"* If WIFEXITED(status) is true, return WEXITSTATUS(status).\n"
+"* If WIFSIGNALED(status) is true, return -WTERMSIG(status).\n"
+"* Otherwise, raise a ValueError.\n"
+"\n"
+"On Windows, return status shifted right by 8 bits.\n"
+"\n"
+"On Unix, if the process is being traced or if waitpid() was called with\n"
+"WUNTRACED option, the caller must first check if WIFSTOPPED(status) is true.\n"
+"This function must not be called if WIFSTOPPED(status) is true.");
+
+#define OS_WAITSTATUS_TO_EXITCODE_METHODDEF \
+ {"waitstatus_to_exitcode", (PyCFunction)(void(*)(void))os_waitstatus_to_exitcode, METH_FASTCALL|METH_KEYWORDS, os_waitstatus_to_exitcode__doc__},
+
+static PyObject *
+os_waitstatus_to_exitcode_impl(PyObject *module, int status);
+
+static PyObject *
+os_waitstatus_to_exitcode(PyObject *module, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames)
+{
+ PyObject *return_value = NULL;
+ static const char * const _keywords[] = {"status", NULL};
+ static _PyArg_Parser _parser = {NULL, _keywords, "waitstatus_to_exitcode", 0};
+ PyObject *argsbuf[1];
+ int status;
+
+ args = _PyArg_UnpackKeywords(args, nargs, NULL, kwnames, &_parser, 1, 1, 0, argsbuf);
+ if (!args) {
+ goto exit;
+ }
+ if (PyFloat_Check(args[0])) {
+ PyErr_SetString(PyExc_TypeError,
+ "integer argument expected, got float" );
+ goto exit;
+ }
+ status = _PyLong_AsInt(args[0]);
+ if (status == -1 && PyErr_Occurred()) {
+ goto exit;
+ }
+ return_value = os_waitstatus_to_exitcode_impl(module, status);
+
+exit:
+ return return_value;
+}
+
+#endif /* (defined(WIFEXITED) || defined(MS_WINDOWS)) */
+
#ifndef OS_TTYNAME_METHODDEF
#define OS_TTYNAME_METHODDEF
#endif /* !defined(OS_TTYNAME_METHODDEF) */
#ifndef OS__REMOVE_DLL_DIRECTORY_METHODDEF
#define OS__REMOVE_DLL_DIRECTORY_METHODDEF
#endif /* !defined(OS__REMOVE_DLL_DIRECTORY_METHODDEF) */
-/*[clinic end generated code: output=5d99f90cead7c0e1 input=a9049054013a1b77]*/
+
+#ifndef OS_WAITSTATUS_TO_EXITCODE_METHODDEF
+ #define OS_WAITSTATUS_TO_EXITCODE_METHODDEF
+#endif /* !defined(OS_WAITSTATUS_TO_EXITCODE_METHODDEF) */
+/*[clinic end generated code: output=4e28994a729eddf9 input=a9049054013a1b77]*/
#endif
+
+/* Only check if WIFEXITED is available: expect that it comes
+ with WEXITSTATUS, WIFSIGNALED, etc.
+
+ os.waitstatus_to_exitcode() is implemented in C and not in Python, so
+ subprocess can safely call it during late Python finalization without
+ risking that used os attributes were set to None by _PyImport_Cleanup(). */
+#if defined(WIFEXITED) || defined(MS_WINDOWS)
+/*[clinic input]
+os.waitstatus_to_exitcode
+
+ status: int
+
+Convert a wait status to an exit code.
+
+On Unix:
+
+* If WIFEXITED(status) is true, return WEXITSTATUS(status).
+* If WIFSIGNALED(status) is true, return -WTERMSIG(status).
+* Otherwise, raise a ValueError.
+
+On Windows, return status shifted right by 8 bits.
+
+On Unix, if the process is being traced or if waitpid() was called with
+WUNTRACED option, the caller must first check if WIFSTOPPED(status) is true.
+This function must not be called if WIFSTOPPED(status) is true.
+[clinic start generated code]*/
+
+static PyObject *
+os_waitstatus_to_exitcode_impl(PyObject *module, int status)
+/*[clinic end generated code: output=c7c2265731f79b7a input=edfa5ca5006276fb]*/
+{
+#ifndef MS_WINDOWS
+ WAIT_TYPE wait_status;
+ WAIT_STATUS_INT(wait_status) = status;
+ int exitcode;
+ if (WIFEXITED(wait_status)) {
+ exitcode = WEXITSTATUS(wait_status);
+ /* Sanity check to provide warranty on the function behavior.
+ It should not occur in practice */
+ if (exitcode < 0) {
+ PyErr_Format(PyExc_ValueError, "invalid WEXITSTATUS: %i", exitcode);
+ return NULL;
+ }
+ }
+ else if (WIFSIGNALED(wait_status)) {
+ int signum = WTERMSIG(wait_status);
+ /* Sanity check to provide warranty on the function behavior.
+ It should not occurs in practice */
+ if (signum <= 0) {
+ PyErr_Format(PyExc_ValueError, "invalid WTERMSIG: %i", signum);
+ return NULL;
+ }
+ exitcode = -signum;
+ } else if (WIFSTOPPED(wait_status)) {
+ /* Status only received if the process is being traced
+ or if waitpid() was called with WUNTRACED option. */
+ int signum = WSTOPSIG(wait_status);
+ PyErr_Format(PyExc_ValueError,
+ "process stopped by delivery of signal %i",
+ signum);
+ return NULL;
+ }
+ else {
+ PyErr_Format(PyExc_ValueError, "invalid wait status: %i", status);
+ return NULL;
+ }
+ return PyLong_FromLong(exitcode);
+#else
+ /* Windows implementation: see os.waitpid() implementation
+ which uses _cwait(). */
+ int exitcode = (status >> 8);
+ return PyLong_FromLong(exitcode);
+#endif
+}
+#endif
+
+
static PyMethodDef posix_methods[] = {
OS_STAT_METHODDEF
OS__ADD_DLL_DIRECTORY_METHODDEF
OS__REMOVE_DLL_DIRECTORY_METHODDEF
#endif
+ OS_WAITSTATUS_TO_EXITCODE_METHODDEF
{NULL, NULL} /* Sentinel */
};
import sys
import sysconfig
from glob import glob
-from _bootsubprocess import _waitstatus_to_exitcode as waitstatus_to_exitcode
try:
def run_command(cmd):
status = os.system(cmd)
- return waitstatus_to_exitcode(status)
+ return os.waitstatus_to_exitcode(status)
# Set common compiler and linker flags derived from the Makefile,
projects / thirdparty / Python / cpython.git / commitdiff
