Log in
benchmarksPublic

Viewing files at commit 3d84c25e26dbCommit detailsView latest

Browse files
tests/test_public_claim_gate.py 265 lines · 10.6 KB · python
1
from __future__ import annotations
2
3
import json
4
import subprocess
5
import sys
6
import tempfile
7
import unittest
8
from pathlib import Path
9
10
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
11
12
import public_claim_gate
13
14
15
def write_json(path: Path, payload: dict) -> Path:
16
    path.write_text(json.dumps(payload, sort_keys=True) + "\n")
17
    return path
18
19
20
def write_rows(path: Path, rows: list[dict]) -> Path:
21
    path.write_text("".join(json.dumps(row, sort_keys=True) + "\n" for row in rows))
22
    return path
23
24
25
def spec(path: Path, claim: dict) -> Path:
26
    return write_json(path, {"schema_version": 1, "claims": [claim]})
27
28
29
def fleet_row(platform: str, elapsed_ms: float, **extra) -> dict:
30
    row = {
31
        "scenario": "branch_fleet_n100",
32
        "operation": "fleet.total",
33
        "platform": platform,
34
        "driver": "cli",
35
        "returncode": 0,
36
        "elapsed_ms": elapsed_ms,
37
        "json_oracle_passed": True,
38
        "cleanup_failures": 0,
39
        "remote_transport": "network",
40
    }
41
    row.update(extra)
42
    return row
43
44
45
class PublicClaimGateTest(unittest.TestCase):
46
    def test_absolute_claim_passes_within_tolerance(self) -> None:
47
        claim = {
48
            "id": "branch_fleet_wall",
49
            "row_filter": {"scenario": "branch_fleet_n100", "operation": "fleet.total", "platform": "oak"},
50
            "metric": "elapsed_ms",
51
            "statistic": "median",
52
            "guards": {"fields_true": ["json_oracle_passed"], "fields_zero": ["cleanup_failures"]},
53
            "expected": {"field": "value", "value": 120.0, "tolerance_pct": 5},
54
        }
55
        result = public_claim_gate.evaluate_claim(
56
            claim,
57
            [fleet_row("oak", 118.0), fleet_row("oak", 122.0)],
58
        )
59
        self.assertEqual(result["verdict"], "PASS")
60
        self.assertEqual(result["value"], 120.0)
61
62
    def test_drift_fails(self) -> None:
63
        claim = {
64
            "id": "branch_fleet_wall",
65
            "row_filter": {"scenario": "branch_fleet_n100", "operation": "fleet.total", "platform": "oak"},
66
            "metric": "elapsed_ms",
67
            "expected": {"field": "value", "value": 100.0, "tolerance_pct": 5},
68
        }
69
        result = public_claim_gate.evaluate_claim(claim, [fleet_row("oak", 120.0)])
70
        self.assertEqual(result["verdict"], "FAIL")
71
        self.assertIn("drifted", "; ".join(result["errors"]))
72
73
    def test_ratio_claim_rederives_baseline_and_rejects_transport_mismatch(self) -> None:
74
        claim = {
75
            "id": "oak_vs_github",
76
            "row_filter": {"scenario": "branch_fleet_n100", "operation": "fleet.total", "platform": "oak"},
77
            "baseline_filter": {
78
                "scenario": "branch_fleet_n100",
79
                "operation": "fleet.total",
80
                "platform": "github",
81
            },
82
            "metric": "elapsed_ms",
83
            "expected": {"field": "ratio", "value": 0.5, "tolerance_pct": 1},
84
        }
85
        result = public_claim_gate.evaluate_claim(
86
            claim,
87
            [
88
                fleet_row("oak", 50.0, remote_transport="network"),
89
                fleet_row("github", 100.0, remote_transport="local_file"),
90
            ],
91
        )
92
        self.assertEqual(result["verdict"], "FAIL")
93
        self.assertIn("transport mismatch", "; ".join(result["errors"]))
94
95
    def test_ratio_claim_can_pass_when_transports_match(self) -> None:
