mission-control-v2/backend/tests/test_deploy_units.py

"""Units, die deploy.sh ausspielt (24.09.2026): Probe-Wiederherstellung und PBS-Sicherung.

Prüft die Listen in deploy.sh gegen die Dateien in deploy/, den Takt der Probe (nie sonntags) und das PBS-Skript mit
einer Attrappe statt proxmox-backup-client — ohne Box, ohne PBS."""

import os
import re
import shutil
import subprocess
import sys
from pathlib import Path

import pytest

DEPLOY = Path(__file__).resolve().parents[2] / "deploy"
DEPLOY_SH = (DEPLOY / "deploy.sh").read_text(encoding="utf-8")
BASH = shutil.which("bash")


def _liste(name: str) -> list[str]:
    m = re.search(rf'^{name}="([^"]*)"', DEPLOY_SH, re.MULTILINE)
    assert m, f"{name} fehlt in deploy.sh"
    return m.group(1).split()


def test_jede_unit_aus_deploy_sh_liegt_im_repo():
    units, aktiv = _liste("UNITS"), _liste("AKTIV")
    assert [u for u in units if not (DEPLOY / u).is_file()] == []
    assert set(aktiv) <= set(units)


def test_probe_timer_laeuft_mit_pbs_nur_von_hand():
    units, aktiv = _liste("UNITS"), _liste("AKTIV")
    assert {"mc2-probe-wiederherstellung.service", "mc2-probe-wiederherstellung.timer",
            "pbs-backup.service", "pbs-backup.timer"} <= set(units)
    assert "mc2-probe-wiederherstellung.timer" in aktiv
    assert "pbs-backup.timer" not in aktiv          # Einschalten ist ein Schritt des Leads (docs/BETRIEB.md)
    start = re.search(r"systemctl --user start (mc2-radar\.timer(?:[^\n]*\\\n)*[^\n]*)", DEPLOY_SH)
    assert start and "mc2-probe-wiederherstellung.timer" in start.group(1)


def _timer(name: str) -> dict[str, str]:
    werte = {}
    for zeile in (DEPLOY / name).read_text(encoding="utf-8").splitlines():
        if "=" in zeile and not zeile.startswith("#"):
            k, v = zeile.split("=", 1)
            werte[k.strip()] = v.strip()
    return werte


def test_probe_monatlich_nie_sonntags_und_nicht_um_die_sicherung():
    t = _timer("mc2-probe-wiederherstellung.timer")
    assert t["OnCalendar"] == "Mon *-*-01..07 05:15:00"      # erster Montag im Monat
    assert t["Persistent"] == "true"
    assert _timer("mc2-backup.timer")["OnCalendar"].endswith("03:30:00")
    assert _timer("mc2-autoupdate.timer")["OnCalendar"].startswith("Sun ")


def test_units_und_skripte_haben_lf():
    dateien = [*DEPLOY.glob("*.service"), *DEPLOY.glob("*.timer"), *DEPLOY.glob("*.sh")]
    assert [p.name for p in dateien if b"\r" in p.read_bytes()] == []


# --- pbs-backup.sh mit Attrappe --------------------------------------------------------------------------

# Unter Windows kann „bash“ die WSL-Hülle sein (System32 oder WindowsApps), die Windows-Pfade nicht versteht.
pytestmark_bash = pytest.mark.skipif(
    BASH is None or (sys.platform == "win32" and any(s in BASH.lower() for s in ("system32", "windowsapps"))),
    reason="braucht eine bash (Git-Bash oder Linux)",
)


@pytest.fixture
def pbs(tmp_path):
    (tmp_path / "hermes").mkdir()
    (tmp_path / "lswap").mkdir()
    env = tmp_path / "pbs.env"
    env.write_text("PBS_REPOSITORY=aibox@pbs@192.0.2.1:8007:qnap\nPBS_PASSWORD=nur-ein-test\n", encoding="utf-8")
    client = tmp_path / "client"
    client.write_text('#!/usr/bin/env bash\nprintf "%s\\n" "$@" > "$STUB_ARGS"\n'
                      'printf "%s\\n" "${PBS_REPOSITORY:-}" > "$STUB_ENV"\n', encoding="utf-8", newline="\n")
    client.chmod(0o755)
    umgebung = {**os.environ, "MC_PBS_ENV": env.as_posix(), "MC_PBS_CLIENT": client.as_posix(),
                "HERMES_HOME": (tmp_path / "hermes").as_posix(), "MC_PBS_LLAMASWAP_DIR": (tmp_path / "lswap").as_posix(),
                "MC_PBS_VAULT_DIR": (tmp_path / "vault").as_posix(),
                "STUB_ARGS": (tmp_path / "args.txt").as_posix(), "STUB_ENV": (tmp_path / "env.txt").as_posix()}
    return tmp_path, umgebung


def _pbs(umgebung: dict) -> subprocess.CompletedProcess:
    return subprocess.run([BASH, (DEPLOY / "pbs-backup.sh").as_posix()], env=umgebung,
                          capture_output=True, text=True, encoding="utf-8", timeout=60)


@pytestmark_bash
def test_pbs_sichert_ohne_mem0_und_ueberspringt_fehlende_sammlung(pbs):
    tmp, umgebung = pbs
    r = _pbs(umgebung)
    assert r.returncode == 0, r.stdout + r.stderr
    args = (tmp / "args.txt").read_text(encoding="utf-8").splitlines()
    assert args[0] == "backup" and args[-2:] == ["--backup-id", "aibox"]
    assert [a.split(":", 1)[0] for a in args[1:-2]] == ["hermes.pxar", "llamaswap.pxar"]   # kein mem0.pxar mehr
    assert "wird übersprungen" in r.stdout
    assert (tmp / "env.txt").read_text(encoding="utf-8").strip() == "aibox@pbs@192.0.2.1:8007:qnap"

    (tmp / "vault").mkdir()
    assert _pbs(umgebung).returncode == 0
    assert "vault.pxar" in (tmp / "args.txt").read_text(encoding="utf-8")


@pytestmark_bash
def test_pbs_bricht_ohne_zugang_oder_pflichtordner_ab(pbs):
    tmp, umgebung = pbs
    r = _pbs({**umgebung, "MC_PBS_ENV": (tmp / "fehlt.env").as_posix()})
    assert r.returncode == 1 and "! Zugangsdatei fehlt" in r.stdout
    r = _pbs({**umgebung, "HERMES_HOME": (tmp / "weg").as_posix()})
    assert r.returncode == 1 and "! Ordner fehlt" in r.stdout
    assert not (tmp / "args.txt").exists()