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)