scripts.run_tests_streaming

Reference › scripts › run_tests_streaming

scripts/run_tests_streaming.py is a sanctioned (RUL-009) streaming parallel test runner: it runs test modules concurrently and prints a PASS or FAIL line for each module as it finishes, rather than waiting for a whole batch. A human runs it when live feedback matters more than the full orchestration of run_tests; tests/scripts/test_run_tests_streaming.py covers it. It builds on orchestrata_ultimate.core.threading.test_executor.

Options: positional paths (default tests/; _expand turns directories into modules), -j/--workers (default 4), -t/--timeout (per module, default 420 seconds, matching the authoritative runner's per-batch budget because the slowest module measured about 300 seconds alone) and -a/--pytest-arg (repeatable). _run_one runs one module in a subprocess and _summary prints the totals. Before dispatching workers main calls _ensure_run_identity_env from the test executor, which sets the run id and head commit only if an outer caller has not already, so duration records carry a run id. As RUL-009 requires of every sanctioned runner, it injects the live pytest observer plugin into the pytest arguments, so observe_pytest can attach and per-test durations are recorded for the scheduler cache.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/run_tests_streaming.py

#!/usr/bin/env python3
# mypy: ignore-errors
"""Streaming parallel test runner — live per-module PASS/FAIL.

Uses subprocess per module (for real-time per-module feedback).
For CI/bulk runs use ``scripts/run_tests_threadpool.py`` which uses
in-process threads.

Usage:
    python scripts/run_tests_streaming.py tests/ -j 4 2>&1 | tee /tmp/stream.log
"""

from __future__ import annotations

import argparse
import glob
import os
import queue
import re
import signal
import subprocess
import sys
import time
from concurrent.futures import ThreadPoolExecutor

_SUMMARY_RE = re.compile(
    r"\d+ (?:passed|failed|error|errors|skipped|deselected|xfailed|xpassed)[^\n]*"
)


def _to_text(v: bytes | str | None) -> str:
    """TimeoutExpired carries partial output as BYTES even under text=True
    (CPython raises before the final decode) — normalize for the str-only
    consumers (_SUMMARY_RE, the failure printer)."""
    if isinstance(v, bytes):
        return v.decode("utf-8", errors="replace")
    return v or ""


def _run_one(cmd: list[str], timeout: float, slot: int) -> dict:
    env = dict(os.environ)
    env["ORCHESTRATA_WORKER_SLOT"] = f"slot{slot}"
    # Use Popen with start_new_session=True to create a process group, so
    # timeout kills the WHOLE tree (pytest + grandchildren like gateway/kernel).
    # Bare subprocess.run(capture_output=True) only kills the direct child,
    # leaving orphaned grandchildren holding stdout/stderr pipes open, which
    # blocks communicate() forever after the kill (the 40-minute hang at
    # session 270: a gateway test spawned a long-lived subprocess that inherited
    # the pipe; pytest was killed but the grandchild kept the pipe open).
    try:
        with subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            env=env,
            start_new_session=True,
        ) as proc:
            try:
                stdout, stderr = proc.communicate(timeout=timeout)
                return {
                    "returncode": proc.returncode,
                    "stdout": stdout,
                    "stderr": stderr,
                }
            except subprocess.TimeoutExpired:
                # Kill the whole process group, not just the direct child.
                try:
                    os.killpg(proc.pid, signal.SIGKILL)
                except ProcessLookupError, PermissionError:
                    proc.kill()
                # Drain pipes after kill (with a short timeout to avoid another hang).
                try:
                    stdout, stderr = proc.communicate(timeout=30)
                except subprocess.TimeoutExpired:
                    stdout, stderr = "", ""
                return {
                    "returncode": -1,
                    "stdout": stdout or "",
                    "stderr": stderr or "",
                    "error": f"Timeout after {timeout}s",
                }
    except Exception as exc:
        return {"returncode": -1, "stdout": "", "stderr": str(exc), "error": str(exc)}


