audit-core/tests/test_backend_conformance.py

"""Behaviour every durable audit backend must satisfy.

One suite, run against every backend. This is what makes "the Postgres backend
is done" mean something: it is the same contract SQLite already passes, not a
parallel set of tests that happen to be green.

Postgres tests are skipped unless a server is reachable. Point
``AUDIT_CORE_TEST_DATABASE_URL`` at one, or run ``make pg-test-up`` to start a
throwaway container.
"""

from __future__ import annotations

import hashlib
import json
import os
import threading
import uuid

import pytest

from audit_core.interface import (
    AuditEvent,
    BackendUnavailableError,
    EventConflictError,
)
from audit_core.redaction import Finding
from audit_core.sqlite_backend import SQLiteAuditBackend

PG_URL = os.environ.get("AUDIT_CORE_TEST_DATABASE_URL")


def _postgres_available() -> bool:
    if not PG_URL:
        return False
    try:
        import psycopg

        with psycopg.connect(PG_URL, connect_timeout=3):
            return True
    except Exception:
        return False


HAVE_PG = _postgres_available()


@pytest.fixture(params=["sqlite", "postgres"])
def backend(request, tmp_path):
    if request.param == "sqlite":
        yield SQLiteAuditBackend(str(tmp_path / "conformance.db"))
        return
    if not HAVE_PG:
        pytest.skip("no PostgreSQL reachable (set AUDIT_CORE_TEST_DATABASE_URL)")
    from audit_core.postgres_backend import PostgresAuditBackend

    schema = f"conf_{uuid.uuid4().hex[:12]}"
    instance = PostgresAuditBackend(PG_URL, schema=schema)
    try:
        yield instance
    finally:
        with instance.pool.connection() as conn:
            conn.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')
        instance.close()


def make_event(event_id="evt-1", **kw):
    fields = dict(
        event_id=event_id, source="user-engine", action="membership.added",
        resource="membership-1", outcome="recorded", tenant="tenant:friendly:binky",
        scope="tenant", details={"correlation_id": "corr-1", "data": {"a": 1}},
        observed_at="2026-08-09T00:00:00+00:00",
    )
    fields.update(kw)
    return AuditEvent(**fields)


def digest(event: AuditEvent) -> str:
    return hashlib.sha256(
        json.dumps(event.as_record(), sort_keys=True).encode()
    ).hexdigest()


# --- the custody contract ---------------------------------------------------

def test_declares_a_retention_policy(backend):
    policy = backend.retention_policy
    assert policy.durable is True
    assert policy.custody_class in ("development", "archive", "hot_search")
    # A backend claiming tamper evidence must also claim immutability;
    # the reverse is allowed.
    if policy.tamper_evidence:
        assert policy.immutable


def test_accepts_then_reports_duplicate(backend):
    event = make_event()
    first = backend.accept(event, digest(event))
    assert first.duplicate is False
    second = backend.accept(event, digest(event))
    assert second.duplicate is True
    assert second.reference == first.reference


def test_same_id_different_payload_conflicts(backend):
    event = make_event()
    backend.accept(event, digest(event))
    with pytest.raises(EventConflictError):
        backend.accept(event, "a-different-hash")


def test_rejects_an_invalid_event(backend):
    from audit_core.interface import EventValidationError

    with pytest.raises((EventValidationError, ValueError)):
        backend.accept(make_event(tenant=""), "hash")


# --- concurrency ------------------------------------------------------------

