#!/usr/bin/env python3
"""Algo Dojo judge: runs every solution against its problem's tests.
Contract (see README.md):
problems/<slug>/statement.md
problems/<slug>/tests.json [{"in": "stdin...", "out": "expected stdout..."}]
solutions/<slug>/<name>.py|.js|.rs
Prints `FAIL solutions/<slug>/<file> ...` lines for the build gate, writes
a full report to judge_report.md, exits 0 only if everything passes.
ALWAYS run via the pixi env: `pixi run python judge.py` (the house rule —
no system python). Boss console `judge` does it for you.
"""
import json
import shutil
import subprocess
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parent
TIMEOUT = 10
def norm(text: str) -> str:
return "\n".join(line.rstrip() for line in text.strip().splitlines())
def command_for(src: Path):
"""Return the command to run a solution file (or None if unsupported)."""
ext = src.suffix.lower()
if ext == ".py":
py = shutil.which("python3") or shutil.which("python")
return [py, str(src)] if py else None
if ext == ".js":
node = shutil.which("node")
return [node, str(src)] if node else None
if ext == ".rs":
binary = src.with_suffix("")
if not binary.exists():
rustc = shutil.which("rustc")
if not rustc:
return None
build = subprocess.run(
[rustc, "-O", str(src), "-o", str(binary)],
capture_output=True, text=True, timeout=90,
)
if build.returncode != 0:
return ("compile", build.stderr[-400:])
return [str(binary)]
return None
def run_case(cmd, stdin_text: str, timeout: int):
try:
p = subprocess.run(
cmd, input=stdin_text, capture_output=True, text=True, timeout=timeout
)
return p.returncode == 0, p.stdout, p, False
except subprocess.TimeoutExpired:
return False, "", None, True
def main():
problems_root = ROOT / "problems"
solutions_root = ROOT / "solutions"
lines = []
pass_n = fail_n = 0
if not problems_root.is_dir():
print("no problems/ yet — nothing to judge")
return 0
for tests_path in sorted(problems_root.glob("*/tests.json")):
slug_dir = tests_path.parent
slug = slug_dir.name
sol_dir = solutions_root / slug
try:
tests = json.loads(tests_path.read_text())
if not isinstance(tests, list) or not tests:
raise ValueError("empty/invalid tests array")
except (ValueError, OSError) as e:
fail_n += 1
print(f"FAIL problems/{slug}/tests.json (invalid: {type(e).__name__})")
lines.append(f"## {slug} — BAD TESTS\n`{tests_path.name}`: {e}\n")
continue
lines.append(f"## {slug} — {len(tests)} test(s)")
if not sol_dir.is_dir():
lines.append(f"- unsolved (no solutions/{slug}/ yet)")
lines.append("")
continue
for src in sorted(sol_dir.iterdir()):
if src.is_dir() or src.suffix.lower() in (".md", ".txt", ""):
continue
got = command_for(src)
if got is None:
lines.append(f"- `[unsupported]` {src.name}")
continue
if isinstance(got, tuple):
fail_n += 1
print(f"FAIL solutions/{slug}/{src.name} (compile error)")
lines.append(f"- **FAIL (compile)** `{src.name}`\n```\n{got[1]}\n```")
continue
rel = f"solutions/{slug}/{src.name}"
failed = False
detail = []
for i, case in enumerate(tests, 1):
ok, out, proc, timed_out = run_case(got, case.get("in", ""), TIMEOUT)
want = norm(case.get("out", ""))
if timed_out:
failed = True
detail.append(f"case {i}: TIMEOUT after {TIMEOUT}s")
elif proc.returncode != 0:
failed = True
err = (proc.stderr or "").strip().splitlines()
detail.append(
f"case {i}: exited {proc.returncode}: "
+ (err[-1] if err else "(no output)")
)
elif norm(out) != want:
failed = True
detail.append(f'case {i}: got "{norm(out)[:80]}", want "{want[:80]}"')
if failed:
fail_n += 1
print(f"FAIL {rel} ({len(detail)}/{len(tests)} cases)")
lines.append(f"- **FAIL** `{src.name}`\n" + "\n".join(f" {d}" for d in detail))
else:
pass_n += 1
print(f"PASS {rel} ({len(tests)}/{len(tests)} cases)")
lines.append(f"- PASS `{src.name}` ({len(tests)}/{len(tests)})")
lines.append("")
report = "\n".join(lines) or "(nothing to judge)"
(ROOT / "judge_report.md").write_text(
f"# judge report\n\n{pass_n} pass, {fail_n} fail\n\n{report}\n"
)
console = f"judge: {pass_n} pass, {fail_n} fail"
print(console)
return 1 if fail_n else 0
if __name__ == "__main__":
sys.exit(main())