diff --git a/src/pobsync_backend/forms.py b/src/pobsync_backend/forms.py index 8be9435..ec2f6cc 100644 --- a/src/pobsync_backend/forms.py +++ b/src/pobsync_backend/forms.py @@ -249,12 +249,18 @@ class RetentionApplyForm(forms.Form): kind = forms.ChoiceField(choices=(("scheduled", "Scheduled"), ("manual", "Manual"), ("all", "All"))) protect_bases = forms.BooleanField(required=False) max_delete = forms.IntegerField(min_value=0, initial=10) + confirm_delete_count = forms.IntegerField(min_value=0) confirm_host = forms.CharField() - def __init__(self, *args, host_name: str, **kwargs) -> None: + def __init__(self, *args, host_name: str, expected_delete_count: int | None = None, **kwargs) -> None: self.host_name = host_name + self.expected_delete_count = expected_delete_count super().__init__(*args, **kwargs) self.fields["confirm_host"].help_text = f"Type {host_name} to confirm deletion." + if expected_delete_count is not None: + self.fields["confirm_delete_count"].help_text = ( + f"Type {expected_delete_count} to confirm the current number of planned deletions." + ) def clean_confirm_host(self) -> str: value = self.cleaned_data["confirm_host"].strip() @@ -262,6 +268,12 @@ class RetentionApplyForm(forms.Form): raise forms.ValidationError(f"Type {self.host_name} to confirm.") return value + def clean_confirm_delete_count(self) -> int: + value = self.cleaned_data["confirm_delete_count"] + if self.expected_delete_count is not None and value != self.expected_delete_count: + raise forms.ValidationError(f"Type {self.expected_delete_count} to confirm the delete count.") + return value + class ScheduleConfigForm(forms.ModelForm): cron_expr = forms.CharField( diff --git a/src/pobsync_backend/templates/pobsync_backend/retention_plan.html b/src/pobsync_backend/templates/pobsync_backend/retention_plan.html index 0d94a11..ccacb38 100644 --- a/src/pobsync_backend/templates/pobsync_backend/retention_plan.html +++ b/src/pobsync_backend/templates/pobsync_backend/retention_plan.html @@ -123,6 +123,13 @@
{{ apply_form.confirm_host.help_text }}
+
+ {{ apply_form.confirm_delete_count.errors }} + + {{ apply_form.confirm_delete_count }} +
{{ apply_form.confirm_delete_count.help_text }}
+
+
diff --git a/src/pobsync_backend/tests/test_views.py b/src/pobsync_backend/tests/test_views.py index 2dbdf40..c6e9cc5 100644 --- a/src/pobsync_backend/tests/test_views.py +++ b/src/pobsync_backend/tests/test_views.py @@ -1412,6 +1412,8 @@ class ViewTests(TestCase): self.assertContains(response, "newest") self.assertContains(response, "Would Delete") self.assertContains(response, "outside retention policy") + self.assertContains(response, "Confirm delete count") + self.assertContains(response, "Type 1 to confirm the current number of planned deletions.") def test_retention_plan_warns_when_scheduled_prune_limit_is_exceeded(self) -> None: self.client.force_login(self.staff_user) @@ -1546,6 +1548,7 @@ class ViewTests(TestCase): "kind": "scheduled", "max_delete": "1", "confirm_host": host.host, + "confirm_delete_count": "1", }, follow=True, ) @@ -1569,6 +1572,7 @@ class ViewTests(TestCase): "kind": "scheduled", "max_delete": "1", "confirm_host": "wrong", + "confirm_delete_count": "1", }, follow=True, ) @@ -1577,6 +1581,34 @@ class ViewTests(TestCase): self.assertContains(response, "Retention apply confirmation is invalid.") self.assertEqual(SnapshotRecord.objects.count(), 1) + def test_retention_apply_rejects_mismatched_delete_count_confirmation(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, + ) + self._snapshot(host, "20260518-021500Z__OLDSNAP") + self._snapshot(host, "20260519-021500Z__NEWSNAP") + + response = self.client.post( + reverse("apply_host_retention", args=[host.host]), + { + "kind": "scheduled", + "max_delete": "1", + "confirm_host": host.host, + "confirm_delete_count": "0", + }, + follow=True, + ) + + self.assertRedirects(response, reverse("host_retention_plan", args=[host.host])) + self.assertContains(response, "Retention apply confirmation is invalid.") + self.assertEqual(SnapshotRecord.objects.count(), 2) + def test_retention_apply_requires_post(self) -> None: self.client.force_login(self.staff_user) host = HostConfig.objects.create(host="web-01", address="web-01.example.test") diff --git a/src/pobsync_backend/views.py b/src/pobsync_backend/views.py index 44d7715..cf469a4 100644 --- a/src/pobsync_backend/views.py +++ b/src/pobsync_backend/views.py @@ -542,10 +542,12 @@ def host_retention_plan(request, host: str): "scheduled_prune_exceeded": scheduled_prune_limit is not None and delete_count > scheduled_prune_limit, "apply_form": RetentionApplyForm( host_name=host_config.host, + expected_delete_count=delete_count, initial={ "kind": kind, "protect_bases": protect_bases, "max_delete": delete_count, + "confirm_delete_count": delete_count, }, ), } @@ -556,7 +558,26 @@ def host_retention_plan(request, host: str): @require_POST def apply_host_retention(request, host: str): host_config = get_object_or_404(HostConfig, host=host) - form = RetentionApplyForm(request.POST, host_name=host_config.host) + raw_kind = request.POST.get("kind", "scheduled") + raw_protect_bases = request.POST.get("protect_bases") in {"1", "true", "on", "yes"} + expected_delete_count = None + if raw_kind in {"scheduled", "manual", "all"}: + try: + plan = run_sql_retention_plan( + host=host_config.host, + kind=raw_kind, + protect_bases=raw_protect_bases, + ) + except PobsyncError as exc: + messages.error(request, str(exc)) + return redirect("host_retention_plan", host=host_config.host) + expected_delete_count = len(plan.get("delete") or []) + + form = RetentionApplyForm( + request.POST, + host_name=host_config.host, + expected_delete_count=expected_delete_count, + ) if not form.is_valid(): messages.error(request, "Retention apply confirmation is invalid.") return redirect("host_retention_plan", host=host_config.host)