| 1 | """Correctness attestation for benchmark rows.
|
| 2 |
|
| 3 | These tests pin the failure mode where a subject binary exits 0 without
|
| 4 | persisting repository state.
|
| 5 | """
|
| 6 |
|
| 7 | from __future__ import annotations
|
| 8 |
|
| 9 | import hashlib
|
| 10 | import os
|
| 11 | import shutil
|
| 12 | import subprocess
|
| 13 | import sys
|
| 14 | import tempfile
|
| 15 | import unittest
|
| 16 | from pathlib import Path
|
| 17 |
|
| 18 | sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
| 19 |
|
| 20 | import bench
|
| 21 | import workflow_ab
|
| 22 | from oakbench.integrity import (
|
| 23 | content_integrity_fields,
|
| 24 | content_integrity_public_trust_passed,
|
| 25 | parse_manifest,
|
| 26 | worktree_payload_hashes,
|
| 27 | )
|
| 28 | from oakbench.subjects import Subject
|
| 29 |
|
| 30 | GIT = shutil.which("git")
|
| 31 | LFS_POINTER = (
|
| 32 | b"version https://git-lfs.github.com/spec/v1\n"
|
| 33 | b"oid sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef\n"
|
| 34 | b"size 14\n"
|
| 35 | )
|
| 36 |
|
| 37 |
|
| 38 | class ManifestParsingTest(unittest.TestCase):
|
| 39 | def test_parse_manifest_accepts_sha256sum_spacing_and_lowercases_digest(self) -> None:
|
| 40 | digest = "A" * 64
|
| 41 | self.assertEqual(
|
| 42 | parse_manifest(f"{digest} README.md\n{'b' * 64} docs/one.md\ninvalid\n"),
|
| 43 | {
|
| 44 | "README.md": digest.lower(),
|
| 45 | "docs/one.md": "b" * 64,
|
| 46 | },
|
| 47 | )
|
| 48 |
|
| 49 |
|
| 50 | class ContentIntegrityHelperTest(unittest.TestCase):
|
| 51 | def test_content_integrity_fields_detects_hash_mismatch(self) -> None:
|
| 52 | with tempfile.TemporaryDirectory(prefix="oakbench-content-helper-") as raw:
|
| 53 | root = Path(raw)
|
| 54 | payload = root / "payload.txt"
|
| 55 | payload.write_text("expected\n")
|
| 56 | expected = worktree_payload_hashes(root, ["payload.txt"])
|
| 57 | payload.write_text("corrupt\n")
|
| 58 |
|
| 59 | fields = content_integrity_fields(
|
| 60 | root,
|
| 61 | operation="snapshot.initial",
|
| 62 | process_returncode=0,
|
| 63 | expected_hashes=expected,
|
| 64 | )
|
| 65 |
|
| 66 | self.assertFalse(fields["content_integrity_check_passed"])
|
| 67 | self.assertIn("sha256_mismatch", fields["content_integrity_failure_reason"])
|
| 68 | self.assertEqual(fields["content_expected_files"], 1)
|
| 69 | self.assertEqual(fields["content_verified_files"], 0)
|
| 70 | self.assertFalse(fields["content_integrity_public_trust_passed"])
|
| 71 |
|
| 72 | def test_worktree_source_is_measured_but_not_public_trust(self) -> None:
|
| 73 | with tempfile.TemporaryDirectory(prefix="oakbench-content-helper-") as raw:
|
| 74 | root = Path(raw)
|
| 75 | payload = root / "payload.txt"
|
| 76 | payload.write_text("expected\n")
|
| 77 |
|
| 78 | fields = content_integrity_fields(
|
| 79 | root,
|
| 80 | operation="snapshot.initial",
|
| 81 | process_returncode=0,
|
| 82 | expected_hashes=worktree_payload_hashes(root, ["payload.txt"]),
|
| 83 | )
|
| 84 |
|
| 85 | self.assertTrue(fields["content_integrity_check_passed"])
|
| 86 | self.assertEqual(fields["content_integrity_source"], "worktree_payload_sha256")
|
| 87 | self.assertFalse(fields["content_integrity_public_trust_passed"])
|
| 88 | self.assertFalse(content_integrity_public_trust_passed(fields))
|
| 89 |
|
| 90 | @unittest.skipIf(GIT is None, "git not installed")
|
| 91 | def test_mixed_head_and_lfs_sources_report_weakest_source(self) -> None:
|
| 92 | with tempfile.TemporaryDirectory(prefix="oakbench-content-helper-") as raw:
|
| 93 | root = Path(raw)
|
| 94 | subprocess.run([GIT, "init", "-q"], cwd=root, check=True)
|
| 95 | subprocess.run([GIT, "config", "user.email", "[email protected]"], cwd=root, check=True)
|
| 96 | subprocess.run([GIT, "config", "user.name", "Oak Bench"], cwd=root, check=True)
|
| 97 | (root / "asset.bin").write_bytes(LFS_POINTER)
|
| 98 | (root / "normal.txt").write_text("normal committed bytes\n")
|
| 99 | subprocess.run([GIT, "add", "asset.bin", "normal.txt"], cwd=root, check=True)
|
| 100 | subprocess.run([GIT, "commit", "-q", "-m", "fixture"], cwd=root, check=True)
|
| 101 | (root / "asset.bin").write_bytes(b"hydrated bytes\n")
|
| 102 |
|
| 103 | fields = content_integrity_fields(
|
| 104 | root,
|
| 105 | operation="snapshot.initial",
|
| 106 | process_returncode=0,
|
| 107 | expected_hashes={
|
| 108 | "asset.bin": "a" * 64,
|
| 109 | "normal.txt": "b" * 64,
|
| 110 | },
|
| 111 | )
|
| 112 |
|
| 113 | self.assertFalse(fields["content_integrity_check_passed"])
|
| 114 | self.assertEqual(fields["content_integrity_source"], "mixed_payload_sources")
|
| 115 | self.assertEqual(
|
| 116 | fields["content_integrity_sources"],
|
| 117 | ["git_head_payload_sha256", "git_lfs_worktree_payload_sha256"],
|
| 118 | )
|
| 119 | self.assertFalse(fields["content_integrity_public_trust_passed"])
|
| 120 |
|
| 121 | @unittest.skipIf(GIT is None, "git not installed")
|
| 122 | def test_mixed_head_and_lfs_sources_can_pass_measurement_not_public_trust(self) -> None:
|
| 123 | with tempfile.TemporaryDirectory(prefix="oakbench-content-helper-") as raw:
|
| 124 | root = Path(raw)
|
| 125 | subprocess.run([GIT, "init", "-q"], cwd=root, check=True)
|
| 126 | subprocess.run([GIT, "config", "user.email", "[email protected]"], cwd=root, check=True)
|
| 127 | subprocess.run([GIT, "config", "user.name", "Oak Bench"], cwd=root, check=True)
|
| 128 | (root / "asset.bin").write_bytes(LFS_POINTER)
|
| 129 | (root / "normal.txt").write_text("normal committed bytes\n")
|
| 130 | subprocess.run([GIT, "add", "asset.bin", "normal.txt"], cwd=root, check=True)
|
| 131 | subprocess.run([GIT, "commit", "-q", "-m", "fixture"], cwd=root, check=True)
|
| 132 | (root / "asset.bin").write_bytes(b"hydrated bytes\n")
|
| 133 |
|
| 134 | fields = content_integrity_fields(
|
| 135 | root,
|
| 136 | operation="snapshot.initial",
|
| 137 | process_returncode=0,
|
| 138 | expected_hashes={
|
| 139 | "asset.bin": hashlib.sha256(b"hydrated bytes\n").hexdigest(),
|
| 140 | "normal.txt": hashlib.sha256(b"normal committed bytes\n").hexdigest(),
|
| 141 | },
|
| 142 | )
|
| 143 |
|
| 144 | self.assertTrue(fields["content_integrity_check_passed"])
|
| 145 | self.assertEqual(fields["content_integrity_source"], "mixed_payload_sources")
|
| 146 | self.assertFalse(fields["content_integrity_public_trust_passed"])
|
| 147 | self.assertFalse(content_integrity_public_trust_passed(fields))
|
| 148 |
|
| 149 | @unittest.skipIf(GIT is None, "git not installed")
|
| 150 | def test_git_lfs_worktree_source_is_measured_but_not_public_trust(self) -> None:
|
| 151 | with tempfile.TemporaryDirectory(prefix="oakbench-content-helper-") as raw:
|
| 152 | root = Path(raw)
|
| 153 | subprocess.run([GIT, "init", "-q"], cwd=root, check=True)
|
| 154 | subprocess.run([GIT, "config", "user.email", "[email protected]"], cwd=root, check=True)
|
| 155 | subprocess.run([GIT, "config", "user.name", "Oak Bench"], cwd=root, check=True)
|
| 156 | (root / "asset.bin").write_bytes(LFS_POINTER)
|
| 157 | subprocess.run([GIT, "add", "asset.bin"], cwd=root, check=True)
|
| 158 | subprocess.run([GIT, "commit", "-q", "-m", "fixture"], cwd=root, check=True)
|
| 159 | (root / "asset.bin").write_bytes(b"hydrated bytes\n")
|
| 160 |
|
| 161 | fields = content_integrity_fields(
|
| 162 | root,
|
| 163 | operation="snapshot.initial",
|
| 164 | process_returncode=0,
|
| 165 | expected_hashes={"asset.bin": hashlib.sha256(b"hydrated bytes\n").hexdigest()},
|
| 166 | )
|
| 167 |
|
| 168 | self.assertTrue(fields["content_integrity_check_passed"])
|
| 169 | self.assertEqual(fields["content_integrity_source"], "git_lfs_worktree_payload_sha256")
|
| 170 | self.assertFalse(fields["content_integrity_public_trust_passed"])
|
| 171 | self.assertFalse(content_integrity_public_trust_passed(fields))
|
| 172 |
|
| 173 |
|
| 174 | class CoreIntegrityAttestationTest(unittest.TestCase):
|
| 175 | def setUp(self) -> None:
|
| 176 | self.tmp = Path(tempfile.mkdtemp(prefix="oakbench-integrity-core-"))
|
| 177 |
|
| 178 | def tearDown(self) -> None:
|
| 179 | shutil.rmtree(self.tmp, ignore_errors=True)
|
| 180 |
|
| 181 | def fake_binary(self, body: str) -> Path:
|
| 182 | path = self.tmp / "fake-vcs"
|
| 183 | path.write_text("#!/bin/sh\n" + body)
|
| 184 | os.chmod(path, 0o755)
|
| 185 | return path
|
| 186 |
|
| 187 | def test_noop_snapshot_success_is_marked_failed(self) -> None:
|
| 188 | fake = self.fake_binary("exit 0\n")
|
| 189 | subject = Subject(name="fake_git", kind="git", label="Fake Git", bin=fake)
|
| 190 | scenario = bench.Scenario("tiny_text", 3, 64, 1, False, 1)
|
| 191 | fixture = bench.ensure_fixture(self.tmp, scenario)
|
| 192 | rows = bench.run_subject_scenario(
|
| 193 | subject,
|
| 194 | scenario,
|
| 195 | fixture,
|
| 196 | run_index=0,
|
| 197 | run_root=self.tmp / "runs",
|
| 198 | metadata={
|
| 199 | "bench_id": "test",
|
| 200 | "profile": "micro",
|
| 201 | "benchmark_track": "agent-default",
|
| 202 | "command_semantics_version": bench.COMMAND_SEMANTICS_VERSION,
|
| 203 | "timestamp_utc": "test",
|
| 204 | },
|
| 205 | skip_diff=True,
|
| 206 | admitted_output_chars=20_000,
|
| 207 | track="agent-default",
|
| 208 | skip_determinism_probe=True,
|
| 209 | skip_remote=True,
|
| 210 | )
|
| 211 |
|
| 212 | snapshot = next(row for row in rows if row["operation"] == "snapshot.initial")
|
| 213 | self.assertEqual(snapshot["process_returncode"], 0)
|
| 214 | self.assertEqual(snapshot["returncode"], 1)
|
| 215 | self.assertFalse(snapshot["integrity_check_passed"])
|
| 216 | self.assertIn("repository_metadata", snapshot["integrity_failure_reason"])
|
| 217 |
|
| 218 | def test_snapshot_success_with_payload_corruption_is_marked_failed(self) -> None:
|
| 219 | fake = self.fake_binary(
|
| 220 | "if [ \"$1\" = \"init\" ]; then mkdir -p .oak; echo init > .oak/state; exit 0; fi\n"
|
| 221 | "if [ \"$1\" = \"commit\" ]; then\n"
|
| 222 | " echo commit >> .oak/state\n"
|
| 223 | " printf 'corrupt\\n' > group-000/file-0000.txt\n"
|
| 224 | " exit 0\n"
|
| 225 | "fi\n"
|
| 226 | "exit 0\n"
|
| 227 | )
|
| 228 | subject = Subject(name="fake_oak", kind="oak", label="Fake Oak", bin=fake)
|
| 229 | scenario = bench.Scenario("tiny_text", 3, 64, 1, False, 1)
|
| 230 | fixture = bench.ensure_fixture(self.tmp, scenario)
|
| 231 | rows = bench.run_subject_scenario(
|
| 232 | subject,
|
| 233 | scenario,
|
| 234 | fixture,
|
| 235 | run_index=0,
|
| 236 | run_root=self.tmp / "runs",
|
| 237 | metadata={
|
| 238 | "bench_id": "test",
|
| 239 | "profile": "micro",
|
| 240 | "benchmark_track": "agent-default",
|
| 241 | "command_semantics_version": bench.COMMAND_SEMANTICS_VERSION,
|
| 242 | "timestamp_utc": "test",
|
| 243 | },
|
| 244 | skip_diff=True,
|
| 245 | admitted_output_chars=20_000,
|
| 246 | track="agent-default",
|
| 247 | skip_determinism_probe=True,
|
| 248 | skip_remote=True,
|
| 249 | )
|
| 250 |
|
| 251 | snapshot = next(row for row in rows if row["operation"] == "snapshot.initial")
|
| 252 | self.assertEqual(snapshot["process_returncode"], 0)
|
| 253 | self.assertTrue(snapshot["integrity_check_passed"])
|
| 254 | self.assertEqual(snapshot["returncode"], 1)
|
| 255 | self.assertFalse(snapshot["content_integrity_check_passed"])
|
| 256 | self.assertIn("sha256_mismatch", snapshot["content_integrity_failure_reason"])
|
| 257 |
|
| 258 |
|
| 259 | @unittest.skipIf(GIT is None, "git not installed")
|
| 260 | class WorkflowIntegrityAttestationTest(unittest.TestCase):
|
| 261 | def setUp(self) -> None:
|
| 262 | self.tmp = Path(tempfile.mkdtemp(prefix="oakbench-integrity-workflow-"))
|
| 263 |
|
| 264 | def tearDown(self) -> None:
|
| 265 | shutil.rmtree(self.tmp, ignore_errors=True)
|
| 266 |
|
| 267 | def fake_git_that_noops_agent_commit(self) -> Path:
|
| 268 | path = self.tmp / "fake-git"
|
| 269 | path.write_text(
|
| 270 | "#!/bin/sh\n"
|
| 271 | "for arg in \"$@\"; do\n"
|
| 272 | " if [ \"$arg\" = \"agent task\" ]; then\n"
|
| 273 | " exit 0\n"
|
| 274 | " fi\n"
|
| 275 | "done\n"
|
| 276 | f"exec {GIT} \"$@\"\n"
|
| 277 | )
|
| 278 | os.chmod(path, 0o755)
|
| 279 | return path
|
| 280 |
|
| 281 | def fake_git_that_commits_wrong_agent_payload(self) -> Path:
|
| 282 | path = self.tmp / "fake-git"
|
| 283 | path.write_text(
|
| 284 | "#!/bin/sh\n"
|
| 285 | "if [ \"$1\" = \"commit\" ]; then\n"
|
| 286 | " for arg in \"$@\"; do\n"
|
| 287 | " if [ \"$arg\" = \"agent task\" ]; then\n"
|
| 288 | " cat > app/pricing.py <<'PY'\n"
|
| 289 | "def apply_discount(cents: int, percent: int) -> int:\n"
|
| 290 | " return cents - percent\n"
|
| 291 | "PY\n"
|
| 292 | f" {GIT} add app/pricing.py\n"
|
| 293 | f" exec {GIT} \"$@\"\n"
|
| 294 | " fi\n"
|
| 295 | " done\n"
|
| 296 | "fi\n"
|
| 297 | f"exec {GIT} \"$@\"\n"
|
| 298 | )
|
| 299 | os.chmod(path, 0o755)
|
| 300 | return path
|
| 301 |
|
| 302 | def test_workflow_total_fails_when_final_commit_noops(self) -> None:
|
| 303 | subject = Subject(name="fake_git", kind="git", label="Fake Git", bin=self.fake_git_that_noops_agent_commit())
|
| 304 | workflow = workflow_ab.WORKFLOWS["bugfix_test_loop"]
|
| 305 | fixture = workflow_ab.prepare_fixture(self.tmp, workflow)
|
| 306 | rows = workflow_ab.run_subject_workflow(
|
| 307 | subject,
|
| 308 | workflow,
|
| 309 | fixture,
|
| 310 | run_index=0,
|
| 311 | run_root=self.tmp / "runs",
|
| 312 | metadata={
|
| 313 | "bench_id": "test",
|
| 314 | "profile": "workflow-ab",
|
| 315 | "benchmark_track": "agent-default",
|
| 316 | "command_semantics_version": workflow_ab.COMMAND_SEMANTICS_VERSION,
|
| 317 | "timestamp_utc": "test",
|
| 318 | },
|
| 319 | admitted_output_chars=20_000,
|
| 320 | track="agent-default",
|
| 321 | )
|
| 322 |
|
| 323 | commit = next(row for row in rows if row["operation"] == "vcs.snapshot.commit")
|
| 324 | self.assertEqual(commit["process_returncode"], 0)
|
| 325 | self.assertEqual(commit["returncode"], 1)
|
| 326 | self.assertFalse(commit["integrity_check_passed"])
|
| 327 |
|
| 328 | total = next(row for row in rows if row["operation"] == "workflow.total")
|
| 329 | self.assertEqual(total["returncode"], 1)
|
| 330 | self.assertFalse(total["workflow_integrity_passed"])
|
| 331 | self.assertFalse(total["correctness_passed"])
|
| 332 | self.assertEqual(total["steps_failed"], 1)
|
| 333 | self.assertGreater(total["step_failure_rate"], 0)
|
| 334 | self.assertIn("repository_metadata_unchanged", total["workflow_integrity_failure_reason"])
|
| 335 |
|
| 336 | def test_workflow_commit_fails_when_committed_payload_is_wrong(self) -> None:
|
| 337 | subject = Subject(
|
| 338 | name="fake_git",
|
| 339 | kind="git",
|
| 340 | label="Fake Git",
|
| 341 | bin=self.fake_git_that_commits_wrong_agent_payload(),
|
| 342 | )
|
| 343 | workflow = workflow_ab.WORKFLOWS["bugfix_test_loop"]
|
| 344 | fixture = workflow_ab.prepare_fixture(self.tmp, workflow)
|
| 345 | rows = workflow_ab.run_subject_workflow(
|
| 346 | subject,
|
| 347 | workflow,
|
| 348 | fixture,
|
| 349 | run_index=0,
|
| 350 | run_root=self.tmp / "runs",
|
| 351 | metadata={
|
| 352 | "bench_id": "test",
|
| 353 | "profile": "workflow-ab",
|
| 354 | "benchmark_track": "agent-default",
|
| 355 | "command_semantics_version": workflow_ab.COMMAND_SEMANTICS_VERSION,
|
| 356 | "timestamp_utc": "test",
|
| 357 | },
|
| 358 | admitted_output_chars=20_000,
|
| 359 | track="agent-default",
|
| 360 | )
|
| 361 |
|
| 362 | commit = next(row for row in rows if row["operation"] == "vcs.snapshot.commit")
|
| 363 | self.assertEqual(commit["process_returncode"], 0)
|
| 364 | self.assertTrue(commit["integrity_check_passed"])
|
| 365 | self.assertEqual(commit["returncode"], 1)
|
| 366 | self.assertFalse(commit["content_integrity_check_passed"])
|
| 367 | self.assertIn("app/pricing.py", commit["content_integrity_failure_reason"])
|
| 368 |
|
| 369 | total = next(row for row in rows if row["operation"] == "workflow.total")
|
| 370 | self.assertEqual(total["returncode"], 1)
|
| 371 | self.assertFalse(total["workflow_integrity_passed"])
|
| 372 | self.assertIn("app/pricing.py", total["workflow_integrity_failure_reason"])
|
| 373 |
|
| 374 | def test_summary_keeps_failed_workflow_totals_visible(self) -> None:
|
| 375 | subject = Subject(name="fake_git", kind="git", label="Fake Git", bin=Path(GIT))
|
| 376 | workflow = workflow_ab.WORKFLOWS["bugfix_test_loop"]
|
| 377 | summary = workflow_ab.summary_text(
|
| 378 | [
|
| 379 | {
|
| 380 | "benchmark_track": "agent-default",
|
| 381 | "operation": "workflow.total",
|
| 382 | "returncode": 1,
|
| 383 | "subject": subject.name,
|
| 384 | "subject_kind": subject.kind,
|
| 385 | "workflow": workflow.name,
|
| 386 | "elapsed_ms": 12.0,
|
| 387 | }
|
| 388 | ],
|
| 389 | [subject],
|
| 390 | [workflow],
|
| 391 | )
|
| 392 |
|
| 393 | self.assertIn("| `bugfix_test_loop` | `fake_git` | 1 | 0 | 1 | 100% |", summary)
|
| 394 |
|
| 395 |
|
| 396 | if __name__ == "__main__":
|
| 397 | unittest.main()
|