PVE-Backup-Report/tests/test_cli.py

from pathlib import Path

from pve_backup_report.models import PbsAccessUser, PbsBackupSnapshotSummary, ReportData
from pve_backup_report.cli import (
    build_parser,
    configured_pbs_clients,
    datastore_name_from_raw,
    ensure_report_output_dir_writable,
    ensure_writable_directory,
    run,
)
from pve_backup_report.config import AppConfig, PbsServerConfig


def test_cli_check_config(tmp_path, monkeypatch) -> None:
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "PVE_API_URL=https://pve.example.invalid:8006",
                "PVE_API_TOKEN_ID=backup-report@pve!report",
                "PVE_API_TOKEN_SECRET=secret",
            ]
        ),
        encoding="utf-8",
    )
    monkeypatch.chdir(tmp_path)

    assert run(["--check-config"]) == 0


def test_cli_has_dump_pbs_storage_usages() -> None:
    args = build_parser().parse_args(["--dump-pbs-storage-usages"])

    assert args.dump_pbs_storage_usages is True


def test_cli_has_dump_pbs_users() -> None:
    args = build_parser().parse_args(["--dump-pbs-users"])

    assert args.dump_pbs_users is True


def test_dump_report_data_does_not_export_sensitive_raw_fields(
    monkeypatch,
    capsys,
) -> None:
    config = AppConfig(
        pve_api_url="https://pve.example.invalid:8006",
        pve_api_token_id="backup-report@pve!report",
        pve_api_token_secret="secret",
        report_output_dir=Path("reports"),
        report_timezone="Europe/Paris",
        pve_verify_tls=True,
        pve_ca_bundle=None,
        pve_timeout_seconds=30,
        pve_backup_jobs_endpoint="/cluster/backup",
        pve_task_history_limit=500,
        pve_task_log_limit=5000,
        pbs_hostnames={},
        pbs_servers=(),
        log_level="INFO",
        report_filename_prefix="rapport-sauvegardes-pve",
    )
    report_data = ReportData(
        pbs_access_users=[
            PbsAccessUser(
                server_name="PBS01",
                auth_id="backup@pbs",
                user_id="backup@pbs",
                storage_id="BACKUP-PROD",
                raw={
                    "Authorization": "PBSAPIToken=abc:secret",
                    "password": "secret-password",
                },
            )
        ],
        pbs_snapshot_summaries={
            ("PBS01", "RAID5", "serveurs", "qemu", 100): PbsBackupSnapshotSummary(
                server_name="PBS01",
                vmid=100,
                guest_type="qemu",
                datastore="RAID5",
                namespace="serveurs",
                snapshot_count=1,
                raw={
                    "fingerprint": "aa:bb:cc",
                    "files": [{"filename": "index.json.blob"}],
                    "raw": {"secret": "secret-value"},
                },
            )
        },
    )

    monkeypatch.setattr("pve_backup_report.cli.load_config", lambda: config)
    monkeypatch.setattr(
        "pve_backup_report.cli.collect_data_or_log_error",
        lambda loaded_config, label: report_data,
    )

    assert run(["--dump-report-data"]) == 0
    output = capsys.readouterr().out

    assert "fingerprint" not in output
    assert '"raw"' not in output
    assert '"files"' not in output
    assert "PBSAPIToken=abc:secret" not in output
    assert "secret-password" not in output
    assert "secret-value" not in output
    assert '"pbs_access_users"' in output
    assert '"auth_id": "backup@pbs"' in output
    assert '"pbs_snapshot_summaries"' in output
    assert '"snapshot_count": 1' in output


def test_datastore_name_from_raw() -> None:
    assert datastore_name_from_raw({"name": "RAID5"}) == "RAID5"
    assert datastore_name_from_raw({"store": "PBS2RAID5"}) == "PBS2RAID5"
    assert datastore_name_from_raw({"datastore": "BACKUPSTORAGE"}) == "BACKUPSTORAGE"
    assert datastore_name_from_raw({}) is None


def test_report_output_dir_must_be_directory(tmp_path) -> None:
    output_file = tmp_path / "reports"
    output_file.write_text("not a directory", encoding="utf-8")

    try:
        ensure_report_output_dir_writable(output_file)
    except OSError as exc:
        assert "REPORT_OUTPUT_DIR doit pointer vers un repertoire" in str(exc)
    else:
        raise AssertionError("OSError attendu")


def test_docker_report_output_dir_falls_back_to_local_reports(
    tmp_path,
    monkeypatch,
) -> None:
    def fake_ensure_writable_directory(path: Path) -> None:
        if path == Path("/reports"):
            raise OSError("permission denied")
        ensure_writable_directory(path)

    monkeypatch.chdir(tmp_path)
    monkeypatch.setattr(
        "pve_backup_report.cli.ensure_writable_directory",
        fake_ensure_writable_directory,
    )

    assert ensure_report_output_dir_writable(Path("/reports")) == Path("reports")
    assert (tmp_path / "reports").is_dir()


def test_configured_pbs_clients_uses_every_configured_server() -> None:
    config = AppConfig(
        pve_api_url="https://pve.example.invalid:8006",
        pve_api_token_id="backup-report@pve!report",
        pve_api_token_secret="secret",
        report_output_dir=Path("reports"),
        report_timezone="Europe/Paris",
        pve_verify_tls=True,
        pve_ca_bundle=None,
        pve_timeout_seconds=30,
        pve_backup_jobs_endpoint="/cluster/backup",
        pve_task_history_limit=500,
        pve_task_log_limit=5000,
        pbs_hostnames={},
        pbs_servers=(
            PbsServerConfig(
                prefix="PBS01",
                name="PBS01",
                api_url="https://backup-a.example.invalid:8007",
                api_token_id="backup-report@pbs!report",
                api_token_secret="secret",
                verify_tls=True,
                ca_bundle=None,
                timeout_seconds=30,
            ),
            PbsServerConfig(
                prefix="PBS04",
                name="PBS04",
                api_url="https://backup-d.example.invalid:8007",
                api_token_id="backup-report@pbs!report",
                api_token_secret="secret4",
                verify_tls=True,
                ca_bundle=None,
                timeout_seconds=30,
            ),
        ),
        log_level="INFO",
        report_filename_prefix="rapport-sauvegardes-pve",
    )

    clients = configured_pbs_clients(config)

    try:
        assert [client.server_name for client in clients] == ["PBS01", "PBS04"]
    finally:
        for client in clients:
            client.close()