from dataclasses import replace from pathlib import Path from pve_backup_report.config import AppConfig, PbsServerConfig from pve_backup_report.coverage import STATUS_MISSING, STATUS_PBS_PLANNED from pve_backup_report.models import ( BackupCoverage, BackupJob, Guest, LastBackupResult, PbsAccessUser, PbsBackupSnapshotSummary, PbsRetentionPolicy, PbsStorage, ReportData, ) from pve_backup_report.report_data import build_report_summary, prepare_report_data, report_data_to_dict 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=(), log_level="INFO", report_filename_prefix="rapport-sauvegardes-pve", ) def test_build_report_summary() -> None: guest_vm = Guest(vmid=100, name="srv-a", guest_type="qemu") guest_ct = Guest(vmid=101, name="ct-a", guest_type="lxc") active_job = BackupJob(job_id="backup-a", enabled=True) inactive_job = BackupJob(job_id="backup-b", enabled=False) report_data = ReportData( pbs_storages=[PbsStorage(storage_id="backup-storage")], guests=[guest_vm, guest_ct], backup_jobs=[active_job, inactive_job], coverage=[ BackupCoverage(guest=guest_vm, status=STATUS_PBS_PLANNED), BackupCoverage(guest=guest_ct, status=STATUS_MISSING), ], ) summary = build_report_summary(report_data, make_config()) assert summary.total_vm == 1 assert summary.total_ct == 1 assert summary.total_guests == 2 assert summary.pbs_storage_count == 1 assert summary.backup_job_count == 2 assert summary.active_backup_job_count == 1 assert summary.inactive_backup_job_count == 1 assert summary.pbs_planned_count == 1 assert summary.missing_count == 1 def test_report_data_to_dict_keeps_pdf_inputs() -> None: guest = Guest(vmid=100, name="srv-a", guest_type="qemu", node="pve01") job = BackupJob(job_id="backup-a", storage="backup-storage", schedule="23:00") report_data = ReportData( pbs_storages=[PbsStorage(storage_id="backup-storage", server="backup.example.invalid")], pbs_access_users=[ PbsAccessUser( server_name="PBS01", auth_id="backup@pbs", user_id="backup@pbs", storage_id="backup-storage", permissions={"Datastore.Backup": True}, raw={ "Authorization": "PBSAPIToken=backup@pbs!report:secret", "password": "secret", "token": "secret", }, ) ], pbs_retention_policies=[ PbsRetentionPolicy( policy_id="prune-prod", server_name="PBS01", datastore="RAID5", namespace="serveurs-internes", keep_daily=14, ) ], pbs_snapshot_summaries={ ("PBS01", "RAID5", "serveurs-internes", "qemu", 100): PbsBackupSnapshotSummary( server_name="PBS01", vmid=100, guest_type="qemu", datastore="RAID5", namespace="serveurs-internes", snapshot_count=3, raw={ "fingerprint": "aa:bb:cc", "files": [{"filename": "index.json.blob"}], "owner": "backup@pbs", }, ) }, guests=[guest], backup_jobs=[job], coverage=[ BackupCoverage( guest=guest, status=STATUS_PBS_PLANNED, jobs=[job], storages=["backup-storage"], ) ], last_backup_results={100: LastBackupResult(vmid=100, status="succes")}, ) data = report_data_to_dict(report_data) assert data["pbs_storages"][0]["id"] == "backup-storage" assert data["pbs_access_users"][0]["auth_id"] == "backup@pbs" assert data["pbs_access_users"][0]["server"] == "PBS01" assert data["pbs_access_users"][0]["user_id"] == "backup@pbs" assert data["pbs_access_users"][0]["permissions"] == {"Datastore.Backup": True} assert "raw" not in data["pbs_access_users"][0] assert data["pbs_server_names"] == [] assert data["pbs_retention_policies"][0]["id"] == "prune-prod" assert data["pbs_retention_policies"][0]["keep_daily"] == 14 assert data["pbs_snapshot_summaries"][0]["snapshot_count"] == 3 assert data["pbs_snapshot_summaries"][0]["server"] == "PBS01" assert data["pbs_snapshot_summaries"][0]["type"] == "qemu" assert "raw" not in data["pbs_snapshot_summaries"][0] assert data["backup_jobs"][0]["id"] == "backup-a" assert data["coverage"][0]["vmid"] == 100 assert data["coverage"][0]["jobs"] == ["backup-a"] assert data["coverage"][0]["last_backup"]["status"] == "succes" def test_report_data_to_dict_redacts_sensitive_nested_fields() -> None: report_data = ReportData( 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={ "nested": { "api_token_secret": "secret", "safe": "visible", }, "ticket": "secret-ticket", }, ) } ) data = report_data_to_dict(report_data) text = str(data) assert "secret" not in text assert "ticket" not in text assert "api_token_secret" not in text def test_prepare_report_data_keeps_only_configured_pbs_servers() -> None: config = replace( make_config(), pbs_servers=( PbsServerConfig( prefix="PBS01", name="PBS01", api_url="https://backup.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-extra.example.invalid:8007", api_token_id="backup-report@pbs!report", api_token_secret="secret4", verify_tls=True, ca_bundle=None, timeout_seconds=30, ), ), ) access_user = PbsAccessUser( server_name="PBS01", auth_id="backup@pbs", user_id="backup@pbs", storage_id="backup-storage", ) report_data = prepare_report_data( ReportData( pbs_server_names=["PBS01", "PBS02", "PBS03", "PBS04"], pbs_access_users=[access_user], ), config, ) assert report_data.pbs_server_names == ["PBS01", "PBS04"] assert report_data.pbs_access_users == [access_user]