bpo-31173: Rewrite WSTOPSIG test of test_subprocess (#3055) (#3070)

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 7b7c6dcfff)
This commit is contained in:
Victor Stinner 2017-08-11 02:36:30 +02:00 committed by GitHub
parent 270c3c62ed
commit bc69d00288
2 changed files with 40 additions and 32 deletions

View File

@ -29,6 +29,11 @@ try:
except ImportError:
threading = None
try:
import _testcapi
except ImportError:
_testcapi = None
if support.PGO:
raise unittest.SkipTest("test is not helpful for PGO")
@ -2567,42 +2572,24 @@ class POSIXProcessTestCase(BaseTestCase):
proc.communicate(timeout=999)
mock_proc_stdin.close.assert_called_once_with()
@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')
args = [sys.executable, '-c', 'pass']
proc = subprocess.Popen(args)
code = textwrap.dedent(f"""
import ctypes
import faulthandler
from test.support import SuppressCrashReport
# Wait until the real process completes to avoid zombie process
pid = proc.pid
pid, status = os.waitpid(pid, 0)
self.assertEqual(status, 0)
libc = ctypes.CDLL({libc_name!r})
libc.ptrace({PTRACE_TRACEME}, 0, 0)
""")
status = _testcapi.W_STOPCODE(3)
with mock.patch('subprocess.os.waitpid', return_value=(pid, status)):
returncode = proc.wait()
child = subprocess.Popen([sys.executable, '-c', code])
if child.wait() != 0:
raise unittest.SkipTest('ptrace() failed - unable to test')
code += textwrap.dedent(f"""
with SuppressCrashReport():
# Crash the process
faulthandler._sigsegv()
""")
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.
self.assertEqual(returncode, -3)
@unittest.skipUnless(mswindows, "Windows specific tests")

View File

@ -18,6 +18,10 @@
# include <winsock2.h> /* struct timeval */
#endif
#ifdef HAVE_SYS_WAIT_H
#include <sys/wait.h> /* For W_STOPCODE */
#endif
#ifdef WITH_THREAD
#include "pythread.h"
#endif /* WITH_THREAD */
@ -4148,6 +4152,20 @@ raise_SIGINT_then_send_None(PyObject *self, PyObject *args)
}
#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},
{"raise_memoryerror", (PyCFunction)raise_memoryerror, METH_NOARGS},
@ -4355,6 +4373,9 @@ static PyMethodDef TestMethods[] = {
{"pyobject_fastcalldict", test_pyobject_fastcalldict, METH_VARARGS},
{"pyobject_fastcallkeywords", test_pyobject_fastcallkeywords, METH_VARARGS},
{"raise_SIGINT_then_send_None", raise_SIGINT_then_send_None, METH_VARARGS},
#ifdef W_STOPCODE
{"W_STOPCODE", py_w_stopcode, METH_VARARGS},
#endif
{NULL, NULL} /* sentinel */
};