audit-core/tests/test_cli.py

import json
import sys
import types

from audit_core.cli import build_parser
from audit_core.ingestion import build_backend
from audit_core.integrity import ChainReport, GENESIS
from audit_core.interface import AcceptResult, RetentionPolicy


def _install_fake_postgres(monkeypatch, factory):
    """Avoid importing psycopg just to stub the constructor."""
    existing = sys.modules.get("audit_core.postgres_backend")
    if existing is not None and getattr(existing, "PostgresAuditBackend", None):
        monkeypatch.setattr(existing, "PostgresAuditBackend", factory)
        return
    mod = types.ModuleType("audit_core.postgres_backend")
    mod.PostgresAuditBackend = factory
    monkeypatch.setitem(sys.modules, "audit_core.postgres_backend", mod)


class _FakePostgres:
    def __init__(self, dsn, **kwargs):
        self.dsn = dsn
        self.kwargs = kwargs
        self.closed = False
        self.applied = ["0001-events"]

    def migrate(self):
        return list(self.applied)

    def replay(self, event_id):
        if event_id == "missing":
            raise KeyError(event_id)
        return AcceptResult(duplicate=True, reference=f"audit:{event_id}")

    def close(self):
        self.closed = True

    def verify_chain(self, attestation=None):
        return ChainReport(
            intact=True,
            events=0,
            head=GENESIS,
            head_event_id=None,
            head_accepted_at=None,
            first_break=None,
        )

    @property
    def retention_policy(self):
        return RetentionPolicy(
            custody_class="archive",
            retention_days=None,
            immutable=True,
            tamper_evidence=False,
            durable=True,
        )


def test_build_backend_disables_auto_migrate_when_asked(monkeypatch, tmp_path):
    captured = {}

    def fake(dsn, **kwargs):
        captured["dsn"] = dsn
        captured.update(kwargs)
        return _FakePostgres(dsn, **kwargs)

    monkeypatch.setenv("AUDIT_CORE_AUTO_MIGRATE", "0")
    monkeypatch.setenv("AUDIT_CORE_CREDENTIAL_DIR", str(tmp_path))
    _install_fake_postgres(monkeypatch, fake)
    backend = build_backend()
    assert captured["migrate"] is False
    assert captured["credential_dir"] == str(tmp_path)
    assert isinstance(backend, _FakePostgres)


def test_build_backend_auto_migrates_by_default(monkeypatch, tmp_path):
    captured = {}

    def fake(dsn, **kwargs):
        captured.update(kwargs)
        return _FakePostgres(dsn, **kwargs)

    monkeypatch.delenv("AUDIT_CORE_AUTO_MIGRATE", raising=False)
    monkeypatch.setenv("AUDIT_CORE_CREDENTIAL_DIR", str(tmp_path))
    _install_fake_postgres(monkeypatch, fake)
    build_backend()
    assert captured["migrate"] is True


def test_migrate_command_applies_pending_schema(monkeypatch, capsys):
    seen = {}

    def fake(dsn, **kwargs):
        backend = _FakePostgres(dsn, **kwargs)
        seen["backend"] = backend
        return backend

    monkeypatch.setenv("AUDIT_CORE_DATABASE_URL", "postgresql://example/audit_core")
    _install_fake_postgres(monkeypatch, fake)
    args = build_parser().parse_args(["migrate", "--schema", "audit_core"])
    assert args.func(args) == 0
    body = json.loads(capsys.readouterr().out)
    assert body == {"applied": ["0001-events"], "ok": True, "schema": "audit_core"}
    assert seen["backend"].kwargs["migrate"] is False
    assert seen["backend"].closed is True


def test_replay_command_reconciles(monkeypatch, capsys):
    def fake(dsn, **kwargs):
        return _FakePostgres(dsn, **kwargs)

    monkeypatch.setenv("AUDIT_CORE_DATABASE_URL", "postgresql://example/audit_core")
    _install_fake_postgres(monkeypatch, fake)
    args = build_parser().parse_args(["replay", "--event-id", "evt-1"])
    assert args.func(args) == 0
    body = json.loads(capsys.readouterr().out)
    assert body["duplicate"] is True
    assert body["reference"] == "audit:evt-1"


def test_verify_chain_command(monkeypatch, capsys):
    def fake(dsn, **kwargs):
        return _FakePostgres(dsn, **kwargs)

    monkeypatch.setenv("AUDIT_CORE_DATABASE_URL", "postgresql://example/audit_core")
    _install_fake_postgres(monkeypatch, fake)
    args = build_parser().parse_args(["verify-chain", "--schema", "audit_core"])
    assert args.func(args) == 0
    body = json.loads(capsys.readouterr().out)
    assert body["intact"] is True
    assert body["head"] == GENESIS


def test_replay_command_missing_event(monkeypatch, capsys):
    def fake(dsn, **kwargs):
        return _FakePostgres(dsn, **kwargs)

    monkeypatch.setenv("AUDIT_CORE_DATABASE_URL", "postgresql://example/audit_core")
    _install_fake_postgres(monkeypatch, fake)
    args = build_parser().parse_args(["replay", "--event-id", "missing"])
    assert args.func(args) == 1
    assert json.loads(capsys.readouterr().out)["error"] == "not_found"