[Python-checkins] bpo-31173: Rewrite WSTOPSIG test of test_subprocess (#3055) (#3071)
Victor Stinner
webhook-mailer at python.org
Fri Aug 11 08:38:43 EDT 2017
https://github.com/python/cpython/commit/4dea06531ece28dffc1452de2694fb22e99b45f9
commit: 4dea06531ece28dffc1452de2694fb22e99b45f9
branch: 2.7
author: Victor Stinner <victor.stinner at gmail.com>
committer: GitHub <noreply at github.com>
date: 2017-08-11T14:38:37+02:00
summary:
bpo-31173: Rewrite WSTOPSIG test of test_subprocess (#3055) (#3071)
The current test_child_terminated_in_stopped_state() function test
creates a child process which calls ptrace(PTRACE_TRACEME, 0, 0) and
then crash (SIGSEGV). The problem is that calling os.waitpid() in the
parent process is not enough to close the process: the child process
remains alive and so the unit test leaks a child process in a
strange state. Closing the child process requires non-trivial code,
maybe platform specific.
Remove the functional test and replaces it with an unit test which
mocks os.waitpid() using a new _testcapi.W_STOPCODE() function to
test the WIFSTOPPED() path.
(cherry picked from commit 7b7c6dcfff6a35333988a3c74c895ed19dff2e09)
files:
M Lib/test/test_subprocess.py
M Modules/_testcapimodule.c
diff --git a/Lib/test/test_subprocess.py b/Lib/test/test_subprocess.py
index 77633376eb5..ce3372354cf 100644
--- a/Lib/test/test_subprocess.py
+++ b/Lib/test/test_subprocess.py
@@ -28,6 +28,11 @@
except ImportError:
threading = None
+try:
+ import _testcapi
+except ImportError:
+ _testcapi = None
+
mswindows = (sys.platform == "win32")
#
@@ -1265,40 +1270,28 @@ def test_pipe_cloexec(self):
self.assertEqual(p2.returncode, 0, "Unexpected error: " + repr(stderr))
- @unittest.skipIf(not ctypes, 'ctypes module required')
- @unittest.skipIf(not sys.executable, 'Test requires sys.executable')
- def test_child_terminated_in_stopped_state(self):
+ @unittest.skipUnless(_testcapi is not None
+ and hasattr(_testcapi, 'W_STOPCODE'),
+ 'need _testcapi.W_STOPCODE')
+ def test_stopped(self):
"""Test wait() behavior when waitpid returns WIFSTOPPED; issue29335."""
- PTRACE_TRACEME = 0 # From glibc and MacOS (PT_TRACE_ME).
- libc_name = ctypes.util.find_library('c')
- libc = ctypes.CDLL(libc_name)
- if not hasattr(libc, 'ptrace'):
- raise unittest.SkipTest('ptrace() required')
-
- code = textwrap.dedent("""
- import ctypes
- from test.support import _crash_python
-
- libc = ctypes.CDLL({libc_name!r})
- libc.ptrace({PTRACE_TRACEME}, 0, 0)
- """.format(libc_name=libc_name, PTRACE_TRACEME=PTRACE_TRACEME))
-
- child = subprocess.Popen([sys.executable, '-c', code])
- if child.wait() != 0:
- raise unittest.SkipTest('ptrace() failed - unable to test')
-
- code += textwrap.dedent("""
- # Crash the process
- _crash_python()
- """)
- child = subprocess.Popen([sys.executable, '-c', code])
- try:
- returncode = child.wait()
- except:
- child.kill() # Clean up the hung stopped process.
- raise
- self.assertNotEqual(0, returncode)
- self.assertLess(returncode, 0) # signal death, likely SIGSEGV.
+ args = [sys.executable, '-c', 'pass']
+ proc = subprocess.Popen(args)
+
+ # Wait until the real process completes to avoid zombie process
+ pid = proc.pid
+ pid, status = os.waitpid(pid, 0)
+ self.assertEqual(status, 0)
+
+ status = _testcapi.W_STOPCODE(3)
+
+ def mock_waitpid(pid, flags):
+ return (pid, status)
+
+ with test_support.swap_attr(os, 'waitpid', mock_waitpid):
+ returncode = proc.wait()
+
+ self.assertEqual(returncode, -3)
@unittest.skipUnless(mswindows, "Windows specific tests")
diff --git a/Modules/_testcapimodule.c b/Modules/_testcapimodule.c
index a68d839a7f7..a5e3f07667d 100644
--- a/Modules/_testcapimodule.c
+++ b/Modules/_testcapimodule.c
@@ -15,6 +15,10 @@
# include <crtdbg.h>
#endif
+#ifdef HAVE_SYS_WAIT_H
+#include <sys/wait.h> /* For W_STOPCODE */
+#endif
+
#ifdef WITH_THREAD
#include "pythread.h"
#endif /* WITH_THREAD */
@@ -2523,6 +2527,7 @@ msvcrt_CrtSetReportMode(PyObject* self, PyObject *args)
return PyInt_FromLong(res);
}
+
static PyObject*
msvcrt_CrtSetReportFile(PyObject* self, PyObject *args)
{
@@ -2540,6 +2545,20 @@ msvcrt_CrtSetReportFile(PyObject* self, PyObject *args)
#endif
+#ifdef W_STOPCODE
+static PyObject*
+py_w_stopcode(PyObject *self, PyObject *args)
+{
+ int sig, status;
+ if (!PyArg_ParseTuple(args, "i", &sig)) {
+ return NULL;
+ }
+ status = W_STOPCODE(sig);
+ return PyLong_FromLong(status);
+}
+#endif
+
+
static PyMethodDef TestMethods[] = {
{"raise_exception", raise_exception, METH_VARARGS},
{"set_errno", set_errno, METH_VARARGS},
@@ -2656,6 +2675,9 @@ static PyMethodDef TestMethods[] = {
{"CrtSetReportMode", (PyCFunction)msvcrt_CrtSetReportMode, METH_VARARGS},
{"CrtSetReportFile", (PyCFunction)msvcrt_CrtSetReportFile, METH_VARARGS},
#endif
+#ifdef W_STOPCODE
+ {"W_STOPCODE", py_w_stopcode, METH_VARARGS},
+#endif
{NULL, NULL} /* sentinel */
};
More information about the Python-checkins
mailing list