pobsync/src/pobsync_backend/tests/test_sql_retention.py

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,
        )