## Summary
- Fall back to filesystem measurement when snapshot storage metadata is missing. - Prefer `data/` inside snapshot directories so incomplete snapshot metadata/log files are not counted as backup data. - Add stats and dashboard rendering coverage for incomplete snapshots without recorded storage metadata. ## Tests - .venv/bin/python manage.py test src.pobsync_backend.tests.test_stats_summary src.pobsync_backend.tests.test_views.ViewTests.test_dashboard_host_cards_measure_incomplete_data_without_snapshot_metadata --verbosity 2 - .venv/bin/python manage.py check - git diff --check - .venv/bin/python manage.py test src.pobsync_backend --verbosity 2 Closes #62
This commit is contained in:
@@ -1,9 +1,12 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from tempfile import TemporaryDirectory
|
||||
|
||||
from django.test import TestCase
|
||||
|
||||
from pobsync.run_stats import tree_usage
|
||||
from pobsync_backend.models import HostConfig, SnapshotRecord
|
||||
from pobsync_backend.stats_summary import collect_dashboard_stats, collect_host_stats
|
||||
|
||||
@@ -42,6 +45,42 @@ class StatsSummaryTests(TestCase):
|
||||
self.assertEqual(stats["backup_data"]["total"]["count"], 4)
|
||||
self.assertEqual(stats["backup_data"]["total"]["allocated_size_bytes"], 1000)
|
||||
|
||||
def test_collect_host_stats_falls_back_to_filesystem_usage_for_snapshots_without_metadata(self) -> None:
|
||||
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
|
||||
with TemporaryDirectory() as tmp:
|
||||
incomplete_dir = Path(tmp) / host.host / ".incomplete" / "20260519-051500Z__BROKEN1"
|
||||
data_dir = incomplete_dir / "data"
|
||||
meta_dir = incomplete_dir / "meta"
|
||||
data_dir.mkdir(parents=True)
|
||||
meta_dir.mkdir()
|
||||
data_dir.joinpath("partial-file").write_text("interrupted backup data\n", encoding="utf-8")
|
||||
meta_dir.joinpath("rsync.log").write_text("not part of the backup data total\n", encoding="utf-8")
|
||||
expected_usage = tree_usage(data_dir)
|
||||
SnapshotRecord.objects.create(
|
||||
host=host,
|
||||
kind=SnapshotRecord.Kind.INCOMPLETE,
|
||||
dirname=incomplete_dir.name,
|
||||
path=str(incomplete_dir),
|
||||
status="failed",
|
||||
metadata={},
|
||||
)
|
||||
|
||||
stats = collect_host_stats(host=host)
|
||||
|
||||
self.assertEqual(stats["backup_data"]["incomplete"]["count"], 1)
|
||||
self.assertEqual(
|
||||
stats["backup_data"]["incomplete"]["allocated_size_bytes"],
|
||||
expected_usage["allocated_size_bytes"],
|
||||
)
|
||||
self.assertEqual(
|
||||
stats["backup_data"]["incomplete"]["apparent_size_bytes"],
|
||||
expected_usage["apparent_size_bytes"],
|
||||
)
|
||||
self.assertEqual(
|
||||
stats["backup_data"]["total"]["allocated_size_bytes"],
|
||||
expected_usage["allocated_size_bytes"],
|
||||
)
|
||||
|
||||
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(
|
||||
|
||||
@@ -8,9 +8,11 @@ from unittest.mock import patch
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.core.files.uploadedfile import SimpleUploadedFile
|
||||
from django.template.defaultfilters import filesizeformat
|
||||
from django.test import TestCase, override_settings
|
||||
from django.urls import reverse
|
||||
|
||||
from pobsync.run_stats import tree_usage
|
||||
from pobsync.util import write_yaml_atomic
|
||||
from pobsync_backend.models import (
|
||||
BackupRun,
|
||||
@@ -248,6 +250,30 @@ class ViewTests(TestCase):
|
||||
self.assertContains(response, "300 bytes", html=True)
|
||||
self.assertContains(response, "600 bytes", html=True)
|
||||
|
||||
def test_dashboard_host_cards_measure_incomplete_data_without_snapshot_metadata(self) -> None:
|
||||
self.client.force_login(self.staff_user)
|
||||
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
|
||||
with TemporaryDirectory() as tmp:
|
||||
incomplete_dir = Path(tmp) / host.host / ".incomplete" / "20260519-041500Z__BROKEN1"
|
||||
data_dir = incomplete_dir / "data"
|
||||
data_dir.mkdir(parents=True)
|
||||
data_dir.joinpath("partial-file").write_text("interrupted backup data\n", encoding="utf-8")
|
||||
expected_usage = tree_usage(data_dir)
|
||||
SnapshotRecord.objects.create(
|
||||
host=host,
|
||||
kind=SnapshotRecord.Kind.INCOMPLETE,
|
||||
dirname=incomplete_dir.name,
|
||||
path=str(incomplete_dir),
|
||||
status="failed",
|
||||
metadata={},
|
||||
)
|
||||
|
||||
response = self.client.get(reverse("dashboard_hosts_live"))
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertContains(response, "Incomplete data")
|
||||
self.assertContains(response, filesizeformat(expected_usage["allocated_size_bytes"]))
|
||||
|
||||
def test_hosts_list_renders_host_cards_and_controls(self) -> None:
|
||||
self.client.force_login(self.staff_user)
|
||||
web = HostConfig.objects.create(host="web-01", address="web-01.example.test")
|
||||
|
||||
Reference in New Issue
Block a user