audit-core/tests/test_migrate_store.py

"""Migration fidelity tests (AUDIT-WP-0005-T04).

The controlling property is that migrating audit records does not alter them.
An event that arrives in the destination with a rewritten acceptance time has
been damaged, not moved.
"""

from __future__ import annotations

import hashlib
import json
import os
import uuid

import pytest

from audit_core.ingestion import IngestionApplication
from audit_core.migrate_store import migrate
from audit_core.sqlite_backend import SQLiteAuditBackend

from tests.test_backend_conformance import HAVE_PG, PG_URL, make_event

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


@pytest.fixture
def destination():
    from audit_core.postgres_backend import PostgresAuditBackend

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


@pytest.fixture
def populated_sqlite(tmp_path):
    """A SQLite store holding events, a dead letter, and a finding."""
    path = str(tmp_path / "legacy.db")
    source = SQLiteAuditBackend(path)
    for i in range(3):
        event = make_event(f"legacy-{i}")
        source.accept(event, hashlib.sha256(f"h{i}".encode()).hexdigest())
    source.record_rejection(
        event_id="bad-1", reason="source_not_allowed", payload_hash="hh",
        sender="user-engine", payload='{"source":"nope"}',
    )
    from audit_core.redaction import Finding

    source.count_secret_findings(
        sender="user-engine", source="user-engine", action="membership.added",
        outcome="redacted", findings=[Finding("data.auth_token", True)],
    )
    return path, source


def test_migration_preserves_identity_hash_and_time(populated_sqlite, destination):
    path, source = populated_sqlite
    original = {
        row[0]: (row[1], row[2])
        for row in source.db.execute(
            "SELECT event_id, payload_hash, accepted_at FROM events"
        ).fetchall()
    }

    report = migrate(path, destination)

    assert report.events_read == 3
    assert report.events_written == 3
    assert report.conflicts == []
    assert report.verified is True
    assert report.ok is True

    for event_id, (payload_hash, accepted_at) in original.items():
        rows = destination._query(
            f"SELECT payload_hash, accepted_at FROM {destination._events} "
            "WHERE event_id = %s", (event_id,),
        )
        assert rows, f"{event_id} missing"
        assert rows[0][0] == payload_hash
        # Acceptance time is evidence; rewriting it would destroy what the
        # migration exists to preserve.
        assert rows[0][1].isoformat().startswith(accepted_at[:16])


def test_migration_carries_dead_letters_and_findings(populated_sqlite, destination):
    path, _ = populated_sqlite
    report = migrate(path, destination)

    assert report.dead_letters_written == 1
    assert report.findings_written == 1
    assert destination.dead_letters()[0]["reason"] == "source_not_allowed"
    assert destination.secret_findings()[0]["field_path"] == "data.auth_token"


def test_migration_is_idempotent(populated_sqlite, destination):
    path, _ = populated_sqlite
    first = migrate(path, destination)
    second = migrate(path, destination)

    assert first.events_written == 3
    assert second.events_written == 0
    assert second.events_already_present == 3
    assert second.ok is True

    count = destination._query(
        f"SELECT count(*) FROM {destination._events}", ()
    )[0][0]
    assert count == 3


def test_a_diverging_destination_record_is_a_conflict_not_an_overwrite(
    populated_sqlite, destination
):
    """The destination wins and a human decides.

    Silently overwriting a stored audit record with a differing one would be
    the same class of failure as losing it.
    """
    path, _ = populated_sqlite
    migrate(path, destination)

    # Simulate divergence: same id, different payload hash in the source.
    source = SQLiteAuditBackend(path)
    source.db.execute("BEGIN IMMEDIATE")
    source.db.execute(
        "UPDATE events SET payload_hash = 'diverged' WHERE event_id = 'legacy-1'"
    )
    source.db.execute("COMMIT")

    report = migrate(path, destination)
    assert report.conflicts, "divergence must be reported"
    assert "legacy-1" in report.conflicts[0]
    assert report.ok is False

    survived = destination._query(
        f"SELECT payload_hash FROM {destination._events} WHERE event_id = 'legacy-1'", ()
    )[0][0]
    assert survived != "diverged", "destination record must not be overwritten"


def test_migrated_events_remain_readable_and_append_only(populated_sqlite, destination):
    path, _ = populated_sqlite
    migrate(path, destination)

    record = destination.get("legacy-0")
    assert record["tenant"] == "tenant:friendly:binky"
    assert destination.retention_policy.immutable is True

    from audit_core.interface import BackendUnavailableError

    with pytest.raises(BackendUnavailableError):
        destination._execute(
            f"DELETE FROM {destination._events} WHERE event_id = %s", ("legacy-0",)
        )


def test_empty_source_migrates_cleanly(tmp_path, destination):
    """Today's actual case: nothing to move."""
    path = str(tmp_path / "empty.db")
    SQLiteAuditBackend(path)

    report = migrate(path, destination)
    assert report.events_read == 0
    assert report.events_written == 0
    assert report.ok is True