ed/manual_slop
Private
Public Access
0
0
Fork 0
Code Activity

review batch script

Browse Source
This commit is contained in:
ed
2026-06-15 20:02:36 -04:00
parent 25d047fa75
commit 4419922bce
1 changed files with 70 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/run_tests_batched.py
+69 -84
View File
@@ -12,7 +12,7 @@ try:
except ImportError: except ImportError:
_HAS_XDIST = False _HAS_XDIST = False
_SCRIPT_DIR = Path(__file__).resolve().parent _SCRIPT_DIR = Path(__file__).resolve().parent
_PROJECT_ROOT = _SCRIPT_DIR.parent _PROJECT_ROOT = _SCRIPT_DIR.parent
sys.path.insert(0, str(_PROJECT_ROOT / "tests")) sys.path.insert(0, str(_PROJECT_ROOT / "tests"))
@@ -26,19 +26,19 @@ if _USE_COLOR and os.name == "nt":
_USE_COLOR = False _USE_COLOR = False
class _C: class _C:
RESET = "\033[0m" RESET = "\033[0m"
BOLD = "\033[1m" BOLD = "\033[1m"
DIM = "\033[2m" DIM = "\033[2m"
RED = "\033[31m" RED = "\033[31m"
GREEN = "\033[32m" GREEN = "\033[32m"
YELLOW = "\033[33m" YELLOW = "\033[33m"
BLUE = "\033[34m" BLUE = "\033[34m"
MAGENTA = "\033[35m" MAGENTA = "\033[35m"
CYAN = "\033[36m" CYAN = "\033[36m"
BOLD_GREEN = "\033[1;32m" BOLD_GREEN = "\033[1;32m"
BOLD_RED = "\033[1;31m" BOLD_RED = "\033[1;31m"
BOLD_YELLOW = "\033[1;33m" BOLD_YELLOW = "\033[1;33m"
BOLD_CYAN = "\033[1;36m" BOLD_CYAN = "\033[1;36m"
def _c(text: str, color: str) -> str: def _c(text: str, color: str) -> str:
if not _USE_COLOR: if not _USE_COLOR:
@@ -110,138 +110,123 @@ def _format_pytest_line(line: str) -> str | None:
return None return None
if stripped.startswith(("tests/", "tests\\")) and "::" in stripped and len(stripped.split()) == 1: if stripped.startswith(("tests/", "tests\\")) and "::" in stripped and len(stripped.split()) == 1:
return None return None
if " PASSED " in stripped and "[gw" in stripped: if " PASSED " in stripped and "[gw" in stripped: return _c(stripped, _C.GREEN)
return _c(stripped, _C.GREEN) if " FAILED " in stripped and "[gw" in stripped: return _c(stripped, _C.BOLD_RED)
if " FAILED " in stripped and "[gw" in stripped: if " ERROR " in stripped and "[gw" in stripped: return _c(stripped, _C.BOLD_RED)
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 stripped.startswith(("tests/", "tests\\")) and "::" in stripped:
if " PASSED" in stripped: if " PASSED" in stripped: return _c(stripped, _C.GREEN)
return _c(stripped, _C.GREEN) if " FAILED" in stripped: return _c(stripped, _C.BOLD_RED)
if " FAILED" in stripped: if " ERROR" in stripped: return _c(stripped, _C.BOLD_RED)
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: if stripped.startswith(("PASSED", "FAILED", "ERROR")) and "::" in stripped:
status = stripped.split()[0] status = stripped.split()[0]
rest = stripped[len(status):] res = stripped[len(status):]
if status == "PASSED": if status == "PASSED": return _c(f"{status}{rest}", _C.GREEN)
return _c(f"{status}{rest}", _C.GREEN)
return _c(f"{status}{rest}", _C.BOLD_RED) return _c(f"{status}{rest}", _C.BOLD_RED)
if stripped.startswith(("passed", "failed", "error")) and " in " in stripped and stripped.endswith("s"): if stripped.startswith(("passed", "failed", "error")) and " in " in stripped and stripped.endswith("s"):
return _c(stripped, _C.BOLD) return _c(stripped, _C.BOLD)
return stripped return stripped
def _run_batch(b: Batch, durations: dict[str, float]) -> tuple[int, float, dict[str, float]]: def _run_batch(b: Batch, durations: dict[str, float]) -> tuple[int, float, dict[str, float]]:
if b.skip_reason: if b.skip_reason: return 0, 0.0, {}
return 0, 0.0, {}
args = list(b.pytest_args) args = list(b.pytest_args)
if not _HAS_XDIST: if not _HAS_XDIST: args = [a for a in args if a not in {"-n", "auto"}]
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] 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)) 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) proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1)
captured: list[str] = [] captured: list[str] = []
assert proc.stdout is not None assert proc.stdout is not None
for line in proc.stdout: for line in proc.stdout:
captured.append(line) captured.append(line)
formatted = _format_pytest_line(line) formatted = _format_pytest_line(line)
if formatted is None: if formatted is None: continue
continue
print(formatted) print(formatted)
proc.wait() proc.wait()
elapsed = time.monotonic() - t0 elapsed = time.monotonic() - t0
new_durs = _parse_durations_from_pytest_output("".join(captured)) new_durs = _parse_durations_from_pytest_output("".join(captured))
captured_text = "".join(captured) captured_text = "".join(captured)
saw_failure = "FAILED " in captured_text or " stopping after " in captured_text 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) effective_code = proc.returncode if proc.returncode != 0 else (1 if saw_failure else 0)
if effective_code == 0: if effective_code == 0: print(_c(f"<<< {b.label} PASS in {elapsed:.1f}s", _C.BOLD_GREEN))
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))
else:
print(_c(f"<<< {b.label} FAIL (exit {effective_code}) in {elapsed:.1f}s", _C.BOLD_RED))
return effective_code, elapsed, new_durs return effective_code, elapsed, new_durs
def _print_summary(results: list[tuple[Batch, int, float]]) -> int: def _print_summary(results: list[tuple[Batch, int, float]]) -> int:
print() print()
rows: list[tuple[str, str, str, int, float, int]] = [] rows: list[tuple[str, str, str, int, float, int]] = []
worst = 0 worst = 0
total_files = 0 total_files = 0
total_time = 0.0 total_time = 0.0
passed_count = 0 passed_count = 0
failed_count = 0 failed_count = 0
skipped_count = 0 skipped_count = 0
for b, code, elapsed in results: for b, code, elapsed in results:
n = len(b.files) n = len(b.files)
total_files += n total_files += n
total_time += elapsed total_time += elapsed
if b.skip_reason: if b.skip_reason:
status_text = "SKIPPED" status_text = "SKIPPED"
skipped_count += 1 skipped_count += 1
elif code == 0: elif code == 0:
status_text = "PASS" status_text = "PASS"
passed_count += 1 passed_count += 1
else: else:
status_text = "FAIL" status_text = "FAIL"
failed_count += 1 failed_count += 1
worst = max(worst, code) worst = max(worst, code)
rows.append((b.tier, b.label, status_text, n, elapsed, 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)) 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)) 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)) 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)) 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)) 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} " 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) sep = "─" * len(header)
print(_c(sep, _C.DIM)) print(_c(sep, _C.DIM))
print(_c(header, _C.BOLD)) 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: for tier, label, status_text, n, elapsed, _code in rows:
if status_text == "PASS": if status_text == "PASS": status = _c(status_text, _C.BOLD_GREEN)
status = _c(status_text, _C.BOLD_GREEN) elif status_text == "FAIL": status = _c(status_text, _C.BOLD_RED)
elif status_text == "FAIL": else: status = _c(status_text, _C.BOLD_YELLOW)
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) 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(f"{tier_colored} │ {label:<{label_w}s} │ {status} │ {n:>{files_w}d} │ {elapsed:>{time_w - 1}.1f}s")
print(_c(sep, _C.DIM)) print(_c(sep, _C.DIM))
if failed_count: if failed_count:
overall_text = f"{failed_count} FAILED" overall_text = f"{failed_count} FAILED"
overall = _c(overall_text, _C.BOLD_RED) overall = _c(overall_text, _C.BOLD_RED)
elif passed_count: elif passed_count:
overall_text = f"ALL {passed_count} PASS" overall_text = f"ALL {passed_count} PASS"
overall = _c(overall_text, _C.BOLD_GREEN) overall = _c(overall_text, _C.BOLD_GREEN)
else: else:
overall_text = "NO BATCHES RUN" overall_text = "NO BATCHES RUN"
overall = _c(overall_text, _C.BOLD_YELLOW) overall = _c(overall_text, _C.BOLD_YELLOW)
total_label = _c(f" {'TOTAL':<{tier_w}s}", _C.BOLD) 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(f"{total_label} │ {'':<{label_w}s} │ {overall} │ {total_files:>{files_w}d} │ {total_time:>{time_w - 1}.1f}s")
print(_c(sep, _C.DIM)) print(_c(sep, _C.DIM))
return worst return worst
def main() -> int: def main() -> int:
p = argparse.ArgumentParser() p = argparse.ArgumentParser()
p.add_argument("--tests-dir", default=str(_PROJECT_ROOT / "tests")) 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("--registry", default=str(_PROJECT_ROOT / "tests" / "test_categories.toml"))
p.add_argument("--tiers", default="1,2,3,H") p.add_argument("--tiers", default="1,2,3,H")
p.add_argument("--include-opt-in", action="store_true") p.add_argument("--include-opt-in", action="store_true")
p.add_argument("--no-xdist", action="store_true") p.add_argument("--no-xdist", action="store_true")
p.add_argument("--plan", action="store_true") p.add_argument("--plan", action="store_true")
p.add_argument("--audit", action="store_true") p.add_argument("--audit", action="store_true")
p.add_argument("--strict", 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("--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-color", action="store_true", help="Disable ANSI color output")
options = p.parse_args() options = p.parse_args()
if options.no_color: if options.no_color:
global _USE_COLOR global _USE_COLOR
_USE_COLOR = False _USE_COLOR = False
tiers = _parse_tiers(options.tiers) tiers = _parse_tiers(options.tiers)
tests_dir = Path(options.tests_dir) if Path(options.tests_dir).is_absolute() else (_PROJECT_ROOT / options.tests_dir) 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_path = _durations_path(tests_dir)
durations = _load_durations(durations_path) durations = _load_durations(durations_path)
records = categorize_all(tests_dir, Path(options.registry)) records = categorize_all(tests_dir, Path(options.registry))
if options.audit: if options.audit:
auto = [r for r in records if r.source == "auto"] auto = [r for r in records if r.source == "auto"]
print(f"Auto-inferred (unclassified) records: {len(auto)}") print(f"Auto-inferred (unclassified) records: {len(auto)}")