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

"""Wöchentliches Suchen (services/homelab/pflege.py, 24.09.2026): Der Homelab-Teil erneuert alte Paketlisten in den
Gästen selbst — nachts, einmal je Gast und Tag, nie neben einem Update. Mit dem echten Bericht vom 24.09.
(tests/fixtures/pve-bericht.json): AdGuard, NPMplus, NetBird, PVE Scripts Local und Gitea haben alte Listen, die des
PBS sind frisch."""

import json
import os
import subprocess
import sys
import time
from collections import Counter
from datetime import datetime
from pathlib import Path

import pytest
from kern import einstellungen as einstellungen_mod
from kern.zeit import LOCAL_TZ
from services import waechter
from services.homelab import kanal, pflege, updates

BERICHT = json.loads((Path(__file__).parent / "fixtures" / "pve-bericht.json").read_text(encoding="utf-8"))
BACKEND = Path(__file__).resolve().parents[1]
NACHT = datetime(2026, 9, 25, 3, 0, tzinfo=LOCAL_TZ).timestamp()
TAG = datetime(2026, 9, 25, 14, 0, tzinfo=LOCAL_TZ).timestamp()
NAECHSTE_NACHT = datetime(2026, 9, 26, 3, 0, tzinfo=LOCAL_TZ).timestamp()
ALTE_LISTEN = {100, 101, 102, 103, 104}


@pytest.fixture
def uhr(tmp_path, monkeypatch):
    """Daten in tmp; der Ausführer hat zuletzt um uhr["zuletzt"] berichtet."""
    monkeypatch.setenv("MC_DATEN_DIR", str(tmp_path))
    einstellungen_mod.einstellungen.cache_clear()
    stand = {"zuletzt": NACHT}
    monkeypatch.setattr(kanal, "zuletzt", lambda: stand["zuletzt"])
    monkeypatch.setattr(waechter, "TROCKEN", False)
    kanal.bericht_speichern(BERICHT)
    yield stand
    einstellungen_mod.einstellungen.cache_clear()


def _suchen() -> Counter:
    """Wie viele Aufträge „suchen“ es je Gast gibt."""
    return Counter(a["parameter"]["vmid"] for a in kanal.liste() if a["aktion"] == "suchen")


def _ausfuehrer_antwortet(scheitert: set[int] = frozenset()) -> None:
    while (a := kanal.naechster()):
        vmid = a["parameter"].get("vmid")
        if vmid in scheitert:
            kanal.ergebnis(a["id"], 100, "Hit:1 http://deb.debian.org/debian trixie InRelease\n"
                                         "E: Failed to fetch http://security.debian.org  Temporary failure resolving")
        else:
            kanal.ergebnis(a["id"], 0, "Hit:1 http://deb.debian.org/debian trixie InRelease")


def test_nachts_einmal_je_gast_und_tag(uhr):
    assert pflege.pruefe_paketlisten(jetzt=NACHT) == []
    assert set(_suchen()) == ALTE_LISTEN                   # PBS: frische Listen; die Arcane-VM ist kein Container
    pflege.pruefe_paketlisten(jetzt=NACHT + 60)            # nächster Takt: offen, nichts doppelt
    _ausfuehrer_antwortet()
    pflege.pruefe_paketlisten(jetzt=NACHT + 120)           # erledigt, Listen im Bericht noch alt: heute nicht mehr
    assert _suchen() == Counter({v: 1 for v in ALTE_LISTEN})
    zustand = pflege.zustand()
    assert zustand["nacht"] == "2026-09-25"
    assert zustand["gaeste"]["100"]["name"] == "AdGuard Home" and zustand["gaeste"]["100"]["fehlschlaege"] == 0
    uhr["zuletzt"] = NAECHSTE_NACHT                          # am nächsten Tag wieder, falls die Listen alt bleiben
    pflege.pruefe_paketlisten(jetzt=NAECHSTE_NACHT)
    assert _suchen() == Counter({v: 2 for v in ALTE_LISTEN})


def test_tagsueber_nur_wenn_die_nacht_verpasst_wurde(uhr):
    uhr["zuletzt"] = TAG
    pflege._schreiben({"nacht": "2026-09-25", "gaeste": {}})       # die Instanz war letzte Nacht da
    pflege.pruefe_paketlisten(jetzt=TAG)
    assert _suchen() == Counter()                                 # also bis zur nächsten Nacht warten
    pflege._schreiben({"nacht": "2026-09-24", "gaeste": {}})       # letzte Nacht war sie aus
    pflege.pruefe_paketlisten(jetzt=TAG)
    assert set(_suchen()) == ALTE_LISTEN                          # dann eben jetzt
    assert pflege.zustand()["nacht"] == "2026-09-24"              # tagsüber gilt keine Nacht als gesehen


def test_vor_zwei_uhr_zaehlt_die_nacht_von_gestern(uhr):
    kurz_vor_zwei = datetime(2026, 9, 26, 1, 30, tzinfo=LOCAL_TZ).timestamp()
    uhr["zuletzt"] = kurz_vor_zwei
    pflege._schreiben({"nacht": "2026-09-25", "gaeste": {}})
    pflege.pruefe_paketlisten(jetzt=kurz_vor_zwei)
    assert _suchen() == Counter()


