(self, stdin_no, stdout_no, stderr_no)
| 2768 | os.close(fd) |
| 2769 | |
| 2770 | def check_swap_fds(self, stdin_no, stdout_no, stderr_no): |
| 2771 | # open up some temporary files |
| 2772 | temps = [tempfile.mkstemp() for i in range(3)] |
| 2773 | temp_fds = [fd for fd, fname in temps] |
| 2774 | try: |
| 2775 | # unlink the files -- we won't need to reopen them |
| 2776 | for fd, fname in temps: |
| 2777 | os.unlink(fname) |
| 2778 | |
| 2779 | # save a copy of the standard file descriptors |
| 2780 | saved_fds = self._save_fds(range(3)) |
| 2781 | try: |
| 2782 | # duplicate the temp files over the standard fd's 0, 1, 2 |
| 2783 | for fd, temp_fd in enumerate(temp_fds): |
| 2784 | os.dup2(temp_fd, fd) |
| 2785 | |
| 2786 | # write some data to what will become stdin, and rewind |
| 2787 | os.write(stdin_no, b"STDIN") |
| 2788 | os.lseek(stdin_no, 0, 0) |
| 2789 | |
| 2790 | # now use those files in the given order, so that subprocess |
| 2791 | # has to rearrange them in the child |
| 2792 | p = subprocess.Popen([sys.executable, "-c", |
| 2793 | 'import sys; got = sys.stdin.read();' |
| 2794 | 'sys.stdout.write("got %s"%got); sys.stderr.write("err")'], |
| 2795 | stdin=stdin_no, |
| 2796 | stdout=stdout_no, |
| 2797 | stderr=stderr_no) |
| 2798 | p.wait() |
| 2799 | |
| 2800 | for fd in temp_fds: |
| 2801 | os.lseek(fd, 0, 0) |
| 2802 | |
| 2803 | out = os.read(stdout_no, 1024) |
| 2804 | err = os.read(stderr_no, 1024).strip() |
| 2805 | finally: |
| 2806 | self._restore_fds(saved_fds) |
| 2807 | |
| 2808 | self.assertEqual(out, b"got STDIN") |
| 2809 | self.assertEqual(err, b"err") |
| 2810 | |
| 2811 | finally: |
| 2812 | for fd in temp_fds: |
| 2813 | os.close(fd) |
| 2814 | |
| 2815 | # When duping fds, if there arises a situation where one of the fds is |
| 2816 | # either 0, 1 or 2, it is possible that it is overwritten (#12607). |
no test coverage detected