def test_concurrent_accept_yields_exactly_one_first(backend):
    """The assertion the service rests on, checked per backend.

    An early SQLite implementation passed every serial test while telling two
    concurrent callers they were both first, so this is not theoretical.
    """
    event = make_event("race-1")
    payload_hash = digest(event)
    results: list = []
    lock = threading.Lock()
    barrier = threading.Barrier(12)

    def submit():
        barrier.wait()
        try:
            outcome = backend.accept(event, payload_hash).duplicate
        except Exception as exc:  # recorded, not swallowed
            outcome = type(exc).__name__
        with lock:
            results.append(outcome)

    threads = [threading.Thread(target=submit) for _ in range(12)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    assert results.count(False) == 1, results
    assert results.count(True) == 11, results


# --- durability and reads ---------------------------------------------------

def test_lookup_by_event_id_and_correlation(backend):
    backend.accept(make_event("e1"), "h1")
    backend.accept(make_event("e2"), "h2")

    record = backend.get("e1")
    assert record["event_id"] == "e1"
    assert record["tenant"] == "tenant:friendly:binky"
    assert record["accepted_at"]

    assert backend.get("missing") is None
    assert {e["event_id"] for e in backend.by_correlation("corr-1")} == {"e1", "e2"}


def test_dead_letters_withhold_secret_payloads(backend):
    backend.record_rejection(
        event_id="e9", reason="secret_shaped_field", payload_hash="h",
        sender="user-engine", payload='{"password":"hunter2"}',
    )
    backend.record_rejection(
        event_id="e8", reason="source_not_allowed", payload_hash="h2",
        sender="user-engine", payload='{"source":"nope"}',
    )
    entries = {d["event_id"]: d for d in backend.dead_letters()}
    assert entries["e9"]["payload"] is None
    assert entries["e9"]["payload_withheld"] is True
    assert entries["e8"]["payload"] is not None
    assert entries["e8"]["payload_withheld"] is False


def test_secret_findings_count_per_path(backend):
    findings = [Finding("data.auth_token", True)]
    for _ in range(3):
        backend.count_secret_findings(
            sender="user-engine", source="user-engine",
            action="membership.added", outcome="redacted", findings=findings,
        )
    row = backend.secret_findings()[0]
    assert row["field_path"] == "data.auth_token"
    assert row["occurrences"] == 3
    assert row["persisted"] is True


def test_health_passes_on_a_live_backend(backend):
    backend.health()


# --- postgres-specific guarantees -------------------------------------------

pg_only = pytest.mark.skipif(not HAVE_PG, reason="needs PostgreSQL")


@pg_only
def test_replay_reconciles_rather_than_duplicating():
    """Replay must never mint a second custody record for one source event."""
    from audit_core.postgres_backend import PostgresAuditBackend

    schema = f"conf_{uuid.uuid4().hex[:12]}"
    backend = PostgresAuditBackend(PG_URL, schema=schema)
    try:
        event = make_event("replay-1")
        backend.accept(event, digest(event))

        outcome = backend.replay("replay-1")
        assert outcome.duplicate is True

        rows = backend._query(
            f'SELECT count(*) FROM "{schema}".events WHERE event_id = %s', ("replay-1",)
        )
        assert rows[0][0] == 1

        with pytest.raises(KeyError):
            backend.replay("never-stored")
    finally:
        with backend.pool.connection() as conn:
            conn.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')
        backend.close()


@pg_only
def test_stored_events_are_append_only():
    """The basis for declaring immutable=True.

    Without this, a leaked runtime credential could rewrite or erase the audit
    trail — and the retention policy would be claiming a guarantee it does not
    have.
    """
    from audit_core.postgres_backend import PostgresAuditBackend

    schema = f"conf_{uuid.uuid4().hex[:12]}"
    backend = PostgresAuditBackend(PG_URL, schema=schema)
    try:
        event = make_event("immutable-1")
        backend.accept(event, digest(event))
        assert backend.retention_policy.immutable is True

        with pytest.raises(BackendUnavailableError):
            backend._execute(
                f'UPDATE "{schema}".events SET tenant = %s WHERE event_id = %s',
                ("tenant:coulomb", "immutable-1"),
            )
        with pytest.raises(BackendUnavailableError):
            backend._execute(
                f'DELETE FROM "{schema}".events WHERE event_id = %s', ("immutable-1",)
            )
        assert backend.get("immutable-1")["tenant"] == "tenant:friendly:binky"
    finally:
        with backend.pool.connection() as conn:
            conn.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')
        backend.close()


@pg_only
def test_migrations_are_idempotent_and_recorded():
    from audit_core.postgres_backend import MIGRATIONS, PostgresAuditBackend

    schema = f"conf_{uuid.uuid4().hex[:12]}"
    backend = PostgresAuditBackend(PG_URL, schema=schema, migrate=False)
    try:
        applied = backend.migrate()
        assert applied == [m[0] for m in MIGRATIONS]
        assert backend.migrate() == []  # second call is a no-op
    finally:
        with backend.pool.connection() as conn:
            conn.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')
        backend.close()


@pg_only
def test_connects_from_a_brokered_libpq_environment(monkeypatch):
    """The credential broker injects PG* vars into the child process rather
    than handing over a DSN, so an empty conninfo must work.

    This keeps the credential out of audit-core's configuration entirely.
    """
    import urllib.parse

    from audit_core.postgres_backend import PostgresAuditBackend

    parsed = urllib.parse.urlparse(PG_URL)
    monkeypatch.setenv("PGHOST", parsed.hostname or "127.0.0.1")
    monkeypatch.setenv("PGPORT", str(parsed.port or 5432))
    monkeypatch.setenv("PGUSER", parsed.username or "postgres")
    monkeypatch.setenv("PGPASSWORD", parsed.password or "")
    monkeypatch.setenv("PGDATABASE", (parsed.path or "/postgres").lstrip("/"))
    monkeypatch.delenv("AUDIT_CORE_DATABASE_URL", raising=False)

    schema = f"conf_{uuid.uuid4().hex[:12]}"
    backend = PostgresAuditBackend(schema=schema)
    try:
        event = make_event("brokered-1")
        assert backend.accept(event, digest(event)).duplicate is False
        assert backend.get("brokered-1")["event_id"] == "brokered-1"
    finally:
        with backend.pool.connection() as conn:
            conn.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')
        backend.close()


def test_missing_connection_information_is_a_clear_error(monkeypatch):
    for var in ("AUDIT_CORE_DATABASE_URL", "PGHOST", "PGUSER"):
        monkeypatch.delenv(var, raising=False)
    from audit_core.postgres_backend import PostgresAuditBackend

    with pytest.raises(ValueError, match="no connection information"):
        PostgresAuditBackend()