PVE-Backup-Report/tests/test_pbs_client.py

from __future__ import annotations

from pathlib import Path

from pve_backup_report.config import AppConfig, PbsServerConfig
from pve_backup_report.pbs_client import PbsClient, PbsConnectionError, PbsHttpError


class FakeResponse:
    def __init__(self, status_code: int, payload: dict, reason: str = "OK") -> None:
        self.status_code = status_code
        self._payload = payload
        self.reason = reason

    def json(self) -> dict:
        return self._payload


class FakeSession:
    def __init__(self, response: FakeResponse) -> None:
        self.headers: dict[str, str] = {}
        self.response = response
        self.calls: list[tuple[str, dict[str, object] | None, int, bool | str]] = []

    def get(
        self,
        url: str,
        params: dict[str, object] | None,
        timeout: int,
        verify: bool | str,
    ) -> FakeResponse:
        self.calls.append((url, params, timeout, verify))
        return self.response

    def close(self) -> None:
        pass


def make_config() -> AppConfig:
    return 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=False,
        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=False,
                ca_bundle=None,
                timeout_seconds=30,
            ),
            PbsServerConfig(
                prefix="PBS02",
                name="PBS02",
                api_url="https://backup-b.example.invalid:8007",
                api_token_id="backup-report@pbs!report",
                api_token_secret="secret2",
                verify_tls=False,
                ca_bundle=None,
                timeout_seconds=30,
            ),
            PbsServerConfig(
                prefix="PBS10",
                name="PBS10",
                api_url="https://backup-j.example.invalid:8007",
                api_token_id="backup-report@pbs!report",
                api_token_secret="secret10",
                verify_tls=False,
                ca_bundle=None,
                timeout_seconds=120,
            ),
        ),
        log_level="INFO",
        report_filename_prefix="rapport-sauvegardes-pve",
    )


def test_get_prune_jobs_uses_pbs_token_auth() -> None:
    session = FakeSession(FakeResponse(200, {"data": []}))
    client = PbsClient(make_config(), session=session)  # type: ignore[arg-type]

    assert client.get_prune_jobs() == []
    assert session.calls == [
        ("https://backup-a.example.invalid:8007/api2/json/config/prune", None, 30, False)
    ]
    assert session.headers["Authorization"].startswith(
        "PBSAPIToken=backup-report@pbs!report:"
    )
    assert "secret" in session.headers["Authorization"]


def test_get_datastore_snapshots_uses_namespace_param() -> None:
    session = FakeSession(FakeResponse(200, {"data": []}))
    client = PbsClient(make_config(), session=session)  # type: ignore[arg-type]

    assert client.get_datastore_snapshots("RAID5", "serveurs-internes") == []
    assert session.calls == [
        (
            "https://backup-a.example.invalid:8007/api2/json/admin/datastore/RAID5/snapshots",
            {"ns": "serveurs-internes"},
            30,
            False,
        )
    ]


def test_get_datastores() -> None:
    session = FakeSession(FakeResponse(200, {"data": []}))
    client = PbsClient(make_config(), session=session)  # type: ignore[arg-type]

    assert client.get_datastores() == []
    assert session.calls == [
        ("https://backup-a.example.invalid:8007/api2/json/config/datastore", None, 30, False)
    ]


def test_get_datastore_status() -> None:
    session = FakeSession(FakeResponse(200, {"data": {"total": 100, "used": 40, "avail": 60}}))
    client = PbsClient(make_config(), session=session)  # type: ignore[arg-type]

    assert client.get_datastore_status("RAID5") == {"total": 100, "used": 40, "avail": 60}
    assert session.calls == [
        ("https://backup-a.example.invalid:8007/api2/json/admin/datastore/RAID5/status", None, 30, False)
    ]


