Log in
benchmarksPublic

Viewing files at commit a3ea065926c9Commit detailsView latest

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