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

(feature) queue manual backups from the Django host page

Browse Source 
Add a staff-only manual backup form to host detail pages with safe
dry-run defaults and optional retention settings.

Queue manual BackupRun records through the existing worker-backed runner
path instead of executing backups inside the web request.

Validate disabled hosts, missing global config, and invalid methods with
view tests covering the new UI flow.
This commit is contained in:
Peter van Arkel
2026-05-19 13:04:50 +02:00
parent fe8e65e12e
commit 3da877eb8a
5 changed files with 141 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
src/pobsync_backend/forms.py
View File

@@ -117,6 +117,23 @@ class GlobalConfigForm(forms.ModelForm):
return instance
class ManualBackupForm(forms.Form):
dry_run = forms.BooleanField(
required=False,
initial=True,
help_text="Queue rsync in dry-run mode without writing a snapshot.",
)
prune = forms.BooleanField(
required=False,
help_text="Apply retention after a successful non-dry-run backup.",
)
prune_max_delete = forms.IntegerField(min_value=0, initial=10)
prune_protect_bases = forms.BooleanField(
required=False,
help_text="Keep snapshots that are used as bases by other snapshots.",
)
class ScheduleConfigForm(forms.ModelForm):
cron_expr = forms.CharField(
label="Cron expression",

21
src/pobsync_backend/templates/pobsync_backend/host_detail.html
View File

@@ -51,6 +51,27 @@
</section>
</div>
<section class="panel">
<h2>Queue Manual Backup</h2>
<form method="post" action="{% url 'queue_manual_backup' host.host %}" class="form-grid">
{% csrf_token %}
{{ manual_backup_form.non_field_errors }}
{% for field in manual_backup_form %}
<div class="field">
{{ field.errors }}
<label for="{{ field.id_for_label }}">{{ field.label }}</label>
{{ field }}
{% if field.help_text %}<div class="helptext">{{ field.help_text }}</div>{% endif %}
</div>
{% endfor %}
<div class="actions">
<button type="submit">Queue backup</button>
</div>
</form>
</section>
<section class="panel">
<h2>Latest Runs</h2>
<table>

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

@@ -180,6 +180,8 @@ class ViewTests(TestCase):
self.assertContains(response, "Discover snapshots")
self.assertContains(response, "Edit schedule")
self.assertContains(response, "Edit config")
self.assertContains(response, "Queue Manual Backup")
self.assertContains(response, reverse("queue_manual_backup", args=[host.host]))
self.assertContains(response, reverse("run_detail", args=[BackupRun.objects.get().id]))
self.assertContains(response, reverse("snapshot_detail", args=[snapshot.id]))
@@ -190,6 +192,66 @@ class ViewTests(TestCase):
self.assertEqual(response.status_code, 404)
def test_queue_manual_backup_creates_queued_run_and_redirects_to_run_detail(self) -> None:
self.client.force_login(self.staff_user)
GlobalConfig.objects.create(name="default", backup_root="/backups")
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
response = self.client.post(
reverse("queue_manual_backup", args=[host.host]),
{
"dry_run": "on",
"prune": "on",
"prune_max_delete": "4",
"prune_protect_bases": "on",
},
follow=True,
)
run = BackupRun.objects.get(host=host)
self.assertRedirects(response, reverse("run_detail", args=[run.id]))
self.assertContains(response, f"Queued manual backup run {run.id} for web-01.")
self.assertEqual(run.status, BackupRun.Status.QUEUED)
self.assertEqual(run.run_type, BackupRun.RunType.MANUAL)
self.assertEqual(
run.result["requested"],
{
"dry_run": True,
"prune": True,
"prune_max_delete": 4,
"prune_protect_bases": True,
},
)
def test_queue_manual_backup_requires_default_global_config(self) -> None:
self.client.force_login(self.staff_user)
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
response = self.client.post(reverse("queue_manual_backup", args=[host.host]), {"dry_run": "on"}, follow=True)
self.assertRedirects(response, reverse("host_detail", args=[host.host]))
self.assertContains(response, "Create the default global config before queueing backups.")
self.assertFalse(BackupRun.objects.exists())
def test_queue_manual_backup_rejects_disabled_host(self) -> None:
self.client.force_login(self.staff_user)
GlobalConfig.objects.create(name="default", backup_root="/backups")
host = HostConfig.objects.create(host="web-01", address="web-01.example.test", enabled=False)
response = self.client.post(reverse("queue_manual_backup", args=[host.host]), {"dry_run": "on"}, follow=True)
self.assertRedirects(response, reverse("host_detail", args=[host.host]))
self.assertContains(response, "Cannot queue backup for disabled host web-01.")
self.assertFalse(BackupRun.objects.exists())
def test_queue_manual_backup_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("queue_manual_backup", args=[host.host]))
self.assertEqual(response.status_code, 405)
def test_run_detail_renders_result_payload(self) -> None:
self.client.force_login(self.staff_user)
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")