def test_get_datastore_gc_status() -> None:
    session = FakeSession(FakeResponse(200, {"data": {"upid": "UPID:pbs:gc"}}))
    client = PbsClient(make_config(), session=session)  # type: ignore[arg-type]

    assert client.get_datastore_gc_status("RAID5") == {"upid": "UPID:pbs:gc"}
    assert session.calls == [
        ("https://backup-a.example.invalid:8007/api2/json/admin/datastore/RAID5/gc", None, 30, False)
    ]


def test_get_datastore_namespaces() -> None:
    session = FakeSession(FakeResponse(200, {"data": []}))
    client = PbsClient(make_config(), session=session)  # type: ignore[arg-type]

    assert client.get_datastore_namespaces("RAID5") == []
    assert session.calls == [
        ("https://backup-a.example.invalid:8007/api2/json/admin/datastore/RAID5/namespace", None, 30, False)
    ]


def test_get_access_users() -> None:
    session = FakeSession(FakeResponse(200, {"data": []}))
    client = PbsClient(make_config(), session=session)  # type: ignore[arg-type]

    assert client.get_access_users() == []
    assert session.calls == [
        ("https://backup-a.example.invalid:8007/api2/json/access/users", None, 30, False)
    ]


def test_get_access_permissions() -> None:
    session = FakeSession(FakeResponse(200, {"data": {"/datastore/RAID5": {}}}))
    client = PbsClient(make_config(), session=session)  # type: ignore[arg-type]

    assert client.get_access_permissions("backup@pbs", "/datastore/RAID5") == {
        "/datastore/RAID5": {}
    }
    assert session.calls == [
        (
            "https://backup-a.example.invalid:8007/api2/json/access/permissions",
            {"auth-id": "backup@pbs", "path": "/datastore/RAID5"},
            30,
            False,
        )
    ]


def test_pbs02_client_uses_pbs02_settings() -> None:
    session = FakeSession(FakeResponse(200, {"data": []}))
    config = make_config()
    client = PbsClient(config, server=config.pbs_servers[1], session=session)  # type: ignore[arg-type]

    assert client.get_prune_jobs() == []
    assert session.calls == [
        ("https://backup-b.example.invalid:8007/api2/json/config/prune", None, 30, False)
    ]
    assert session.headers["Authorization"].startswith(
        "PBSAPIToken=backup-report@pbs!report:"
    )
    assert "secret2" in session.headers["Authorization"]


def test_pbs10_client_uses_dynamically_discovered_settings() -> None:
    session = FakeSession(FakeResponse(200, {"data": []}))
    config = make_config()
    client = PbsClient(config, server=config.pbs_servers[2], session=session)  # type: ignore[arg-type]

    assert client.get_prune_jobs() == []
    assert session.calls == [
        ("https://backup-j.example.invalid:8007/api2/json/config/prune", None, 120, False)
    ]
    assert session.headers["Authorization"].startswith(
        "PBSAPIToken=backup-report@pbs!report:"
    )
    assert "secret10" in session.headers["Authorization"]


def test_pbs_sanitize_exception_masks_sensitive_values() -> None:
    message = PbsClient._sanitize_exception(
        PbsConnectionError(
            "PBSAPIToken=backup@pbs!report:secret password=secret2 secret=secret3"
        )
    )

    assert "report:secret" not in message
    assert "secret2" not in message
    assert "secret3" not in message
    assert "PBSAPIToken=***" in message
    assert "password=***" in message
    assert "secret=***" in message


def test_pbs_http_error_message_is_sanitized() -> None:
    session = FakeSession(
        FakeResponse(
            500,
            {"data": "PBSAPIToken=backup@pbs!report:secret password=secret2"},
            "Server Error",
        )
    )
    client = PbsClient(make_config(), session=session)  # type: ignore[arg-type]

    try:
        client.get_prune_jobs()
    except PbsHttpError as exc:
        assert "report:secret" not in exc.message
        assert "secret2" not in exc.message
        assert exc.message == "PBSAPIToken=*** password=***"
    else:
        raise AssertionError("PbsHttpError attendu")