diff --git a/scripts/run_tests_batched.py b/scripts/run_tests_batched.py index a77a40e3..7cc26a55 100644 --- a/scripts/run_tests_batched.py +++ b/scripts/run_tests_batched.py @@ -85,13 +85,50 @@ def _parse_durations_from_pytest_output(stdout: str) -> dict[str, float]: continue return out +_NOISE_PREFIXES: tuple[str, ...] = ( + "[LogPruner]", + "[startup]", + "created: ", + "=========", +) + +_NOISE_SUBSTRINGS: tuple[str, ...] = ( + "[WinError", +) + +def _format_pytest_line(line: str) -> str | None: + stripped = line.rstrip() + if not stripped: + return None + for prefix in _NOISE_PREFIXES: + if stripped.startswith(prefix): + return None + for sub in _NOISE_SUBSTRINGS: + if sub in stripped: + 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 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) + 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, {} args = list(b.pytest_args) if not _HAS_XDIST: args = [a for a in args if a not in {"-n", "auto"}] - cmd = ["uv", "run", "pytest", "-v", "--durations=0"] + args + [str(f) for f in b.files] + 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() proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1) @@ -99,7 +136,10 @@ def _run_batch(b: Batch, durations: dict[str, float]) -> tuple[int, float, dict[ assert proc.stdout is not None for line in proc.stdout: captured.append(line) - print(line, end="") + formatted = _format_pytest_line(line) + if formatted is None: + continue + print(formatted) proc.wait() elapsed = time.monotonic() - t0 new_durs = _parse_durations_from_pytest_output("".join(captured)) @@ -110,20 +150,61 @@ def _run_batch(b: Batch, durations: dict[str, float]) -> tuple[int, float, dict[ return proc.returncode, elapsed, new_durs def _print_summary(results: list[tuple[Batch, int, float]]) -> int: - print("\n" + "=" * 60) - print("SUMMARY") - print("=" * 60) + 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 + skipped_count = 0 for b, code, elapsed in results: - if b.skip_reason: - status = _c("SKIPPED", _C.BOLD_YELLOW) - elif code == 0: - status = _c("PASS", _C.BOLD_GREEN) - else: - status = _c("FAIL", _C.BOLD_RED) - worst = max(worst, code) n = len(b.files) - print(f"[{b.tier}] {b.label:40s} {status} {n} files {elapsed:6.1f}s") + total_files += n + total_time += elapsed + if b.skip_reason: + status_text = "SKIPPED" + skipped_count += 1 + elif code == 0: + status_text = "PASS" + passed_count += 1 + else: + 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)) + print(_c(header, _C.BOLD)) + 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) + 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) + elif passed_count: + overall_text = f"ALL {passed_count} PASS" + 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) + 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: