From d1586445670a9df9292db3438ff828f0dc213cae Mon Sep 17 00:00:00 2001 From: Peter van Arkel Date: Tue, 19 May 2026 11:39:10 +0200 Subject: [PATCH] Improve Django admin navigation for backup data Add linked admin summaries for hosts, snapshots, and backup runs so the SQL-first backup state is easier to inspect from the Django admin. Hosts now link to their filtered snapshot and run lists, backup runs link back to their snapshot, and snapshots show base/run relationships without requiring filesystem inspection. Cover the new admin display helpers with focused tests. --- src/pobsync_backend/admin.py | 94 +++++++++++++++++++++++- src/pobsync_backend/tests/test_admin.py | 96 +++++++++++++++++++++++++ 2 files changed, 187 insertions(+), 3 deletions(-) create mode 100644 src/pobsync_backend/tests/test_admin.py diff --git a/src/pobsync_backend/admin.py b/src/pobsync_backend/admin.py index 3ad39f7..c3b0ab9 100644 --- a/src/pobsync_backend/admin.py +++ b/src/pobsync_backend/admin.py @@ -1,6 +1,10 @@ from __future__ import annotations from django.contrib import admin +from django.db.models import Count +from django.urls import reverse +from django.utils.html import format_html +from django.utils.http import urlencode from .models import BackupRun, GlobalConfig, HostConfig, ScheduleConfig, SnapshotRecord @@ -33,7 +37,16 @@ class GlobalConfigAdmin(admin.ModelAdmin): @admin.register(HostConfig) class HostConfigAdmin(admin.ModelAdmin): - list_display = ("host", "address", "enabled", "updated_at") + list_display = ( + "host", + "address", + "enabled", + "schedule_state", + "snapshot_count_link", + "backup_run_count_link", + "latest_run_state", + "updated_at", + ) list_filter = ("enabled",) search_fields = ("host", "address") readonly_fields = ("created_at", "updated_at") @@ -47,18 +60,66 @@ class HostConfigAdmin(admin.ModelAdmin): ("Timestamps", {"fields": ("created_at", "updated_at"), "classes": ("collapse",)}), ) + def get_queryset(self, request): + return super().get_queryset(request).annotate( + snapshot_count=Count("snapshots", distinct=True), + backup_run_count=Count("runs", distinct=True), + ) + + @admin.display(description="Schedule") + def schedule_state(self, obj: HostConfig) -> str: + try: + schedule = obj.schedule + except ScheduleConfig.DoesNotExist: + return "none" + if not schedule.enabled: + return "disabled" + return schedule.cron_expr + + @admin.display(description="Snapshots", ordering="snapshot_count") + def snapshot_count_link(self, obj: HostConfig) -> str: + count = getattr(obj, "snapshot_count", None) + if count is None: + count = obj.snapshots.count() + url = _admin_changelist_url("pobsync_backend", "snapshotrecord", {"host__id__exact": obj.pk}) + return format_html('{}', url, count) + + @admin.display(description="Runs", ordering="backup_run_count") + def backup_run_count_link(self, obj: HostConfig) -> str: + count = getattr(obj, "backup_run_count", None) + if count is None: + count = obj.runs.count() + url = _admin_changelist_url("pobsync_backend", "backuprun", {"host__id__exact": obj.pk}) + return format_html('{}', url, count) + + @admin.display(description="Latest run") + def latest_run_state(self, obj: HostConfig) -> str: + latest = obj.runs.order_by("-created_at").first() + if latest is None: + return "none" + return f"{latest.status} {latest.started_at:%Y-%m-%d %H:%M}" if latest.started_at else latest.status + @admin.register(BackupRun) class BackupRunAdmin(admin.ModelAdmin): - list_display = ("host", "run_type", "status", "started_at", "ended_at", "snapshot") + list_display = ("host", "run_type", "status", "started_at", "ended_at", "snapshot_link") list_filter = ("run_type", "status", "started_at") search_fields = ("host__host", "snapshot_path", "snapshot__dirname", "snapshot__path") autocomplete_fields = ("snapshot",) + list_select_related = ("host", "snapshot") + date_hierarchy = "started_at" + + @admin.display(description="Snapshot", ordering="snapshot__dirname") + def snapshot_link(self, obj: BackupRun) -> str: + if obj.snapshot is None: + return "" + url = reverse("admin:pobsync_backend_snapshotrecord_change", args=[obj.snapshot.pk]) + return format_html('{}', url, obj.snapshot.dirname) @admin.register(SnapshotRecord) class SnapshotRecordAdmin(admin.ModelAdmin): - list_display = ("host", "kind", "dirname", "status", "base", "started_at", "discovered_at") + list_display = ("host", "kind", "dirname", "status", "base_link", "backup_run_count_link", "started_at", "discovered_at") list_filter = ("kind", "status", "base_kind", "started_at", "discovered_at") search_fields = ( "host__host", @@ -69,11 +130,38 @@ class SnapshotRecordAdmin(admin.ModelAdmin): "base_snapshot_id", ) autocomplete_fields = ("base",) + list_select_related = ("host", "base") readonly_fields = ("discovered_at",) + def get_queryset(self, request): + return super().get_queryset(request).annotate(backup_run_count=Count("backup_runs")) + + @admin.display(description="Base", ordering="base__dirname") + def base_link(self, obj: SnapshotRecord) -> str: + if obj.base is not None: + url = reverse("admin:pobsync_backend_snapshotrecord_change", args=[obj.base.pk]) + return format_html('{}', url, obj.base.dirname) + if obj.base_dirname: + return obj.base_dirname + return "" + + @admin.display(description="Runs", ordering="backup_run_count") + def backup_run_count_link(self, obj: SnapshotRecord) -> str: + count = getattr(obj, "backup_run_count", None) + if count is None: + count = obj.backup_runs.count() + url = _admin_changelist_url("pobsync_backend", "backuprun", {"snapshot__id__exact": obj.pk}) + return format_html('{}', url, count) + @admin.register(ScheduleConfig) class ScheduleConfigAdmin(admin.ModelAdmin): list_display = ("host", "cron_expr", "enabled", "prune", "last_status", "last_started_at", "updated_at") list_filter = ("enabled", "prune", "last_status") search_fields = ("host__host", "cron_expr") + list_select_related = ("host",) + + +def _admin_changelist_url(app_label: str, model_name: str, params: dict[str, object]) -> str: + base_url = reverse(f"admin:{app_label}_{model_name}_changelist") + return f"{base_url}?{urlencode(params)}" diff --git a/src/pobsync_backend/tests/test_admin.py b/src/pobsync_backend/tests/test_admin.py new file mode 100644 index 0000000..1263dbf --- /dev/null +++ b/src/pobsync_backend/tests/test_admin.py @@ -0,0 +1,96 @@ +from __future__ import annotations + +from datetime import datetime, timezone + +from django.contrib.admin.sites import AdminSite +from django.test import TestCase + +from pobsync_backend.admin import BackupRunAdmin, HostConfigAdmin, SnapshotRecordAdmin +from pobsync_backend.models import BackupRun, HostConfig, ScheduleConfig, SnapshotRecord + + +class AdminDisplayTests(TestCase): + def test_host_admin_links_to_related_snapshots_and_runs(self) -> None: + site = AdminSite() + admin = HostConfigAdmin(HostConfig, site) + host = HostConfig.objects.create(host="web-01", address="web-01.example.test") + snapshot = SnapshotRecord.objects.create( + host=host, + kind="scheduled", + dirname="20260519-021500Z__ABCDEFGH", + path="/backups/web-01/scheduled/20260519-021500Z__ABCDEFGH", + ) + BackupRun.objects.create(host=host, status=BackupRun.Status.SUCCESS, snapshot=snapshot) + + snapshot_link = str(admin.snapshot_count_link(host)) + run_link = str(admin.backup_run_count_link(host)) + + self.assertIn("/admin/pobsync_backend/snapshotrecord/", snapshot_link) + self.assertIn(f"host__id__exact={host.pk}", snapshot_link) + self.assertIn(">1<", snapshot_link) + self.assertIn("/admin/pobsync_backend/backuprun/", run_link) + self.assertIn(f"host__id__exact={host.pk}", run_link) + self.assertIn(">1<", run_link) + + def test_host_admin_summarizes_schedule_and_latest_run(self) -> None: + site = AdminSite() + admin = HostConfigAdmin(HostConfig, site) + host = HostConfig.objects.create(host="web-01", address="web-01.example.test") + + self.assertEqual(admin.schedule_state(host), "none") + + ScheduleConfig.objects.create(host=host, cron_expr="15 2 * * *", enabled=True) + host.refresh_from_db() + BackupRun.objects.create( + host=host, + status=BackupRun.Status.SUCCESS, + started_at=datetime(2026, 5, 19, 2, 15, tzinfo=timezone.utc), + ) + + self.assertEqual(admin.schedule_state(host), "15 2 * * *") + self.assertEqual(admin.latest_run_state(host), "success 2026-05-19 02:15") + + def test_snapshot_admin_links_to_base_and_backup_runs(self) -> None: + site = AdminSite() + admin = SnapshotRecordAdmin(SnapshotRecord, site) + host = HostConfig.objects.create(host="web-01", address="web-01.example.test") + base = SnapshotRecord.objects.create( + host=host, + kind="scheduled", + dirname="20260518-021500Z__BASESNAP", + path="/backups/web-01/scheduled/20260518-021500Z__BASESNAP", + ) + child = SnapshotRecord.objects.create( + host=host, + kind="scheduled", + dirname="20260519-021500Z__CHILDSNP", + path="/backups/web-01/scheduled/20260519-021500Z__CHILDSNP", + base=base, + ) + BackupRun.objects.create(host=host, status=BackupRun.Status.SUCCESS, snapshot=child) + + base_link = str(admin.base_link(child)) + run_link = str(admin.backup_run_count_link(child)) + + self.assertIn(f"/admin/pobsync_backend/snapshotrecord/{base.pk}/change/", base_link) + self.assertIn(base.dirname, base_link) + self.assertIn("/admin/pobsync_backend/backuprun/", run_link) + self.assertIn(f"snapshot__id__exact={child.pk}", run_link) + self.assertIn(">1<", run_link) + + def test_backup_run_admin_links_to_snapshot(self) -> None: + site = AdminSite() + admin = BackupRunAdmin(BackupRun, site) + host = HostConfig.objects.create(host="web-01", address="web-01.example.test") + snapshot = SnapshotRecord.objects.create( + host=host, + kind="scheduled", + dirname="20260519-021500Z__ABCDEFGH", + path="/backups/web-01/scheduled/20260519-021500Z__ABCDEFGH", + ) + run = BackupRun.objects.create(host=host, status=BackupRun.Status.SUCCESS, snapshot=snapshot) + + snapshot_link = str(admin.snapshot_link(run)) + + self.assertIn(f"/admin/pobsync_backend/snapshotrecord/{snapshot.pk}/change/", snapshot_link) + self.assertIn(snapshot.dirname, snapshot_link)