from __future__ import annotations import json import stat 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["keep_items"]], [new.dirname]) self.assertEqual([item["dirname"] for item in plan["delete"]], [old.dirname]) self.assertEqual(plan["delete"][0]["reason"], "outside retention policy") self.assertEqual(plan["incomplete"], []) 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_deletes_snapshot_with_readonly_data_directory(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" old_data = old_dir / "data" old_data.mkdir(parents=True) old_data.joinpath("etc").mkdir() old_data.joinpath("etc", "config").write_text("preserved permissions\n") old_data.chmod(stat.S_IREAD | stat.S_IEXEC | stat.S_IRGRP | stat.S_IXGRP | stat.S_IROTH | stat.S_IXOTH) new_dir = Path(tmp) / "backups" / host.host / "scheduled" / "20260519-021500Z__NEW" new_dir.mkdir(parents=True) old = self._snapshot(host, old_dir.name, path=str(old_data)) 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.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, )