96
        claim = {
97
            "id": "oak_vs_github",
98
            "row_filter": {"scenario": "branch_fleet_n100", "operation": "fleet.total", "platform": "oak"},
99
            "baseline_filter": {
100
                "scenario": "branch_fleet_n100",
101
                "operation": "fleet.total",
102
                "platform": "github",
103
            },
104
            "metric": "elapsed_ms",
105
            "expected": {"field": "ratio", "value": 0.5, "tolerance_pct": 1},
106
        }
107
        result = public_claim_gate.evaluate_claim(
108
            claim,
109
            [fleet_row("oak", 50.0), fleet_row("github", 100.0)],
110
        )
111
        self.assertEqual(result["verdict"], "PASS")
112
        self.assertEqual(result["ratio"], 0.5)
113
114
    def test_oracle_and_cleanup_guards_fail_closed(self) -> None:
115
        claim = {
116
            "id": "branch_fleet_oracle",
117
            "row_filter": {"scenario": "branch_fleet_n100", "operation": "fleet.total", "platform": "oak"},
118
            "metric": "elapsed_ms",
119
            "guards": {"fields_true": ["json_oracle_passed"], "fields_zero": ["cleanup_failures"]},
120
            "expected": {"field": "value", "min": 1},
121
        }
122
        result = public_claim_gate.evaluate_claim(
123
            claim,
124
            [fleet_row("oak", 120.0, json_oracle_passed=False, cleanup_failures=1)],
125
        )
126
        self.assertEqual(result["verdict"], "FAIL")
127
        errors = "; ".join(result["errors"])
128
        self.assertIn("json_oracle_passed", errors)
129
        self.assertIn("cleanup_failures", errors)
130
131
    def test_oracle_and_cleanup_fields_fail_closed_even_without_optional_guards(self) -> None:
132
        claim = {
133
            "id": "branch_fleet_oracle_default",
134
            "row_filter": {"scenario": "branch_fleet_n100", "operation": "fleet.total", "platform": "oak"},
135
            "metric": "elapsed_ms",
136
            "expected": {"field": "value", "min": 1},
137
        }
138
        result = public_claim_gate.evaluate_claim(
139
            claim,
140
            [fleet_row("oak", 120.0, json_oracle_passed=False, cleanup_failures=1)],
141
        )
142
        self.assertEqual(result["verdict"], "FAIL")
143
        errors = "; ".join(result["errors"])
144
        self.assertIn("json_oracle_passed", errors)
145
        self.assertIn("cleanup_failures", errors)
146
147
    def test_cleanup_failure_count_field_fails_closed(self) -> None:
148
        claim = {
149
            "id": "branch_fleet_cleanup_count",
150
            "row_filter": {"scenario": "branch_fleet_n100", "operation": "fleet.total", "platform": "oak"},
151
            "metric": "elapsed_ms",
152
            "expected": {"field": "value", "min": 1},
153
        }
154
        result = public_claim_gate.evaluate_claim(
155
            claim,
156
            [fleet_row("oak", 120.0, cleanup_failure_count=1)],
157
        )
158
        self.assertEqual(result["verdict"], "FAIL")
159
        self.assertIn("cleanup_failure_count", "; ".join(result["errors"]))
160
161
    def test_fake_provider_rows_are_rejected_by_default(self) -> None:
162
        claim = {
163
            "id": "branch_fleet_fake",
164
            "row_filter": {"scenario": "branch_fleet_n100", "operation": "fleet.total"},
165
            "metric": "elapsed_ms",
166
            "expected": {"field": "value", "min": 0},
167
        }
168
        result = public_claim_gate.evaluate_claim(
169
            claim,
170
            [fleet_row("oak", 0.1, driver="fake-provider")],
171
        )
172
        self.assertEqual(result["verdict"], "FAIL")
173
        self.assertIn("fake-provider", "; ".join(result["errors"]))
174
175
    def test_invalid_filter_operator_fails_closed(self) -> None:
176
        claim = {
177
            "id": "bad_filter",
178
            "row_filter": {"scenario": {"equals": "branch_fleet_n100"}},
179
            "metric": "elapsed_ms",
180
            "expected": {"field": "value", "min": 1},
181
        }
