(feature) Add schedule editing view for hosts

Add a staff-only Django form for creating and updating host schedules using the
SQL-backed ScheduleConfig model. Link the form from host detail pages, validate
cron expressions with the existing scheduler parser, and preserve scheduler/CLI
behavior by writing to the same source of truth.

Cover default rendering, schedule creation, updates, and invalid cron handling
with view tests.
This commit is contained in:
2026-05-19 12:13:12 +02:00
parent 123583a502
commit 6d7bf531ac
8 changed files with 200 additions and 0 deletions

View File

@@ -155,6 +155,7 @@ Staff-only JSON endpoints expose service status, hosts, snapshots, and backup ru
Staff-only dashboard views expose the same operational state through Django templates. Staff-only dashboard views expose the same operational state through Django templates.
Host pages include a safe snapshot discovery action that records existing snapshots into SQL. Host pages include a safe snapshot discovery action that records existing snapshots into SQL.
Host pages also include a read-only SQL retention plan view before any destructive pruning action. Host pages also include a read-only SQL retention plan view before any destructive pruning action.
Schedules can be created or updated from host pages using the same SQL-backed scheduler model.
The remaining internal engine code still contains reusable backup primitives: The remaining internal engine code still contains reusable backup primitives:

View File

@@ -0,0 +1,33 @@
from __future__ import annotations
from django import forms
from .models import ScheduleConfig
from .scheduler import parse_cron_expr
class ScheduleConfigForm(forms.ModelForm):
cron_expr = forms.CharField(
label="Cron expression",
help_text='Five-field cron expression, for example "15 2 * * *".',
)
prune_max_delete = forms.IntegerField(min_value=0)
class Meta:
model = ScheduleConfig
fields = (
"cron_expr",
"user",
"enabled",
"prune",
"prune_max_delete",
"prune_protect_bases",
)
def clean_cron_expr(self) -> str:
cron_expr = self.cleaned_data["cron_expr"].strip()
try:
parse_cron_expr(cron_expr)
except ValueError as exc:
raise forms.ValidationError(str(exc)) from exc
return cron_expr

View File

