160 lines
4.3 KiB
Python
160 lines
4.3 KiB
Python
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]
|