PacketStation/tests/server/test_backup_check.py

"""The backup-check job: the newest dump and the markers under BACKUP_DIR."""

from __future__ import annotations

import os
import time
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

import pytest
from pshelpers import make_settings, needs_pg

from packetstation.settings import Settings
from packetstation.worker import jobs, runs
from packetstation.worker.jobs import backup_check
from packetstation.worker.scheduler import Scheduler

HOUR = 3600


def put(folder: Path, name: str, data: bytes = b"PGDMP dump", hours_ago: float = 1.0) -> Path:
    p = folder / name
    p.write_bytes(data)
    t = time.time() - hours_ago * HOUR
    os.utime(p, (t, t))
    return p


def test_fresh_dump_is_fine(tmp_path: Path) -> None:
    put(tmp_path, "packetstation-test-20261020.dump", hours_ago=30)
    put(tmp_path, "packetstation-test-20261021.dump", data=b"x" * 2048, hours_ago=2)
    put(tmp_path, "last-success", b"", hours_ago=2)
    put(tmp_path, "last-failure", b"", hours_ago=50)
    put(tmp_path, "notes.txt", hours_ago=0)
    counts, problems = backup_check.inspect(tmp_path)
    assert problems == []
    assert counts["dumps"] == 2 and counts["newest"] == "packetstation-test-20261021.dump"
    assert counts["size"] == 2048 and 1.9 <= counts["age_hours"] <= 2.1


@pytest.mark.parametrize(
    "setup,expect",
    [
        (lambda d: None, "No backup dump"),
        (lambda d: put(d, "a.dump", hours_ago=27), "27 hours old"),
        (lambda d: put(d, "a.dump", b"", hours_ago=1), "is empty"),
        (
            lambda d: (put(d, "a.dump"), put(d, "last-success", b"", 5), put(d, "last-failure", b"", 0.5)),
            "last backup run failed",
        ),
        (lambda d: (put(d, "a.dump"), put(d, "last-failure", b"", 0.5)), "last backup run failed"),
    ],
)
def test_problems(tmp_path: Path, setup: Any, expect: str) -> None:
    setup(tmp_path)
    _, problems = backup_check.inspect(tmp_path)
    assert any(expect in p for p in problems), problems


def test_missing_folder(tmp_path: Path) -> None:
    counts, problems = backup_check.inspect(tmp_path / "nope")
    assert counts["dumps"] == 0 and problems and "not there" in problems[0]


def test_age_uses_the_given_clock(tmp_path: Path) -> None:
    p = put(tmp_path, "a.dump", hours_ago=0)
    at = datetime.fromtimestamp(p.stat().st_mtime, UTC)
    _, problems = backup_check.inspect(tmp_path, now=at.replace(year=at.year + 1))
    assert problems and "hours old" in problems[0]


@needs_pg
def test_job_raises_then_resolves(pg_settings: Settings, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    calls: list[tuple[str, str, dict[str, Any]]] = []
    monkeypatch.setattr(jobs, "raise_event", lambda conn, t, **kw: calls.append(("raise", t, kw)))
    monkeypatch.setattr(jobs, "resolve_event", lambda conn, t, **kw: calls.append(("resolve", t, kw)))
    folder = tmp_path / "backups"
    folder.mkdir()
    s = pg_settings.model_copy(update={"backup_dir": folder})
    sch = Scheduler(s, build="pytest", jobs={"backup-check": backup_check.SPEC})

    row = runs.run_job(sch, "backup-check", "schedule")
    assert row["status"] == "ok" and row["counts"]["dumps"] == 0 and row["counts"]["problems"]
    ((kind, etype, kw),) = calls
    assert (kind, etype, kw["subject"], kw["condition"]) == ("raise", "backup_stale", "ci", True)
    assert kw["title"] == "Backup failed or stale (ci)" and "No backup dump" in kw["body"]

    calls.clear()
    put(folder, "ps.dump")
    put(folder, "last-success", b"")
    row = runs.run_job(sch, "backup-check", "schedule")
    assert row["counts"]["problems"] == [] and row["counts"]["newest"] == "ps.dump"
    assert calls == [("resolve", "backup_stale", {"subject": "ci", "title": "Backup failed or stale (ci)"})]


def test_settings_backup_dir_default(tmp_path: Path) -> None:
    assert make_settings(tmp_path).backup_dir == tmp_path / "no-backups"