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")