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_dashboard_links_latest_snapshot_for_each_host(self) -> None: self.client.force_login(self.staff_user) host = HostConfig.objects.create(host="web-01", address="web-01.example.test") old_snapshot = self._snapshot(host, "20260518-021500Z__OLDSNAP") latest_snapshot = self._snapshot(host, "20260519-021500Z__NEWSNAP") response = self.client.get(reverse("dashboard")) self.assertEqual(response.status_code, 200) self.assertContains(response, "Latest Snapshot") self.assertContains(response, latest_snapshot.dirname) self.assertContains(response, reverse("snapshot_detail", args=[latest_snapshot.id])) self.assertNotContains(response, reverse("snapshot_detail", args=[old_snapshot.id])) def test_dashboard_prompts_for_global_config_when_database_is_empty(self) -> None: self.client.force_login(self.staff_user) response = self.client.get(reverse("dashboard")) self.assertEqual(response.status_code, 200) self.assertContains(response, "No default global config exists yet.") self.assertContains(response, reverse("edit_global_config")) self.assertContains(response, "Create global config") def test_dashboard_prompts_for_first_host_after_global_config_exists(self) -> None: self.client.force_login(self.staff_user) GlobalConfig.objects.create(name="default", backup_root="/opt/pobsync/backups") response = self.client.get(reverse("dashboard")) self.assertEqual(response.status_code, 200) self.assertContains(response, "Global config is ready.") self.assertContains(response, reverse("create_host_config")) self.assertContains(response, "Add first host") def test_global_config_form_creates_default_config(self) -> None: self.client.force_login(self.staff_user) response = self.client.post( reverse("edit_global_config"), { "name": "default", "ssh_user": "backup", "ssh_port": "2222", "ssh_options": "StrictHostKeyChecking=no\nBatchMode=yes", "rsync_binary": "rsync", "rsync_args": "-a\n--numeric-ids", "rsync_extra_args": "--delete", "rsync_timeout_seconds": "60", "rsync_bwlimit_kbps": "1000", "default_source_root": "/srv", "default_destination_subdir": "rootfs", "excludes_default": "*.tmp\ncache/", "retention_daily": "7", "retention_weekly": "4", "retention_monthly": "2", "retention_yearly": "1", }, follow=True, ) self.assertRedirects(response, reverse("dashboard")) self.assertContains(response, "Global config saved for default.") config = GlobalConfig.objects.get(name="default") self.assertEqual(config.backup_root, "/backups") self.assertEqual(config.pobsync_home, "/opt/pobsync") self.assertEqual(config.ssh_user, "backup") self.assertEqual(config.ssh_port, 2222) self.assertEqual(config.ssh_options, ["StrictHostKeyChecking=no", "BatchMode=yes"]) self.assertEqual(config.rsync_args, ["-a", "--numeric-ids"]) self.assertEqual(config.rsync_extra_args, ["--delete"]) self.assertEqual(config.excludes_default, ["*.tmp", "cache/"]) self.assertEqual(config.retention_daily, 7) self.assertEqual(config.retention_yearly, 1) def test_global_config_form_renders_static_container_backup_root_on_edit(self) -> None: self.client.force_login(self.staff_user) GlobalConfig.objects.create( name="default", backup_root="/mnt/pobsync/backups", pobsync_home="/custom/legacy/home", ) response = self.client.get(reverse("edit_global_config")) self.assertEqual(response.status_code, 200) self.assertContains(response, "Backup root:") self.assertContains(response, "/backups") self.assertNotContains(response, "/mnt/pobsync/backups") self.assertNotContains(response, "/opt/pobsync/backups") self.assertNotContains(response, "Pobsync home") def test_global_config_form_resets_backup_root_to_static_container_path(self) -> None: self.client.force_login(self.staff_user) GlobalConfig.objects.create( name="default", backup_root="/mnt/pobsync/backups", pobsync_home="/custom/legacy/home", ) response = self.client.post( reverse("edit_global_config"), { "name": "default", "ssh_user": "root", "ssh_port": "22", "ssh_options": "", "rsync_binary": "rsync", "rsync_args": "", "rsync_extra_args": "", "rsync_timeout_seconds": "0", "rsync_bwlimit_kbps": "0", "default_source_root": "/", "default_destination_subdir": "", "excludes_default": "", "retention_daily": "14", "retention_weekly": "8", "retention_monthly": "12", "retention_yearly": "0", }, follow=True, ) self.assertRedirects(response, reverse("dashboard")) config = GlobalConfig.objects.get(name="default") self.assertEqual(config.backup_root, "/backups") self.assertEqual(config.pobsync_home, "/opt/pobsync") def test_create_host_config_form_creates_host(self) -> None: self.client.force_login(self.staff_user) response = self.client.post( reverse("create_host_config"), { "host": "web-01", "address": "web-01.example.test", "enabled": "on", "ssh_user": "backup", "ssh_port": "2222", "source_root": "/srv", "includes": "/srv/www\n/srv/db", "excludes_add": "*.tmp", "excludes_replace": "", "rsync_extra_args": "--numeric-ids", "retention_daily": "7", "retention_weekly": "4", "retention_monthly": "2", "retention_yearly": "1", }, follow=True, ) self.assertRedirects(response, reverse("host_detail", args=["web-01"])) self.assertContains(response, "Host config created for web-01.") host = HostConfig.objects.get(host="web-01") self.assertEqual(host.address, "web-01.example.test") self.assertEqual(host.ssh_user, "backup") self.assertEqual(host.includes, ["/srv/www", "/srv/db"]) self.assertEqual(host.excludes_add, ["*.tmp"]) self.assertEqual(host.rsync_extra_args, ["--numeric-ids"]) self.assertEqual(host.retention_weekly, 4) 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") self.assertContains(response, "Queue Manual Backup") self.assertContains(response, "Snapshot Discovery") 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])) def test_host_detail_renders_discovery_status_for_existing_snapshot_dirs(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") (backup_root / host.host / "scheduled" / "20260519-021500Z__ABCDEFGH").mkdir(parents=True) (backup_root / host.host / ".incomplete" / "20260519-031500Z__BROKEN01").mkdir(parents=True) response = self.client.get(reverse("host_detail", args=[host.host])) self.assertEqual(response.status_code, 200) self.assertContains(response, f"Host root: {backup_root / host.host}") self.assertContains(response, "Found 2 snapshot directories") self.assertContains(response, "scheduled 1") self.assertContains(response, "incomplete 1") 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_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") snapshot = self._snapshot(host, "20260519-021500Z__ABCDEFGH") run = BackupRun.objects.create( host=host, status=BackupRun.Status.SUCCESS, snapshot=snapshot, snapshot_path=snapshot.path, base_path="/backups/web-01/scheduled/base", rsync_exit_code=0, result={"ok": True, "snapshot": snapshot.path}, ) response = self.client.get(reverse("run_detail", args=[run.id])) self.assertEqual(response.status_code, 200) self.assertContains(response, "Run") self.assertContains(response, "web-01") self.assertContains(response, "success") self.assertContains(response, "ABCDEFGH") self.assertContains(response, ""ok": true") self.assertContains(response, reverse("snapshot_detail", args=[snapshot.id])) def test_snapshot_detail_renders_metadata_runs_and_children(self) -> None: self.client.force_login(self.staff_user) host = HostConfig.objects.create(host="web-01", address="web-01.example.test") base = self._snapshot(host, "20260518-021500Z__BASESNAP") base.metadata = {"status": "success", "snapshot_id": "BASESNAP"} base.save(update_fields=["metadata"]) child = self._snapshot(host, "20260519-021500Z__CHILDSNP") child.base = base child.base_dirname = base.dirname child.save(update_fields=["base", "base_dirname"]) run = BackupRun.objects.create(host=host, status=BackupRun.Status.SUCCESS, snapshot=base) response = self.client.get(reverse("snapshot_detail", args=[base.id])) self.assertEqual(response.status_code, 200) self.assertContains(response, base.dirname) self.assertContains(response, "BASESNAP") self.assertContains(response, child.dirname) self.assertContains(response, f"Run {run.id}") self.assertContains(response, reverse("run_detail", args=[run.id])) self.assertContains(response, reverse("snapshot_detail", args=[child.id])) 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_warns_when_host_root_is_missing(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") 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 0 items") self.assertContains(response, "Host backup directory does not exist yet") self.assertFalse(SnapshotRecord.objects.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, )