test_honesty_large_ledger.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Stress tests for large honesty ledgers.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from tests.support import honesty_artifact_hash, load_honesty_config |
| 6 | from tools.honesty.ledger import verify_ledger_file |
| 7 | from tools.honesty.ledger_io import serialize_entry |
| 8 | from tools.honesty.types import LedgerAppendOptions |
| 9 | from tools.honesty.ledger import append_entry |
| 10 | |
| 11 | |
| 12 | def test_verify_10k_lines(repo_root) -> None: |
| 13 | config = load_honesty_config(repo_root) |
| 14 | artifact_hash = honesty_artifact_hash(repo_root) |
| 15 | ledger = repo_root / ".overseer" / "honesty" / "VERDICT-LEDGER.jsonl" |
| 16 | ledger.parent.mkdir(parents=True, exist_ok=True) |
| 17 | |
| 18 | body = { |
| 19 | "actor_role": "verifier", |
| 20 | "actor_session_id": "stress-verifier", |
| 21 | "artifact_sha256": artifact_hash, |
| 22 | "passed": True, |
| 23 | "evidence": {"reexecuted": ["verify-step:stress"]}, |
| 24 | } |
| 25 | append_entry( |
| 26 | config=config, |
| 27 | repo_root=repo_root, |
| 28 | options=LedgerAppendOptions(kind="verdict", body=body), |
| 29 | ) |
| 30 | |
| 31 | from tools.honesty.canonical import compute_entry_hash |
| 32 | import json |
| 33 | |
| 34 | text = ledger.read_text(encoding="utf-8") |
| 35 | last = json.loads(text.strip().splitlines()[-1]) |
| 36 | expected_prev = last["entry_hash"] |
| 37 | extra_lines: list[str] = [] |
| 38 | for index in range(9998): |
| 39 | entry = { |
| 40 | "v": 1, |
| 41 | "ts": "2026-01-01T00:00:00Z", |
| 42 | "kind": "hook_check", |
| 43 | "actor_role": "overseer", |
| 44 | "actor_session_id": f"s-{index}", |
| 45 | "hook": "register", |
| 46 | "ok": True, |
| 47 | "prev_hash": expected_prev, |
| 48 | } |
| 49 | entry["entry_hash"] = compute_entry_hash(entry) |
| 50 | expected_prev = entry["entry_hash"] |
| 51 | extra_lines.append(serialize_entry(entry)) |
| 52 | |
| 53 | with ledger.open("a", encoding="utf-8") as handle: |
| 54 | handle.write("".join(extra_lines)) |
| 55 | |
| 56 | result = verify_ledger_file(config=config, repo_root=repo_root) |
| 57 | assert result.exit_code == 0 |
| 58 | assert len(ledger.read_text(encoding="utf-8").strip().splitlines()) == 10000 |