def _expand(paths: list[str]) -> list[str]:
    mods: list[str] = []
    for p in paths:
        # A pytest node id (``tests/foo_test.py::test_bar``) is a single
        # literal target, not a directory — globbing it as one silently
        # matched ZERO files (the ``::`` segment doesn't exist on disk) and
        # `main()` printed "no test modules matched", exit 0: a false green
        # for the exact single-test-during-debugging invocation the
        # route_pytest_to_threads.py guard now routes here (2026-08-19, the
        # single-module/nodeid exemption it used to grant was the loophole
        # that shipped duration records with no run_id/batch_id/head_commit
        # and skipped the required kernel gRPC verification).
        if p.endswith(".py") or "::" in p:
            mods.append(p)
        else:
            mods.extend(
                sorted(glob.glob(f"{p.rstrip('/')}/**/test_*.py", recursive=True))
            )
    seen: set[str] = set()
    out: list[str] = []
    for m in mods:
        if m not in seen:
            seen.add(m)
            out.append(m)
    return out


def _summary(res: dict) -> str:
    text = (res.get("stdout") or "") + (res.get("stderr") or "")
    hits = _SUMMARY_RE.findall(text)
    return hits[-1] if hits else res.get("error") or "no summary"


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Streaming parallel pytest runner (live per-module results)."
    )
    ap.add_argument(
        "paths", nargs="*", default=["tests/"], help="test files or directories"
    )
    ap.add_argument("-j", "--workers", type=int, default=4)
    # Default matches the authoritative runner's per-batch budget (420s): the
    # suite's slowest module measured 298s SOLO (test_gen_brief_pack.py,
    # 2026-09-10), so 300s here was a guaranteed false timeout under load.
    ap.add_argument("-t", "--timeout", type=float, default=420.0)
    ap.add_argument("-a", "--pytest-arg", action="append", default=[])
    args = ap.parse_args()

    # Defect found 2026-08-19: profiling reports/test-durations.ndjson's
    # `kind: run` headers found ~19% missing run_id entirely -- every OTHER
    # caller of the batch-subprocess machinery besides run_tests.py's own
    # main() never established ORCH_RUN_ID/ORCH_HEAD_COMMIT, including this
    # "sanctioned" (RUL-009, below) streaming runner. Same fix, same idiom
    # as ParallelTestExecutor.__init__: establish it once here, before any
    # worker subprocess is dispatched -- setdefault, so an outer caller
    # (e.g. run_tests.py) that already set ORCH_RUN_ID keeps it. Imported
    # locally to avoid pulling core.threading.test_executor's heavier
    # import chain into every invocation of this script's --help.
    from orchestrata_ultimate.core.threading.test_executor import (
        _ensure_run_identity_env,
    )

    _ensure_run_identity_env()

    # RUL-009: every sanctioned runner must inject the live observer, exactly as
    # ``core.threading.test_executor`` does for the batched runner. Without it
    # this runner printed its own per-module PASS/FAIL but produced NO observer
    # session (``observe_pytest.py attach`` had nothing to attach to) and NO
    # per-test duration records -- so ``reports/test-durations.ndjson`` and the
    # LPT scheduler cache never learned anything from a streaming run, while the
    # force-test-observation hook still accepted the command as "observed".
    pytest_args = [
        "-q",
        "--no-header",
        "--timeout=120",
        "-p",
        "orchestrata_ultimate.core.threading.pytest_live_observer",
        # Same marker expression as scripts/run_tests.py: override
        # pyproject.toml's default `-m "not slow and not llm_integration and not
        # incus_live"` so SLOW tests RUN, while llm_integration/incus_live stay
        # opt-in lanes. Without this override the runner inherited the default
        # and silently DESELECTED every @pytest.mark.slow test (47 tests across
        # 20 modules) while still reporting the module green -- a false green,
        # and a different test set from the authoritative gate.
        "-m",
        "not llm_integration and not incus_live",
        *args.pytest_arg,
    ]
    modules = _expand(args.paths)
    total = len(modules)
    print(
        f"[streaming] {total} modules · {args.workers} workers · timeout {args.timeout:.0f}s",
        flush=True,
    )
    if not total:
        print("[streaming] no test modules matched", flush=True)
        return 0

    slot_q: queue.Queue[int] = queue.Queue()
    for i in range(args.workers):
        slot_q.put(i)

    start = time.monotonic()
    done = passed = failed = 0
    failures: list[tuple[str, dict]] = []

    with ThreadPoolExecutor(max_workers=args.workers) as ex:
        futs = {}
        submitted = 0
        last_heartbeat = time.monotonic()
        pending = set()

        # Submit modules and process results concurrently (not submit-all-then-wait).
        # The submit loop blocks on slot_q.get() when all workers are busy, so we
        # interleave submission with result processing to print per-module output
        # as soon as each completes (the original 40-minute hang at session 270
        # was caused by the submit loop blocking while 1843 modules were queued).
        for m in modules:
            # Check for completed futures while waiting for a slot
            while True:
                try:
                    slot = slot_q.get(timeout=1.0)
                    break
                except queue.Empty:
                    # No slot available — process any completed futures
                    completed = [f for f in pending if f.done()]
                    for fut in completed:
                        pending.remove(fut)
                        module, slot = futs[fut]
                        res = fut.result()
                        done += 1
                        # Exit 5 = "no tests collected": a fully-deselected module (e.g.
                        # every test carries an excluded marker like llm_integration) is
                        # a legitimate empty outcome, not a failure — same rule as the
                        # batched executor's _pytest_no_tests_collected.
                        rc = res.get("returncode")
                        ok = rc == 0 or rc == 5
                        if ok:
                            passed += 1
                        else:
                            failed += 1
                            failures.append((module, res))
                        tag = ("PASS" if rc == 0 else "EMPTY") if ok else "FAIL"
                        elapsed = time.monotonic() - start
                        print(
                            f"[{done:>4}/{total}] {tag}  {module}  ({_summary(res)})  t+{elapsed:5.0f}s",
                            flush=True,
                        )

                    # Heartbeat: print progress every 30s if nothing completed
                    now = time.monotonic()
                    if now - last_heartbeat > 30:
                        elapsed = now - start
                        print(
                            f"[streaming] heartbeat: {done}/{total} done, "
                            f"{submitted - done} running, t+{elapsed:.0f}s",
                            flush=True,
                        )
                        last_heartbeat = now

            # Submit the next module
            cmd = [sys.executable, "-m", "pytest", m, *pytest_args]
            fut = ex.submit(_run_one, cmd, args.timeout, slot)
            fut.add_done_callback(lambda _f, _slot=slot: slot_q.put(_slot))
            futs[fut] = (m, slot)
            pending.add(fut)
            submitted += 1

        # Drain any remaining pending futures
        while pending:
            completed = [f for f in pending if f.done()]
            if not completed:
                # Wait a bit for futures to complete
                time.sleep(0.5)
                # Heartbeat while waiting
                now = time.monotonic()
                if now - last_heartbeat > 30:
                    elapsed = now - start
                    print(
                        f"[streaming] heartbeat: {done}/{total} done, "
                        f"{len(pending)} running, t+{elapsed:.0f}s",
                        flush=True,
                    )
                    last_heartbeat = now
                continue

            for fut in completed:
                pending.remove(fut)
                module, slot = futs[fut]
                res = fut.result()
                done += 1
                rc = res.get("returncode")
                ok = rc == 0 or rc == 5
                if ok:
                    passed += 1
                else:
                    failed += 1
                    failures.append((module, res))
                tag = ("PASS" if rc == 0 else "EMPTY") if ok else "FAIL"
                elapsed = time.monotonic() - start
                print(
                    f"[{done:>4}/{total}] {tag}  {module}  ({_summary(res)})  t+{elapsed:5.0f}s",
                    flush=True,
                )

    elapsed = time.monotonic() - start
    print(
        f"\n[streaming] DONE in {elapsed:.0f}s · total={total} passed={passed} failed={failed}",
        flush=True,
    )
    for module, res in failures:
        out = (res.get("stdout") or "") + (res.get("stderr") or "")
        print(f"\n--- FAIL {module} (rc={res.get('returncode')}) ---", flush=True)
        print(out[-1200:], flush=True)
    return 0 if failed == 0 else 1


if __name__ == "__main__":
    sys.exit(main())