from __future__ import annotations from datetime import datetime, timezone from pathlib import Path from tempfile import TemporaryDirectory from django.contrib.auth import get_user_model from django.test import TestCase from django.urls import reverse from pobsync.util import write_yaml_atomic from pobsync_backend.models import BackupRun, GlobalConfig, HostConfig, ScheduleConfig, SnapshotRecord class ViewTests(TestCase): def setUp(self) -> None: user_model = get_user_model() self.staff_user = user_model.objects.create_user( username="admin", password="secret", is_staff=True, is_superuser=True, ) def test_dashboard_requires_staff_login(self) -> None: response = self.client.get(reverse("dashboard")) self.assertEqual(response.status_code, 302) self.assertIn("/admin/login/", response["Location"]) 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") snapshot = self._snapshot(host, "20260519-021500Z__ABCDEFGH") BackupRun.objects.create( host=host, status=BackupRun.Status.SUCCESS, snapshot=snapshot, started_at=datetime(2026, 5, 19, 2, 15, tzinfo=timezone.utc), ) response = self.client.get(reverse("dashboard")) self.assertEqual(response.status_code, 200) self.assertContains(response, "Dashboard") self.assertContains(response, "web-01") self.assertContains(response, "20260519-021500Z__ABCDEFGH") self.assertContains(response, "success") def test_host_detail_renders_config_schedule_runs_and_snapshots(self) -> None: self.client.force_login(self.staff_user) host = HostConfig.objects.create( host="web-01", address="web-01.example.test", source_root="/srv", retention_daily=7, ) ScheduleConfig.objects.create(host=host, cron_expr="15 2 * * *", prune=True, last_status="success") snapshot = self._snapshot(host, "20260519-021500Z__ABCDEFGH") BackupRun.objects.create(host=host, status=BackupRun.Status.SUCCESS, snapshot=snapshot) response = self.client.get(reverse("host_detail", args=[host.host])) self.assertEqual(response.status_code, 200) self.assertContains(response, "web-01") self.assertContains(response, "web-01.example.test") self.assertContains(response, "15 2 * * *") self.assertContains(response, "20260519-021500Z__ABCDEFGH") self.assertContains(response, "Discover snapshots") self.assertContains(response, "Edit schedule") self.assertContains(response, "Edit config") def test_host_detail_returns_404_for_unknown_host(self) -> None: self.client.force_login(self.staff_user) response = self.client.get(reverse("host_detail", args=["missing-host"])) self.assertEqual(response.status_code, 404) def test_discover_host_snapshots_action_discovers_and_redirects(self) -> None: self.client.force_login(self.staff_user) 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") 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"}) response = self.client.post(reverse("discover_host_snapshots", args=[host.host]), follow=True) self.assertRedirects(response, reverse("host_detail", args=[host.host])) self.assertContains(response, "Snapshot discovery scanned 1 items") self.assertTrue(SnapshotRecord.objects.filter(host=host, dirname=snapshot_dir.name).exists()) def test_discover_host_snapshots_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("discover_host_snapshots", args=[host.host])) self.assertEqual(response.status_code, 405) def test_retention_plan_renders_keep_and_delete_sets(self) -> None: self.client.force_login(self.staff_user) 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_snapshot = self._snapshot(host, "20260518-021500Z__OLDSNAP") new_snapshot = self._snapshot(host, "20260519-021500Z__NEWSNAP") response = self.client.get(reverse("host_retention_plan", args=[host.host])) self.assertEqual(response.status_code, 200) self.assertContains(response, "Retention Plan: web-01") self.assertContains(response, old_snapshot.dirname) self.assertContains(response, new_snapshot.dirname) self.assertContains(response, "newest") self.assertContains(response, "Would Delete") def test_retention_plan_can_enable_base_protection(self) -> None: self.client.force_login(self.staff_user) host = HostConfig.objects.create( host="web-01", address="web-01.example.test", retention_daily=0, retention_weekly=0, retention_monthly=0, retention_yearly=0, ) base = self._snapshot(host, "20260518-021500Z__BASESNAP") child = self._snapshot(host, "20260519-021500Z__CHILDSNP") child.base = base child.save(update_fields=["base"]) response = self.client.get(reverse("host_retention_plan", args=[host.host]), {"protect_bases": "1"}) self.assertEqual(response.status_code, 200) self.assertContains(response, "Protect bases: yes") self.assertContains(response, f"base-of:{child.dirname}") def test_retention_plan_rejects_invalid_kind(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("host_retention_plan", args=[host.host]), {"kind": "weird"}, follow=True) self.assertRedirects(response, reverse("host_detail", args=[host.host])) 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 test_host_config_form_renders_existing_values(self) -> None: self.client.force_login(self.staff_user) host = HostConfig.objects.create( host="web-01", address="web-01.example.test", includes=["/srv"], excludes_add=["*.tmp"], rsync_extra_args=["--numeric-ids"], ) response = self.client.get(reverse("edit_host_config", args=[host.host])) self.assertEqual(response.status_code, 200) self.assertContains(response, "Config: web-01") self.assertContains(response, "web-01.example.test") self.assertContains(response, "/srv") self.assertContains(response, "*.tmp") self.assertContains(response, "--numeric-ids") def test_host_config_form_updates_host_config(self) -> None: self.client.force_login(self.staff_user) host = HostConfig.objects.create(host="web-01", address="old.example.test") response = self.client.post( reverse("edit_host_config", args=[host.host]), { "address": "new.example.test", "enabled": "on", "ssh_user": "backup", "ssh_port": "2222", "source_root": "/srv", "includes": "/srv/www\n/srv/db", "excludes_add": "*.tmp\ncache/", "excludes_replace": "", "rsync_extra_args": "--numeric-ids\n--delete", "retention_daily": "7", "retention_weekly": "4", "retention_monthly": "2", "retention_yearly": "1", }, follow=True, ) self.assertRedirects(response, reverse("host_detail", args=[host.host])) self.assertContains(response, "Host config saved for web-01.") host.refresh_from_db() self.assertEqual(host.address, "new.example.test") self.assertEqual(host.ssh_user, "backup") self.assertEqual(host.ssh_port, 2222) self.assertEqual(host.source_root, "/srv") self.assertEqual(host.includes, ["/srv/www", "/srv/db"]) self.assertEqual(host.excludes_add, ["*.tmp", "cache/"]) self.assertIsNone(host.excludes_replace) self.assertEqual(host.rsync_extra_args, ["--numeric-ids", "--delete"]) self.assertEqual(host.retention_daily, 7) self.assertEqual(host.retention_yearly, 1) def test_host_config_form_can_replace_global_excludes(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_config", args=[host.host]), { "address": host.address, "ssh_user": "", "ssh_port": "", "source_root": "", "includes": "", "excludes_add": "", "excludes_replace": "*.cache\nnode_modules/", "rsync_extra_args": "", "retention_daily": "14", "retention_weekly": "8", "retention_monthly": "12", "retention_yearly": "0", }, follow=True, ) self.assertRedirects(response, reverse("host_detail", args=[host.host])) host.refresh_from_db() self.assertFalse(host.enabled) self.assertEqual(host.excludes_add, []) self.assertEqual(host.excludes_replace, ["*.cache", "node_modules/"]) 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) return SnapshotRecord.objects.create( host=host, kind=SnapshotRecord.Kind.SCHEDULED, dirname=dirname, path=f"/backups/{host.host}/scheduled/{dirname}", status="success", started_at=started_at, )