(release) Add purged snapshot audit overview
Record snapshot purge history whenever retention or incomplete cleanup removes snapshot directories and SQL records. Store the purge reason, original kind, path, action source, and triggering operator so manual, scheduled, CLI, and incomplete cleanup actions remain auditable after the original snapshot record is deleted. Add a staff-only Purged Snapshots page with host/action filters and register the audit model in Django admin. Refs #16 Refs #8
This commit is contained in:
@@ -12,7 +12,15 @@ from django.test import TestCase, override_settings
|
||||
from django.urls import reverse
|
||||
|
||||
from pobsync.util import write_yaml_atomic
|
||||
from pobsync_backend.models import BackupRun, GlobalConfig, HostConfig, ScheduleConfig, SnapshotRecord, SshCredential
|
||||
from pobsync_backend.models import (
|
||||
BackupRun,
|
||||
GlobalConfig,
|
||||
HostConfig,
|
||||
PurgedSnapshot,
|
||||
ScheduleConfig,
|
||||
SnapshotRecord,
|
||||
SshCredential,
|
||||
)
|
||||
|
||||
|
||||
class ViewTests(TestCase):
|
||||
@@ -328,6 +336,61 @@ class ViewTests(TestCase):
|
||||
self.assertIn("--since", command)
|
||||
self.assertIn("6 hours ago", command)
|
||||
|
||||
def test_purged_snapshots_view_renders_history(self) -> None:
|
||||
self.client.force_login(self.staff_user)
|
||||
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
|
||||
PurgedSnapshot.objects.create(
|
||||
host=host,
|
||||
host_name=host.host,
|
||||
kind=SnapshotRecord.Kind.SCHEDULED,
|
||||
dirname="20260518-021500Z__OLDSNAP",
|
||||
path=f"/backups/{host.host}/scheduled/20260518-021500Z__OLDSNAP",
|
||||
reason="outside retention policy",
|
||||
action=PurgedSnapshot.Action.SCHEDULED,
|
||||
triggered_by="pobsync-scheduler",
|
||||
)
|
||||
|
||||
response = self.client.get(reverse("purged_snapshots"))
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertContains(response, "Purged Snapshots")
|
||||
self.assertContains(response, "20260518-021500Z__OLDSNAP")
|
||||
self.assertContains(response, "outside retention policy")
|
||||
self.assertContains(response, "Scheduled")
|
||||
self.assertContains(response, "pobsync-scheduler")
|
||||
|
||||
def test_purged_snapshots_view_filters_by_host_and_action(self) -> None:
|
||||
self.client.force_login(self.staff_user)
|
||||
web = HostConfig.objects.create(host="web-01", address="web-01.example.test")
|
||||
db = HostConfig.objects.create(host="db-01", address="db-01.example.test")
|
||||
PurgedSnapshot.objects.create(
|
||||
host=web,
|
||||
host_name=web.host,
|
||||
kind=SnapshotRecord.Kind.SCHEDULED,
|
||||
dirname="20260518-021500Z__WEBOLD",
|
||||
path=f"/backups/{web.host}/scheduled/20260518-021500Z__WEBOLD",
|
||||
reason="outside retention policy",
|
||||
action=PurgedSnapshot.Action.MANUAL,
|
||||
)
|
||||
PurgedSnapshot.objects.create(
|
||||
host=db,
|
||||
host_name=db.host,
|
||||
kind=SnapshotRecord.Kind.INCOMPLETE,
|
||||
dirname="20260518-021500Z__DBBROKEN",
|
||||
path=f"/backups/{db.host}/.incomplete/20260518-021500Z__DBBROKEN",
|
||||
reason="manual incomplete cleanup",
|
||||
action=PurgedSnapshot.Action.INCOMPLETE_CLEANUP,
|
||||
)
|
||||
|
||||
response = self.client.get(
|
||||
reverse("purged_snapshots"),
|
||||
{"host": db.host, "action": PurgedSnapshot.Action.INCOMPLETE_CLEANUP},
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertContains(response, "20260518-021500Z__DBBROKEN")
|
||||
self.assertNotContains(response, "20260518-021500Z__WEBOLD")
|
||||
|
||||
def test_ssh_credentials_view_creates_key(self) -> None:
|
||||
self.client.force_login(self.staff_user)
|
||||
|
||||
@@ -1881,6 +1944,10 @@ class ViewTests(TestCase):
|
||||
self.assertContains(response, "Retention deleted 1 snapshot(s) for web-01.")
|
||||
self.assertFalse(SnapshotRecord.objects.filter(pk=old_snapshot.pk).exists())
|
||||
self.assertTrue(SnapshotRecord.objects.filter(pk=new_snapshot.pk).exists())
|
||||
purged = PurgedSnapshot.objects.get(dirname=old_snapshot.dirname)
|
||||
self.assertEqual(purged.action, PurgedSnapshot.Action.MANUAL)
|
||||
self.assertEqual(purged.triggered_by, self.staff_user.username)
|
||||
self.assertEqual(purged.reason, "outside retention policy")
|
||||
|
||||
def test_retention_apply_rejects_bad_confirmation(self) -> None:
|
||||
self.client.force_login(self.staff_user)
|
||||
|
||||
Reference in New Issue
Block a user