(feature) Add staff-only JSON inspection API
Expose lightweight Django JSON endpoints for hosts, snapshots, and backup runs using the existing admin/staff authentication boundary. Include filters for snapshot and run inspection, return resolved snapshot base metadata, and document the new /api/ entrypoint. Add endpoint tests for authentication, host summaries, snapshot lineage payloads, and run filtering.
This commit is contained in:
103
src/pobsync_backend/tests/test_api.py
Normal file
103
src/pobsync_backend/tests/test_api.py
Normal file
@@ -0,0 +1,103 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.test import TestCase
|
||||
|
||||
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["hosts"], "http://testserver/api/hosts/")
|
||||
self.assertEqual(endpoints["snapshots"], "http://testserver/api/snapshots/")
|
||||
self.assertEqual(endpoints["runs"], "http://testserver/api/runs/")
|
||||
|
||||
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,
|
||||
)
|
||||
Reference in New Issue
Block a user