From 4419922bce8af5302f6c36c93a8096f55beab017 Mon Sep 17 00:00:00 2001 From: Ed_ Date: Mon, 15 Jun 2026 20:02:36 -0400 Subject: [PATCH] review batch script --- scripts/run_tests_batched.py | 155 ++++++++++++++++------------------- 1 file changed, 70 insertions(+), 85 deletions(-) diff --git a/scripts/run_tests_batched.py b/scripts/run_tests_batched.py index 13d06708..9bf6da0a 100644 --- a/scripts/run_tests_batched.py +++ b/scripts/run_tests_batched.py @@ -12,7 +12,7 @@ try: except ImportError: _HAS_XDIST = False -_SCRIPT_DIR = Path(__file__).resolve().parent +_SCRIPT_DIR = Path(__file__).resolve().parent _PROJECT_ROOT = _SCRIPT_DIR.parent sys.path.insert(0, str(_PROJECT_ROOT / "tests")) @@ -26,19 +26,19 @@ if _USE_COLOR and os.name == "nt": _USE_COLOR = False class _C: - RESET = "\033[0m" - BOLD = "\033[1m" - DIM = "\033[2m" - RED = "\033[31m" - GREEN = "\033[32m" - YELLOW = "\033[33m" - BLUE = "\033[34m" - MAGENTA = "\033[35m" - CYAN = "\033[36m" - BOLD_GREEN = "\033[1;32m" - BOLD_RED = "\033[1;31m" + RESET = "\033[0m" + BOLD = "\033[1m" + DIM = "\033[2m" + RED = "\033[31m" + GREEN = "\033[32m" + YELLOW = "\033[33m" + BLUE = "\033[34m" + MAGENTA = "\033[35m" + CYAN = "\033[36m" + BOLD_GREEN = "\033[1;32m" + BOLD_RED = "\033[1;31m" BOLD_YELLOW = "\033[1;33m" - BOLD_CYAN = "\033[1;36m" + BOLD_CYAN = "\033[1;36m" def _c(text: str, color: str) -> str: if not _USE_COLOR: @@ -110,138 +110,123 @@ def _format_pytest_line(line: str) -> str | None: return None if stripped.startswith(("tests/", "tests\\")) and "::" in stripped and len(stripped.split()) == 1: return None - if " PASSED " in stripped and "[gw" in stripped: - return _c(stripped, _C.GREEN) - if " FAILED " in stripped and "[gw" in stripped: - return _c(stripped, _C.BOLD_RED) - if " ERROR " in stripped and "[gw" in stripped: - return _c(stripped, _C.BOLD_RED) + if " PASSED " in stripped and "[gw" in stripped: return _c(stripped, _C.GREEN) + if " FAILED " in stripped and "[gw" in stripped: return _c(stripped, _C.BOLD_RED) + if " ERROR " in stripped and "[gw" in stripped: return _c(stripped, _C.BOLD_RED) if stripped.startswith(("tests/", "tests\\")) and "::" in stripped: - if " PASSED" in stripped: - return _c(stripped, _C.GREEN) - if " FAILED" in stripped: - return _c(stripped, _C.BOLD_RED) - if " ERROR" in stripped: - return _c(stripped, _C.BOLD_RED) + if " PASSED" in stripped: return _c(stripped, _C.GREEN) + if " FAILED" in stripped: return _c(stripped, _C.BOLD_RED) + if " ERROR" in stripped: return _c(stripped, _C.BOLD_RED) if stripped.startswith(("PASSED", "FAILED", "ERROR")) and "::" in stripped: status = stripped.split()[0] - rest = stripped[len(status):] - if status == "PASSED": - return _c(f"{status}{rest}", _C.GREEN) + res = stripped[len(status):] + if status == "PASSED": return _c(f"{status}{rest}", _C.GREEN) return _c(f"{status}{rest}", _C.BOLD_RED) if stripped.startswith(("passed", "failed", "error")) and " in " in stripped and stripped.endswith("s"): return _c(stripped, _C.BOLD) return stripped def _run_batch(b: Batch, durations: dict[str, float]) -> tuple[int, float, dict[str, float]]: - if b.skip_reason: - return 0, 0.0, {} + if b.skip_reason: return 0, 0.0, {} args = list(b.pytest_args) - if not _HAS_XDIST: - args = [a for a in args if a not in {"-n", "auto"}] + if not _HAS_XDIST: args = [a for a in args if a not in {"-n", "auto"}] cmd = ["uv", "run", "pytest", "-v", "--durations=3"] + args + [str(f) for f in b.files] print(_c(f"\n>>> Running {b.label} ({len(b.files)} files)", _C.BOLD_CYAN)) - t0 = time.monotonic() + t0 = time.monotonic() proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1) captured: list[str] = [] assert proc.stdout is not None for line in proc.stdout: captured.append(line) formatted = _format_pytest_line(line) - if formatted is None: - continue + if formatted is None: continue print(formatted) proc.wait() - elapsed = time.monotonic() - t0 - new_durs = _parse_durations_from_pytest_output("".join(captured)) - captured_text = "".join(captured) - saw_failure = "FAILED " in captured_text or " stopping after " in captured_text + elapsed = time.monotonic() - t0 + new_durs = _parse_durations_from_pytest_output("".join(captured)) + captured_text = "".join(captured) + saw_failure = "FAILED " in captured_text or " stopping after " in captured_text effective_code = proc.returncode if proc.returncode != 0 else (1 if saw_failure else 0) - if effective_code == 0: - print(_c(f"<<< {b.label} PASS in {elapsed:.1f}s", _C.BOLD_GREEN)) - else: - print(_c(f"<<< {b.label} FAIL (exit {effective_code}) in {elapsed:.1f}s", _C.BOLD_RED)) + if effective_code == 0: print(_c(f"<<< {b.label} PASS in {elapsed:.1f}s", _C.BOLD_GREEN)) + else: print(_c(f"<<< {b.label} FAIL (exit {effective_code}) in {elapsed:.1f}s", _C.BOLD_RED)) return effective_code, elapsed, new_durs def _print_summary(results: list[tuple[Batch, int, float]]) -> int: print() rows: list[tuple[str, str, str, int, float, int]] = [] - worst = 0 - total_files = 0 - total_time = 0.0 - passed_count = 0 - failed_count = 0 + worst = 0 + total_files = 0 + total_time = 0.0 + passed_count = 0 + failed_count = 0 skipped_count = 0 for b, code, elapsed in results: n = len(b.files) total_files += n - total_time += elapsed + total_time += elapsed if b.skip_reason: - status_text = "SKIPPED" + status_text = "SKIPPED" skipped_count += 1 elif code == 0: - status_text = "PASS" + status_text = "PASS" passed_count += 1 else: - status_text = "FAIL" + status_text = "FAIL" failed_count += 1 worst = max(worst, code) rows.append((b.tier, b.label, status_text, n, elapsed, code)) - tier_w = max(len("TIER"), max(len(r[0]) for r in rows)) - label_w = max(len("BATCH LABEL"), max(len(r[1]) for r in rows)) - status_w = max(len("STATUS"), max(len(r[2]) for r in rows)) - files_w = max(len("FILES"), max(len(str(r[3])) for r in rows)) - time_w = max(len("TIME"), max(len(f"{r[4]:.1f}s") for r in rows)) - header = f" {'TIER':{tier_w}s} │ {'BATCH LABEL':{label_w}s} │ {'STATUS':{status_w}s} │ {'FILES':>{files_w}s} │ {'TIME':>{time_w}s} " - sep = "─" * len(header) - print(_c(sep, _C.DIM)) + tier_w = max(len("TIER"), max(len(r[0]) for r in rows)) + label_w = max(len("BATCH LABEL"), max(len(r[1]) for r in rows)) + status_w = max(len("STATUS"), max(len(r[2]) for r in rows)) + files_w = max(len("FILES"), max(len(str(r[3])) for r in rows)) + time_w = max(len("TIME"), max(len(f"{r[4]:.1f}s") for r in rows)) + header = f" {'TIER':{tier_w}s} │ {'BATCH LABEL':{label_w}s} │ {'STATUS':{status_w}s} │ {'FILES':>{files_w}s} │ {'TIME':>{time_w}s} " + sep = "─" * len(header) + print(_c(sep, _C.DIM)) print(_c(header, _C.BOLD)) - print(_c(sep, _C.DIM)) + print(_c(sep, _C.DIM)) for tier, label, status_text, n, elapsed, _code in rows: - if status_text == "PASS": - status = _c(status_text, _C.BOLD_GREEN) - elif status_text == "FAIL": - status = _c(status_text, _C.BOLD_RED) - else: - status = _c(status_text, _C.BOLD_YELLOW) + if status_text == "PASS": status = _c(status_text, _C.BOLD_GREEN) + elif status_text == "FAIL": status = _c(status_text, _C.BOLD_RED) + else: status = _c(status_text, _C.BOLD_YELLOW) tier_colored = _c(f" {tier:<{tier_w}s}", _C.CYAN) print(f"{tier_colored} │ {label:<{label_w}s} │ {status} │ {n:>{files_w}d} │ {elapsed:>{time_w - 1}.1f}s") print(_c(sep, _C.DIM)) if failed_count: overall_text = f"{failed_count} FAILED" - overall = _c(overall_text, _C.BOLD_RED) + overall = _c(overall_text, _C.BOLD_RED) elif passed_count: overall_text = f"ALL {passed_count} PASS" - overall = _c(overall_text, _C.BOLD_GREEN) + overall = _c(overall_text, _C.BOLD_GREEN) else: overall_text = "NO BATCHES RUN" - overall = _c(overall_text, _C.BOLD_YELLOW) - total_label = _c(f" {'TOTAL':<{tier_w}s}", _C.BOLD) + overall = _c(overall_text, _C.BOLD_YELLOW) + total_label = _c(f" {'TOTAL':<{tier_w}s}", _C.BOLD) print(f"{total_label} │ {'':<{label_w}s} │ {overall} │ {total_files:>{files_w}d} │ {total_time:>{time_w - 1}.1f}s") print(_c(sep, _C.DIM)) return worst def main() -> int: p = argparse.ArgumentParser() - p.add_argument("--tests-dir", default=str(_PROJECT_ROOT / "tests")) - p.add_argument("--registry", default=str(_PROJECT_ROOT / "tests" / "test_categories.toml")) - p.add_argument("--tiers", default="1,2,3,H") + p.add_argument("--tests-dir", default=str(_PROJECT_ROOT / "tests")) + p.add_argument("--registry", default=str(_PROJECT_ROOT / "tests" / "test_categories.toml")) + p.add_argument("--tiers", default="1,2,3,H") p.add_argument("--include-opt-in", action="store_true") - p.add_argument("--no-xdist", action="store_true") - p.add_argument("--plan", action="store_true") - p.add_argument("--audit", action="store_true") - p.add_argument("--strict", action="store_true") - p.add_argument("--durations", action="store_true", help="Record per-test durations to .test_durations.json") - p.add_argument("--no-color", action="store_true", help="Disable ANSI color output") + p.add_argument("--no-xdist", action="store_true") + p.add_argument("--plan", action="store_true") + p.add_argument("--audit", action="store_true") + p.add_argument("--strict", action="store_true") + p.add_argument("--durations", action="store_true", help="Record per-test durations to .test_durations.json") + p.add_argument("--no-color", action="store_true", help="Disable ANSI color output") options = p.parse_args() if options.no_color: global _USE_COLOR _USE_COLOR = False - tiers = _parse_tiers(options.tiers) - tests_dir = Path(options.tests_dir) if Path(options.tests_dir).is_absolute() else (_PROJECT_ROOT / options.tests_dir) + tiers = _parse_tiers(options.tiers) + tests_dir = Path(options.tests_dir) if Path(options.tests_dir).is_absolute() else (_PROJECT_ROOT / options.tests_dir) durations_path = _durations_path(tests_dir) - durations = _load_durations(durations_path) - records = categorize_all(tests_dir, Path(options.registry)) + durations = _load_durations(durations_path) + records = categorize_all(tests_dir, Path(options.registry)) if options.audit: auto = [r for r in records if r.source == "auto"] print(f"Auto-inferred (unclassified) records: {len(auto)}") @@ -251,7 +236,7 @@ def main() -> int: bad = [r for r in auto if len(r.subsystems) > 1] if bad: print(f"STRICT: {len(bad)} auto-inferred files have multiple subsystems:") - for r in bad: + for r in bad: print(f" {r.filename}: subs={r.subsystems}") return 1 return 0