From 2da64e78401fffa0bebee2bb498106bd41765f30 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Tue, 21 Jul 2026 10:51:22 +0530 Subject: [PATCH] refactor: drop platform kwarg, fix PTY test cleanup - Remove redundant platform= test seam from _terminal_may_leak_cpr(); use monkeypatch.setattr(sys, 'platform', ...) consistently in both test files. - Wrap PTY tests in try/finally for fd cleanup on assertion failure. - Guard select.select() in terminal thread against OSError after fd close (fixes PytestUnhandledThreadExceptionWarning). - Trim PR-number reference from test module docstring. --- cli.py | 10 +- tests/cli/test_cli_init.py | 13 ++- tests/cli/test_cpr_local_leak.py | 155 +++++++++++++++++-------------- 3 files changed, 99 insertions(+), 79 deletions(-) diff --git a/cli.py b/cli.py index a4d0116c81ac..287c2de25837 100644 --- a/cli.py +++ b/cli.py @@ -3270,14 +3270,13 @@ def _disable_prompt_toolkit_cpr_warning(app) -> None: pass -def _terminal_may_leak_cpr(*, platform: str | None = None) -> bool: +def _terminal_may_leak_cpr() -> bool: """Whether classic CLI should suppress prompt_toolkit CPR (ESC[6n) queries. Delayed CPR replies (``ESC[;R`` / visible ``^[[;R``) leak into the status line and can freeze input when the reply is slow - (#13870 on SSH/slow PTYs). The same race hits **local POSIX** TTYs under - heavy subagent / status-line load — deterministic delayed-CPR PTY harness - in ``tests/cli/test_cpr_local_leak.py``. + (#13870 on SSH/slow PTYs). The same race hits local POSIX TTYs under + heavy subagent / status-line load — see ``tests/cli/test_cpr_local_leak.py``. Policy: - ``PROMPT_TOOLKIT_NO_CPR=1`` → always suppress @@ -3288,8 +3287,7 @@ def _terminal_may_leak_cpr(*, platform: str | None = None) -> bool: """ if os.environ.get("PROMPT_TOOLKIT_NO_CPR", "") == "1": return True - plat = sys.platform if platform is None else platform - if plat == "win32": + if sys.platform == "win32": return False return True diff --git a/tests/cli/test_cli_init.py b/tests/cli/test_cli_init.py index 52f54a419682..48de5b7c95f4 100644 --- a/tests/cli/test_cli_init.py +++ b/tests/cli/test_cli_init.py @@ -295,17 +295,22 @@ class TestPromptToolkitTerminalCompatibility: Broader coverage (Application wiring + delayed-CPR PTY repro) lives in ``tests/cli/test_cpr_local_leak.py``. """ + import sys as _sys + from cli import _terminal_may_leak_cpr for var in ("SSH_CONNECTION", "SSH_CLIENT", "SSH_TTY", "PROMPT_TOOLKIT_NO_CPR"): monkeypatch.delenv(var, raising=False) - assert _terminal_may_leak_cpr(platform="linux") is True - assert _terminal_may_leak_cpr(platform="darwin") is True - assert _terminal_may_leak_cpr(platform="win32") is False + monkeypatch.setattr(_sys, "platform", "linux") + assert _terminal_may_leak_cpr() is True + monkeypatch.setattr(_sys, "platform", "darwin") + assert _terminal_may_leak_cpr() is True + monkeypatch.setattr(_sys, "platform", "win32") + assert _terminal_may_leak_cpr() is False monkeypatch.setenv("PROMPT_TOOLKIT_NO_CPR", "1") - assert _terminal_may_leak_cpr(platform="win32") is True + assert _terminal_may_leak_cpr() is True class TestSingleQueryState: diff --git a/tests/cli/test_cpr_local_leak.py b/tests/cli/test_cpr_local_leak.py index c2201f3d2039..efd174196ff4 100644 --- a/tests/cli/test_cpr_local_leak.py +++ b/tests/cli/test_cpr_local_leak.py @@ -1,7 +1,5 @@ """Local CPR leak reproduction + classic-CLI Application output selection. -Addresses review on #67377: - * Deterministic local-PTY proof that delayed CPR replies leak as ``ESC[row;colR`` / ``^[[row;colR`` when ``enable_cpr=True`` (no SSH). * Integration-level assertion that, with no SSH env vars, classic CLI @@ -92,83 +90,102 @@ class TestDelayedCprLocalPtyLeak: from prompt_toolkit.output.vt100 import Vt100_Output master, slave = _openpty_or_skip() - tty.setraw(slave) - slave_w = os.fdopen(os.dup(slave), "w", buffering=1) - stop = threading.Event() - queries = 0 + try: + tty.setraw(slave) + slave_w = os.fdopen(os.dup(slave), "w", buffering=1) + stop = threading.Event() + queries = 0 - def terminal() -> None: - nonlocal queries - buf = b"" - while not stop.is_set(): - r, _, _ = select.select([master], [], [], 0.05) - if not r: - continue - try: - chunk = os.read(master, 4096) - except OSError: - break - if not chunk: - break - buf += chunk - while True: - idx = buf.find(b"\x1b[6n") - if idx < 0: - buf = buf[-8:] if len(buf) > 8 else buf + def terminal() -> None: + nonlocal queries + buf = b"" + while not stop.is_set(): + try: + r, _, _ = select.select([master], [], [], 0.05) + except OSError: break - buf = buf[idx + 4 :] - queries += 1 - time.sleep(0.12) - os.write(master, b"\x1b[39;1R") + if not r: + continue + try: + chunk = os.read(master, 4096) + except OSError: + break + if not chunk: + break + buf += chunk + while True: + idx = buf.find(b"\x1b[6n") + if idx < 0: + buf = buf[-8:] if len(buf) > 8 else buf + break + buf = buf[idx + 4 :] + queries += 1 + time.sleep(0.12) + try: + os.write(master, b"\x1b[39;1R") + except OSError: + pass - threading.Thread(target=terminal, daemon=True).start() - out = Vt100_Output( - slave_w, lambda: Size(rows=40, columns=80), enable_cpr=True - ) - out.ask_for_cpr() - out.flush() - for i in range(4): - slave_w.write(f"\rgpt-5.6-sol Q {i}\n") - slave_w.flush() - time.sleep(0.02) - time.sleep(0.3) + threading.Thread(target=terminal, daemon=True).start() + out = Vt100_Output( + slave_w, lambda: Size(rows=40, columns=80), enable_cpr=True + ) + out.ask_for_cpr() + out.flush() + for i in range(4): + slave_w.write(f"\rgpt-5.6-sol Q {i}\n") + slave_w.flush() + time.sleep(0.02) + time.sleep(0.3) - data = b"" - while True: - r, _, _ = select.select([slave], [], [], 0.05) - if not r: - break - data += os.read(slave, 4096) + data = b"" + while True: + r, _, _ = select.select([slave], [], [], 0.05) + if not r: + break + data += os.read(slave, 4096) - stop.set() - slave_w.close() - os.close(slave) - os.close(master) + stop.set() + slave_w.close() - assert queries >= 1 - assert b"\x1b[39;1R" in data + assert queries >= 1 + assert b"\x1b[39;1R" in data + finally: + try: + os.close(slave) + except OSError: + pass + try: + os.close(master) + except OSError: + pass def test_cpr_disabled_output_sends_no_query(self): """Hermes CPR-disabled builder must not emit ESC[6n.""" master, slave = _openpty_or_skip() - slave_w = os.fdopen(slave, "w", buffering=1) - out = _build_cpr_disabled_output(slave_w) - assert out is not None - assert out.enable_cpr is False + try: + slave_w = os.fdopen(slave, "w", buffering=1) + out = _build_cpr_disabled_output(slave_w) + assert out is not None + assert out.enable_cpr is False - seen = b"" + seen = b"" - def reader() -> None: - nonlocal seen - r, _, _ = select.select([master], [], [], 0.25) - if r: - seen = os.read(master, 4096) + def reader() -> None: + nonlocal seen + r, _, _ = select.select([master], [], [], 0.25) + if r: + seen = os.read(master, 4096) - threading.Thread(target=reader, daemon=True).start() - slave_w.write("status ok\n") - slave_w.flush() - # Do not call ask_for_cpr — renderer skips it when NOT_SUPPORTED. - time.sleep(0.3) - slave_w.close() - os.close(master) - assert b"\x1b[6n" not in seen + threading.Thread(target=reader, daemon=True).start() + slave_w.write("status ok\n") + slave_w.flush() + # Do not call ask_for_cpr — renderer skips it when NOT_SUPPORTED. + time.sleep(0.3) + slave_w.close() + assert b"\x1b[6n" not in seen + finally: + try: + os.close(master) + except OSError: + pass