(argv: list[str], cwd: Path, env: dict[str, str], timeout_s: float, settle_s: float)
| 205 | |
| 206 | |
| 207 | def launch_interactive(argv: list[str], cwd: Path, env: dict[str, str], timeout_s: float, settle_s: float) -> SessionLaunch: |
| 208 | master_fd, slave_fd = pty.openpty() |
| 209 | proc = subprocess.Popen( |
| 210 | argv, |
| 211 | cwd=str(cwd), |
| 212 | env=env, |
| 213 | stdin=slave_fd, |
| 214 | stdout=slave_fd, |
| 215 | stderr=slave_fd, |
| 216 | preexec_fn=os.setsid, |
| 217 | ) |
| 218 | os.close(slave_fd) |
| 219 | os.set_blocking(master_fd, False) |
| 220 | start = time.perf_counter() |
| 221 | buf = b"" |
| 222 | ready = False |
| 223 | input_ready = False |
| 224 | probe_sent = False |
| 225 | excerpt = None |
| 226 | while time.perf_counter() - start < timeout_s: |
| 227 | rlist, _, _ = select.select([master_fd], [], [], 0.05) |
| 228 | if rlist: |
| 229 | try: |
| 230 | chunk = os.read(master_fd, 65536) |
| 231 | except BlockingIOError: |
| 232 | chunk = b"" |
| 233 | if chunk: |
| 234 | buf += chunk |
| 235 | buf = reply_queries(master_fd, buf) |
| 236 | plain = strip_ansi(buf.decode("utf-8", "replace")) |
| 237 | excerpt = first_meaningful_line(plain) |
| 238 | if excerpt: |
| 239 | ready = True |
| 240 | if not probe_sent: |
| 241 | try: |
| 242 | os.write(master_fd, PROBE.encode()) |
| 243 | probe_sent = True |
| 244 | except OSError: |
| 245 | break |
| 246 | if probe_sent and PROBE in plain: |
| 247 | input_ready = True |
| 248 | break |
| 249 | if proc.poll() is not None: |
| 250 | break |
| 251 | if input_ready or ready: |
| 252 | time.sleep(settle_s) |
| 253 | elapsed = time.perf_counter() - start |
| 254 | return SessionLaunch( |
| 255 | root_pid=proc.pid, |
| 256 | pgid=os.getpgid(proc.pid), |
| 257 | master_fd=master_fd, |
| 258 | ready=ready, |
| 259 | input_ready=input_ready, |
| 260 | excerpt=excerpt, |
| 261 | seconds_to_visible=elapsed if ready else None, |
| 262 | seconds_to_input_ready=elapsed if input_ready else None, |
| 263 | buffer_excerpt=(strip_ansi(buf.decode("utf-8", "replace"))[:300] or None), |
| 264 | ) |
no test coverage detected