| 100 | temp_dir.cleanup() |
| 101 | |
| 102 | def _run_repl( |
| 103 | self, |
| 104 | repl_input: str | list[str], |
| 105 | *, |
| 106 | env: dict | None, |
| 107 | cmdline_args: list[str] | None, |
| 108 | cwd: str, |
| 109 | skip: bool, |
| 110 | timeout: float, |
| 111 | exit_on_output: str | None, |
| 112 | ) -> tuple[str, int]: |
| 113 | assert pty |
| 114 | master_fd, slave_fd = pty.openpty() |
| 115 | cmd = [sys.executable, "-i", "-u"] |
| 116 | if env is None: |
| 117 | cmd.append("-I") |
| 118 | elif "PYTHON_HISTORY" not in env: |
| 119 | env["PYTHON_HISTORY"] = os.path.join(cwd, ".regrtest_history") |
| 120 | if cmdline_args is not None: |
| 121 | cmd.extend(cmdline_args) |
| 122 | |
| 123 | try: |
| 124 | import termios |
| 125 | except ModuleNotFoundError: |
| 126 | pass |
| 127 | else: |
| 128 | term_attr = termios.tcgetattr(slave_fd) |
| 129 | term_attr[6][termios.VREPRINT] = 0 # pass through CTRL-R |
| 130 | term_attr[6][termios.VINTR] = 0 # pass through CTRL-C |
| 131 | termios.tcsetattr(slave_fd, termios.TCSANOW, term_attr) |
| 132 | |
| 133 | process = subprocess.Popen( |
| 134 | cmd, |
| 135 | stdin=slave_fd, |
| 136 | stdout=slave_fd, |
| 137 | stderr=slave_fd, |
| 138 | cwd=cwd, |
| 139 | text=True, |
| 140 | close_fds=True, |
| 141 | env=env if env else os.environ, |
| 142 | ) |
| 143 | os.close(slave_fd) |
| 144 | if isinstance(repl_input, list): |
| 145 | repl_input = "\n".join(repl_input) + "\n" |
| 146 | os.write(master_fd, repl_input.encode("utf-8")) |
| 147 | |
| 148 | output = [] |
| 149 | while select.select([master_fd], [], [], timeout)[0]: |
| 150 | try: |
| 151 | data = os.read(master_fd, 1024).decode("utf-8") |
| 152 | if not data: |
| 153 | break |
| 154 | except OSError: |
| 155 | break |
| 156 | output.append(data) |
| 157 | if exit_on_output is not None: |
| 158 | output = ["".join(output)] |
| 159 | if exit_on_output in output[0]: |