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

(feature) Add run completion notifications

Browse Source 
Add email and webhook notification targets with delivery tracking, and send
notifications when backup runs reach a terminal status.

Expose notification target management in the Django UI and keep delivery
failures recorded without failing the backup worker.
This commit is contained in:
Peter van Arkel
2026-05-28 21:20:38 +02:00
parent 1f5c4e0756
commit 67ffd6101b
14 changed files with 819 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
src/pobsync_backend/tests/test_backup_worker.py
View File

@@ -11,7 +11,7 @@ from django.utils import timezone
from pobsync.util import write_yaml_atomic
from pobsync_backend.backup_runner import queue_backup_run, reconcile_running_runs
from pobsync_backend.management.commands.run_pobsync_worker import Command
from pobsync_backend.models import BackupRun, GlobalConfig, HostConfig, SnapshotRecord
from pobsync_backend.models import BackupRun, GlobalConfig, HostConfig, NotificationDelivery, NotificationTarget, SnapshotRecord
class BackupWorkerTests(TestCase):
@@ -116,6 +116,42 @@ class BackupWorkerTests(TestCase):
self.assertEqual(run.rsync_exit_code, 24)
self.assertEqual(run.result["warning"]["category"], "vanished")
def test_worker_sends_notification_after_completed_run(self) -> None:
with TemporaryDirectory() as tmp:
backup_root = Path(tmp) / "backups"
GlobalConfig.objects.create(name="default", backup_root=str(backup_root))
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
NotificationTarget.objects.create(
name="ops",
channel=NotificationTarget.Channel.EMAIL,
email_to="ops@example.test",
)
snapshot_dir = backup_root / host.host / "scheduled" / "20260519-021500Z__ABCDEFGH"
meta_dir = snapshot_dir / "meta"
meta_dir.mkdir(parents=True)
write_yaml_atomic(meta_dir / "meta.yaml", {"status": "success", "started_at": "2026-05-19T02:15:00Z"})
run = queue_backup_run(host=host)
with (
patch("pobsync_backend.backup_runner.run_scheduled") as run_scheduled,
patch("pobsync_backend.notifications.send_mail", return_value=1) as send_mail,
):
run_scheduled.return_value = {
"ok": True,
"dry_run": False,
"host": host.host,
"snapshot": str(snapshot_dir),
"base": None,
"rsync": {"exit_code": 0},
}
Command()._run_once(prefix=Path(tmp) / "home")
run.refresh_from_db()
self.assertEqual(run.status, BackupRun.Status.SUCCESS)
self.assertEqual(NotificationDelivery.objects.get(run=run).status, NotificationDelivery.Status.SENT)
send_mail.assert_called_once()
def test_worker_refreshes_heartbeat_while_run_is_active(self) -> None:
with TemporaryDirectory() as tmp:
GlobalConfig.objects.create(name="default", backup_root=str(Path(tmp) / "backups"))

125
src/pobsync_backend/tests/test_notifications.py Normal file
View File

@@ -0,0 +1,125 @@
from __future__ import annotations
from datetime import timedelta
from unittest.mock import Mock, patch
from django.test import TestCase
from django.utils import timezone
from pobsync_backend.models import BackupRun, HostConfig, NotificationDelivery, NotificationTarget
from pobsync_backend.notifications import notify_backup_run_completed
class NotificationTests(TestCase):
def test_email_notification_is_sent_for_matching_status(self) -> None:
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
target = NotificationTarget.objects.create(
name="ops",
channel=NotificationTarget.Channel.EMAIL,
statuses=[BackupRun.Status.FAILED],
email_to="ops@example.test",
)
run = BackupRun.objects.create(
host=host,
status=BackupRun.Status.FAILED,
run_type=BackupRun.RunType.MANUAL,
started_at=timezone.now() - timedelta(minutes=5),
ended_at=timezone.now(),
rsync_exit_code=12,
result={"failure": {"message": "rsync failed"}},
)
with patch("pobsync_backend.notifications.send_mail", return_value=1) as send_mail:
results = notify_backup_run_completed(run)
self.assertEqual(len(results), 1)
self.assertTrue(results[0].sent)
send_mail.assert_called_once()
subject, message, _from_email, recipients = send_mail.call_args.args
self.assertEqual(subject, f"pobsync failed: web-01 run {run.id}")
self.assertIn("Failure: rsync failed", message)
self.assertEqual(recipients, ["ops@example.test"])
delivery = NotificationDelivery.objects.get(target=target, run=run)
self.assertEqual(delivery.status, NotificationDelivery.Status.SENT)
target.refresh_from_db()
self.assertEqual(target.last_status, NotificationDelivery.Status.SENT)
self.assertEqual(target.last_error, "")
self.assertIsNotNone(target.last_sent_at)
def test_webhook_notification_posts_payload(self) -> None:
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
target = NotificationTarget.objects.create(
name="discord",
channel=NotificationTarget.Channel.WEBHOOK,
webhook_url="https://hooks.example.test/pobsync",
webhook_headers={"X-Token": "secret"},
)
run = BackupRun.objects.create(host=host, status=BackupRun.Status.SUCCESS, run_type=BackupRun.RunType.SCHEDULED)
response = Mock()
response.status = 204
response.__enter__ = Mock(return_value=response)
response.__exit__ = Mock(return_value=False)
with patch("pobsync_backend.notifications.urllib.request.urlopen", return_value=response) as urlopen:
notify_backup_run_completed(run)
request = urlopen.call_args.args[0]
self.assertEqual(request.full_url, "https://hooks.example.test/pobsync")
self.assertEqual(request.get_method(), "POST")
self.assertEqual(request.headers["X-token"], "secret")
self.assertIn(f'"id": {run.id}', request.data.decode("utf-8"))
self.assertEqual(NotificationDelivery.objects.get(target=target, run=run).status, NotificationDelivery.Status.SENT)
def test_notification_filters_statuses(self) -> None:
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
NotificationTarget.objects.create(
name="failures-only",
channel=NotificationTarget.Channel.EMAIL,
statuses=[BackupRun.Status.FAILED],
email_to="ops@example.test",
)
run = BackupRun.objects.create(host=host, status=BackupRun.Status.SUCCESS)
with patch("pobsync_backend.notifications.send_mail") as send_mail:
results = notify_backup_run_completed(run)
self.assertEqual(results, [])
send_mail.assert_not_called()
self.assertEqual(NotificationDelivery.objects.count(), 0)
def test_notification_delivery_is_idempotent_per_run_and_target(self) -> None:
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
target = NotificationTarget.objects.create(
name="ops",
channel=NotificationTarget.Channel.EMAIL,
email_to="ops@example.test",
)
run = BackupRun.objects.create(host=host, status=BackupRun.Status.WARNING)
with patch("pobsync_backend.notifications.send_mail", return_value=1) as send_mail:
notify_backup_run_completed(run)
notify_backup_run_completed(run)
self.assertEqual(NotificationDelivery.objects.filter(target=target, run=run).count(), 1)
send_mail.assert_called_once()
def test_failed_delivery_is_recorded_without_raising(self) -> None:
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
target = NotificationTarget.objects.create(
name="broken",
channel=NotificationTarget.Channel.EMAIL,
email_to="ops@example.test",
)
run = BackupRun.objects.create(host=host, status=BackupRun.Status.FAILED)
with patch("pobsync_backend.notifications.send_mail", side_effect=RuntimeError("smtp down")):
results = notify_backup_run_completed(run)
self.assertEqual(len(results), 1)
self.assertFalse(results[0].sent)
delivery = NotificationDelivery.objects.get(target=target, run=run)
self.assertEqual(delivery.status, NotificationDelivery.Status.FAILED)
self.assertEqual(delivery.error, "smtp down")
target.refresh_from_db()
self.assertEqual(target.last_status, NotificationDelivery.Status.FAILED)
self.assertEqual(target.last_error, "smtp down")

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

