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