hippogrief/pobsync
1
0
Fork 0
Code Issues 21 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

(feature) improve snapshot discovery visibility in Django

Browse Source 
Add a discovery preflight that reports the configured backup root, host
root, and snapshot directory counts before importing anything.

Show discovery status on host detail pages so missing mounts or mismatched
host directories are visible from the UI.

Warn clearly when discovery scans zero snapshots, including whether the
host backup directory is missing or simply empty.
This commit is contained in:
Peter van Arkel
2026-05-19 13:21:31 +02:00
parent 573177e118
commit 1d90454109
7 changed files with 133 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -8,3 +8,4 @@ backups/
*.egg-info/
build/
dist/
.env

48
src/pobsync_backend/snapshot_discovery.py
View File

@@ -65,6 +65,54 @@ def discover_snapshots(
}
def inspect_snapshot_discovery(
*,
host: HostConfig,
global_config: GlobalConfig | None = None,
kinds: list[str] | None = None,
) -> dict[str, Any]:
try:
global_config = global_config or GlobalConfig.objects.get(name="default")
except GlobalConfig.DoesNotExist:
return {
"ok": False,
"reason": "missing_global_config",
"message": "Create the default global config before discovering snapshots.",
"backup_root": "",
"host_root": "",
"host_root_exists": False,
"kind_counts": {},
"total_candidates": 0,
}
kinds = kinds or ["scheduled", "manual", "incomplete"]
host_root = resolve_host_root(global_config.backup_root, host.host)
kind_counts = {kind: len(list(iter_snapshot_dirs(host_root, kind))) for kind in kinds}
total_candidates = sum(kind_counts.values())
host_root_exists = host_root.exists()
if not host_root_exists:
reason = "missing_host_root"
message = f"Host backup directory does not exist yet: {host_root}"
elif total_candidates == 0:
reason = "no_snapshots"
message = f"No snapshot directories found below {host_root}."
else:
reason = "ready"
message = f"Found {total_candidates} snapshot directories below {host_root}."
return {
"ok": True,
"reason": reason,
"message": message,
"backup_root": str(global_config.backup_root),
"host_root": str(host_root),
"host_root_exists": host_root_exists,
"kind_counts": kind_counts,
"total_candidates": total_candidates,
}
def upsert_snapshot_record(*, host: HostConfig, kind: str, snapshot_dir: Path) -> tuple[SnapshotRecord, bool]:
meta = read_snapshot_meta(snapshot_dir)
base_defaults = _base_defaults_from_meta(meta)

1
src/pobsync_backend/templates/pobsync_backend/base.html
View File

