gizmoduck/tests/test_backup.py

import json
from hashlib import sha256
from pathlib import Path
from types import SimpleNamespace

import pytest

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 stdout is not None:
            stdout.write(self.output)
        return SimpleNamespace(stdout=b"pg_dump (PostgreSQL) 17.6\n")


@pytest.fixture
def database_settings(settings):
    settings.DATABASES["default"] = {
        "ENGINE": "django.db.backends.postgresql",
        "HOST": "live-db",
        "PORT": "5432",
        "NAME": "gizmoduck",
        "USER": "gizmoduck_migrate",
        "PASSWORD": "synthetic-secret",
    }
    return settings.DATABASES["default"]


@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},
    )
    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 not None
    assert capture_output is False
    assert result.path.read_bytes() == runner.output


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