gizmoduck/tests/test_backup.py

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)
                )
            )