@@ -109,6 +109,7 @@
}
.message.success { border-color: #a7d8b9; background: #edf8f1; color: var(--success); }
.message.error { border-color: #e8b4b4; background: #fff0f0; color: var(--failed); }
.message.warning { border-color: #e7cf8a; background: #fff8df; color: var(--running); }
.form-grid { display: grid; gap: 14px; max-width: 680px; }
.field { display: grid; gap: 5px; }
.field label { font-weight: 650; }

16
src/pobsync_backend/templates/pobsync_backend/host_detail.html
View File

@@ -51,6 +51,22 @@
</section>
</div>
<section class="panel">
<h2>Snapshot Discovery</h2>
<div class="stack">
<div><strong>Backup root:</strong> {{ discovery.backup_root|default:"" }}</div>
<div><strong>Host root:</strong> {{ discovery.host_root|default:"" }}</div>
<div><strong>Status:</strong> {{ discovery.message }}</div>
{% if discovery.kind_counts %}
<div><strong>On disk:</strong>
scheduled {{ discovery.kind_counts.scheduled|default:0 }},
manual {{ discovery.kind_counts.manual|default:0 }},
incomplete {{ discovery.kind_counts.incomplete|default:0 }}
</div>
{% endif %}
</div>
</section>
<section class="panel">
<h2>Queue Manual Backup</h2>
<form method="post" action="{% url 'queue_manual_backup' host.host %}" class="form-grid">

25
src/pobsync_backend/tests/test_snapshot_discovery.py
View File

@@ -12,6 +12,7 @@ from pobsync.util import write_yaml_atomic
from pobsync_backend.models import GlobalConfig, HostConfig, SnapshotRecord
from pobsync_backend.snapshot_discovery import (
discover_snapshots,
inspect_snapshot_discovery,
parse_snapshot_datetime,
resolve_base_links,
upsert_snapshot_record,
@@ -63,6 +64,30 @@ class SnapshotDiscoveryTests(TestCase):
self.assertEqual(record.kind, "scheduled")
self.assertEqual(record.started_at, datetime(2026, 5, 19, 2, 15, tzinfo=timezone.utc))
def test_inspect_snapshot_discovery_reports_missing_global_config(self) -> None:
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
result = inspect_snapshot_discovery(host=host)
self.assertFalse(result["ok"])
self.assertEqual(result["reason"], "missing_global_config")
self.assertEqual(result["total_candidates"], 0)
def test_inspect_snapshot_discovery_counts_snapshot_directories(self) -> None:
with TemporaryDirectory() as tmp:
backup_root = Path(tmp) / "backups"
GlobalConfig.objects.create(name="default", backup_root=str(backup_root))
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
(backup_root / host.host / "scheduled" / "20260519-021500Z__ABCDEFGH").mkdir(parents=True)
(backup_root / host.host / "manual" / "20260519-031500Z__MANUAL01").mkdir(parents=True)
result = inspect_snapshot_discovery(host=host)
self.assertTrue(result["ok"])
self.assertEqual(result["reason"], "ready")
self.assertEqual(result["total_candidates"], 2)
self.assertEqual(result["kind_counts"], {"scheduled": 1, "manual": 1, "incomplete": 0})
def test_discovery_links_snapshot_to_base_record(self) -> None:
with TemporaryDirectory() as tmp:
backup_root = Path(tmp) / "backups"

32
src/pobsync_backend/tests/test_views.py
View File

@@ -218,10 +218,28 @@ class ViewTests(TestCase):
self.assertContains(response, "Edit schedule")
self.assertContains(response, "Edit config")
self.assertContains(response, "Queue Manual Backup")
self.assertContains(response, "Snapshot Discovery")
self.assertContains(response, reverse("queue_manual_backup", args=[host.host]))
self.assertContains(response, reverse("run_detail", args=[BackupRun.objects.get().id]))
self.assertContains(response, reverse("snapshot_detail", args=[snapshot.id]))
def test_host_detail_renders_discovery_status_for_existing_snapshot_dirs(self) -> None:
self.client.force_login(self.staff_user)
with TemporaryDirectory() as tmp:
backup_root = Path(tmp) / "backups"
GlobalConfig.objects.create(name="default", backup_root=str(backup_root))
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
(backup_root / host.host / "scheduled" / "20260519-021500Z__ABCDEFGH").mkdir(parents=True)
(backup_root / host.host / ".incomplete" / "20260519-031500Z__BROKEN01").mkdir(parents=True)
response = self.client.get(reverse("host_detail", args=[host.host]))
self.assertEqual(response.status_code, 200)
self.assertContains(response, f"Host root:</strong> {backup_root / host.host}")
self.assertContains(response, "Found 2 snapshot directories")
self.assertContains(response, "scheduled 1")
self.assertContains(response, "incomplete 1")
def test_host_detail_returns_404_for_unknown_host(self) -> None:
self.client.force_login(self.staff_user)
@@ -352,6 +370,20 @@ class ViewTests(TestCase):
self.assertContains(response, "Snapshot discovery scanned 1 items")
self.assertTrue(SnapshotRecord.objects.filter(host=host, dirname=snapshot_dir.name).exists())
def test_discover_host_snapshots_warns_when_host_root_is_missing(self) -> None:
self.client.force_login(self.staff_user)
with TemporaryDirectory() as tmp:
backup_root = Path(tmp) / "backups"
GlobalConfig.objects.create(name="default", backup_root=str(backup_root))
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
response = self.client.post(reverse("discover_host_snapshots", args=[host.host]), follow=True)
self.assertRedirects(response, reverse("host_detail", args=[host.host]))
self.assertContains(response, "Snapshot discovery scanned 0 items")
self.assertContains(response, "Host backup directory does not exist yet")
self.assertFalse(SnapshotRecord.objects.exists())
def test_discover_host_snapshots_requires_post(self) -> None:
self.client.force_login(self.staff_user)
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")

17
src/pobsync_backend/views.py
View File

@@ -15,7 +15,7 @@ from .backup_runner import queue_backup_run
from .forms import CreateHostConfigForm, GlobalConfigForm, HostConfigForm, ManualBackupForm, ScheduleConfigForm
from .models import BackupRun, GlobalConfig, HostConfig, ScheduleConfig, SnapshotRecord
from .retention import run_sql_retention_plan
from .snapshot_discovery import discover_snapshots
from .snapshot_discovery import discover_snapshots, inspect_snapshot_discovery
@staff_member_required
@@ -93,6 +93,7 @@ def host_detail(request, host: str):
context = {
"host": host_config,
"schedule": _schedule_for_host(host_config),
"discovery": inspect_snapshot_discovery(host=host_config),
"manual_backup_form": ManualBackupForm(initial=_default_manual_backup_initial(host_config)),
"latest_runs": host_config.runs.select_related("snapshot").order_by("-created_at")[:10],
"snapshots": host_config.snapshots.select_related("base").order_by("-started_at", "dirname")[:20],
@@ -167,13 +168,15 @@ def discover_host_snapshots(request, host: str):
except Exception as exc:
messages.error(request, f"Snapshot discovery failed for {host_config.host}: {exc}")
else:
messages.success(
request,
(
f"Snapshot discovery scanned {result['scanned']} items for {host_config.host}: "
f"{result['created']} created, {result['updated']} updated."
),
summary = (
f"Snapshot discovery scanned {result['scanned']} items for {host_config.host}: "
f"{result['created']} created, {result['updated']} updated."
)
if result["scanned"]:
messages.success(request, summary)
else:
discovery = inspect_snapshot_discovery(host=host_config)
messages.warning(request, f"{summary} {discovery['message']}")
return redirect("host_detail", host=host_config.host)