Issue #12607: In subprocess, fix issue where if stdin, stdout or stderr is
given as a low fd, it gets overwritten.
diff --git a/Lib/test/test_subprocess.py b/Lib/test/test_subprocess.py
index 3440b45..8382c72 100644
--- a/Lib/test/test_subprocess.py
+++ b/Lib/test/test_subprocess.py
@@ -1106,6 +1106,64 @@
for fd in temp_fds:
os.close(fd)
+ def check_swap_fds(self, stdin_no, stdout_no, stderr_no):
+ # open up some temporary files
+ temps = [mkstemp() for i in range(3)]
+ temp_fds = [fd for fd, fname in temps]
+ try:
+ # unlink the files -- we won't need to reopen them
+ for fd, fname in temps:
+ os.unlink(fname)
+
+ # save a copy of the standard file descriptors
+ saved_fds = [os.dup(fd) for fd in range(3)]
+ try:
+ # duplicate the temp files over the standard fd's 0, 1, 2
+ for fd, temp_fd in enumerate(temp_fds):
+ os.dup2(temp_fd, fd)
+
+ # write some data to what will become stdin, and rewind
+ os.write(stdin_no, b"STDIN")
+ os.lseek(stdin_no, 0, 0)
+
+ # now use those files in the given order, so that subprocess
+ # has to rearrange them in the child
+ p = subprocess.Popen([sys.executable, "-c",
+ 'import sys; got = sys.stdin.read();'
+ 'sys.stdout.write("got %s"%got); sys.stderr.write("err")'],
+ stdin=stdin_no,
+ stdout=stdout_no,
+ stderr=stderr_no)
+ p.wait()
+
+ for fd in temp_fds:
+ os.lseek(fd, 0, 0)
+
+ out = os.read(stdout_no, 1024)
+ err = support.strip_python_stderr(os.read(stderr_no, 1024))
+ finally:
+ for std, saved in enumerate(saved_fds):
+ os.dup2(saved, std)
+ os.close(saved)
+
+ self.assertEqual(out, b"got STDIN")
+ self.assertEqual(err, b"err")
+
+ finally:
+ for fd in temp_fds:
+ os.close(fd)
+
+ # When duping fds, if there arises a situation where one of the fds is
+ # either 0, 1 or 2, it is possible that it is overwritten (#12607).
+ # This tests all combinations of this.
+ def test_swap_fds(self):
+ self.check_swap_fds(0, 1, 2)
+ self.check_swap_fds(0, 2, 1)
+ self.check_swap_fds(1, 0, 2)
+ self.check_swap_fds(1, 2, 0)
+ self.check_swap_fds(2, 0, 1)
+ self.check_swap_fds(2, 1, 0)
+
def test_surrogates_error_message(self):
def prepare():
raise ValueError("surrogate:\uDCff")