scripts.run_tests_threadpool

Reference › scripts › run_tests_threadpool

scripts/run_tests_threadpool.py is a parallel test runner that uses a ThreadPoolExecutor of pytest subprocesses instead of pytest-xdist. A human or the pre-push hook runs it; tests/scripts/test_run_tests_threadpool.py covers it. It uses the batch executor in orchestrata_ultimate.core.threading.test_executor and the live reader in orchestrata_ultimate.core.threading.pytest_live_reader. The fuller runner is run_tests.

Options: positional paths (default tests/), -j/--workers (default 4; higher risks shared-service contention timeouts), -t/--timeout (per batch, default 600 seconds), --pytest-arg (repeatable) and --ignore (repeatable; filters the globbed module list, because pytest's own --ignore only affects one process). A file path that no longer exists is skipped rather than passed to pytest: the pre-push hook derives its list from git diff, which includes test files deleted or renamed later, and a missing file would be counted as a failure and wrongly block the push.

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_threadpool.py

#!/usr/bin/env python3
"""Parallel test runner using ThreadPoolExecutor instead of xdist."""

import argparse
import glob
import os
import sys

from orchestrata_ultimate.core.threading.pytest_live_reader import (
    SANCTIONED_RUNNER_ENV,
)
from orchestrata_ultimate.core.threading.test_executor import (
    TEST_TOTALS_KEY,
    pop_aggregates,
    run_tests_in_parallel,
)


def main() -> int:
    parser = argparse.ArgumentParser(description="Run tests via ThreadPoolExecutor")
    parser.add_argument("paths", nargs="*", default=["tests/"], help="Test paths")
    parser.add_argument(
        "-j",
        "--workers",
        type=int,
        default=4,
        help="Max parallel pytest processes (default: 4 — higher risks "
        "shared-service contention timeouts; see run_tests.py --workers help)",
    )
    parser.add_argument(
        "-t", "--timeout", type=int, default=600, help="Timeout per batch (seconds)"
    )
    parser.add_argument(
        "--pytest-arg", action="append", default=[], help="Extra pytest args"
    )
    parser.add_argument(
        "--ignore",
        action="append",
        default=[],
        help="Directory or file prefix to exclude from collection (repeatable). "
        "Replaces the pytest --ignore flag, which only affects a single pytest "
        "process; here it filters the globbed module list.",
    )
    args = parser.parse_args()

    def _ignored(path: str) -> bool:
        norm = os.path.normpath(path)
        return any(
            norm == os.path.normpath(ig)
            or norm.startswith(os.path.normpath(ig) + os.sep)
            for ig in args.ignore
        )

    modules = []
    for p in args.paths:
        if p.endswith(".py"):
            # Skip paths that no longer exist: the pre-push hook derives its
            # target list from `git diff origin/master...HEAD`, which includes
            # test files DELETED/renamed in later commits. Passing a missing
            # file to pytest is a hard "file not found" error that would count
            # as a failure and wrongly block the push.
            if not os.path.exists(p):
                print(f"skipping missing module: {p}", file=sys.stderr)
            elif _ignored(p):
                pass
            else:
                modules.append(p)
        else:
            modules.extend(
                m
                for m in sorted(glob.glob(f"{p}/**/test_*.py", recursive=True))
                if not _ignored(m)
            )

    if not modules:
        print("No test modules found.", file=sys.stderr)
        return 1

    results = run_tests_in_parallel(
        test_paths=modules,
        pytest_args=args.pytest_arg or ["-q", "--no-header", "--timeout=120"],
        max_workers=args.workers,
        timeout_per_test=args.timeout,
        # RUL-038: a deliberate multi-file runner stamps its child pytest so the
        # live observer's multi-file refusal does not stop it at collection.
        # This runner shares the core executor with run_tests.py but was never
        # added to the sanctioned list — the stamp is its sanction.
        env={SANCTIONED_RUNNER_ENV: "run_tests_threadpool.py"},
    )

    # run_tests_in_parallel injects non-module aggregates (test-level totals,
    # the Phase-2B rerun bookkeeping, the batch plan). None carries a
    # "returncode" key, so any left in `results` reads as an extra FAILED
    # module in the tally below. Popped via the executor's own helper rather
    # than by naming them here: this file listed the two keys it knew about
    # and silently miscounted the moment a third was added (2026-08-07).
    totals = pop_aggregates(results).get(TEST_TOTALS_KEY)

    passed = sum(1 for r in results.values() if r.get("returncode") == 0)
    failed = len(results) - passed
    summary = f"\nResults: {passed} passed, {failed} failed / {len(results)} modules"
    if totals:
        summary += (
            f"  (tests: {totals.get('passed', 0)} passed, "
            f"{totals.get('failed', 0)} failed, {totals.get('errors', 0)} errors)"
        )
    print(summary)

    for path, res in results.items():
        if res.get("returncode", 1) != 0:
            print(f"\n--- {path} ---")
            output = res.get("stderr") or res.get("stdout") or ""
            print(output[-800:] if output else "(no output)")

    return 0 if failed == 0 else 1


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