import json import os import subprocess from contextlib import contextmanager from hashlib import sha256 from pathlib import Path from types import SimpleNamespace import pytest import psycopg from django.db import connection from psycopg import connect, sql from core.backup import ( BACKUP_FORMAT, BACKUP_VERSION, BackupError, DatabaseSnapshot, DatabaseTarget, create_backup, verify_backup, ) class RecordingRunner: def __init__(self, output=b"PGDMP synthetic"): self.calls = [] self.output = output def __call__(self, argv, *, env, check, stdout=None, capture_output=False): self.calls.append((list(argv), dict(env), check, stdout, capture_output)) if argv[0] == "pg_dump" and argv != ["pg_dump", "--version"]: output_name = next(value.removeprefix("--file=") for value in argv if value.startswith("--file=")) Path(output_name).write_bytes(self.output) return SimpleNamespace(stdout=b"pg_dump (PostgreSQL) 17.6\n") @pytest.fixture def database_settings(settings, monkeypatch): configured = { "ENGINE": "django.db.backends.postgresql", "HOST": "live-db", "PORT": "5432", "NAME": "gizmoduck", "USER": "gizmoduck_migrate", "PASSWORD": "synthetic-secret", } monkeypatch.setitem(settings.DATABASES, "default", configured) return configured @pytest.fixture def backup_root(tmp_path, settings): settings.BACKUP_ROOT = tmp_path return tmp_path @pytest.fixture def synthetic_snapshot(monkeypatch): snapshot = DatabaseSnapshot( migrations=(("finance", "0001_initial"), ("ingestion", "0001_initial")), row_counts={"finance_account": 2, "ingestion_importbatch": 1}, ) @contextmanager def exported_snapshot(_target): yield "synthetic-snapshot", snapshot monkeypatch.setattr("core.backup._exported_snapshot", exported_snapshot) monkeypatch.setattr("core.backup._read_snapshot", lambda _target: snapshot) return snapshot def test_pg_dump_receives_password_only_in_environment( backup_root, database_settings, synthetic_snapshot ): runner = RecordingRunner() result = create_backup("synthetic.dump", runner=runner) argv, env, check, stdout, capture_output = runner.calls[0] assert argv[:4] == ["pg_dump", "--format=custom", "--no-owner", "--no-acl"] assert database_settings["PASSWORD"] not in argv assert env["PGPASSWORD"] == database_settings["PASSWORD"] assert check is True assert stdout is None assert capture_output is True assert result.path.read_bytes() == runner.output def test_pg_dump_uses_the_exported_snapshot( backup_root, database_settings, synthetic_snapshot ): runner = RecordingRunner() create_backup("synthetic.dump", runner=runner) assert "--snapshot=synthetic-snapshot" in runner.calls[0][0] assert runner.calls[0][4] is True def test_backup_writes_versioned_checksum_manifest( backup_root, database_settings, synthetic_snapshot ): runner = RecordingRunner() result = create_backup("synthetic.dump", runner=runner) manifest = json.loads(result.manifest_path.read_text(encoding="utf-8")) assert manifest == { "application_version": "0.1.0", "byte_size": len(runner.output), "created_at": result.created_at.isoformat(), "database_name": "gizmoduck", "format": BACKUP_FORMAT, "migrations": [ ["finance", "0001_initial"], ["ingestion", "0001_initial"], ], "postgresql_client_version": result.postgresql_client_version, "row_counts": { "finance_account": 2, "ingestion_importbatch": 1, }, "sha256": sha256(runner.output).hexdigest(), "version": BACKUP_VERSION, } @pytest.mark.parametrize( "name", ["../outside.dump", "nested/file.dump", "nested\\file.dump", ".dump", "file.zip"], ) def test_backup_rejects_non_plain_dump_names( backup_root, database_settings, synthetic_snapshot, name ): with pytest.raises(BackupError, match="invalid_name"): create_backup(name, runner=RecordingRunner()) assert list(backup_root.iterdir()) == [] def test_backup_does_not_replace_existing_artifacts( backup_root, database_settings, synthetic_snapshot ): dump = backup_root / "synthetic.dump" manifest = backup_root / "synthetic.dump.json" dump.write_bytes(b"existing dump") manifest.write_text("existing manifest", encoding="utf-8") with pytest.raises(BackupError, match="existing_destination"): create_backup("synthetic.dump", runner=RecordingRunner()) assert dump.read_bytes() == b"existing dump" assert manifest.read_text(encoding="utf-8") == "existing manifest" def test_backup_cleans_private_temporary_files_on_dump_failure( backup_root, database_settings, synthetic_snapshot ): def failing_runner(argv, **kwargs): raise OSError("synthetic private detail") with pytest.raises(BackupError, match="backup_failed"): create_backup("synthetic.dump", runner=failing_runner) assert list(backup_root.iterdir()) == [] def test_backup_sanitizes_subprocess_failure_and_captures_diagnostics( backup_root, database_settings, synthetic_snapshot ): def failing_runner(argv, *, capture_output, **kwargs): assert capture_output is True raise subprocess.CalledProcessError( 1, argv, output=b"SYNTHETIC PRIVATE STDOUT", stderr=b"SYNTHETIC PRIVATE STDERR" ) with pytest.raises(BackupError, match="backup_failed") as error: create_backup("synthetic.dump", runner=failing_runner) assert "PRIVATE" not in str(error.value) assert list(backup_root.iterdir()) == [] def test_second_temporary_unlink_failure_preserves_complete_committed_pair( backup_root, database_settings, synthetic_snapshot, monkeypatch ): original_unlink = Path.unlink failed = False def fail_once(path, *args, **kwargs): nonlocal failed if not failed and path.name.endswith(".dump.tmp"): failed = True raise OSError("synthetic unlink failure") return original_unlink(path, *args, **kwargs) monkeypatch.setattr(Path, "unlink", fail_once) with pytest.raises(BackupError, match="cleanup_failed"): create_backup("synthetic.dump", runner=RecordingRunner()) assert (backup_root / "synthetic.dump").read_bytes() == b"PGDMP synthetic" assert (backup_root / "synthetic.dump.json").is_file() assert sorted(path.name for path in backup_root.iterdir()) == [ "synthetic.dump", "synthetic.dump.json", ] def test_directory_fsync_failure_rolls_back_both_owned_finals( backup_root, database_settings, synthetic_snapshot, monkeypatch ): monkeypatch.setattr( "core.backup._fsync_directory", lambda _root: (_ for _ in ()).throw(OSError("synthetic fsync failure")), ) with pytest.raises(BackupError, match="publication_failed"): create_backup("synthetic.dump", runner=RecordingRunner()) assert list(backup_root.iterdir()) == [] def test_publication_never_deletes_concurrent_final_files(tmp_path, monkeypatch): import core.backup as backup manifest_temp = tmp_path / ".manifest.tmp" dump_temp = tmp_path / ".dump.tmp" manifest_final = tmp_path / "synthetic.dump.json" dump_final = tmp_path / "synthetic.dump" manifest_temp.write_bytes(b"our manifest") dump_temp.write_bytes(b"our dump") original_link = os.link calls = 0 def competing_link(source, destination): nonlocal calls calls += 1 if calls == 2: manifest_final.unlink() manifest_final.write_bytes(b"other manifest") dump_final.write_bytes(b"other dump") raise FileExistsError return original_link(source, destination) monkeypatch.setattr(os, "link", competing_link) with pytest.raises(BackupError, match="publication_failed"): backup._publish_pair( tmp_path, ((manifest_temp, manifest_final), (dump_temp, dump_final)), ) assert manifest_final.read_bytes() == b"other manifest" assert dump_final.read_bytes() == b"other dump" assert sorted(path.name for path in tmp_path.iterdir()) == [ "synthetic.dump", "synthetic.dump.json", ] def test_verification_rejects_live_database( backup_root, database_settings, synthetic_snapshot ): target = DatabaseTarget.from_default(database_settings) with pytest.raises(BackupError, match="live_target"): verify_backup("synthetic.dump", target, runner=RecordingRunner()) def test_verification_requires_explicit_allow_flag( backup_root, database_settings, synthetic_snapshot, settings ): settings.BACKUP_VERIFY_ALLOWED = False target = DatabaseTarget( host="restore-db", port="5432", name="gizmoduck_restore", user="restore_user", password="restore-secret", ) with pytest.raises(BackupError, match="verification_not_allowed"): verify_backup("synthetic.dump", target, runner=RecordingRunner()) def test_verification_rejects_changed_dump_before_restore( backup_root, database_settings, synthetic_snapshot, settings ): settings.BACKUP_VERIFY_ALLOWED = True result = create_backup("synthetic.dump", runner=RecordingRunner()) result.path.write_bytes(b"PGDMP tampered!") restore_runner = RecordingRunner() target = DatabaseTarget( host="restore-db", port="5432", name="gizmoduck_restore", user="restore_user", password="restore-secret", ) with pytest.raises(BackupError, match="digest_mismatch"): verify_backup(result.name, target, runner=restore_runner) assert restore_runner.calls == [] def test_verification_sanitizes_digest_read_failure( backup_root, database_settings, synthetic_snapshot, settings, monkeypatch ): settings.BACKUP_VERIFY_ALLOWED = True result = create_backup("synthetic.dump", runner=RecordingRunner()) monkeypatch.setattr( "core.backup._file_digest", lambda _path: (_ for _ in ()).throw(OSError("SYNTHETIC PRIVATE PATH")), ) target = DatabaseTarget( host="restore-db", port="5432", name="restore", user="user", password="secret" ) with pytest.raises(BackupError, match="verification_failed") as error: verify_backup(result.name, target, runner=RecordingRunner()) assert "PRIVATE" not in str(error.value) def test_verification_rejects_manifest_with_wrong_field_type( backup_root, database_settings, synthetic_snapshot, settings ): settings.BACKUP_VERIFY_ALLOWED = True result = create_backup("synthetic.dump", runner=RecordingRunner()) manifest = json.loads(result.manifest_path.read_text(encoding="utf-8")) manifest["application_version"] = 1 result.manifest_path.write_text(json.dumps(manifest), encoding="utf-8") target = DatabaseTarget( host="restore-db", port="5432", name="gizmoduck_restore", user="restore_user", password="restore-secret", ) with pytest.raises(BackupError, match="invalid_manifest"): verify_backup(result.name, target, runner=RecordingRunner()) def test_backup_rejects_nonmatching_postgresql_client_major( backup_root, database_settings, synthetic_snapshot ): class WrongVersionRunner(RecordingRunner): def __call__(self, argv, **kwargs): result = super().__call__(argv, **kwargs) if argv == ["pg_dump", "--version"]: result.stdout = b"pg_dump (PostgreSQL) 16.9\n" return result with pytest.raises(BackupError, match="client_version_failed"): create_backup("synthetic.dump", runner=WrongVersionRunner()) assert list(backup_root.iterdir()) == [] def test_pg_restore_receives_password_only_in_environment( backup_root, database_settings, synthetic_snapshot, settings, monkeypatch ): settings.BACKUP_VERIFY_ALLOWED = True result = create_backup("synthetic.dump", runner=RecordingRunner()) snapshots = iter( [DatabaseSnapshot(migrations=(), row_counts={}), synthetic_snapshot] ) monkeypatch.setattr("core.backup._read_snapshot", lambda _target: next(snapshots)) runner = RecordingRunner() target = DatabaseTarget( host="restore-db", port="5432", name="gizmoduck_restore", user="restore_user", password="synthetic-restore-secret", ) verification = verify_backup(result.name, target, runner=runner) argv, env, check, stdout, capture_output = runner.calls[0] assert argv[:5] == [ "pg_restore", "--exit-on-error", "--single-transaction", "--no-owner", "--no-acl", ] assert target.password not in argv assert env["PGPASSWORD"] == target.password assert check is True assert stdout is None assert capture_output is True assert verification.row_counts == synthetic_snapshot.row_counts @pytest.mark.django_db(transaction=True) def test_snapshot_counts_tables_in_non_system_schemas(settings): from core.backup import _read_snapshot target = DatabaseTarget.from_default(settings.DATABASES["default"]) with connection.cursor() as cursor: cursor.execute("CREATE SCHEMA synthetic_extra") cursor.execute("CREATE TABLE synthetic_extra.marker (id integer)") cursor.execute("INSERT INTO synthetic_extra.marker VALUES (1)") try: snapshot = _read_snapshot(target) assert snapshot.row_counts["synthetic_extra.marker"] == 1 finally: with connection.cursor() as cursor: cursor.execute("DROP SCHEMA synthetic_extra CASCADE") def test_manifest_rejects_boolean_integer_fields( backup_root, database_settings, synthetic_snapshot, settings ): settings.BACKUP_VERIFY_ALLOWED = True result = create_backup("synthetic.dump", runner=RecordingRunner()) manifest = json.loads(result.manifest_path.read_text(encoding="utf-8")) manifest["byte_size"] = True result.manifest_path.write_text(json.dumps(manifest), encoding="utf-8") target = DatabaseTarget( host="restore-db", port="5432", name="restore", user="user", password="secret" ) with pytest.raises(BackupError, match="invalid_manifest"): verify_backup(result.name, target, runner=RecordingRunner()) def test_manifest_rejects_boolean_version( backup_root, database_settings, synthetic_snapshot, settings ): settings.BACKUP_VERIFY_ALLOWED = True result = create_backup("synthetic.dump", runner=RecordingRunner()) manifest = json.loads(result.manifest_path.read_text(encoding="utf-8")) manifest["version"] = True result.manifest_path.write_text(json.dumps(manifest), encoding="utf-8") target = DatabaseTarget( host="restore-db", port="5432", name="restore", user="user", password="secret" ) with pytest.raises(BackupError, match="invalid_manifest"): verify_backup(result.name, target, runner=RecordingRunner()) def test_synthetic_preflight_database_error_is_sanitized(monkeypatch, settings): import core.backup as backup settings.BACKUP_VERIFY_ALLOWED = True for name, value in { "BACKUP_SOURCE_HOST": "source", "BACKUP_SOURCE_NAME": "source", "BACKUP_SOURCE_USER": "user", "BACKUP_SOURCE_PASSWORD": "SYNTHETIC PRIVATE PASSWORD", }.items(): monkeypatch.setenv(name, value) monkeypatch.setattr( backup, "_read_snapshot", lambda _target: (_ for _ in ()).throw( psycopg.OperationalError("SYNTHETIC PRIVATE DIAGNOSTIC") ), ) with pytest.raises(BackupError, match="integration_verification_failed") as error: backup.run_synthetic_restore_verification() assert "PRIVATE" not in str(error.value) def test_cleanup_attempts_every_resource_and_reports_safe_failure(monkeypatch, tmp_path): import core.backup as backup attempts = [] dump = tmp_path / "synthetic.dump" manifest = tmp_path / "synthetic.dump.json" dump.write_bytes(b"dump") manifest.write_text("{}", encoding="utf-8") artifact = SimpleNamespace(path=dump, manifest_path=manifest) source = DatabaseTarget("source", "5432", "source", "user", "password") target = DatabaseTarget("target", "5432", "target", "user", "password") original_unlink = Path.unlink def unlink(path, *args, **kwargs): attempts.append(f"unlink:{path.name}") if path == dump: raise OSError("private path") return original_unlink(path, *args, **kwargs) def clear(database): attempts.append(f"clear:{database.name}") if database == source: raise BackupError("cleanup_failed") monkeypatch.setattr(Path, "unlink", unlink) monkeypatch.setattr(backup, "_clear_disposable_database", clear) with pytest.raises(BackupError, match="cleanup_failed"): backup._cleanup_synthetic_resources(artifact, source, target) assert attempts == [ "unlink:synthetic.dump", "unlink:synthetic.dump.json", "clear:source", "clear:target", ] @pytest.mark.django_db(transaction=True) def test_concurrent_insert_is_excluded_from_both_manifest_and_dump(tmp_path, settings): from finance.models import Account settings.BACKUP_ROOT = tmp_path settings.BACKUP_VERIFY_ALLOWED = True source = DatabaseTarget.from_default(settings.DATABASES["default"]) Account.objects.create(name="Synthetic Snapshot First", kind="BANK", currency="BRL") inserted = False def runner(argv, **kwargs): nonlocal inserted if argv[0] == "pg_dump" and argv != ["pg_dump", "--version"]: assert any(value.startswith("--snapshot=") for value in argv) Account.objects.create( name="Synthetic Snapshot Concurrent", kind="BANK", currency="BRL" ) inserted = True return subprocess.run(argv, **kwargs) database_name = f"snapshot_restore_{os.getpid()}" admin = dict( host=source.host, port=source.port, dbname="postgres", user=source.user, password=source.password, autocommit=True, ) with connect(**admin) as database: database.execute(sql.SQL("CREATE DATABASE {}").format(sql.Identifier(database_name))) target = DatabaseTarget( host=source.host, port=source.port, name=database_name, user=source.user, password=source.password, ) try: result = create_backup("concurrent.dump", runner=runner) verification = verify_backup(result.name, target) assert inserted is True assert result.row_counts["finance_account"] == 1 assert verification.row_counts["finance_account"] == 1 assert Account.objects.count() == 2 finally: with connect(**admin) as database: database.execute( sql.SQL("DROP DATABASE {} WITH (FORCE)").format( sql.Identifier(database_name) ) )