From 4fb33eca6cd7a2e573e1588f71713b1c5785355f Mon Sep 17 00:00:00 2001 From: Peter van Arkel Date: Tue, 19 May 2026 13:54:15 +0200 Subject: [PATCH] (feature) Add Django retention apply flow Expose retention apply from the host retention plan page so planned snapshot deletions can be executed from the Django UI. The form requires explicit host confirmation, carries through the selected retention kind and base-protection setting, and uses max_delete as a deletion guard. The view delegates to the SQL retention apply service and reports predictable pobsync errors back through Django messages instead of surfacing a server error. Add view coverage for confirmed deletion, invalid confirmation, and POST-only enforcement. --- src/pobsync_backend/forms.py | 18 +++++ .../pobsync_backend/retention_plan.html | 35 +++++++++ src/pobsync_backend/tests/test_views.py | 71 ++++++++++++++++++- src/pobsync_backend/views.py | 57 +++++++++++++-- src/pobsync_server/urls.py | 1 + 5 files changed, 177 insertions(+), 5 deletions(-) diff --git a/src/pobsync_backend/forms.py b/src/pobsync_backend/forms.py index c57379f..3895084 100644 --- a/src/pobsync_backend/forms.py +++ b/src/pobsync_backend/forms.py @@ -133,6 +133,24 @@ class ManualBackupForm(forms.Form): ) +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_host = forms.CharField() + + def __init__(self, *args, host_name: str, **kwargs) -> None: + self.host_name = host_name + super().__init__(*args, **kwargs) + self.fields["confirm_host"].help_text = f"Type {host_name} to confirm deletion." + + def clean_confirm_host(self) -> str: + value = self.cleaned_data["confirm_host"].strip() + if value != self.host_name: + raise forms.ValidationError(f"Type {self.host_name} to confirm.") + return value + + class ScheduleConfigForm(forms.ModelForm): cron_expr = forms.CharField( label="Cron expression", diff --git a/src/pobsync_backend/templates/pobsync_backend/retention_plan.html b/src/pobsync_backend/templates/pobsync_backend/retention_plan.html index dac85a7..1b1f9ec 100644 --- a/src/pobsync_backend/templates/pobsync_backend/retention_plan.html +++ b/src/pobsync_backend/templates/pobsync_backend/retention_plan.html @@ -59,6 +59,41 @@ + {% if plan.delete %} +
+

Apply Retention

+
+ {% csrf_token %} + {{ apply_form.non_field_errors }} + {{ apply_form.kind.as_hidden }} + +
+ {{ apply_form.max_delete.errors }} + + {{ apply_form.max_delete }} +
Must be at least the number of snapshots shown in Would Delete.
+
+ +
+ {{ apply_form.protect_bases.errors }} + + {{ apply_form.protect_bases }} +
+ +
+ {{ apply_form.confirm_host.errors }} + + {{ apply_form.confirm_host }} +
{{ apply_form.confirm_host.help_text }}
+
+ +
+ +
+
+
+ {% endif %} +

Keep Reasons

diff --git a/src/pobsync_backend/tests/test_views.py b/src/pobsync_backend/tests/test_views.py index 31feeb0..d2bd253 100644 --- a/src/pobsync_backend/tests/test_views.py +++ b/src/pobsync_backend/tests/test_views.py @@ -5,7 +5,7 @@ from pathlib import Path from tempfile import TemporaryDirectory from django.contrib.auth import get_user_model -from django.test import TestCase +from django.test import TestCase, override_settings from django.urls import reverse from pobsync.util import write_yaml_atomic @@ -458,6 +458,75 @@ class ViewTests(TestCase): self.assertRedirects(response, reverse("host_detail", args=[host.host])) self.assertContains(response, "Retention kind must be scheduled, manual, or all.") + def test_retention_apply_deletes_planned_snapshot_after_confirmation(self) -> None: + self.client.force_login(self.staff_user) + with TemporaryDirectory() as tmp: + home = Path(tmp) / "home" + 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_dir = Path(tmp) / "backups" / host.host / "scheduled" / "20260518-021500Z__OLDSNAP" + new_dir = Path(tmp) / "backups" / host.host / "scheduled" / "20260519-021500Z__NEWSNAP" + old_dir.mkdir(parents=True) + new_dir.mkdir(parents=True) + old_snapshot = self._snapshot(host, old_dir.name) + old_snapshot.path = str(old_dir) + old_snapshot.save(update_fields=["path"]) + new_snapshot = self._snapshot(host, new_dir.name) + new_snapshot.path = str(new_dir) + new_snapshot.save(update_fields=["path"]) + + with override_settings(POBSYNC_HOME=str(home)): + response = self.client.post( + reverse("apply_host_retention", args=[host.host]), + { + "kind": "scheduled", + "max_delete": "1", + "confirm_host": host.host, + }, + follow=True, + ) + + self.assertFalse(old_dir.exists()) + self.assertTrue(new_dir.exists()) + + self.assertRedirects(response, f"{reverse('host_retention_plan', args=[host.host])}?kind=scheduled") + 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()) + + def test_retention_apply_rejects_bad_confirmation(self) -> None: + self.client.force_login(self.staff_user) + host = HostConfig.objects.create(host="web-01", address="web-01.example.test") + self._snapshot(host, "20260518-021500Z__OLDSNAP") + + response = self.client.post( + reverse("apply_host_retention", args=[host.host]), + { + "kind": "scheduled", + "max_delete": "1", + "confirm_host": "wrong", + }, + 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(), 1) + + 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") + + response = self.client.get(reverse("apply_host_retention", args=[host.host])) + + self.assertEqual(response.status_code, 405) + def test_schedule_form_renders_defaults_for_new_schedule(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 bde74bb..a838456 100644 --- a/src/pobsync_backend/views.py +++ b/src/pobsync_backend/views.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +from pathlib import Path from django.contrib import messages from django.contrib.admin.views.decorators import staff_member_required @@ -9,12 +10,19 @@ from django.db.models import Count from django.shortcuts import get_object_or_404, redirect, render from django.views.decorators.http import require_POST -from pobsync.errors import ConfigError +from pobsync.errors import PobsyncError from .backup_runner import queue_backup_run -from .forms import CreateHostConfigForm, GlobalConfigForm, HostConfigForm, ManualBackupForm, ScheduleConfigForm +from .forms import ( + CreateHostConfigForm, + GlobalConfigForm, + HostConfigForm, + ManualBackupForm, + RetentionApplyForm, + ScheduleConfigForm, +) from .models import BackupRun, GlobalConfig, HostConfig, ScheduleConfig, SnapshotRecord -from .retention import run_sql_retention_plan +from .retention import run_sql_retention_apply, run_sql_retention_plan from .snapshot_discovery import discover_snapshots, inspect_snapshot_discovery @@ -196,7 +204,7 @@ def host_retention_plan(request, host: str): protect_bases = request.GET.get("protect_bases") in {"1", "true", "on", "yes"} try: plan = run_sql_retention_plan(host=host_config.host, kind=kind, protect_bases=protect_bases) - except ConfigError as exc: + except PobsyncError as exc: messages.error(request, str(exc)) return redirect("host_detail", host=host_config.host) context = { @@ -204,10 +212,51 @@ def host_retention_plan(request, host: str): "kind": kind, "protect_bases": protect_bases, "plan": plan, + "apply_form": RetentionApplyForm( + host_name=host_config.host, + initial={ + "kind": kind, + "protect_bases": protect_bases, + "max_delete": len(plan["delete"]), + }, + ), } return render(request, "pobsync_backend/retention_plan.html", context) +@staff_member_required +@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) + if not form.is_valid(): + messages.error(request, "Retention apply confirmation is invalid.") + return redirect("host_retention_plan", host=host_config.host) + + kind = form.cleaned_data["kind"] + protect_bases = bool(form.cleaned_data["protect_bases"]) + try: + result = run_sql_retention_apply( + prefix=Path(settings.POBSYNC_HOME), + host=host_config.host, + kind=kind, + protect_bases=protect_bases, + yes=True, + max_delete=form.cleaned_data["max_delete"], + ) + except PobsyncError as exc: + messages.error(request, str(exc)) + else: + messages.success(request, f"Retention deleted {len(result['deleted'])} snapshot(s) for {host_config.host}.") + + target = redirect("host_retention_plan", host=host_config.host) + query = f"kind={kind}" + if protect_bases: + query += "&protect_bases=1" + target["Location"] = f"{target['Location']}?{query}" + return target + + @staff_member_required def edit_host_config(request, host: str): host_config = get_object_or_404(HostConfig, host=host) diff --git a/src/pobsync_server/urls.py b/src/pobsync_server/urls.py index 31a6199..6088e15 100644 --- a/src/pobsync_server/urls.py +++ b/src/pobsync_server/urls.py @@ -14,6 +14,7 @@ urlpatterns = [ path("hosts//config/", views.edit_host_config, name="edit_host_config"), path("hosts//queue-backup/", views.queue_manual_backup, name="queue_manual_backup"), path("hosts//discover-snapshots/", views.discover_host_snapshots, name="discover_host_snapshots"), + path("hosts//retention-apply/", views.apply_host_retention, name="apply_host_retention"), path("hosts//retention-plan/", views.host_retention_plan, name="host_retention_plan"), path("hosts//schedule/", views.edit_host_schedule, name="edit_host_schedule"), path("runs//", views.run_detail, name="run_detail"),