"""Tests for tools/failfast_probe.ps1 -- the REQ-018 Slice 2b-C fail-fast probe that launches the caller's program, classifies it as "exited fast" (probe window crossed before the process finished -> discard-and-rebuild candidate) vs. "still running" (the user's real, possibly long-running program -- never touched again, no Kill() anywhere), and captures stdout/stderr to caller-specified files. No importable functions exist (the script reads its inputs from env vars) -- this is exercised end-to-end via a real `pwsh` subprocess, mirroring tests/test_pyproj_deps.py's subprocess-based pattern for a flat script. Per docs/plan-cli-interactive-verification.md Finding 6, the script now live-echoes the child's stdout/stderr to its own stdout/stderr AS IT ARRIVES (instead of only writing captured output to disk after exit) and writes its "$exceeded|$exitcode" classification result to a dedicated result file (HP_PROBE_RESULT, default ~probe_result.txt) instead of printing it as the final stdout line -- the caller no longer wraps this script in a `for /f` stdout-capture, which would otherwise swallow the live-teed lines and corrupt the result parsing. Tests read the result file, not `proc.stdout`, for the classification outcome; `proc.stdout`/`proc.stderr` are asserted to contain the live-teed child output instead. Covers the fast-exit / probe-window-exceeded classification, exit-code passthrough in both cases, default vs. caller-specified output-path behavior, stdout/stderr capture content (both the live-teed process output and the ~run.out.txt/~run.err.txt file capture), and the base64 HP_FAILFAST_PROBE payload sync (byte-equality of the embedded payload vs this source, CRLF/LF-normalized -- mirrors test_embed_tier.py's PayloadSync pattern for a .ps1 canonical source, since *.ps1 carries `.gitattributes`' `eol=crlf`). """ import base64 import os import queue import re import shutil import subprocess import sys import tempfile import threading import time import unittest from pathlib import Path REPO = Path(__file__).resolve().parent.parent SOURCE = REPO / "tools" / "failfast_probe.ps1" PWSH = shutil.which("pwsh") FAST_SCRIPT = """ import sys sys.exit(0) """ ARGV_ECHO_SCRIPT = """ import sys print("ARGV:" + "|".join(sys.argv[1:])) sys.exit(0) """ FAST_NONZERO_SCRIPT = """ import sys sys.exit(5) """ SLOW_SCRIPT = """ import sys import time print("hello-stdout") print("hello-stderr", file=sys.stderr) time.sleep(1.5) sys.exit(7) """ # The owner's actual target program shape for this repo: no launch arguments, asks setup # questions via input(), loops on stdin until a quit command. HP_PROBE_ARGS points at this file # on disk (not stdin -- see test_exe_smokerun.py's module docstring for why a script fed via # stdin can't also read its own input() calls from that same, already-consumed stream), so # stdin stays completely free for the script's own prompts. INTERACTIVE_SCRIPT = """ name = input("Enter your name: ") print("Hello, " + name + "!") while True: cmd = input("Type 'ping' or 'exit': ") if cmd == "exit": print("Goodbye!") break elif cmd == "ping": print("pong") else: print("unknown command: " + cmd) """ def _run_probe(d, env_extra, timeout=15, stdin=None): env = { "PATH": "/usr/bin:/bin:/usr/local/bin", "HP_FAILFAST_PROBE_MS": "300", } env.update(env_extra) proc = subprocess.run( [PWSH, "-NoProfile", "-NonInteractive", "-File", str(SOURCE)], cwd=d, env=env, stdin=stdin, capture_output=True, text=True, timeout=timeout, ) return proc def _result(d, env_extra): # HP_PROBE_RESULT defaults to ~probe_result.txt in the cwd; honor a caller override the same # way the script does, so callers testing a custom path can still retrieve it. override = env_extra.get("HP_PROBE_RESULT") result_path = Path(override) if override else (Path(d) / "~probe_result.txt") return result_path.read_text(encoding="ascii").strip() def _start_reader_thread(fileobj): """Background thread doing blocking os.read() on fileobj's fd, pushing each chunk onto a queue.Queue -- the portable (Windows + POSIX) way to get timed/non-blocking reads on a subprocess pipe. select.select() only supports socket objects on Windows (a pipe raises OSError: [WinError 10038] An operation was attempted on something that is not a socket -- confirmed via a real Windows CI failure), so this thread+queue approach replaces it. A blocking os.read() returning near-instantly and pushing onto the queue preserves the same live-timing sensitivity select() gave on POSIX: the caller's queue.get(timeout=N) still distinguishes "visible almost immediately" from "never became visible within N seconds." """ q = queue.Queue() def _reader(): try: while True: chunk = os.read(fileobj.fileno(), 65536) q.put(chunk) if chunk == b"": break except OSError: q.put(b"") t = threading.Thread(target=_reader, daemon=True) t.start() return q def _read_available(q, timeout): """Returns None if nothing arrived within timeout (keep waiting), or the bytes read (b"" specifically means EOF -- distinct from None/timeout). See _start_reader_thread for why this reads from a queue fed by a background thread instead of select().""" try: return q.get(timeout=timeout) except queue.Empty: return None @unittest.skipUnless(PWSH, "pwsh not available") class FastExitClassification(unittest.TestCase): def test_fast_exit_zero_not_exceeded(self): with tempfile.TemporaryDirectory() as d: script = Path(d) / "fast.py" script.write_text(FAST_SCRIPT, encoding="utf-8") env = { "HP_PROBE_EXE": sys.executable, "HP_PROBE_ARGS": str(script), "HP_PROBE_CWD": d, } proc = _run_probe(d, env) self.assertEqual(proc.returncode, 0, proc.stderr) self.assertEqual(_result(d, env), "0|0") def test_fast_exit_nonzero_passthrough(self): # derived requirement: HP_PROBE_ARGS is a SINGLE path argument only (the script's own # header comment: $si.Arguments = '"' + $rawArgs + '"' wraps the whole string as one # quoted token) -- the nonzero exit code is baked into the script itself rather than # passed as a second CLI token, since a second token would be swallowed into the same # quoted argument as the path and misparsed as part of the filename. with tempfile.TemporaryDirectory() as d: script = Path(d) / "fast_nonzero.py" script.write_text(FAST_NONZERO_SCRIPT, encoding="utf-8") env = { "HP_PROBE_EXE": sys.executable, "HP_PROBE_ARGS": str(script), "HP_PROBE_CWD": d, } proc = _run_probe(d, env) self.assertEqual(proc.returncode, 0, proc.stderr) self.assertEqual(_result(d, env), "0|5") @unittest.skipUnless(PWSH, "pwsh not available") class ArgvPassthrough(unittest.TestCase): # [REQ-026] argv passthrough (docs/plan-cli-interactive-verification.md P1): HP_PROBE_ARGS is # now a full, already-quoted Windows Arguments string used verbatim as $si.Arguments -- # superseding the old contract (this class used to be ArgsIsSingleArgumentOnly, documenting # that a multi-token HP_PROBE_ARGS was misparsed as one literal path because the script # wrapped the whole value in one extra pair of quotes). The caller (run_setup.bat's # HP_APP_ARGS) is responsible for quoting each individual token. def test_multiple_pre_quoted_tokens_forwarded_as_separate_argv(self): with tempfile.TemporaryDirectory() as d: script = Path(d) / "argv.py" script.write_text(ARGV_ECHO_SCRIPT, encoding="utf-8") env = { "HP_PROBE_EXE": sys.executable, "HP_PROBE_ARGS": '"{}" "--foo" "bar baz"'.format(script), "HP_PROBE_CWD": d, } proc = _run_probe(d, env) self.assertEqual(proc.returncode, 0, proc.stderr) self.assertEqual(_result(d, env), "0|0") out = (Path(d) / "~run.out.txt").read_text(encoding="utf-8") self.assertIn("ARGV:--foo|bar baz", out) def test_unquoted_token_with_space_is_caller_responsibility_not_script_bug(self): # Documents the contract, not a bug: this script does zero re-quoting or re-tokenizing # of HP_PROBE_ARGS -- if the caller forgets to quote a token that itself contains a # space, the CHILD's own argv parser splits it there, same as typing it unquoted at a # real command line. run_setup.bat's HP_APP_ARGS always quotes each token individually # specifically to avoid this; it is exercised here to prove the underlying mechanism # behaves exactly as a normal Windows command line would, not something special-cased. with tempfile.TemporaryDirectory() as d: script = Path(d) / "argv.py" script.write_text(ARGV_ECHO_SCRIPT, encoding="utf-8") env = { "HP_PROBE_EXE": sys.executable, "HP_PROBE_ARGS": '"{}" bar baz'.format(script), "HP_PROBE_CWD": d, } proc = _run_probe(d, env) self.assertEqual(proc.returncode, 0, proc.stderr) self.assertEqual(_result(d, env), "0|0") out = (Path(d) / "~run.out.txt").read_text(encoding="utf-8") self.assertIn("ARGV:bar|baz", out) @unittest.skipUnless(PWSH, "pwsh not available") class SlowExitClassification(unittest.TestCase): def test_probe_window_exceeded_but_final_rc_captured(self): with tempfile.TemporaryDirectory() as d: script = Path(d) / "slow.py" script.write_text(SLOW_SCRIPT, encoding="utf-8") env = { "HP_PROBE_EXE": sys.executable, "HP_PROBE_ARGS": str(script), "HP_PROBE_CWD": d, } proc = _run_probe(d, env) self.assertEqual(proc.returncode, 0, proc.stderr) self.assertEqual(_result(d, env), "1|7") @unittest.skipUnless(PWSH, "pwsh not available") class LiveTee(unittest.TestCase): def test_child_output_reaches_scripts_own_stdout_stderr_live(self): # docs/plan-cli-interactive-verification.md Finding 5b/6: the whole point of this # requirement is that a real user watching the console sees the child's prompts as they # happen, not only after it exits. This can't assert *timing* through a captured # subprocess.run() call the way the sandbox pwsh repro did, but it can assert the # mechanism used (Register-ObjectEvent + Write-Host/[Console]::Error.WriteLine) actually # lands the child's own stdout/stderr on this script's own stdout/stderr at all -- the # previous ReadToEndAsync()-only design never wrote anything to its own stdout/stderr. with tempfile.TemporaryDirectory() as d: script = Path(d) / "slow.py" script.write_text(SLOW_SCRIPT, encoding="utf-8") proc = _run_probe(d, { "HP_PROBE_EXE": sys.executable, "HP_PROBE_ARGS": str(script), "HP_PROBE_CWD": d, }) self.assertEqual(proc.returncode, 0, proc.stderr) self.assertIn("hello-stdout", proc.stdout) self.assertIn("hello-stderr", proc.stderr) @unittest.skipUnless(PWSH, "pwsh not available") class ProcessIdMessage(unittest.TestCase): # A user whose program stays silent forever past the probe window has no clean way to get # back to the bootstrapper -- Ctrl+C broadcasts to the whole console process group (killing # the bootstrapper too), and this file family deliberately never Kill()s the process itself # (see the module header). Printing the real Windows PID lets a stuck user End Task it via # Task Manager's Details tab -- a precise, single-process kill, unlike Ctrl+C. def test_pid_message_printed_with_real_numeric_pid(self): with tempfile.TemporaryDirectory() as d: script = Path(d) / "fast.py" script.write_text(FAST_SCRIPT, encoding="utf-8") proc = _run_probe(d, { "HP_PROBE_EXE": sys.executable, "HP_PROBE_ARGS": str(script), "HP_PROBE_CWD": d, }) self.assertEqual(proc.returncode, 0, proc.stderr) m = re.search(r"\[INFO\] Process ID (\d+)\.", proc.stdout) self.assertIsNotNone(m, "PID message not found in stdout: " + proc.stdout) self.assertGreater(int(m.group(1)), 0) self.assertIn("Task Manager", proc.stdout) self.assertIn("End Task", proc.stdout) def test_pid_message_never_lands_in_captured_output_file(self): # The PID line is bootstrapper-generated console guidance, not part of the user's own # program output -- it must never contaminate ~run.out.txt (the file downstream checks # read as "what did the user's program print"). with tempfile.TemporaryDirectory() as d: script = Path(d) / "fast.py" script.write_text(FAST_SCRIPT, encoding="utf-8") proc = _run_probe(d, { "HP_PROBE_EXE": sys.executable, "HP_PROBE_ARGS": str(script), "HP_PROBE_CWD": d, }) self.assertEqual(proc.returncode, 0, proc.stderr) captured = (Path(d) / "~run.out.txt").read_text(encoding="ascii") self.assertNotIn("Process ID", captured) @unittest.skipUnless(PWSH, "pwsh not available") class InteractiveRoundTrip(unittest.TestCase): # Answers the actual question this repo's live-echo work exists to answer: not just "is # output visible," but "can a real multi-round input()-driven program be watched AND # answered through this exact mechanism." RedirectStandardInput is never set on the # grandchild in failfast_probe.ps1, so it inherits whatever stdin this pwsh process itself # has -- subprocess.run's own stdin= parameter points that at a file containing only the # ANSWERS (never the script source; see INTERACTIVE_SCRIPT's own comment for why those two # must stay on separate channels). This does not prove a live human's typing timing (that # needs a real Windows double-click, still open -- see # docs/plan-cli-interactive-verification.md requirement 2), but it does prove the mechanism # (Register-ObjectEvent tee + inherited/piped stdin together, not one or the other in # isolation) carries a genuine multi-round conversation correctly end to end. def test_multi_round_input_conversation_completes_correctly(self): with tempfile.TemporaryDirectory() as d: script = Path(d) / "interactive.py" script.write_text(INTERACTIVE_SCRIPT, encoding="utf-8") answers = Path(d) / "answers.txt" answers.write_text("Alice\nping\nexit\n", encoding="utf-8") with open(answers, "r", encoding="utf-8") as stdin_src: proc = _run_probe(d, { "HP_PROBE_EXE": sys.executable, "HP_PROBE_ARGS": str(script), "HP_PROBE_CWD": d, }, stdin=stdin_src) self.assertEqual(proc.returncode, 0, proc.stderr) # Ordering, not just presence: proves each answer was consumed by the RIGHT prompt, # not just that all expected substrings happen to appear somewhere in the output. out = proc.stdout i1 = out.index("Enter your name:") i2 = out.index("Hello, Alice!") i3 = out.index("Type 'ping' or 'exit':") i4 = out.index("pong") i5 = out.index("Goodbye!") self.assertLess(i1, i2) self.assertLess(i2, i3) self.assertLess(i3, i4) self.assertLess(i4, i5) # Same round trip landed in the captured file, not just the live-teed console stream. captured = (Path(d) / "~run.out.txt").read_text(encoding="ascii") self.assertIn("Hello, Alice!", captured) self.assertIn("pong", captured) self.assertIn("Goodbye!", captured) @unittest.skipUnless(PWSH, "pwsh not available") class NoNewlinePromptVisibility(unittest.TestCase): # Regression test for Finding 9 (docs/plan-cli-interactive-verification.md): input("text") # prints its prompt WITHOUT a trailing newline by design (the cursor stays on the same line # waiting for typed input). The old ReadLineAsync()-based reader would not surface that text # until a LATER newline arrived or the process exited -- meaning a real user would see # nothing until after they'd already blindly typed an answer. InteractiveRoundTrip above # cannot catch this: it feeds the whole answer sequence via a pre-loaded file, so the OS # makes it available to the reader near-instantly regardless of read strategy, masking the # timing bug entirely. This test instead drives stdin itself via a live pipe and asserts the # prompt is readable BEFORE any answer is sent -- the one thing a static-file test can't # distinguish. def test_prompt_without_trailing_newline_is_visible_before_input_is_sent(self): with tempfile.TemporaryDirectory() as d: script = Path(d) / "interactive.py" script.write_text(INTERACTIVE_SCRIPT, encoding="utf-8") env = { "PATH": "/usr/bin:/bin:/usr/local/bin", "HP_FAILFAST_PROBE_MS": "300", "HP_PROBE_EXE": sys.executable, "HP_PROBE_ARGS": str(script), "HP_PROBE_CWD": d, } proc = subprocess.Popen( [PWSH, "-NoProfile", "-NonInteractive", "-File", str(SOURCE)], cwd=d, env=env, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, bufsize=0, ) try: q = _start_reader_thread(proc.stdout) # derived requirement: the helper now prints its own "[INFO] Process ID ..." # line (Task Manager kill-by-PID guidance) as the very first output, before the # child process's own prompt -- accumulate reads until the prompt text shows up # instead of assuming the first chunk IS the prompt. preinput = b"" deadline = time.monotonic() + 5.0 while time.monotonic() < deadline and b"Enter your name:" not in preinput: chunk = _read_available(q, max(0.1, deadline - time.monotonic())) if chunk is None: continue preinput += chunk self.assertIn( b"Enter your name:", preinput, "no prompt observed within 5s before any input was sent -- " "Finding 9 regression (prompt invisible without a trailing newline)", ) proc.stdin.write(b"Alice\nping\nexit\n") proc.stdin.flush() proc.stdin.close() all_output = preinput deadline = time.monotonic() + 10 while time.monotonic() < deadline: more = _read_available(q, 1.0) if more is None: continue if more == b"": break all_output += more proc.wait(timeout=5) finally: proc.stdout.close() self.assertEqual(proc.returncode, 0) self.assertIn(b"Hello, Alice!", all_output) self.assertIn(b"pong", all_output) self.assertIn(b"Goodbye!", all_output) @unittest.skipUnless(PWSH, "pwsh not available") class OutputCapture(unittest.TestCase): def test_default_output_paths_used_when_unset(self): with tempfile.TemporaryDirectory() as d: script = Path(d) / "fast.py" script.write_text(FAST_SCRIPT, encoding="utf-8") _run_probe(d, { "HP_PROBE_EXE": sys.executable, "HP_PROBE_ARGS": str(script), "HP_PROBE_CWD": d, }) self.assertTrue((Path(d) / "~run.out.txt").exists()) self.assertTrue((Path(d) / "~run.err.txt").exists()) def test_caller_specified_output_paths_and_content(self): with tempfile.TemporaryDirectory() as d: script = Path(d) / "slow.py" script.write_text(SLOW_SCRIPT, encoding="utf-8") out_path = Path(d) / "custom.out.txt" err_path = Path(d) / "custom.err.txt" result_path = Path(d) / "custom.result.txt" _run_probe(d, { "HP_PROBE_EXE": sys.executable, "HP_PROBE_ARGS": str(script), "HP_PROBE_CWD": d, "HP_PROBE_OUT": str(out_path), "HP_PROBE_ERR": str(err_path), "HP_PROBE_RESULT": str(result_path), }) self.assertIn("hello-stdout", out_path.read_text(encoding="ascii")) self.assertIn("hello-stderr", err_path.read_text(encoding="ascii")) self.assertEqual(result_path.read_text(encoding="ascii").strip(), "1|7") self.assertFalse((Path(d) / "~run.out.txt").exists()) self.assertFalse((Path(d) / "~probe_result.txt").exists()) def test_single_argument_path_with_spaces_quoted_correctly(self): # [REQ-026] argv passthrough changed this contract: HP_PROBE_ARGS is now used verbatim # (no re-quoting by this script), so the CALLER must quote a path containing spaces -- # matching how run_setup.bat's HP_APP_ARGS / "%HP_ENTRY%" construction already does. with tempfile.TemporaryDirectory() as d: subdir = Path(d) / "dir with spaces" subdir.mkdir() script = subdir / "fast.py" script.write_text(FAST_SCRIPT, encoding="utf-8") env = { "HP_PROBE_EXE": sys.executable, "HP_PROBE_ARGS": '"{}"'.format(script), "HP_PROBE_CWD": d, } proc = _run_probe(d, env) self.assertEqual(proc.returncode, 0, proc.stderr) self.assertEqual(_result(d, env), "0|0") class PayloadSync(unittest.TestCase): def test_embedded_base64_matches_source(self): # derived requirement: tools/failfast_probe.ps1 carries .gitattributes' `*.ps1 text # eol=crlf`, so a checkout on Windows CI materializes CRLF line endings regardless of # what the payload was encoded from -- normalize both sides before comparing so this # verifies content, not incidental checkout-time line-ending translation. bat = (REPO / "run_setup.bat").read_text(encoding="ascii", errors="replace") m = re.search(r'set "HP_FAILFAST_PROBE=([A-Za-z0-9+/=]+)"', bat) self.assertIsNotNone(m, "HP_FAILFAST_PROBE payload not found in run_setup.bat") decoded = base64.b64decode(m.group(1)).replace(b"\r\n", b"\n") source = SOURCE.read_bytes().replace(b"\r\n", b"\n") self.assertEqual( decoded, source, "HP_FAILFAST_PROBE base64 is out of sync with tools/failfast_probe.ps1; re-encode it.", ) if __name__ == "__main__": unittest.main()