mirror of
https://github.com/python/cpython.git
synced 2025-08-04 00:48:58 +00:00
bpo-32270: Don't close stdin/out/err in pass_fds (GH-6242)
When subprocess.Popen() stdin= stdout= or stderr= handles are specified
and appear in pass_fds=, don't close the original fds after dup'ing them.
This implementation and unittest primarily came from @izbyshev (see the PR)
See also b89b52f284
This also removes the old manual p2cread, c2pwrite, and errwrite closing logic
as inheritable flags and _close_open_fds takes care of that properly today without special treatment.
This code is within child_exec() where it is the only thread so there is no
race condition between the dup and _Py_set_inheritable_async_safe call.
This commit is contained in:
parent
880d42a3b2
commit
ce34410b8b
3 changed files with 46 additions and 10 deletions
|
@ -2529,6 +2529,36 @@ class POSIXProcessTestCase(BaseTestCase):
|
|||
self.assertEqual(os.get_inheritable(inheritable), True)
|
||||
self.assertEqual(os.get_inheritable(non_inheritable), False)
|
||||
|
||||
|
||||
# bpo-32270: Ensure that descriptors specified in pass_fds
|
||||
# are inherited even if they are used in redirections.
|
||||
# Contributed by @izbyshev.
|
||||
def test_pass_fds_redirected(self):
|
||||
"""Regression test for https://bugs.python.org/issue32270."""
|
||||
fd_status = support.findfile("fd_status.py", subdir="subprocessdata")
|
||||
pass_fds = []
|
||||
for _ in range(2):
|
||||
fd = os.open(os.devnull, os.O_RDWR)
|
||||
self.addCleanup(os.close, fd)
|
||||
pass_fds.append(fd)
|
||||
|
||||
stdout_r, stdout_w = os.pipe()
|
||||
self.addCleanup(os.close, stdout_r)
|
||||
self.addCleanup(os.close, stdout_w)
|
||||
pass_fds.insert(1, stdout_w)
|
||||
|
||||
with subprocess.Popen([sys.executable, fd_status],
|
||||
stdin=pass_fds[0],
|
||||
stdout=pass_fds[1],
|
||||
stderr=pass_fds[2],
|
||||
close_fds=True,
|
||||
pass_fds=pass_fds):
|
||||
output = os.read(stdout_r, 1024)
|
||||
fds = {int(num) for num in output.split(b',')}
|
||||
|
||||
self.assertEqual(fds, {0, 1, 2} | frozenset(pass_fds), f"output={output!a}")
|
||||
|
||||
|
||||
def test_stdout_stdin_are_single_inout_fd(self):
|
||||
with io.open(os.devnull, "r+") as inout:
|
||||
p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"],
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue