codex-app-server/cloud/tests/test_ledger_anchor_check.py

"""Anchor script: exit-code contract for tools/ledger_anchor_check.py.

Runs the script as a subprocess against throwaway SQLite chains
in a TemporaryDirectory (deleted with the test). Pins only what
the operator depends on: 0 pass, 1 verification failure with the
failure named, 2 usage error, and --write-anchor refused on a
failing chain. The script itself never writes the chain; the
tamper/tail cases simulate privileged tampering by dropping the
SQLite append-only triggers first (a normal UPDATE is already
blocked by the trigger, which is itself asserted here).
"""

import json
import os
import sqlite3
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parents[2]
SCRIPT = REPO_ROOT / "tools" / "ledger_anchor_check.py"

try:
    from cloud.audit_wiring import record_audit_verdict
    from cloud.ledger import LedgerStore
except ImportError:
    sys.path.insert(0, str(REPO_ROOT))
    from cloud.audit_wiring import record_audit_verdict
    from cloud.ledger import LedgerStore


def _run(*args):
    proc = subprocess.run(
        [sys.executable, str(SCRIPT), *args],
        capture_output=True, text=True, timeout=60)
    try:
        report = json.loads(proc.stdout) if proc.stdout.strip() else None
    except ValueError:
        report = None
    return proc.returncode, report, proc.stderr


class LedgerAnchorCheckTestCase(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db_path = os.path.join(self.tmp.name, "control.db")
        store = LedgerStore(self.db_path)
        record_audit_verdict(store, "t1", "p1", "apps/web", "v2",
                             "fail", "audit", failures=["ci_green"])
        record_audit_verdict(store, "t1", "p1", "apps/web", "v2",
                             "pass", "audit")
        head = store.export()["head_hash"]
        store.close()
        self.anchor_path = os.path.join(self.tmp.name, "anchor.json")
        with open(self.anchor_path, "w", encoding="utf-8") as handle:
            json.dump({"version": 1, "head_hash": head, "count": 2},
                      handle)

    def tearDown(self):
        self.tmp.cleanup()

    def _copy_db(self, name):
        import shutil
        path = os.path.join(self.tmp.name, name)
        shutil.copyfile(self.db_path, path)
        return path

    def test_pass_without_and_with_anchor(self):
        code, report, _ = _run("--db", self.db_path)
        self.assertEqual(code, 0)
        self.assertTrue(report["ok"])
        code, report, _ = _run("--db", self.db_path,
                               "--anchor", self.anchor_path)
        self.assertEqual(code, 0)
        self.assertTrue(report["ok"])
        self.assertEqual(report["head_advanced_by"], 0)

    def test_normal_update_blocked_by_trigger_then_privileged_tamper_fails(self):
        path = self._copy_db("tamper.db")
        conn = sqlite3.connect(path)
        with self.assertRaises(sqlite3.IntegrityError):
            conn.execute(
                "UPDATE audit_ledger SET detail='x' WHERE id=1")
        conn.rollback()
        conn.execute("DROP TRIGGER audit_ledger_no_update")
        conn.execute("UPDATE audit_ledger SET detail='tampered'"
                     " WHERE id=1")
        conn.commit()
        conn.close()
        code, report, _ = _run("--db", path,
                               "--anchor", self.anchor_path)
        self.assertEqual(code, 1)
        self.assertFalse(report["ok"])
        self.assertTrue(any("entry_hash mismatch" in f
                            for f in report["failures"]))

    def test_tail_deletion_fails_against_old_anchor(self):
        path = self._copy_db("tail.db")
        conn = sqlite3.connect(path)
        conn.execute("DROP TRIGGER audit_ledger_no_delete")
        conn.execute("DELETE FROM audit_ledger WHERE id=2")
        conn.commit()
        conn.close()
        code, report, _ = _run("--db", path,
                               "--anchor", self.anchor_path)
        self.assertEqual(code, 1)
        self.assertFalse(report["ok"])
        self.assertTrue(any("shorter than anchor" in f
                            for f in report["failures"]))

    def test_extended_chain_still_passes_anchor(self):
        store = LedgerStore(self.db_path)
        record_audit_verdict(store, "t1", "p1", "apps/web", "v2",
                             "pass", "audit")
        store.close()
        code, report, _ = _run("--db", self.db_path,
                               "--anchor", self.anchor_path)
        self.assertEqual(code, 0)
        self.assertTrue(report["ok"])
        self.assertEqual(report["head_advanced_by"], 1)

    def test_write_anchor_refused_on_failing_export(self):
        path = self._copy_db("tail2.db")
        conn = sqlite3.connect(path)
        conn.execute("DROP TRIGGER audit_ledger_no_delete")
        conn.execute("DELETE FROM audit_ledger WHERE id=2")
        conn.commit()
        conn.close()
        store = LedgerStore(path)
        export_path = os.path.join(self.tmp.name, "export.json")
        with open(export_path, "w", encoding="utf-8") as handle:
            json.dump(store.export(), handle)
        store.close()
        out_path = os.path.join(self.tmp.name, "new-anchor.json")
        code, _report, _ = _run("--export", export_path,
                                "--anchor", self.anchor_path,
                                "--write-anchor", out_path)
        self.assertEqual(code, 1)
        self.assertFalse(os.path.exists(out_path))

    def test_usage_errors_exit_2(self):
        code, _report, _ = _run(
            "--db", os.path.join(self.tmp.name, "nope.db"))
        self.assertEqual(code, 2)
        code, _report, _ = _run("--db", "postgres://host/db")
        self.assertEqual(code, 2)


if __name__ == "__main__":
    unittest.main()