| 1 | from __future__ import annotations
|
| 2 |
|
| 3 | import contextlib
|
| 4 | import io
|
| 5 | import json
|
| 6 | import sys
|
| 7 | import tempfile
|
| 8 | import unittest
|
| 9 | from pathlib import Path
|
| 10 |
|
| 11 | sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
| 12 |
|
| 13 | import public_readiness
|
| 14 |
|
| 15 |
|
| 16 | def write_jsonl(path: Path, rows: list[dict]) -> None:
|
| 17 | path.parent.mkdir(parents=True, exist_ok=True)
|
| 18 | path.write_text("\n".join(json.dumps(row) for row in rows) + "\n")
|
| 19 |
|
| 20 |
|
| 21 | def row(subject: str, run: int, *, operation: str = "status.clean") -> dict:
|
| 22 | return {
|
| 23 | "bench_id": "test",
|
| 24 | "profile": "standard",
|
| 25 | "subject": subject,
|
| 26 | "subject_kind": "oak" if subject.startswith("oak") else "git",
|
| 27 | "subject_label": subject,
|
| 28 | "scenario": "tiny_text_100",
|
| 29 | "operation": operation,
|
| 30 | "run": run,
|
| 31 | "returncode": 0,
|
| 32 | "elapsed_ms": 10.0 + run,
|
| 33 | "command": [subject, "status"],
|
| 34 | "tool_call_count": 1,
|
| 35 | "env_isolation_version": "1",
|
| 36 | "randomized_subject_order": True,
|
| 37 | "subject_order": ["git", "git_untracked_cache", "git_fsmonitor", "oak_main"],
|
| 38 | "source": {"oak_hash": "abc123", "oak_status": ""},
|
| 39 | }
|
| 40 |
|
| 41 |
|
| 42 | class PublicReadinessTest(unittest.TestCase):
|
| 43 | def test_writes_readiness_bundle_for_same_inputs(self) -> None:
|
| 44 | with tempfile.TemporaryDirectory() as tmp:
|
| 45 | root = Path(tmp)
|
| 46 | results = root / "results"
|
| 47 | out_dir = root / "bundle"
|
| 48 | rows = []
|
| 49 | for run in range(30):
|
| 50 | rows.extend(
|
| 51 | [
|
| 52 | row("git", run),
|
| 53 | row("git_untracked_cache", run),
|
| 54 | row("git_fsmonitor", run),
|
| 55 | row("oak_main", run),
|
| 56 | ]
|
| 57 | )
|
| 58 | write_jsonl(results / "latest.jsonl", rows)
|
| 59 | try:
|
| 60 | from tests.test_conformance_gate import healthy_rows
|
| 61 | except ImportError: # discover -s tests
|
| 62 | from test_conformance_gate import healthy_rows
|
| 63 | from oakbench import conformance_gate
|
| 64 | write_jsonl(results / "conformance" / "latest.jsonl",
|
| 65 | [dict(r, binary_sha256=None) for r in healthy_rows(conformance_gate.load_expected())])
|
| 66 |
|
| 67 | stdout = io.StringIO()
|
| 68 | with contextlib.redirect_stdout(stdout):
|
| 69 | code = public_readiness.main(
|
| 70 | [
|
| 71 | "--results",
|
| 72 | str(results),
|
| 73 | "--out-dir",
|
| 74 | str(out_dir),
|
| 75 | "--bootstrap-samples",
|
| 76 | "10",
|
| 77 | ]
|
| 78 | )
|
| 79 |
|
| 80 | self.assertEqual(code, 0, stdout.getvalue())
|
| 81 | manifest = json.loads((out_dir / "manifest.json").read_text())
|
| 82 | self.assertTrue(manifest["ready"])
|
| 83 | self.assertEqual(manifest["inputs"], [str((results / "latest.jsonl").resolve()),
|
| 84 | str((results / "conformance" / "latest.jsonl").resolve())])
|
| 85 | self.assertTrue((out_dir / "publish_gate.txt").exists())
|
| 86 | self.assertTrue((out_dir / "benchmark_stats.md").exists())
|
| 87 | self.assertTrue((out_dir / "opportunities.md").exists())
|
| 88 | self.assertTrue((out_dir / "opportunities.json").exists())
|
| 89 |
|
| 90 | def test_publish_gate_failure_marks_bundle_not_ready(self) -> None:
|
| 91 | with tempfile.TemporaryDirectory() as tmp:
|
| 92 | root = Path(tmp)
|
| 93 | results = root / "results"
|
| 94 | out_dir = root / "bundle"
|
| 95 | write_jsonl(results / "latest.jsonl", [row("git", 0)])
|
| 96 |
|
| 97 | stdout = io.StringIO()
|
| 98 | with contextlib.redirect_stdout(stdout):
|
| 99 | code = public_readiness.main(
|
| 100 | [
|
| 101 | "--results",
|
| 102 | str(results),
|
| 103 | "--out-dir",
|
| 104 | str(out_dir),
|
| 105 | "--skip-stats",
|
| 106 | "--skip-opportunities",
|
| 107 | ]
|
| 108 | )
|
| 109 |
|
| 110 | self.assertEqual(code, 1)
|
| 111 | manifest = json.loads((out_dir / "manifest.json").read_text())
|
| 112 | self.assertFalse(manifest["ready"])
|
| 113 | gate = next(check for check in manifest["checks"] if check["name"] == "publish_gate")
|
| 114 | self.assertEqual(gate["exit_code"], 1)
|
| 115 |
|
| 116 |
|
| 117 | if __name__ == "__main__":
|
| 118 | unittest.main()
|