@@ -101,6 +101,23 @@
} }
.message.success { border-color: #a7d8b9; background: #edf8f1; color: var(--success); } .message.success { border-color: #a7d8b9; background: #edf8f1; color: var(--success); }
.message.error { border-color: #e8b4b4; background: #fff0f0; color: var(--failed); } .message.error { border-color: #e8b4b4; background: #fff0f0; color: var(--failed); }
.form-grid { display: grid; gap: 14px; max-width: 680px; }
.field { display: grid; gap: 5px; }
.field label { font-weight: 650; }
.field input[type="text"], .field input[type="number"] {
border: 1px solid var(--border);
border-radius: 6px;
font: inherit;
padding: 8px 10px;
width: 100%;
}
.field .helptext { color: var(--muted); font-size: 12px; }
.errorlist {
color: var(--failed);
list-style: none;
margin: 0;
padding: 0;
}
@media (max-width: 800px) { @media (max-width: 800px) {
main { padding: 16px; } main { padding: 16px; }
nav { padding: 0; } nav { padding: 0; }

View File

@@ -11,6 +11,7 @@
<button type="submit">Discover snapshots</button> <button type="submit">Discover snapshots</button>
</form> </form>
<a class="button-link" href="{% url 'host_retention_plan' host.host %}">Plan retention</a> <a class="button-link" href="{% url 'host_retention_plan' host.host %}">Plan retention</a>
<a class="button-link" href="{% url 'edit_host_schedule' host.host %}">Edit schedule</a>
</section> </section>
<section class="grid" aria-label="Host summary"> <section class="grid" aria-label="Host summary">

View File

@@ -0,0 +1,32 @@
{% extends "pobsync_backend/base.html" %}
{% block title %}Schedule | {{ host.host }}{% endblock %}
{% block content %}
<h1>Schedule: {{ host.host }}</h1>
<section class="actions" aria-label="Schedule actions">
<a class="button-link" href="{% url 'host_detail' host.host %}">Back to host</a>
</section>
<section class="panel">
<h2>{% if schedule %}Edit Schedule{% else %}Create Schedule{% endif %}</h2>
<form method="post" class="form-grid">
{% csrf_token %}
{{ form.non_field_errors }}
{% for field in 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">Save schedule</button>
</div>
</form>
</section>
{% endblock %}

View File

@@ -67,6 +67,7 @@ class ViewTests(TestCase):
self.assertContains(response, "15 2 * * *") self.assertContains(response, "15 2 * * *")
self.assertContains(response, "20260519-021500Z__ABCDEFGH") self.assertContains(response, "20260519-021500Z__ABCDEFGH")
self.assertContains(response, "Discover snapshots") self.assertContains(response, "Discover snapshots")
self.assertContains(response, "Edit schedule")
def test_host_detail_returns_404_for_unknown_host(self) -> None: def test_host_detail_returns_404_for_unknown_host(self) -> None:
self.client.force_login(self.staff_user) self.client.force_login(self.staff_user)
@@ -152,6 +153,84 @@ class ViewTests(TestCase):
self.assertRedirects(response, reverse("host_detail", args=[host.host])) self.assertRedirects(response, reverse("host_detail", args=[host.host]))
self.assertContains(response, "Retention kind must be scheduled, manual, or all.") self.assertContains(response, "Retention kind must be scheduled, manual, or all.")
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")
response = self.client.get(reverse("edit_host_schedule", args=[host.host]))
self.assertEqual(response.status_code, 200)
self.assertContains(response, "Create Schedule")
self.assertContains(response, "15 2 * * *")
self.assertContains(response, "Save schedule")
def test_schedule_form_creates_schedule(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("edit_host_schedule", args=[host.host]),
{
"cron_expr": "30 3 * * *",
"user": "root",
"enabled": "on",
"prune": "on",
"prune_max_delete": "4",
"prune_protect_bases": "on",
},
follow=True,
)
self.assertRedirects(response, reverse("host_detail", args=[host.host]))
self.assertContains(response, "Schedule saved for web-01.")
schedule = ScheduleConfig.objects.get(host=host)
self.assertEqual(schedule.cron_expr, "30 3 * * *")
self.assertTrue(schedule.enabled)
self.assertTrue(schedule.prune)
self.assertEqual(schedule.prune_max_delete, 4)
self.assertTrue(schedule.prune_protect_bases)
def test_schedule_form_updates_existing_schedule(self) -> None:
self.client.force_login(self.staff_user)
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
schedule = ScheduleConfig.objects.create(host=host, cron_expr="15 2 * * *", enabled=True, prune=True)
response = self.client.post(
reverse("edit_host_schedule", args=[host.host]),
{
"cron_expr": "45 4 * * 1",
"user": "backup",
"prune_max_delete": "8",
},
follow=True,
)
self.assertRedirects(response, reverse("host_detail", args=[host.host]))
schedule.refresh_from_db()
self.assertEqual(schedule.cron_expr, "45 4 * * 1")
self.assertEqual(schedule.user, "backup")
self.assertFalse(schedule.enabled)
self.assertFalse(schedule.prune)
self.assertEqual(schedule.prune_max_delete, 8)
def test_schedule_form_rejects_invalid_cron(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("edit_host_schedule", args=[host.host]),
{
"cron_expr": "bad cron",
"user": "root",
"enabled": "on",
"prune_max_delete": "10",
},
)
self.assertEqual(response.status_code, 200)
self.assertContains(response, "cron expression must have exactly 5 fields")
self.assertFalse(ScheduleConfig.objects.filter(host=host).exists())
def _snapshot(self, host: HostConfig, dirname: str) -> SnapshotRecord: def _snapshot(self, host: HostConfig, dirname: str) -> SnapshotRecord:
started_at = datetime.strptime(dirname.split("__", 1)[0], "%Y%m%d-%H%M%SZ").replace(tzinfo=timezone.utc) started_at = datetime.strptime(dirname.split("__", 1)[0], "%Y%m%d-%H%M%SZ").replace(tzinfo=timezone.utc)
return SnapshotRecord.objects.create( return SnapshotRecord.objects.create(

View File

@@ -8,6 +8,7 @@ from django.views.decorators.http import require_POST
from pobsync.errors import ConfigError from pobsync.errors import ConfigError
from .forms import ScheduleConfigForm
from .models import BackupRun, HostConfig, ScheduleConfig, SnapshotRecord from .models import BackupRun, HostConfig, ScheduleConfig, SnapshotRecord
from .retention import run_sql_retention_plan from .retention import run_sql_retention_plan
from .snapshot_discovery import discover_snapshots from .snapshot_discovery import discover_snapshots
@@ -95,8 +96,43 @@ def host_retention_plan(request, host: str):
return render(request, "pobsync_backend/retention_plan.html", context) return render(request, "pobsync_backend/retention_plan.html", context)
@staff_member_required
def edit_host_schedule(request, host: str):
host_config = get_object_or_404(HostConfig, host=host)
schedule = _schedule_for_host(host_config)
if request.method == "POST":
form = ScheduleConfigForm(request.POST, instance=schedule)
if form.is_valid():
saved_schedule = form.save(commit=False)
saved_schedule.host = host_config
saved_schedule.save()
messages.success(request, f"Schedule saved for {host_config.host}.")
return redirect("host_detail", host=host_config.host)
else:
form = ScheduleConfigForm(instance=schedule, initial=_default_schedule_initial())
return render(
request,
"pobsync_backend/schedule_form.html",
{
"host": host_config,
"schedule": schedule,
"form": form,
},
)
def _schedule_for_host(host_config: HostConfig) -> ScheduleConfig | None: def _schedule_for_host(host_config: HostConfig) -> ScheduleConfig | None:
try: try:
return host_config.schedule return host_config.schedule
except ScheduleConfig.DoesNotExist: except ScheduleConfig.DoesNotExist:
return None return None
def _default_schedule_initial() -> dict[str, object]:
return {
"cron_expr": "15 2 * * *",
"user": "root",
"enabled": True,
"prune_max_delete": 10,
}

View File

@@ -11,6 +11,7 @@ urlpatterns = [
path("hosts/<str:host>/", views.host_detail, name="host_detail"), path("hosts/<str:host>/", views.host_detail, name="host_detail"),
path("hosts/<str:host>/discover-snapshots/", views.discover_host_snapshots, name="discover_host_snapshots"), path("hosts/<str:host>/discover-snapshots/", views.discover_host_snapshots, name="discover_host_snapshots"),
path("hosts/<str:host>/retention-plan/", views.host_retention_plan, name="host_retention_plan"), path("hosts/<str:host>/retention-plan/", views.host_retention_plan, name="host_retention_plan"),
path("hosts/<str:host>/schedule/", views.edit_host_schedule, name="edit_host_schedule"),
path("api/", api.api_index), path("api/", api.api_index),
path("api/status/", api.status), path("api/status/", api.status),
path("api/hosts/", api.hosts), path("api/hosts/", api.hosts),