Peter van Arkel
123583a502
Add a staff-only retention plan page for each host using the SQL-backed retention service. Link it from the host detail page and show policy settings, keep reasons, and snapshots that would be deleted for scheduled, manual, or all snapshot kinds. Keep the flow non-destructive for now, validate query parameters, and cover the view with tests for rendering, base protection, and invalid kind handling.
165 lines
7.0 KiB
Python
165 lines
7.0 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import TestCase
|
|
from django.urls import reverse
|
|
|
|
from pobsync.util import write_yaml_atomic
|
|
from pobsync_backend.models import BackupRun, GlobalConfig, HostConfig, ScheduleConfig, SnapshotRecord
|
|
|
|
|
|
class ViewTests(TestCase):
|
|
def setUp(self) -> None:
|
|
user_model = get_user_model()
|
|
self.staff_user = user_model.objects.create_user(
|
|
username="admin",
|
|
password="secret",
|
|
is_staff=True,
|
|
is_superuser=True,
|
|
)
|
|
|
|
def test_dashboard_requires_staff_login(self) -> None:
|
|
response = self.client.get(reverse("dashboard"))
|
|
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertIn("/admin/login/", response["Location"])
|
|
|
|
def test_dashboard_renders_hosts_and_latest_runs(self) -> None:
|
|
self.client.force_login(self.staff_user)
|
|
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
|
|
snapshot = self._snapshot(host, "20260519-021500Z__ABCDEFGH")
|
|
BackupRun.objects.create(
|
|
host=host,
|
|
status=BackupRun.Status.SUCCESS,
|
|
snapshot=snapshot,
|
|
started_at=datetime(2026, 5, 19, 2, 15, tzinfo=timezone.utc),
|
|
)
|
|
|
|
response = self.client.get(reverse("dashboard"))
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertContains(response, "Dashboard")
|
|
self.assertContains(response, "web-01")
|
|
self.assertContains(response, "20260519-021500Z__ABCDEFGH")
|
|
self.assertContains(response, "success")
|
|
|
|
def test_host_detail_renders_config_schedule_runs_and_snapshots(self) -> None:
|
|
self.client.force_login(self.staff_user)
|
|
host = HostConfig.objects.create(
|
|
host="web-01",
|
|
address="web-01.example.test",
|
|
source_root="/srv",
|
|
retention_daily=7,
|
|
)
|
|
ScheduleConfig.objects.create(host=host, cron_expr="15 2 * * *", prune=True, last_status="success")
|
|
snapshot = self._snapshot(host, "20260519-021500Z__ABCDEFGH")
|
|
BackupRun.objects.create(host=host, status=BackupRun.Status.SUCCESS, snapshot=snapshot)
|
|
|
|
response = self.client.get(reverse("host_detail", args=[host.host]))
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertContains(response, "web-01")
|
|
self.assertContains(response, "web-01.example.test")
|
|
self.assertContains(response, "15 2 * * *")
|
|
self.assertContains(response, "20260519-021500Z__ABCDEFGH")
|
|
self.assertContains(response, "Discover snapshots")
|
|
|
|
def test_host_detail_returns_404_for_unknown_host(self) -> None:
|
|
self.client.force_login(self.staff_user)
|
|
|
|
response = self.client.get(reverse("host_detail", args=["missing-host"]))
|
|
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
def test_discover_host_snapshots_action_discovers_and_redirects(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")
|
|
snapshot_dir = backup_root / host.host / "scheduled" / "20260519-021500Z__ABCDEFGH"
|
|
meta_dir = snapshot_dir / "meta"
|
|
meta_dir.mkdir(parents=True)
|
|
write_yaml_atomic(meta_dir / "meta.yaml", {"status": "success", "started_at": "2026-05-19T02:15:00Z"})
|
|
|
|
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 1 items")
|
|
self.assertTrue(SnapshotRecord.objects.filter(host=host, dirname=snapshot_dir.name).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")
|
|
|
|
response = self.client.get(reverse("discover_host_snapshots", args=[host.host]))
|
|
|
|
self.assertEqual(response.status_code, 405)
|
|
|
|
def test_retention_plan_renders_keep_and_delete_sets(self) -> None:
|
|
self.client.force_login(self.staff_user)
|
|
host = HostConfig.objects.create(
|
|
host="web-01",
|
|
address="web-01.example.test",
|
|
retention_daily=0,
|
|
retention_weekly=0,
|
|
retention_monthly=0,
|
|
retention_yearly=0,
|
|
)
|
|
old_snapshot = self._snapshot(host, "20260518-021500Z__OLDSNAP")
|
|
new_snapshot = self._snapshot(host, "20260519-021500Z__NEWSNAP")
|
|
|
|
response = self.client.get(reverse("host_retention_plan", args=[host.host]))
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertContains(response, "Retention Plan: web-01")
|
|
self.assertContains(response, old_snapshot.dirname)
|
|
self.assertContains(response, new_snapshot.dirname)
|
|
self.assertContains(response, "newest")
|
|
self.assertContains(response, "Would Delete")
|
|
|
|
def test_retention_plan_can_enable_base_protection(self) -> None:
|
|
self.client.force_login(self.staff_user)
|
|
host = HostConfig.objects.create(
|
|
host="web-01",
|
|
address="web-01.example.test",
|
|
retention_daily=0,
|
|
retention_weekly=0,
|
|
retention_monthly=0,
|
|
retention_yearly=0,
|
|
)
|
|
base = self._snapshot(host, "20260518-021500Z__BASESNAP")
|
|
child = self._snapshot(host, "20260519-021500Z__CHILDSNP")
|
|
child.base = base
|
|
child.save(update_fields=["base"])
|
|
|
|
response = self.client.get(reverse("host_retention_plan", args=[host.host]), {"protect_bases": "1"})
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertContains(response, "Protect bases:</strong> yes")
|
|
self.assertContains(response, f"base-of:{child.dirname}")
|
|
|
|
def test_retention_plan_rejects_invalid_kind(self) -> None:
|
|
self.client.force_login(self.staff_user)
|
|
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
|
|
|
|
response = self.client.get(reverse("host_retention_plan", args=[host.host]), {"kind": "weird"}, follow=True)
|
|
|
|
self.assertRedirects(response, reverse("host_detail", args=[host.host]))
|
|
self.assertContains(response, "Retention kind must be scheduled, manual, or all.")
|
|
|
|
def _snapshot(self, host: HostConfig, dirname: str) -> SnapshotRecord:
|
|
started_at = datetime.strptime(dirname.split("__", 1)[0], "%Y%m%d-%H%M%SZ").replace(tzinfo=timezone.utc)
|
|
return SnapshotRecord.objects.create(
|
|
host=host,
|
|
kind=SnapshotRecord.Kind.SCHEDULED,
|
|
dirname=dirname,
|
|
path=f"/backups/{host.host}/scheduled/{dirname}",
|
|
status="success",
|
|
started_at=started_at,
|
|
)
|