65 lines
3.4 KiB
Python
65 lines
3.4 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
|
||
|
|
from django.test import TestCase
|
||
|
|
|
||
|
|
from pobsync_backend.models import HostConfig, SnapshotRecord
|
||
|
|
from pobsync_backend.stats_summary import collect_dashboard_stats, collect_host_stats
|
||
|
|
|
||
|
|
|
||
|
|
class StatsSummaryTests(TestCase):
|
||
|
|
def test_collect_dashboard_stats_sums_backup_data_across_hosts(self) -> None:
|
||
|
|
web = HostConfig.objects.create(host="web-01", address="web-01.example.test")
|
||
|
|
db = HostConfig.objects.create(host="db-01", address="db-01.example.test")
|
||
|
|
self._snapshot(web, "20260519-021500Z__SCHED01", SnapshotRecord.Kind.SCHEDULED, allocated=100)
|
||
|
|
self._snapshot(web, "20260519-031500Z__MANUAL1", SnapshotRecord.Kind.MANUAL, allocated=200)
|
||
|
|
self._snapshot(db, "20260519-041500Z__SCHED02", SnapshotRecord.Kind.SCHEDULED, allocated=300)
|
||
|
|
self._snapshot(db, "20260519-051500Z__BROKEN1", SnapshotRecord.Kind.INCOMPLETE, allocated=400)
|
||
|
|
|
||
|
|
stats = collect_dashboard_stats(hosts=[web, db], global_config=None)
|
||
|
|
|
||
|
|
self.assertEqual(stats["backup_data"]["scheduled"]["count"], 2)
|
||
|
|
self.assertEqual(stats["backup_data"]["scheduled"]["allocated_size_bytes"], 400)
|
||
|
|
self.assertEqual(stats["backup_data"]["manual"]["allocated_size_bytes"], 200)
|
||
|
|
self.assertEqual(stats["backup_data"]["incomplete"]["allocated_size_bytes"], 400)
|
||
|
|
self.assertEqual(stats["backup_data"]["total"]["count"], 4)
|
||
|
|
self.assertEqual(stats["backup_data"]["total"]["allocated_size_bytes"], 1000)
|
||
|
|
|
||
|
|
def test_collect_host_stats_sums_backup_data_by_snapshot_kind(self) -> None:
|
||
|
|
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
|
||
|
|
self._snapshot(host, "20260519-021500Z__SCHED01", SnapshotRecord.Kind.SCHEDULED, allocated=100)
|
||
|
|
self._snapshot(host, "20260519-031500Z__SCHED02", SnapshotRecord.Kind.SCHEDULED, allocated=200)
|
||
|
|
self._snapshot(host, "20260519-041500Z__MANUAL1", SnapshotRecord.Kind.MANUAL, allocated=300)
|
||
|
|
self._snapshot(host, "20260519-051500Z__BROKEN1", SnapshotRecord.Kind.INCOMPLETE, allocated=400)
|
||
|
|
|
||
|
|
stats = collect_host_stats(host=host)
|
||
|
|
|
||
|
|
self.assertEqual(stats["backup_data"]["scheduled"]["count"], 2)
|
||
|
|
self.assertEqual(stats["backup_data"]["scheduled"]["allocated_size_bytes"], 300)
|
||
|
|
self.assertEqual(stats["backup_data"]["manual"]["allocated_size_bytes"], 300)
|
||
|
|
self.assertEqual(stats["backup_data"]["incomplete"]["allocated_size_bytes"], 400)
|
||
|
|
self.assertEqual(stats["backup_data"]["total"]["count"], 4)
|
||
|
|
self.assertEqual(stats["backup_data"]["total"]["allocated_size_bytes"], 1000)
|
||
|
|
|
||
|
|
def _snapshot(self, host: HostConfig, dirname: str, kind: str, *, allocated: int) -> SnapshotRecord:
|
||
|
|
started_at = datetime.strptime(dirname.split("__", 1)[0], "%Y%m%d-%H%M%SZ").replace(tzinfo=timezone.utc)
|
||
|
|
return SnapshotRecord.objects.create(
|
||
|
|
host=host,
|
||
|
|
kind=kind,
|
||
|
|
dirname=dirname,
|
||
|
|
path=f"/backups/{host.host}/{kind}/{dirname}",
|
||
|
|
status="success",
|
||
|
|
started_at=started_at,
|
||
|
|
metadata={
|
||
|
|
"stats": {
|
||
|
|
"storage": {
|
||
|
|
"snapshot": {
|
||
|
|
"apparent_size_bytes": allocated * 2,
|
||
|
|
"allocated_size_bytes": allocated,
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
},
|
||
|
|
)
|