182
        with self.assertRaises(public_claim_gate.ClaimError):
183
            public_claim_gate.evaluate_claim(claim, [fleet_row("oak", 120.0)])
184
185
    def test_partial_metric_coverage_fails(self) -> None:
186
        claim = {
187
            "id": "partial_metric",
188
            "row_filter": {"scenario": "branch_fleet_n100", "operation": "fleet.total", "platform": "oak"},
189
            "metric": "elapsed_ms",
190
            "expected": {"field": "value", "value": 120.0},
191
        }
192
        unmeasured = fleet_row("oak", 0.0)
193
        unmeasured["elapsed_ms"] = None
194
        result = public_claim_gate.evaluate_claim(claim, [fleet_row("oak", 120.0), unmeasured])
195
        self.assertEqual(result["verdict"], "FAIL")
196
        self.assertIn("measured on 1 of 2", "; ".join(result["errors"]))
197
198
    def test_absolute_claim_rejects_mixed_transports(self) -> None:
199
        claim = {
200
            "id": "mixed_transport",
201
            "row_filter": {"scenario": "branch_fleet_n100", "operation": "fleet.total", "platform": "oak"},
202
            "metric": "elapsed_ms",
203
            "expected": {"field": "value", "min": 1},
204
        }
205
        result = public_claim_gate.evaluate_claim(
206
            claim,
207
            [
208
                fleet_row("oak", 120.0, remote_transport="network"),
209
                fleet_row("oak", 121.0, remote_transport="local_file"),
210
            ],
211
        )
212
        self.assertEqual(result["verdict"], "FAIL")
213
        self.assertIn("mix transport identities", "; ".join(result["errors"]))
214
215
    def test_remote_rows_without_transport_fail(self) -> None:
216
        claim = {
217
            "id": "missing_transport",
218
            "row_filter": {"scenario": "branch_fleet_n100", "operation": "fleet.total", "platform": "oak"},
219
            "metric": "elapsed_ms",
220
            "expected": {"field": "value", "min": 1},
221
        }
222
        row = fleet_row("oak", 120.0, remote_repo="oak/bench-fleet-tmp")
223
        row.pop("remote_transport")
224
        result = public_claim_gate.evaluate_claim(claim, [row])
225
        self.assertEqual(result["verdict"], "FAIL")
226
        self.assertIn("missing transport", "; ".join(result["errors"]))
227
228
    def test_workspace_transport_counts_as_transport_identity(self) -> None:
229
        claim = {
230
            "id": "workspace_transport",
231
            "row_filter": {"scenario": "branch_fleet_n100", "operation": "fleet.total", "platform": "oak"},
232
            "metric": "elapsed_ms",
233
            "expected": {"field": "value", "value": 120.0},
234
        }
235
        row = fleet_row("oak", 120.0, remote_repo="oak/bench-fleet-tmp")
236
        row.pop("remote_transport")
237
        row["workspace_transport"] = "network"
238
        result = public_claim_gate.evaluate_claim(claim, [row])
239
        self.assertEqual(result["verdict"], "PASS")
240
241
    def test_cli_exits_nonzero_for_failed_claim(self) -> None:
242
        with tempfile.TemporaryDirectory() as tmp:
243
            root = Path(tmp)
244
            rows = write_rows(root / "rows.jsonl", [fleet_row("oak", 120.0)])
245
            claims = spec(
246
                root / "claims.json",
247
                {
248
                    "id": "drift",
249
                    "row_filter": {"scenario": "branch_fleet_n100", "operation": "fleet.total", "platform": "oak"},
250
                    "metric": "elapsed_ms",
251
                    "expected": {"field": "value", "value": 100.0},
252
                },
253
            )
254
            proc = subprocess.run(
255
                [sys.executable, "scripts/public_claim_gate.py", "--claims", str(claims), str(rows)],
256
                cwd=Path(__file__).resolve().parents[1],
257
                text=True,
258
                capture_output=True,
259
            )
260
            self.assertEqual(proc.returncode, 1)
261
            self.assertIn("FAIL   drift", proc.stdout)
262
263
264
if __name__ == "__main__":
265
    unittest.main()