def test_nie_neben_einem_update_oder_einem_offenen_auftrag(uhr):
    updates._merken({"id": "l1", "ziel": "ct-104", "baustein": "app", "name": "Gitea", "status": "laeuft",
                     "start": time.time(), "ende": None, "ergebnis": None, "text": None, "schritte": []})
    kanal.anlegen("os_update", {"vmid": 100})
    pflege.pruefe_paketlisten(jetzt=NACHT)
    assert set(_suchen()) == ALTE_LISTEN - {100, 104}


def test_ohne_ausfuehrer_keine_auftraege(uhr):
    uhr["zuletzt"] = NACHT - 3600                                  # letzter Bericht vor einer Stunde
    pflege.pruefe_paketlisten(jetzt=NACHT)
    assert _suchen() == Counter() and pflege.zustand().get("nacht") is None


def test_ausfuehrer_im_nur_lesen_modus_bekommt_keine_auftraege(uhr):
    nur_lesen = json.loads(json.dumps(BERICHT))
    nur_lesen["host"]["nur_lesen"] = True
    kanal.bericht_speichern(nur_lesen)
    pflege.pruefe_paketlisten(jetzt=NACHT)
    assert _suchen() == Counter() and pflege.zustand().get("nacht") is None


def test_trockenlauf_legt_nichts_an(uhr, monkeypatch):
    monkeypatch.setattr(waechter, "TROCKEN", True)
    pflege.pruefe_paketlisten(jetzt=NACHT)
    assert _suchen() == Counter()


def test_wiederholtes_scheitern_wird_ein_gelber_hinweis(uhr):
    pflege.pruefe_paketlisten(jetzt=NACHT)
    _ausfuehrer_antwortet(scheitert={100})
    assert pflege.pruefe_paketlisten(jetzt=NACHT + 60) == []       # einmal ist noch kein Muster
    uhr["zuletzt"] = NAECHSTE_NACHT
    pflege.pruefe_paketlisten(jetzt=NAECHSTE_NACHT)                # neuer Tag, neuer Versuch
    _ausfuehrer_antwortet(scheitert={100})
    befunde = pflege.pruefe_paketlisten(jetzt=NAECHSTE_NACHT + 60)
    assert [(b.id, b.stufe, b.titel, b.sofort) for b in befunde] == [
        ("pflege:ct-100", "gelb", "AdGuard Home: Die Suche nach Updates scheitert", True)]
    assert "2-mal hintereinander" in befunde[0].text and "E: Failed to fetch" in befunde[0].text
    # Listen wieder frisch (etwa von Hand gesucht): Hinweis weg, Zähler vergessen.
    frisch = json.loads(json.dumps(BERICHT))
    next(g for g in frisch["gaeste"] if g["vmid"] == 100)["os_updates"]["listen_stand"] = int(NAECHSTE_NACHT)
    kanal.bericht_speichern(frisch)
    assert pflege.pruefe_paketlisten(jetzt=NAECHSTE_NACHT + 120) == []
    assert pflege.zustand()["gaeste"]["100"]["fehlschlaege"] == 0


def test_der_waechter_zeigt_den_hinweis_ohne_telegram(uhr, tmp_path, monkeypatch):
    pflege._schreiben({"nacht": "2026-09-25", "gaeste": {"100": {
        "name": "AdGuard Home", "tag": "2026-09-25", "auftrag": None, "fehlschlaege": 3, "fehler": "E: kaputt"}}})
    uhr["zuletzt"] = TAG
    monkeypatch.setattr(waechter, "STORE_PATH", tmp_path / "mc2-waechter.json")
    monkeypatch.setattr(waechter, "_stand", {"hinweise": {}, "kandidaten": {}, "verlauf": [], "auto": {}, "stand": 0.0})
    monkeypatch.setattr(waechter, "_update_laeuft", lambda: False)
    meldungen: list[str] = []
    monkeypatch.setattr(waechter, "_telegram", lambda betreff, text: meldungen.append(text))

    def pruefe_paketlisten():
        return pflege.pruefe_paketlisten(jetzt=TAG)

    monkeypatch.setattr(waechter, "PRUEFUNGEN", (pruefe_paketlisten,))
    waechter.takt()
    hinweis = waechter._stand["hinweise"]["pflege:ct-100"]
    assert hinweis["stufe"] == "gelb" and "3-mal" in hinweis["text"] and meldungen == []


@pytest.mark.skipif(os.name != "posix", reason="Die Dateisperre gibt es nur unter Linux (am PC: nur Thread-Sperre).")
def test_auftraege_aus_zwei_prozessen_gehen_nicht_verloren(tmp_path):
    """Oberfläche und Steward schreiben beide in die Auftragsliste; ohne Dateisperre überschrieben sie sich."""
    code = ("from services.homelab import kanal\n"
            "for i in range(40):\n"
            "    kanal.anlegen('suchen', {'vmid': 100 + i})\n")
    env = {**os.environ, "MC_DATEN_DIR": str(tmp_path), "MC_ROLLE": "homelab"}
    prozesse = [subprocess.Popen([sys.executable, "-c", code], cwd=BACKEND, env=env) for _ in range(3)]
    assert all(p.wait(timeout=120) == 0 for p in prozesse)
    auftraege = json.loads((tmp_path / "ausfuehrer-auftraege.json").read_text(encoding="utf-8"))
    assert len(auftraege) == 120