gizmoduck/tests/test_backup_commands.py

from datetime import datetime, timezone
from io import StringIO
from pathlib import Path

import pytest
from django.core.management import call_command
from django.core.management.base import CommandError

from core.backup import BackupError, BackupResult, VerificationResult


def backup_result():
    return BackupResult(
        name="synthetic.dump",
        path=Path("/private/backup/synthetic.dump"),
        manifest_path=Path("/private/backup/synthetic.dump.json"),
        version=1,
        created_at=datetime(2026, 9, 8, tzinfo=timezone.utc),
        postgresql_client_version="pg_dump (PostgreSQL) 17.6",
        row_counts={"finance_account": 2, "finance_transaction": 4},
    )


def test_backup_database_prints_only_safe_summary(monkeypatch):
    monkeypatch.setattr(
        "core.management.commands.backup_database.create_backup",
        lambda name: backup_result(),
    )
    stdout = StringIO()

    call_command("backup_database", output="synthetic.dump", stdout=stdout)

    assert stdout.getvalue().splitlines() == [
        "backup=synthetic.dump format=1",
        "table=finance_account rows=2",
        "table=finance_transaction rows=4",
    ]
    assert "/private/backup" not in stdout.getvalue()


def test_backup_database_converts_service_error_to_safe_code(monkeypatch):
    def fail(_name):
        raise BackupError("backup_failed")

    monkeypatch.setattr("core.management.commands.backup_database.create_backup", fail)

    with pytest.raises(CommandError, match=r"^backup_failed$"):
        call_command("backup_database", output="synthetic.dump")


def test_verify_backup_uses_configured_password_and_prints_safe_summary(
    monkeypatch, settings
):
    monkeypatch.setitem(settings.DATABASES, "default", {
        "HOST": "restore-db",
        "PORT": "5432",
        "NAME": "gizmoduck_restore",
        "USER": "restore_user",
        "PASSWORD": "synthetic-restore-secret",
    })
    captured = {}

    def verify(name, target):
        captured["name"] = name
        captured["target"] = target
        return VerificationResult(
            name=name,
            version=1,
            row_counts={"django_migrations": 12, "finance_account": 2},
        )

    monkeypatch.setattr("core.management.commands.verify_backup.verify_backup", verify)
    stdout = StringIO()

    call_command(
        "verify_backup",
        input="synthetic.dump",
        target_name="gizmoduck_restore",
        target_host="restore-db",
        target_user="restore_user",
        stdout=stdout,
    )

    assert captured["name"] == "synthetic.dump"
    assert captured["target"].password == "synthetic-restore-secret"
    assert stdout.getvalue().splitlines() == [
        "backup=synthetic.dump format=1 verified",
        "table=django_migrations rows=12",
        "table=finance_account rows=2",
    ]
    assert "synthetic-restore-secret" not in stdout.getvalue()


def test_verify_backup_converts_service_error_to_safe_code(monkeypatch, settings):
    monkeypatch.setitem(settings.DATABASES, "default", {
        "HOST": "restore-db",
        "PORT": "5432",
        "NAME": "gizmoduck_restore",
        "USER": "restore_user",
        "PASSWORD": "synthetic-restore-secret",
    })

    def fail(_name, _target):
        raise BackupError("target_not_empty")

    monkeypatch.setattr("core.management.commands.verify_backup.verify_backup", fail)

    with pytest.raises(CommandError, match=r"^target_not_empty$"):
        call_command(
            "verify_backup",
            input="synthetic.dump",
            target_name="gizmoduck_restore",
            target_host="restore-db",
            target_user="restore_user",
        )


def test_backup_restore_acceptance_command_prints_only_safe_summary(monkeypatch):
    calls = 0

    def verify():
        nonlocal calls
        calls += 1
        return VerificationResult(
            name="synthetic-restore-check.dump",
            version=1,
            row_counts={"finance_account": 1, "finance_transaction": 1},
        )

    monkeypatch.setattr(
        "core.management.commands.test_backup_restore.run_synthetic_restore_verification",
        verify,
    )
    stdout = StringIO()

    call_command("test_backup_restore", stdout=stdout)

    assert calls == 2
    assert stdout.getvalue().splitlines() == [
        "backup=synthetic-restore-check.dump format=1 verified",
        "table=finance_account rows=1",
        "table=finance_transaction rows=1",
    ]