Initial commit

2026-05-13 16:04:17 +02:00
commit b66612d672
43 changed files with 10515 additions and 0 deletions
@@ -0,0 +1,159 @@
+from pve_backup_report.coverage import (
+    STATUS_DISABLED_PBS,
+    STATUS_INDETERMINATE,
+    STATUS_MISSING,
+    STATUS_NON_PBS_PLANNED,
+    STATUS_PBS_PLANNED,
+    analyze_backup_coverage,
+    calculate_backup_coverage,
+    calculate_explicit_vmid_coverage,
+    parse_job_vmids,
+)
+from pve_backup_report.models import (
+    BackupJob,
+    Guest,
+    PbsDatastoreUsage,
+    PbsGarbageCollectionStatus,
+    PbsStorage,
+    Pool,
+    ReportData,
+)
+
+
+def test_parse_job_vmids_with_exclusions() -> None:
+    job = BackupJob(
+        job_id="backup-prod",
+        enabled=True,
+        selection="vmid=100,101,102",
+        excluded="101",
+    )
+
+    assert parse_job_vmids(job) == {100, 102}
+
+
+def test_calculate_explicit_vmid_coverage() -> None:
+    guests = [
+        Guest(vmid=100, name="srv-a", guest_type="qemu"),
+        Guest(vmid=101, name="srv-b", guest_type="lxc"),
+    ]
+    jobs = [
+        BackupJob(
+            job_id="backup-prod",
+            enabled=True,
+            selection="vmid=100",
+        )
+    ]
+
+    coverage = calculate_explicit_vmid_coverage(guests, jobs)
+
+    assert coverage[0].status == STATUS_INDETERMINATE
+    assert coverage[0].jobs[0].job_id == "backup-prod"
+    assert coverage[1].status == STATUS_MISSING
+
+
+def test_all_job_covers_all_except_excluded() -> None:
+    guests = [
+        Guest(vmid=100, name="srv-a", guest_type="qemu"),
+        Guest(vmid=101, name="srv-b", guest_type="lxc"),
+    ]
+    jobs = [
+        BackupJob(
+            job_id="backup-all",
+            storage="backup-storage",
+            enabled=True,
+            selection="all=true",
+            excluded="101",
+        )
+    ]
+    storages = [PbsStorage(storage_id="backup-storage", enabled=True)]
+
+    coverage = calculate_backup_coverage(guests, jobs, storages)
+
+    assert coverage[0].status == STATUS_PBS_PLANNED
+    assert coverage[1].status == STATUS_MISSING
+
+
+def test_non_pbs_storage_is_distinguished() -> None:
+    guests = [Guest(vmid=100, name="srv-a", guest_type="qemu")]
+    jobs = [
+        BackupJob(
+            job_id="backup-local",
+            storage="local",
+            enabled=True,
+            selection="vmid=100",
+        )
+    ]
+
+    coverage = calculate_backup_coverage(guests, jobs, pbs_storages=[])
+
+    assert coverage[0].status == STATUS_NON_PBS_PLANNED
+
+
+def test_disabled_pbs_storage_is_reported() -> None:
+    guests = [Guest(vmid=100, name="srv-a", guest_type="qemu")]
+    jobs = [
+        BackupJob(
+            job_id="backup-disabled",
+            storage="pbs-disabled",
+            enabled=True,
+            selection="vmid=100",
+        )
+    ]
+    storages = [PbsStorage(storage_id="pbs-disabled", enabled=False)]
+
+    coverage = calculate_backup_coverage(guests, jobs, storages)
+
+    assert coverage[0].status == STATUS_DISABLED_PBS
+
+
+def test_pool_job_covers_pool_members_except_excluded() -> None:
+    guests = [
+        Guest(vmid=100, name="srv-a", guest_type="qemu"),
+        Guest(vmid=101, name="srv-b", guest_type="lxc"),
+        Guest(vmid=102, name="srv-c", guest_type="qemu"),
+    ]
+    jobs = [
+        BackupJob(
+            job_id="backup-pool",
+            storage="backup-storage",
+            enabled=True,
+            selection="pool=prod",
+            excluded="101",
+        )
+    ]
+    storages = [PbsStorage(storage_id="backup-storage", enabled=True)]
+    pools = [Pool(pool_id="prod", vmids={100, 101})]
+
+    coverage = calculate_backup_coverage(guests, jobs, storages, pools)
+
+    assert coverage[0].status == STATUS_PBS_PLANNED
+    assert coverage[1].status == STATUS_MISSING
+    assert coverage[2].status == STATUS_MISSING
+
+
+def test_analyze_backup_coverage_preserves_pbs_datastore_usages() -> None:
+    usage = PbsDatastoreUsage(
+        server_name="PBS01",
+        datastore="RAID5",
+        total_bytes=100,
+        used_bytes=40,
+        available_bytes=60,
+    )
+    report_data = ReportData(pbs_datastore_usages=[usage])
+
+    analyzed = analyze_backup_coverage(report_data)
+
+    assert analyzed.pbs_datastore_usages == [usage]
+
+
+def test_analyze_backup_coverage_preserves_pbs_gc_statuses() -> None:
+    status = PbsGarbageCollectionStatus(
+        server_name="PBS02",
+        datastore="PBS2RAID5",
+        status="en_cours",
+    )
+    report_data = ReportData(pbs_gc_statuses=[status])
+
+    analyzed = analyze_backup_coverage(report_data)
+
+    assert analyzed.pbs_gc_statuses == [status]