41
src/pobsync_backend/views.py
View File

@@ -10,7 +10,8 @@ from django.views.decorators.http import require_POST
from pobsync.errors import ConfigError
from .forms import CreateHostConfigForm, GlobalConfigForm, HostConfigForm, ScheduleConfigForm
from .backup_runner import queue_backup_run
from .forms import CreateHostConfigForm, GlobalConfigForm, HostConfigForm, ManualBackupForm, ScheduleConfigForm
from .models import BackupRun, GlobalConfig, HostConfig, ScheduleConfig, SnapshotRecord
from .retention import run_sql_retention_plan
from .snapshot_discovery import discover_snapshots
@@ -90,6 +91,7 @@ def host_detail(request, host: str):
context = {
"host": host_config,
"schedule": _schedule_for_host(host_config),
"manual_backup_form": ManualBackupForm(initial=_default_manual_backup_initial(host_config)),
"latest_runs": host_config.runs.select_related("snapshot").order_by("-created_at")[:10],
"snapshots": host_config.snapshots.select_related("base").order_by("-started_at", "dirname")[:20],
"counts": {
@@ -102,6 +104,33 @@ def host_detail(request, host: str):
return render(request, "pobsync_backend/host_detail.html", context)
@staff_member_required
@require_POST
def queue_manual_backup(request, host: str):
host_config = get_object_or_404(HostConfig, host=host)
if not host_config.enabled:
messages.error(request, f"Cannot queue backup for disabled host {host_config.host}.")
return redirect("host_detail", host=host_config.host)
if not GlobalConfig.objects.filter(name="default").exists():
messages.error(request, "Create the default global config before queueing backups.")
return redirect("host_detail", host=host_config.host)
form = ManualBackupForm(request.POST)
if not form.is_valid():
messages.error(request, "Manual backup options are invalid.")
return redirect("host_detail", host=host_config.host)
run = queue_backup_run(
host=host_config,
dry_run=form.cleaned_data["dry_run"],
prune=form.cleaned_data["prune"],
prune_max_delete=form.cleaned_data["prune_max_delete"],
prune_protect_bases=form.cleaned_data["prune_protect_bases"],
)
messages.success(request, f"Queued manual backup run {run.id} for {host_config.host}.")
return redirect("run_detail", run_id=run.id)
@staff_member_required
def run_detail(request, run_id: int):
run = get_object_or_404(BackupRun.objects.select_related("host", "snapshot"), id=run_id)
@@ -257,5 +286,15 @@ def _default_host_initial() -> dict[str, object]:
}
def _default_manual_backup_initial(host_config: HostConfig) -> dict[str, object]:
schedule = _schedule_for_host(host_config)
return {
"dry_run": True,
"prune": bool(schedule.prune) if schedule else False,
"prune_max_delete": schedule.prune_max_delete if schedule else 10,
"prune_protect_bases": bool(schedule.prune_protect_bases) if schedule else False,
}
def _pretty_json(value: object) -> str:
return json.dumps(value or {}, indent=2, sort_keys=True)