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.
This commit is contained in:
@@ -1,6 +1,10 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from django.contrib import admin
|
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
|
from .models import BackupRun, GlobalConfig, HostConfig, ScheduleConfig, SnapshotRecord
|
||||||
|
|
||||||
@@ -33,7 +37,16 @@ class GlobalConfigAdmin(admin.ModelAdmin):
|
|||||||
|
|
||||||
@admin.register(HostConfig)
|
@admin.register(HostConfig)
|
||||||
class HostConfigAdmin(admin.ModelAdmin):
|
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",)
|
list_filter = ("enabled",)
|
||||||
search_fields = ("host", "address")
|
search_fields = ("host", "address")
|
||||||
readonly_fields = ("created_at", "updated_at")
|
readonly_fields = ("created_at", "updated_at")
|
||||||
@@ -47,18 +60,66 @@ class HostConfigAdmin(admin.ModelAdmin):
|
|||||||
("Timestamps", {"fields": ("created_at", "updated_at"), "classes": ("collapse",)}),
|
("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('<a href="{}">{}</a>', 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('<a href="{}">{}</a>', 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)
|
@admin.register(BackupRun)
|
||||||
class BackupRunAdmin(admin.ModelAdmin):
|
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")
|
list_filter = ("run_type", "status", "started_at")
|
||||||
search_fields = ("host__host", "snapshot_path", "snapshot__dirname", "snapshot__path")
|
search_fields = ("host__host", "snapshot_path", "snapshot__dirname", "snapshot__path")
|
||||||
autocomplete_fields = ("snapshot",)
|
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('<a href="{}">{}</a>', url, obj.snapshot.dirname)
|
||||||
|
|
||||||
|
|
||||||
@admin.register(SnapshotRecord)
|
@admin.register(SnapshotRecord)
|
||||||
class SnapshotRecordAdmin(admin.ModelAdmin):
|
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")
|
list_filter = ("kind", "status", "base_kind", "started_at", "discovered_at")
|
||||||
search_fields = (
|
search_fields = (
|
||||||
"host__host",
|
"host__host",
|
||||||
@@ -69,11 +130,38 @@ class SnapshotRecordAdmin(admin.ModelAdmin):
|
|||||||
"base_snapshot_id",
|
"base_snapshot_id",
|
||||||
)
|
)
|
||||||
autocomplete_fields = ("base",)
|
autocomplete_fields = ("base",)
|
||||||
|
list_select_related = ("host", "base")
|
||||||
readonly_fields = ("discovered_at",)
|
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('<a href="{}">{}</a>', 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('<a href="{}">{}</a>', url, count)
|
||||||
|
|
||||||
|
|
||||||
@admin.register(ScheduleConfig)
|
@admin.register(ScheduleConfig)
|
||||||
class ScheduleConfigAdmin(admin.ModelAdmin):
|
class ScheduleConfigAdmin(admin.ModelAdmin):
|
||||||
list_display = ("host", "cron_expr", "enabled", "prune", "last_status", "last_started_at", "updated_at")
|
list_display = ("host", "cron_expr", "enabled", "prune", "last_status", "last_started_at", "updated_at")
|
||||||
list_filter = ("enabled", "prune", "last_status")
|
list_filter = ("enabled", "prune", "last_status")
|
||||||
search_fields = ("host__host", "cron_expr")
|
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)}"
|
||||||
|
|||||||
96
src/pobsync_backend/tests/test_admin.py
Normal file
96
src/pobsync_backend/tests/test_admin.py
Normal file
@@ -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)
|
||||||
Reference in New Issue
Block a user