Method test_double_close_on_error

Lib/test/test_subprocess.py:1536–1563

(self)

        self.assertEqual(proc.returncode, 0)

    def test_double_close_on_error(self):
        # Issue #18851
        fds = []
        def open_fds():
            for i in range(20):
                fds.extend(os.pipe())
                time.sleep(0.001)
        t = threading.Thread(target=open_fds)
        t.start()
        try:
            with self.assertRaises(OSError):
                subprocess.Popen(NONEXISTING_CMD,
                                 stdin=subprocess.PIPE,
                                 stdout=subprocess.PIPE,
                                 stderr=subprocess.PIPE)
        finally:
            t.join()
            exc = None
            for fd in fds:
                # If a double close occurred, some of those fds will
                # already have been closed by mistake, and os.close()
                # here will raise.
                try:
                    os.close(fd)
                except OSError as e:
                    exc = e
            if exc is not None:
                raise exc

    def test_threadsafe_wait(self):
        """Issue21291: Popen.wait() needs to be threadsafe for returncode."""