"""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 def test_historical_import_extends_existing_chain_and_preserves_attestation( populated_sqlite, destination ): path, _ = populated_sqlite event = make_event('native-before-import') destination.accept(event, 'native-hash') attestation = destination.attest_chain() old_link = destination._query( f"SELECT chain_hash FROM {destination._events} WHERE event_id = %s", (event.event_id,), )[0][0] report = migrate(path, destination) assert report.ok assert destination.verify_chain(attestation).intact assert destination.verify_chain(attestation).attestation_match assert destination._query( f"SELECT chain_hash FROM {destination._events} WHERE event_id = %s", (event.event_id,), )[0][0] == old_link destination.accept(make_event('native-after-import'), 'after-hash') assert destination.verify_chain(attestation).intact def test_replaying_side_tables_does_not_duplicate_evidence(populated_sqlite, destination): path, _ = populated_sqlite migrate(path, destination, source_id='stable-source') replay = migrate(path, destination, source_id='stable-source') assert replay.ok assert replay.dead_letters_written == replay.findings_written == 0 assert len(destination.dead_letters()) == 1 assert destination.secret_findings()[0]['occurrences'] == 1 def test_incremental_finding_import_adds_only_new_occurrences(populated_sqlite, destination): from audit_core.redaction import Finding path, source = populated_sqlite migrate(path, destination) source.count_secret_findings( sender='user-engine', source='user-engine', action='membership.added', outcome='redacted', findings=[Finding('data.auth_token', True)], ) assert migrate(path, destination).ok assert migrate(path, destination).ok assert destination.secret_findings()[0]['occurrences'] == 2 def test_import_and_native_accept_share_one_chain_lock(populated_sqlite, destination): from concurrent.futures import ThreadPoolExecutor path, _ = populated_sqlite with ThreadPoolExecutor(max_workers=3) as executor: futures = [executor.submit(migrate, path, destination) for _ in range(2)] futures.append(executor.submit(destination.accept, make_event('concurrent'), 'native')) results = [future.result(timeout=20) for future in futures] assert all(result.ok for result in results[:2]) assert sum(result.events_written for result in results[:2]) == 3 assert destination.verify_chain().intact assert destination.verify_chain().events == 4 assert len(destination.dead_letters()) == 1 assert destination.secret_findings()[0]['occurrences'] == 1 def test_matching_hash_does_not_hide_a_different_acceptance_time(populated_sqlite, destination): path, source = populated_sqlite migrate(path, destination) source.db.execute("UPDATE events SET accepted_at = '2020-01-01T00:00:00+00:00' WHERE event_id = 'legacy-1'") report = migrate(path, destination) assert not report.ok assert any('different historical evidence' in conflict for conflict in report.conflicts) assert destination.verify_chain().intact def test_failed_import_rolls_back_events_and_replay_receipts( populated_sqlite, destination, monkeypatch ): import audit_core.migrate_store as importer import psycopg from audit_core.interface import BackendUnavailableError path, _ = populated_sqlite original = importer._copy_findings def fail_after_side_tables(*args): original(*args) raise psycopg.OperationalError('injected import failure') monkeypatch.setattr(importer, '_copy_findings', fail_after_side_tables) with pytest.raises(BackendUnavailableError): migrate(path, destination) assert destination.verify_chain().events == 0 assert destination.dead_letters() == [] assert destination.secret_findings() == [] assert destination._query(f'SELECT count(*) FROM "{destination.schema}".import_receipts', ())[0][0] == 0 monkeypatch.setattr(importer, '_copy_findings', original) assert migrate(path, destination).ok def test_source_relocation_with_stable_id_preserves_side_table_replay( populated_sqlite, destination, tmp_path ): path, _ = populated_sqlite # Backup captures WAL-backed source data too. import sqlite3 moved = str(tmp_path / 'moved.db') with sqlite3.connect(path) as source, sqlite3.connect(moved) as copy: source.backup(copy) migrate(path, destination, source_id='same-source') replay = migrate(moved, destination, source_id='same-source') assert replay.ok assert replay.events_written == replay.dead_letters_written == replay.findings_written == 0