diff options
Diffstat (limited to 'Lib/test/test_subprocess.py')
| -rw-r--r-- | Lib/test/test_subprocess.py | 116 |
1 files changed, 58 insertions, 58 deletions
diff --git a/Lib/test/test_subprocess.py b/Lib/test/test_subprocess.py index 074f86d..ef0da28 100644 --- a/Lib/test/test_subprocess.py +++ b/Lib/test/test_subprocess.py @@ -945,6 +945,64 @@ class POSIXProcessTestCase(BaseTestCase): self.assertEqual(0, p.returncode, "sigchild_ignore.py exited" " non-zero with this error:\n%s" % stderr) + def test_zombie_fast_process_del(self): + # Issue #12650: on Unix, if Popen.__del__() was called before the + # process exited, it wouldn't be added to subprocess._active, and would + # remain a zombie. + # spawn a Popen, and delete its reference before it exits + p = subprocess.Popen([sys.executable, "-c", + 'import sys, time;' + 'time.sleep(0.2)'], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE) + ident = id(p) + pid = p.pid + del p + # check that p is in the active processes list + self.assertIn(ident, [id(o) for o in subprocess._active]) + + # sleep a little to let the process exit, and create a new Popen: this + # should trigger the wait() of p + time.sleep(1) + with self.assertRaises(EnvironmentError) as c: + with subprocess.Popen(['nonexisting_i_hope'], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE) as proc: + pass + # p should have been wait()ed on, and removed from the _active list + self.assertRaises(OSError, os.waitpid, pid, 0) + self.assertNotIn(ident, [id(o) for o in subprocess._active]) + + def test_leak_fast_process_del_killed(self): + # Issue #12650: on Unix, if Popen.__del__() was called before the + # process exited, and the process got killed by a signal, it would never + # be removed from subprocess._active, which triggered a FD and memory + # leak. + # spawn a Popen, delete its reference and kill it + p = subprocess.Popen([sys.executable, "-c", + 'import time;' + 'time.sleep(3)'], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE) + ident = id(p) + pid = p.pid + del p + os.kill(pid, signal.SIGKILL) + # check that p is in the active processes list + self.assertIn(ident, [id(o) for o in subprocess._active]) + + # let some time for the process to exit, and create a new Popen: this + # should trigger the wait() of p + time.sleep(0.2) + with self.assertRaises(EnvironmentError) as c: + with subprocess.Popen(['nonexisting_i_hope'], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE) as proc: + pass + # p should have been wait()ed on, and removed from the _active list + self.assertRaises(OSError, os.waitpid, pid, 0) + self.assertNotIn(ident, [id(o) for o in subprocess._active]) + @unittest.skipUnless(mswindows, "Windows specific tests") class Win32ProcessTestCase(BaseTestCase): @@ -1080,64 +1138,6 @@ class HelperFunctionTests(unittest.TestCase): subprocess._eintr_retry_call(fake_os_func, 666)) self.assertEqual([(256, 999), (666,), (666,)], record_calls) - def test_zombie_fast_process_del(self): - # Issue #12650: on Unix, if Popen.__del__() was called before the - # process exited, it wouldn't be added to subprocess._active, and would - # remain a zombie. - # spawn a Popen, and delete its reference before it exits - p = subprocess.Popen([sys.executable, "-c", - 'import sys, time;' - 'time.sleep(0.2)'], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE) - ident = id(p) - pid = p.pid - del p - # check that p is in the active processes list - self.assertIn(ident, [id(o) for o in subprocess._active]) - - # sleep a little to let the process exit, and create a new Popen: this - # should trigger the wait() of p - time.sleep(1) - with self.assertRaises(EnvironmentError) as c: - with subprocess.Popen(['nonexisting_i_hope'], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE) as proc: - pass - # p should have been wait()ed on, and removed from the _active list - self.assertRaises(OSError, os.waitpid, pid, 0) - self.assertNotIn(ident, [id(o) for o in subprocess._active]) - - def test_leak_fast_process_del_killed(self): - # Issue #12650: on Unix, if Popen.__del__() was called before the - # process exited, and the process got killed by a signal, it would never - # be removed from subprocess._active, which triggered a FD and memory - # leak. - # spawn a Popen, delete its reference and kill it - p = subprocess.Popen([sys.executable, "-c", - 'import time;' - 'time.sleep(3)'], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE) - ident = id(p) - pid = p.pid - del p - os.kill(pid, signal.SIGKILL) - # check that p is in the active processes list - self.assertIn(ident, [id(o) for o in subprocess._active]) - - # let some time for the process to exit, and create a new Popen: this - # should trigger the wait() of p - time.sleep(0.2) - with self.assertRaises(EnvironmentError) as c: - with subprocess.Popen(['nonexisting_i_hope'], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE) as proc: - pass - # p should have been wait()ed on, and removed from the _active list - self.assertRaises(OSError, os.waitpid, pid, 0) - self.assertNotIn(ident, [id(o) for o in subprocess._active]) - @unittest.skipUnless(mswindows, "mswindows only") class CommandsWithSpaces (BaseTestCase): |