Plan Django retention from snapshot records

src/pobsync_backend/tests/test_sql_retention.py

@@ -0,0 +1,149 @@
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from io import StringIO
+from pathlib import Path
+from tempfile import TemporaryDirectory
+
+from django.core.management import call_command
+from django.test import TestCase
+
+from pobsync.errors import ConfigError
+from pobsync_backend.models import HostConfig, SnapshotRecord
+from pobsync_backend.retention import run_sql_retention_apply, run_sql_retention_plan
+
+
+class SqlRetentionTests(TestCase):
+    def test_plan_uses_snapshot_records(self) -> None:
+        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 = self._snapshot(host, "20260518-021500Z__OLD")
+        new = self._snapshot(host, "20260519-021500Z__NEW")
+
+        plan = run_sql_retention_plan(host=host.host, kind="scheduled", protect_bases=False)
+
+        self.assertEqual(plan["source"], "sql")
+        self.assertEqual(plan["keep"], [new.dirname])
+        self.assertEqual([item["dirname"] for item in plan["delete"]], [old.dirname])
+
+    def test_plan_can_protect_base_snapshot_from_sql_relation(self) -> None:
+        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__BASE")
+        child = self._snapshot(host, "20260519-021500Z__CHILD", base=base)
+
+        plan = run_sql_retention_plan(host=host.host, kind="scheduled", protect_bases=True)
+
+        self.assertEqual(plan["keep"], [base.dirname, child.dirname])
+        self.assertEqual(plan["delete"], [])
+        self.assertEqual(plan["reasons"][base.dirname], [f"base-of:{child.dirname}"])
+
+    def test_apply_deletes_snapshot_directory_and_record(self) -> None:
+        with TemporaryDirectory() as tmp:
+            prefix = Path(tmp) / "home"
+            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_dir = Path(tmp) / "backups" / host.host / "scheduled" / "20260518-021500Z__OLD"
+            new_dir = Path(tmp) / "backups" / host.host / "scheduled" / "20260519-021500Z__NEW"
+            old_dir.mkdir(parents=True)
+            new_dir.mkdir(parents=True)
+            old = self._snapshot(host, old_dir.name, path=str(old_dir))
+            new = self._snapshot(host, new_dir.name, path=str(new_dir))
+
+            result = run_sql_retention_apply(
+                prefix=prefix,
+                host=host.host,
+                kind="scheduled",
+                protect_bases=False,
+                yes=True,
+                max_delete=1,
+                acquire_lock=False,
+            )
+
+            self.assertFalse(old_dir.exists())
+            self.assertTrue(new_dir.exists())
+            self.assertTrue(SnapshotRecord.objects.filter(pk=new.pk).exists())
+            self.assertFalse(SnapshotRecord.objects.filter(pk=old.pk).exists())
+            self.assertEqual(result["deleted"], [{"dirname": old.dirname, "kind": "scheduled", "path": str(old_dir)}])
+
+    def test_apply_respects_max_delete(self) -> None:
+        host = HostConfig.objects.create(
+            host="web-01",
+            address="web-01.example.test",
+            retention_daily=0,
+            retention_weekly=0,
+            retention_monthly=0,
+            retention_yearly=0,
+        )
+        self._snapshot(host, "20260517-021500Z__OLDER")
+        self._snapshot(host, "20260518-021500Z__OLD")
+        self._snapshot(host, "20260519-021500Z__NEW")
+
+        with self.assertRaisesRegex(ConfigError, "exceeds --max-delete=1"):
+            run_sql_retention_apply(
+                prefix=Path("/tmp/pobsync-test"),
+                host=host.host,
+                kind="scheduled",
+                protect_bases=False,
+                yes=True,
+                max_delete=1,
+                acquire_lock=False,
+            )
+
+    def test_management_command_plans_from_sql(self) -> None:
+        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 = self._snapshot(host, "20260518-021500Z__OLD")
+        new = self._snapshot(host, "20260519-021500Z__NEW")
+        stdout = StringIO()
+
+        call_command("run_pobsync_retention", host.host, stdout=stdout)
+
+        result = json.loads(stdout.getvalue())
+        self.assertEqual(result["source"], "sql")
+        self.assertEqual(result["keep"], [new.dirname])
+        self.assertEqual([item["dirname"] for item in result["delete"]], [old.dirname])
+
+    def _snapshot(
+        self,
+        host: HostConfig,
+        dirname: str,
+        *,
+        path: str | None = None,
+        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="scheduled",
+            dirname=dirname,
+            path=path or f"/backups/{host.host}/scheduled/{dirname}",
+            base=base,
+            status="success",
+            started_at=started_at,
+        )