@@ -18,6 +18,8 @@ from pobsync_backend.models import (
BackupRun,
GlobalConfig,
HostConfig,
NotificationDelivery,
NotificationTarget,
PurgedSnapshot,
ScheduleConfig,
SnapshotRecord,
@@ -52,6 +54,7 @@ class ViewTests(TestCase):
self.assertContains(response, reverse("dashboard"))
self.assertContains(response, reverse("hosts_list"))
self.assertContains(response, reverse("ssh_credentials"))
self.assertContains(response, reverse("notification_targets"))
self.assertContains(response, reverse("logs"))
self.assertContains(response, reverse("purged_snapshots"))
self.assertContains(response, reverse("self_check"))
@@ -94,6 +97,70 @@ class ViewTests(TestCase):
self.assertContains(response, "Django control panel")
self.assertContains(response, "Native systemd installer")
def test_notification_targets_view_renders_targets_and_deliveries(self) -> None:
self.client.force_login(self.staff_user)
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
run = BackupRun.objects.create(host=host, status=BackupRun.Status.SUCCESS)
target = NotificationTarget.objects.create(
name="ops",
channel=NotificationTarget.Channel.EMAIL,
email_to="ops@example.test",
last_status=NotificationDelivery.Status.SENT,
)
NotificationDelivery.objects.create(target=target, run=run, status=NotificationDelivery.Status.SENT)
response = self.client.get(reverse("notification_targets"))
self.assertEqual(response.status_code, 200)
self.assertContains(response, "Notifications")
self.assertContains(response, "ops")
self.assertContains(response, "ops@example.test")
self.assertContains(response, f"Run {run.id}")
def test_notification_target_form_creates_email_target(self) -> None:
self.client.force_login(self.staff_user)
response = self.client.post(
reverse("create_notification_target"),
{
"name": "ops",
"enabled": "on",
"channel": NotificationTarget.Channel.EMAIL,
"statuses": [BackupRun.Status.FAILED, BackupRun.Status.WARNING],
"email_to": "ops@example.test, backup@example.test",
"webhook_headers": "{}",
"notes": "Notify ops",
},
follow=True,
)
self.assertRedirects(response, reverse("notification_targets"))
self.assertContains(response, "Notification target ops created.")
target = NotificationTarget.objects.get(name="ops")
self.assertEqual(target.channel, NotificationTarget.Channel.EMAIL)
self.assertEqual(target.statuses, [BackupRun.Status.FAILED, BackupRun.Status.WARNING])
self.assertEqual(target.email_to, "ops@example.test\nbackup@example.test")
def test_notification_target_form_requires_channel_destination(self) -> None:
self.client.force_login(self.staff_user)
response = self.client.post(
reverse("create_notification_target"),
{
"name": "broken",
"enabled": "on",
"channel": NotificationTarget.Channel.WEBHOOK,
"statuses": [BackupRun.Status.FAILED],
"email_to": "",
"webhook_url": "",
"webhook_headers": "{}",
},
)
self.assertEqual(response.status_code, 200)
self.assertContains(response, "Webhook targets need a URL.")
self.assertFalse(NotificationTarget.objects.exists())
def test_dashboard_renders_hosts_and_latest_runs(self) -> None:
self.client.force_login(self.staff_user)
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")