Compare commits
2 Commits
797619acd9
...
d158644567
| Author | SHA1 | Date | |
|---|---|---|---|
| d158644567 | |||
| e16c13a1e7 |
@@ -111,7 +111,7 @@ docker compose up --build web
|
||||
|
||||
This starts Django on:
|
||||
|
||||
- http://127.0.0.1:8000/admin/
|
||||
- http://127.0.0.1:8010/admin/
|
||||
|
||||
Run the scheduler alongside the web admin:
|
||||
|
||||
|
||||
@@ -9,7 +9,7 @@ services:
|
||||
POBSYNC_HOME: "/opt/pobsync"
|
||||
POBSYNC_SQLITE_PATH: "/var/lib/pobsync/pobsync.sqlite3"
|
||||
ports:
|
||||
- "8000:8000"
|
||||
- "8010:8000"
|
||||
volumes:
|
||||
- pobsync_state:/opt/pobsync
|
||||
- pobsync_db:/var/lib/pobsync
|
||||
@@ -45,7 +45,7 @@ services:
|
||||
db:
|
||||
condition: service_healthy
|
||||
ports:
|
||||
- "8000:8000"
|
||||
- "8010:8000"
|
||||
volumes:
|
||||
- pobsync_state:/opt/pobsync
|
||||
|
||||
|
||||
@@ -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('<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)
|
||||
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('<a href="{}">{}</a>', 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('<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)
|
||||
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)}"
|
||||
|
||||
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