FluxIP/tests/test_backup.py

from __future__ import annotations

import base64
import sqlite3
import sys
import tarfile
from pathlib import Path

import pytest

from deploy import backup


def create_database(path: Path) -> None:
    with sqlite3.connect(path) as connection:
        connection.execute("CREATE TABLE sample (value TEXT NOT NULL)")
        connection.execute("INSERT INTO sample VALUES ('ready')")


def create_backup_set(output: Path, kind: str, timestamp: str) -> tuple[Path, Path]:
    archive = output / f"fluxip-{kind}-backup-{timestamp}.tar.gz"
    checksum = archive.with_name(f"{archive.name}.sha256")
    archive.write_bytes(b"archive")
    checksum.write_text("checksum\n", encoding="ascii")
    return archive, checksum


def test_default_backup_excludes_master_key(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    database = tmp_path / "fluxip.db"
    output = tmp_path / "backups"
    create_database(database)
    monkeypatch.setattr(
        sys,
        "argv",
        ["backup.py", "--database", str(database), "--output-dir", str(output)],
    )

    assert backup.main() == 0
    archive = next(output.glob("fluxip-database-backup-*.tar.gz"))
    with tarfile.open(archive, "r:gz") as package:
        assert package.getnames() == ["fluxip.db"]
    assert archive.with_name(f"{archive.name}.sha256").is_file()


def test_complete_backup_requires_explicit_plaintext_confirmation(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    database = tmp_path / "fluxip.db"
    master_key = tmp_path / "master.key"
    output = tmp_path / "backups"
    create_database(database)
    master_key.write_bytes(base64.urlsafe_b64encode(b"k" * 32))
    base_args = [
        "backup.py",
        "--database",
        str(database),
        "--master-key",
        str(master_key),
        "--output-dir",
        str(output),
    ]
    monkeypatch.setattr(sys, "argv", base_args)
    with pytest.raises(RuntimeError, match="拒绝把主密钥写入未加密备份"):
        backup.main()

    monkeypatch.setattr(sys, "argv", [*base_args, "--allow-plaintext-key-archive"])
    assert backup.main() == 0
    archive = next(output.glob("fluxip-complete-backup-*.tar.gz"))
    with tarfile.open(archive, "r:gz") as package:
        assert package.getnames() == ["fluxip.db", "master.key"]


def test_failed_backup_removes_private_temporary_archive(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    database = tmp_path / "fluxip.db"
    output = tmp_path / "backups"
    create_database(database)
    monkeypatch.setattr(
        sys,
        "argv",
        ["backup.py", "--database", str(database), "--output-dir", str(output)],
    )
    monkeypatch.setattr(
        backup.tarfile,
        "open",
        lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("archive failed")),
    )

    with pytest.raises(RuntimeError, match="archive failed"):
        backup.main()

    assert not list(output.glob("*.tmp"))


def test_keep_last_rejects_negative_value(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(sys, "argv", ["backup.py", "--keep-last", "-1"])

    with pytest.raises(SystemExit) as exc_info:
        backup.parse_args()

    assert exc_info.value.code == 2


def test_prune_backups_is_type_scoped_and_stays_in_output_dir(tmp_path: Path) -> None:
    output = tmp_path / "backups"
    output.mkdir()
    timestamps = ["20260101T000000Z", "20260102T000000Z", "20260103T000000Z"]
    database_sets = [create_backup_set(output, "database", value) for value in timestamps]
    complete_sets = [create_backup_set(output, "complete", value) for value in timestamps]
    unrelated = output / "fluxip-database-backup-20250101T000000Z.tar.gz.bak"
    unrelated.write_bytes(b"keep")
    outside = tmp_path / "fluxip-database-backup-20240101T000000Z.tar.gz"
    outside.write_bytes(b"keep")

    assert backup.prune_backups(output, "database", 2) == 1
    assert not database_sets[0][0].exists()
    assert not database_sets[0][1].exists()
    assert all(path.exists() for pair in database_sets[1:] for path in pair)
    assert all(path.exists() for pair in complete_sets for path in pair)
    assert unrelated.exists()
    assert outside.exists()

    assert backup.prune_backups(output, "complete", 1) == 2
    assert all(not path.exists() for pair in complete_sets[:2] for path in pair)
    assert all(path.exists() for path in complete_sets[2])


def test_keep_last_zero_disables_cleanup(tmp_path: Path) -> None:
    output = tmp_path / "backups"
    output.mkdir()
    backup_set = create_backup_set(output, "database", "20260101T000000Z")

    assert backup.prune_backups(output, "database", 0) == 0
    assert all(path.exists() for path in backup_set)


def test_main_applies_retention_after_success(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    database = tmp_path / "fluxip.db"
    output = tmp_path / "backups"
    output.mkdir()
    create_database(database)
    old_set = create_backup_set(output, "database", "20000101T000000Z")
    monkeypatch.setattr(
        sys,
        "argv",
        [
            "backup.py",
            "--database",
            str(database),
            "--output-dir",
            str(output),
            "--keep-last",
            "1",
        ],
    )

    assert backup.main() == 0
    assert all(not path.exists() for path in old_set)
    assert len(list(output.glob("fluxip-database-backup-*.tar.gz"))) == 1