(feature) Link rsync logs from backup run detail

Record the final rsync log path for successful real backup runs, matching
the existing dry-run and failure result payloads.

Add a staff-only run log endpoint and surface the link on run detail pages,
including fallback log discovery for older runs based on snapshot_path.

Cover direct log links and inferred scheduled backup logs with view tests.
This commit is contained in:
2026-05-20 00:09:59 +02:00
parent f41e59e695
commit 0babc57f57
6 changed files with 85 additions and 0 deletions

View File

@@ -242,6 +242,7 @@ class RunScheduledConfigSourceTests(SimpleTestCase):
meta_text = meta_path.read_text(encoding="utf-8")
self.assertTrue(result["ok"])
self.assertEqual(result["log"], str(Path(result["snapshot"]) / "meta" / "rsync.log"))
self.assertEqual(result["stats"]["rsync"]["files_total"], 10)
self.assertEqual(result["stats"]["rsync"]["files_transferred"], 2)
self.assertEqual(result["stats"]["rsync"]["link_dest_estimated_savings_bytes"], 1500)

View File

@@ -895,6 +895,50 @@ class ViewTests(TestCase):
self.assertContains(response, ""ok": true")
self.assertContains(response, reverse("snapshot_detail", args=[snapshot.id]))
def test_run_detail_links_existing_rsync_log(self) -> None:
self.client.force_login(self.staff_user)
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
with TemporaryDirectory() as tmp:
log_path = Path(tmp) / "snapshot" / "meta" / "rsync.log"
log_path.parent.mkdir(parents=True)
log_path.write_text("rsync log line\n", encoding="utf-8")
run = BackupRun.objects.create(
host=host,
status=BackupRun.Status.SUCCESS,
snapshot_path=str(log_path.parent.parent),
result={"ok": True, "log": str(log_path)},
)
response = self.client.get(reverse("run_detail", args=[run.id]))
log_response = self.client.get(reverse("run_rsync_log", args=[run.id]))
log_body = b"".join(log_response.streaming_content)
self.assertContains(response, reverse("run_rsync_log", args=[run.id]))
self.assertContains(response, str(log_path))
self.assertEqual(log_response.status_code, 200)
self.assertEqual(log_body, b"rsync log line\n")
def test_run_detail_infers_rsync_log_from_snapshot_path(self) -> None:
self.client.force_login(self.staff_user)
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
with TemporaryDirectory() as tmp:
snapshot_path = Path(tmp) / "snapshot"
log_path = snapshot_path / "meta" / "rsync.log"
log_path.parent.mkdir(parents=True)
log_path.write_text("scheduled log\n", encoding="utf-8")
run = BackupRun.objects.create(
host=host,
status=BackupRun.Status.SUCCESS,
snapshot_path=str(snapshot_path),
result={"ok": True},
)
response = self.client.get(reverse("run_rsync_log", args=[run.id]))
body = b"".join(response.streaming_content)
self.assertEqual(response.status_code, 200)
self.assertEqual(body, b"scheduled log\n")
def test_run_detail_offers_cancel_for_running_run(self) -> None:
self.client.force_login(self.staff_user)
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")