hippogrief/pobsync
1
0
Fork 0
Code Issues 21 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

(feature) Require review before incomplete cleanup

Browse Source 
Require incomplete snapshots to be marked reviewed before the cleanup action
can delete them, and show review state in the retention plan UI.

Keep cleanup confirmation counts scoped to reviewed incomplete snapshots and
add coverage for blocked, reviewed, and deletion flows.
This commit is contained in:
Peter van Arkel
2026-05-28 21:07:59 +02:00
parent fc6df89370
commit b5e87abad2
5 changed files with 187 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
src/pobsync_backend/tests/test_sql_retention.py
View File

@@ -254,6 +254,8 @@ class SqlRetentionTests(TestCase):
path=str(incomplete_dir),
status="failed",
started_at=datetime(2026, 5, 19, 3, 15, tzinfo=timezone.utc),
reviewed_at=datetime(2026, 5, 19, 4, 15, tzinfo=timezone.utc),
reviewed_by="admin",
)
result = run_incomplete_cleanup(
@@ -291,6 +293,8 @@ class SqlRetentionTests(TestCase):
path=str(incomplete_dir),
status="failed",
started_at=datetime(2026, 5, 19, 3, 15, tzinfo=timezone.utc),
reviewed_at=datetime(2026, 5, 19, 4, 15, tzinfo=timezone.utc),
reviewed_by="admin",
)
result = run_incomplete_cleanup(
@@ -305,6 +309,26 @@ class SqlRetentionTests(TestCase):
self.assertFalse(SnapshotRecord.objects.filter(pk=record.pk).exists())
self.assertEqual(result["deleted"][0]["dirname"], incomplete_dir.name)
def test_incomplete_cleanup_requires_reviewed_snapshots(self) -> None:
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
SnapshotRecord.objects.create(
host=host,
kind=SnapshotRecord.Kind.INCOMPLETE,
dirname="20260519-031500Z__BROKEN01",
path=f"/backups/{host.host}/.incomplete/20260519-031500Z__BROKEN01",
status="failed",
started_at=datetime(2026, 5, 19, 3, 15, tzinfo=timezone.utc),
)
with self.assertRaisesRegex(ConfigError, "have not been reviewed"):
run_incomplete_cleanup(
prefix=Path("/tmp/pobsync-test"),
host=host.host,
yes=True,
max_delete=1,
acquire_lock=False,
)
def test_incomplete_cleanup_respects_max_delete(self) -> None:
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
SnapshotRecord.objects.create(
@@ -314,6 +338,8 @@ class SqlRetentionTests(TestCase):
path=f"/backups/{host.host}/.incomplete/20260519-031500Z__BROKEN01",
status="failed",
started_at=datetime(2026, 5, 19, 3, 15, tzinfo=timezone.utc),
reviewed_at=datetime(2026, 5, 19, 4, 15, tzinfo=timezone.utc),
reviewed_by="admin",
)
with self.assertRaisesRegex(ConfigError, "blocked by --max-delete=0"):
@@ -334,6 +360,8 @@ class SqlRetentionTests(TestCase):
path=f"/backups/{host.host}/scheduled/20260519-031500Z__BROKEN01",
status="failed",
started_at=datetime(2026, 5, 19, 3, 15, tzinfo=timezone.utc),
reviewed_at=datetime(2026, 5, 19, 4, 15, tzinfo=timezone.utc),
reviewed_by="admin",
)
with self.assertRaisesRegex(ConfigError, "unexpected incomplete snapshot path"):

55
src/pobsync_backend/tests/test_views.py
View File

@@ -2226,9 +2226,33 @@ class ViewTests(TestCase):
self.assertContains(response, "Incomplete Snapshots")
self.assertContains(response, "20260519-031500Z__BROKEN01")
self.assertContains(response, "excluded from retention cleanup")
self.assertContains(response, "needs review")
self.assertContains(response, "Cleanup is blocked until all incomplete snapshots are reviewed.")
self.assertContains(response, "Mark incomplete snapshots reviewed")
self.assertContains(response, "delete only incomplete snapshot directories")
self.assertNotContains(response, "Delete incomplete snapshots")
def test_retention_plan_offers_incomplete_cleanup_after_review(self) -> None:
self.client.force_login(self.staff_user)
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
SnapshotRecord.objects.create(
host=host,
kind=SnapshotRecord.Kind.INCOMPLETE,
dirname="20260519-031500Z__BROKEN01",
path=f"/backups/{host.host}/.incomplete/20260519-031500Z__BROKEN01",
status="failed",
started_at=datetime(2026, 5, 19, 3, 15, tzinfo=timezone.utc),
reviewed_at=datetime(2026, 5, 19, 4, 15, tzinfo=timezone.utc),
reviewed_by="admin",
)
response = self.client.get(reverse("host_retention_plan", args=[host.host]))
self.assertEqual(response.status_code, 200)
self.assertContains(response, "reviewed")
self.assertContains(response, "Delete incomplete snapshots")
self.assertContains(response, "Type 1 to confirm the current number of incomplete snapshots.")
self.assertContains(response, "This deletes only incomplete snapshot directories")
self.assertContains(response, "This deletes only reviewed incomplete snapshot directories")
self.assertContains(response, 'class="danger"', html=False)
def test_incomplete_cleanup_deletes_incomplete_snapshot_after_confirmation(self) -> None:
@@ -2246,6 +2270,8 @@ class ViewTests(TestCase):
path=str(incomplete_dir),
status="failed",
started_at=datetime(2026, 5, 19, 3, 15, tzinfo=timezone.utc),
reviewed_at=datetime(2026, 5, 19, 4, 15, tzinfo=timezone.utc),
reviewed_by="admin",
)
with override_settings(POBSYNC_HOME=str(home)):
@@ -2291,6 +2317,33 @@ class ViewTests(TestCase):
self.assertContains(response, "Incomplete cleanup confirmation is invalid.")
self.assertEqual(SnapshotRecord.objects.filter(kind=SnapshotRecord.Kind.INCOMPLETE).count(), 1)
def test_incomplete_cleanup_rejects_unreviewed_snapshots(self) -> None:
self.client.force_login(self.staff_user)
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
SnapshotRecord.objects.create(
host=host,
kind=SnapshotRecord.Kind.INCOMPLETE,
dirname="20260519-031500Z__BROKEN01",
path=f"/backups/{host.host}/.incomplete/20260519-031500Z__BROKEN01",
status="failed",
started_at=datetime(2026, 5, 19, 3, 15, tzinfo=timezone.utc),
)
with TemporaryDirectory() as tmp, override_settings(POBSYNC_HOME=str(Path(tmp) / "home")):
response = self.client.post(
reverse("cleanup_host_incomplete_snapshots", args=[host.host]),
{
"max_delete": "0",
"confirm_host": host.host,
"confirm_delete_count": "0",
},
follow=True,
)
self.assertRedirects(response, reverse("host_retention_plan", args=[host.host]))
self.assertContains(response, "have not been reviewed")
self.assertEqual(SnapshotRecord.objects.filter(kind=SnapshotRecord.Kind.INCOMPLETE).count(), 1)
def test_host_detail_surfaces_retention_warnings(self) -> None:
self.client.force_login(self.staff_user)
host = HostConfig.objects.create(