"""Testing a dump without touching this install (services/backup_verify.py and restore_test_task): the dump is loaded into a temporary database, every table is checked against the counts recorded with the dump, and the temporary database is dropped. The round trips at the bottom dump the test database to an in-process bucket with the real pg_dump and test it with the real pg_restore.""" import json from datetime import UTC, datetime from pathlib import Path import pytest from sqlalchemy import func, select, text from backend.app.models import Artist, BackupRun from backend.app.services import backup_service, backup_verify from backend.app.services.backup_verify import compare, summary from tests.fake_s3 import FakeS3 LIVE = {"alembic": "0121", "tables": {"artist": 12, "image_record": 40, "post": 7}} def _counts(**tables): return {"alembic": "0121", "tables": tables} def test_a_dump_that_comes_back_whole_matches_every_table(): recorded = _counts(artist=10, image_record=40, post=7) result = compare(recorded, _counts(artist=10, image_record=40, post=7), LIVE) assert result["counted"] and result["mismatched"] == [] assert result["rows_restored"] == 57 artist = next(t for t in result["tables"] if t["table"] == "artist") assert artist == {"table": "artist", "dump": 10, "restored": 10, "live": 12} assert summary(result) == "All 3 tables came back with the 57 rows the dump recorded." def test_a_table_with_rows_missing_or_gone_is_named(): recorded = _counts(artist=10, image_record=40, post=7) result = compare(recorded, _counts(artist=10, image_record=39), LIVE) assert result["mismatched"] == ["image_record", "post"] assert summary(result) == ( "2 table(s) came back different: image_record (40 in the dump, 39 restored), " "post (7 in the dump, table missing).") def test_a_dump_without_counts_is_shown_beside_the_live_database(): result = compare(None, _counts(artist=10, image_record=40), LIVE) assert not result["counted"] and result["mismatched"] == [] assert next(t for t in result["tables"] if t["table"] == "post")["restored"] is None assert summary(result).startswith("The dump loaded cleanly: 2 tables, 50 rows.") def test_the_counts_sit_beside_the_dump_and_the_test_database_beside_the_live_one(): key = "database/fabledcurator-20261107T134258Z.dump" assert backup_verify.counts_key(key) == "database/fabledcurator-20261107T134258Z.counts.json" url = "postgresql+psycopg://fc:pw@postgres:5432/fabledcurator" assert backup_verify.scratch_name(url) == "fabledcurator_restore_test" assert backup_verify.scratch_url(url) == ( "postgresql+psycopg://fc:pw@postgres:5432/fabledcurator_restore_test") # ── the task ──────────────────────────────────────────────────────────────── def _local_dump(db_sync, tmp_path, counts): dump = tmp_path / "fc_db_20261107T030000Z.dump" dump.write_bytes(b"a dump") manifest = tmp_path / "fc_db_20261107T030000Z.json" manifest.write_text(json.dumps({"kind": "db", "counts": counts})) row = BackupRun(kind="db", status="ok", triggered_by="manual", sql_path=str(dump), started_at=datetime.now(UTC), size_bytes=6, manifest={"ts": "20261107T030000Z", "manifest_path": str(manifest)}) db_sync.add(row) db_sync.commit() return row @pytest.mark.integration def test_a_dump_on_this_server_is_tested_against_its_manifest(db_sync, tmp_path, monkeypatch): from backend.app.tasks import backup as tasks recorded = _counts(artist=3) src = _local_dump(db_sync, tmp_path, recorded) seen = {} def _trial(*, db_url, dump_path, recorded, note): seen.update(dump_path=dump_path, recorded=recorded) note("comparing") return {**compare(recorded, _counts(artist=3), _counts(artist=5)), "alembic_dump": "0121", "alembic_now": "0121"} monkeypatch.setattr(backup_verify, "trial_restore", _trial) out = tasks.restore_test_task(source_backup_run_id=src.id) db_sync.expire_all() row = db_sync.get(BackupRun, out["backup_run_id"]) assert seen == {"dump_path": Path(src.sql_path), "recorded": recorded} assert (row.kind, row.status, row.error) == ("test", "ok", None) assert row.manifest["summary"] == "All 1 tables came back with the 3 rows the dump recorded." assert row.manifest["dump"]["run_id"] == src.id assert Path(src.sql_path).exists() # the dump itself stays @pytest.mark.integration def test_a_test_that_finds_a_difference_ends_in_error_saying_which(db_sync, tmp_path, monkeypatch): from backend.app.tasks import backup as tasks src = _local_dump(db_sync, tmp_path, _counts(artist=3)) monkeypatch.setattr(backup_verify, "trial_restore", lambda *, recorded, **kw: { **compare(recorded, _counts(artist=2), _counts(artist=2)), "alembic_dump": "0121", "alembic_now": "0121"}) out = tasks.restore_test_task(source_backup_run_id=src.id) db_sync.expire_all() row = db_sync.get(BackupRun, out["backup_run_id"]) assert row.status == "error" assert row.error == "1 table(s) came back different: artist (3 in the dump, 2 restored)." @pytest.mark.integration def test_a_dump_that_wont_load_is_recorded_with_why(db_sync, tmp_path, monkeypatch): from backend.app.tasks import backup as tasks src = _local_dump(db_sync, tmp_path, None) def _fails(**kw): raise backup_service.RestoreFailed("pg_restore: error: could not read input file") monkeypatch.setattr(backup_verify, "trial_restore", _fails) with pytest.raises(backup_service.RestoreFailed): tasks.restore_test_task(source_backup_run_id=src.id) row = db_sync.execute(select(BackupRun).where(BackupRun.kind == "test")).scalar_one() assert row.status == "error" assert row.error == "RestoreFailed: pg_restore: error: could not read input file" @pytest.mark.integration def test_no_test_starts_while_a_restore_or_another_test_is_going(db_sync, tmp_path): from backend.app.tasks import backup as tasks src = _local_dump(db_sync, tmp_path, None) db_sync.add(BackupRun(kind="test", status="running", triggered_by="manual", started_at=datetime.now(UTC))) db_sync.commit() assert tasks.restore_test_task(source_backup_run_id=src.id) == { "skipped": "a test restore is already running"} assert tasks.backup_busy(db_sync) == "a test restore is running" @pytest.mark.integration def test_a_test_cut_off_by_a_restart_is_ended_and_its_database_dropped(db_sync, monkeypatch): from backend.app.tasks import backup as tasks dropped = [] monkeypatch.setattr(backup_verify, "drop_scratch", lambda url: dropped.append(url)) row = BackupRun(kind="test", status="running", triggered_by="manual", started_at=datetime.now(UTC)) db_sync.add(row) db_sync.commit() assert tasks.end_cut_off_tests() == 1 db_sync.expire_all() assert db_sync.get(BackupRun, row.id).status == "error" assert len(dropped) == 1 # ── the screen's endpoints ────────────────────────────────────────────────── @pytest.mark.integration async def test_a_dump_on_this_server_can_be_tested_from_the_screen(client, db_sync, tmp_path, monkeypatch): from backend.app.tasks import backup as tasks queued = [] monkeypatch.setattr(tasks.restore_test_task, "delay", lambda **kw: queued.append(kw)) src = _local_dump(db_sync, tmp_path, None) media = BackupRun(kind="media", status="ok", triggered_by="manual", started_at=datetime.now(UTC)) db_sync.add(media) db_sync.commit() r = await client.post(f"/api/system/backup/runs/{media.id}/test", json={}) assert r.status_code == 400 r = await client.post(f"/api/system/backup/runs/{src.id}/test", json={}) assert r.status_code == 202 assert queued == [{"source_backup_run_id": src.id}] r = await client.post("/api/system/backup/offsite/test", json={"key": "../etc"}) assert r.status_code == 400 # ── round trips with the real pg_dump and pg_restore ──────────────────────── def _scratch_exists(db_sync) -> bool: from backend.app.config import get_config name = backup_verify.scratch_name(get_config().database_url_sync) return db_sync.execute(text("SELECT 1 FROM pg_database WHERE datname = :n"), {"n": name}).first() is not None @pytest.mark.integration def test_the_counts_are_read_in_the_snapshot_the_dump_is_taken_from(db_sync): from backend.app.config import get_config db_sync.add(Artist(name="Bea", slug="bea")) db_sync.commit() with backup_service.dump_snapshot(get_config().database_url_sync) as snap: before = snap.counts() db_sync.add(Artist(name="Later", slug="later")) db_sync.commit() assert snap.counts() == before assert before["tables"]["artist"] == 1 assert before["tables"]["credential"] == 0 assert before["alembic"] @pytest.mark.integration def test_a_dump_tested_from_the_bucket_comes_back_whole_and_leaves_no_trace( db_sync, tmp_path, monkeypatch, ): from backend.app.services import backup_destination from backend.app.tasks import backup as tasks db_sync.add(Artist(name="Bea", slug="bea")) db_sync.commit() monkeypatch.setattr(tasks, "IMAGES_ROOT", tmp_path) with FakeS3() as s3: monkeypatch.setattr(backup_destination, "saved_target", lambda settings: s3.target()) tasks.backup_db_task(triggered_by="manual") dump_key = next(k for k in s3.objects if k.endswith(".dump")) # The database moves on; the test compares with the dump's own counts. db_sync.add(Artist(name="Later", slug="later")) db_sync.commit() out = tasks.restore_test_task(dump_key=dump_key) with tasks._sync_session_factory()() as s: row = s.get(BackupRun, out["backup_run_id"]) assert row.status == "ok", row.error m = row.manifest assert m["counted"] and m["mismatched"] == [] artist = next(t for t in m["tables"] if t["table"] == "artist") assert (artist["dump"], artist["restored"], artist["live"]) == (1, 1, 2) assert m["alembic_dump"] == m["alembic_now"] assert m["offsite"]["key"] == dump_key # This install is as it was, and the test left nothing behind. assert s.execute(select(func.count()).select_from(Artist)).scalar() == 2 assert not _scratch_exists(s) assert not list((tmp_path / "_backups").glob("*")) @pytest.mark.integration def test_a_dump_whose_counts_dont_match_fails_and_one_without_is_compared_with_live( db_sync, tmp_path, monkeypatch, ): from backend.app.services import backup_destination from backend.app.tasks import backup as tasks db_sync.add(Artist(name="Bea", slug="bea")) db_sync.commit() monkeypatch.setattr(tasks, "IMAGES_ROOT", tmp_path) with FakeS3() as s3: monkeypatch.setattr(backup_destination, "saved_target", lambda settings: s3.target()) tasks.backup_db_task(triggered_by="manual") dump_key = next(k for k in s3.objects if k.endswith(".dump")) counts = json.loads(s3.objects[backup_verify.counts_key(dump_key)]) counts["tables"]["artist"] = 5 s3.objects[backup_verify.counts_key(dump_key)] = json.dumps(counts).encode() wrong = tasks.restore_test_task(dump_key=dump_key) del s3.objects[backup_verify.counts_key(dump_key)] uncounted = tasks.restore_test_task(dump_key=dump_key) with tasks._sync_session_factory()() as s: row = s.get(BackupRun, wrong["backup_run_id"]) assert row.status == "error" assert row.error == "1 table(s) came back different: artist (5 in the dump, 1 restored)." row = s.get(BackupRun, uncounted["backup_run_id"]) assert row.status == "ok" and row.manifest["counted"] is False assert not _scratch_exists(s)