from __future__ import annotations from datetime import datetime, timezone from django.contrib.auth import get_user_model from django.test import TestCase from django.utils import timezone as django_timezone from pobsync_backend.models import BackupRun, HostConfig, ScheduleConfig, SnapshotRecord class ApiTests(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_api_requires_staff_login(self) -> None: response = self.client.get("/api/hosts/") self.assertEqual(response.status_code, 302) self.assertIn("/admin/login/", response["Location"]) def test_hosts_endpoint_returns_counts_and_schedule(self) -> None: self.client.force_login(self.staff_user) host = HostConfig.objects.create(host="web-01", address="web-01.example.test") ScheduleConfig.objects.create(host=host, cron_expr="15 2 * * *", enabled=True, prune=True) snapshot = self._snapshot(host, "20260519-021500Z__ABCDEFGH") BackupRun.objects.create(host=host, status=BackupRun.Status.SUCCESS, snapshot=snapshot) response = self.client.get("/api/hosts/") self.assertEqual(response.status_code, 200) payload = response.json() self.assertTrue(payload["ok"]) self.assertEqual(payload["hosts"][0]["host"], "web-01") self.assertEqual(payload["hosts"][0]["snapshot_count"], 1) self.assertEqual(payload["hosts"][0]["run_count"], 1) self.assertEqual(payload["hosts"][0]["schedule"]["cron_expr"], "15 2 * * *") self.assertTrue(payload["hosts"][0]["schedule"]["prune"]) def test_snapshots_endpoint_filters_and_returns_base_payload(self) -> None: self.client.force_login(self.staff_user) host = HostConfig.objects.create(host="web-01", address="web-01.example.test") other_host = HostConfig.objects.create(host="db-01", address="db-01.example.test") base = self._snapshot(host, "20260518-021500Z__BASESNAP") self._snapshot(other_host, "20260519-021500Z__OTHERSNP") child = self._snapshot(host, "20260519-021500Z__CHILDSNP", base=base) response = self.client.get("/api/snapshots/", {"host": host.host, "kind": "scheduled"}) self.assertEqual(response.status_code, 200) snapshots = response.json()["snapshots"] self.assertEqual([snapshot["dirname"] for snapshot in snapshots], [child.dirname, base.dirname]) self.assertEqual(snapshots[0]["base"]["dirname"], base.dirname) self.assertTrue(snapshots[0]["base"]["resolved"]) def test_runs_endpoint_filters_by_status_and_limit(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.FAILED) BackupRun.objects.create(host=host, status=BackupRun.Status.SUCCESS, snapshot=snapshot) response = self.client.get("/api/runs/", {"host": host.host, "status": "success", "limit": "1"}) self.assertEqual(response.status_code, 200) runs = response.json()["runs"] self.assertEqual(len(runs), 1) self.assertEqual(runs[0]["status"], "success") self.assertEqual(runs[0]["snapshot"]["dirname"], snapshot.dirname) def test_api_index_lists_endpoints(self) -> None: self.client.force_login(self.staff_user) response = self.client.get("/api/") self.assertEqual(response.status_code, 200) endpoints = response.json()["endpoints"] self.assertEqual(endpoints["status"], "http://testserver/api/status/") self.assertEqual(endpoints["hosts"], "http://testserver/api/hosts/") self.assertEqual(endpoints["snapshots"], "http://testserver/api/snapshots/") self.assertEqual(endpoints["runs"], "http://testserver/api/runs/") def test_status_endpoint_returns_counts_and_latest_activity(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, last_due_key="202605190215", last_status="success", last_started_at=django_timezone.now(), ) 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("/api/status/") self.assertEqual(response.status_code, 200) payload = response.json() self.assertTrue(payload["ok"]) self.assertEqual(payload["database"]["vendor"], "sqlite") self.assertEqual(payload["counts"]["hosts"], 1) self.assertEqual(payload["counts"]["enabled_hosts"], 1) self.assertEqual(payload["counts"]["enabled_schedules"], 1) self.assertEqual(payload["counts"]["snapshots"], 1) self.assertEqual(payload["counts"]["runs"], 1) self.assertEqual(payload["latest_run"]["host"], host.host) self.assertEqual(payload["latest_run"]["snapshot"]["dirname"], snapshot.dirname) self.assertEqual(payload["latest_schedule"]["host"], host.host) self.assertEqual(payload["latest_schedule"]["last_due_key"], schedule.last_due_key) def _snapshot( self, host: HostConfig, dirname: str, *, base: SnapshotRecord | None = None, ) -> 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, base=base, )