"""Rückfragen an den Commander (Ausbauplan Welle 3, Punkt 15, 25.09.2026). User-Entscheid: Ein „ja“ in Telegram löst genau die gestellte Frage aus — auch den Neustart des Proxmox-Hosts. Darum nur Knöpfe der Oberfläche, ausgeführt wird die gespeicherte Aktion, einmal, höchstens 24 Stunden lang.""" import json import httpx import pytest from fastapi import FastAPI from fastapi.testclient import TestClient from kern import einstellungen as einstellungen_mod from services import rueckfragen T0 = 1_790_000_000.0 class Antwort: def __init__(self, status: int, daten: object): self.status_code, self._daten = status, daten self.content = json.dumps(daten).encode() def json(self): return self._daten @pytest.fixture def umgebung(tmp_path, monkeypatch): """Eigener Datenordner, mitgeschriebene Meldungen und ein nachgestelltes MC2 für das Ausführen.""" monkeypatch.setenv("MC_DATEN_DIR", str(tmp_path)) einstellungen_mod.einstellungen.cache_clear() gemeldet: list[dict] = [] ausgefuehrt: list[tuple[str, dict]] = [] antwort = {"status": 200, "daten": {"ok": True, "lauf": "l1"}} hinweise = {"stand": T0, "aktiv": True, "hinweise": []} monkeypatch.setattr(rueckfragen.announce, "add", lambda text, **k: gemeldet.append({"text": text, **k})) monkeypatch.setattr(rueckfragen.announce, "notify_telegram", lambda betreff, text, dringend=False, knoepfe=None: gemeldet.append( {"betreff": betreff, "text": text, "dringend": dringend, "knoepfe": knoepfe})) def post(url, json=None, timeout=None): ausgefuehrt.append((url.removeprefix(rueckfragen.EIGENE_URL), json)) return Antwort(antwort["status"], antwort["daten"]) def get(url, timeout=None): return Antwort(200, hinweise) monkeypatch.setattr(rueckfragen.httpx, "post", post) monkeypatch.setattr(rueckfragen.httpx, "get", get) yield {"gemeldet": gemeldet, "ausgefuehrt": ausgefuehrt, "antwort": antwort, "hinweise": hinweise, "pfad": tmp_path / "lucy-rueckfragen.json"} einstellungen_mod.einstellungen.cache_clear() def _telegram(u) -> list[dict]: return [m for m in u["gemeldet"] if "betreff" in m] @pytest.mark.parametrize("pfad", [ "/api/homelab/snapshots/rueckwege-loeschen", "/api/homelab/sicherungen/ct-104/x.vma.zst/loeschen", "/api/maintenance/geheimnisse", "/api/homelab/einstellungen/arcane-echt", "/api/boxwart/modelle/aufraeumen/loeschen", "/api/homelab/snapshots/ct-104/vor-dem-umbau/zuruecksetzen", "/api/homelab/ziele/ct-104/update/../../alle", "", ]) def test_nur_knoepfe_der_oberflaeche(umgebung, pfad): ergebnis = rueckfragen.stellen("Soll ich?", pfad, jetzt=T0) assert not ergebnis["ok"] and "nicht per Rückfrage" in ergebnis["detail"] assert not umgebung["pfad"].exists() def test_stellen_meldet_mit_antworttasten(umgebung): r = rueckfragen.stellen("Soll ich Gitea 1.27.3 einspielen?", "/api/homelab/ziele/ct-104/update", text="Snapshot vorher, Prüfung danach.", von="wochenbericht", jetzt=T0) assert r["ok"] and r["neu"] and r["rueckfrage"]["nr"] == 1 assert r["rueckfrage"]["pfad"] == "/api/partner/homelab/ziele/ct-104/update" # von der Box aus gesehen assert _telegram(umgebung) == [{"betreff": "[Rückfrage]", "dringend": False, "knoepfe": ["ja 1", "nein 1"], "text": "Soll ich Gitea 1.27.3 einspielen? Snapshot vorher, Prüfung danach.\n" "Antworte „ja 1“ oder „nein 1“."}] # Dieselbe Aktion noch einmal: keine zweite Frage, dieselbe Nummer. zweimal = rueckfragen.stellen("Soll ich Gitea einspielen?", "/api/homelab/ziele/ct-104/update", jetzt=T0 + 60) assert zweimal["ok"] and zweimal["neu"] is False and zweimal["rueckfrage"]["nr"] == 1 assert len(_telegram(umgebung)) == 1 anders = rueckfragen.stellen("Host neu starten?", "/api/homelab/ziele/pve/neustart", stufe="dringend", jetzt=T0) assert anders["rueckfrage"]["nr"] == 2 and _telegram(umgebung)[-1]["dringend"] is True def test_still_und_leise_schicken_nichts(umgebung): assert rueckfragen.stellen("A?", "/api/maintenance/check-updates", still=True, jetzt=T0)["ok"] assert rueckfragen.stellen("B?", "/api/radar/suche", stufe="leise", jetzt=T0)["ok"] assert umgebung["gemeldet"] == [] assert [f["nr"] for f in rueckfragen.liste(jetzt=T0)["offen"]] == [2, 1] # neueste zuerst def test_ja_fuehrt_genau_die_gespeicherte_aktion_aus(umgebung): rueckfragen.stellen("Soll ich den Motor neu starten?", "/api/maintenance/restart", {"service": "llama-swap"}, jetzt=T0) ergebnis = rueckfragen.beantworten(1, True, wer="lucy", jetzt=T0 + 30) assert ergebnis == {"ok": True, "text": "Rückfrage 1 ausgeführt: Gestartet; das Ergebnis kommt als Meldung."} assert umgebung["ausgefuehrt"] == [("/api/maintenance/restart", {"service": "llama-swap"})] f = rueckfragen.liste(jetzt=T0 + 30)["zuletzt"][0] assert f["status"] == "erledigt" and f["antwort"] == {"zeit": T0 + 30, "wer": "lucy", "ja": True} # Ein zweites Ja tut nichts mehr. zweites = rueckfragen.beantworten(1, True, jetzt=T0 + 40) assert not zweites["ok"] and "schon erledigt" in zweites["detail"] and len(umgebung["ausgefuehrt"]) == 1 def test_nein_und_unbekannte_nummer(umgebung): rueckfragen.stellen("Soll ich suchen?", "/api/radar/suche", jetzt=T0) assert rueckfragen.beantworten(1, False, jetzt=T0)["text"].startswith("Rückfrage 1 abgelehnt") assert umgebung["ausgefuehrt"] == [] assert "gibt es nicht" in rueckfragen.beantworten(9, True, jetzt=T0)["detail"] def test_ja_ohne_nummer(umgebung): assert rueckfragen.beantworten(None, True, jetzt=T0)["detail"] == "Es ist keine Rückfrage offen." rueckfragen.stellen("Soll ich suchen?", "/api/radar/suche", jetzt=T0) rueckfragen.stellen("Soll ich testen?", "/api/radar/qwen4:q4/testen", jetzt=T0) mehrere = rueckfragen.beantworten(None, True, jetzt=T0) assert not mehrere["ok"] and "1: Soll ich suchen?" in mehrere["detail"] and "2: Soll ich testen?" in mehrere["detail"] rueckfragen.beantworten(1, False, jetzt=T0) assert rueckfragen.beantworten(None, True, jetzt=T0)["ok"] assert umgebung["ausgefuehrt"] == [("/api/radar/qwen4:q4/testen", {})] def test_nach_24_stunden_abgelaufen(umgebung): rueckfragen.stellen("Host neu starten?", "/api/homelab/ziele/pve/neustart", jetzt=T0) ergebnis = rueckfragen.beantworten(1, True, jetzt=T0 + 24 * 3600 + 1) assert not ergebnis["ok"] and "abgelaufen" in ergebnis["detail"] and umgebung["ausgefuehrt"] == [] def test_abgelehnte_aktion(umgebung): umgebung["antwort"].update(status=409, daten={"detail": "Es läuft gerade ‚Alle aktualisieren‘."}) rueckfragen.stellen("Gitea?", "/api/homelab/ziele/ct-104/update", jetzt=T0) ergebnis = rueckfragen.beantworten(1, True, jetzt=T0) assert ergebnis == {"ok": False, "detail": "Rückfrage 1: Das ging nicht — Es läuft gerade ‚Alle aktualisieren‘."} assert rueckfragen.liste(jetzt=T0)["zuletzt"][0]["status"] == "fehler" def test_hinweis_schon_weg_heisst_hinfaellig(umgebung): befund = {"bereich": "box", "id": "dienst:lucy-stimme", "titel": "Lucys Stimme ist gestoppt"} rueckfragen.stellen("Starten?", "/api/hinweise/dienst:lucy-stimme/aktion/neustart", befund=befund, still=True, jetzt=T0) ergebnis = rueckfragen.beantworten(1, True, jetzt=T0) assert ergebnis["ok"] and "hat sich erledigt" in ergebnis["text"] and umgebung["ausgefuehrt"] == [] assert rueckfragen.liste(jetzt=T0)["zuletzt"][0]["status"] == "hinfaellig" def test_befund_besteht(umgebung): befund = {"bereich": "homelab", "id": "gast:104"} h = umgebung["hinweise"] assert rueckfragen.befund_besteht(befund) is False h["hinweise"] = [{"id": "gast:104", "zuletzt": T0}] assert rueckfragen.befund_besteht(befund) is True h["hinweise"] = [{"id": "gast:104", "zuletzt": T0 - 60}] # im letzten Takt nicht mehr gesehen: im Puffer assert rueckfragen.befund_besteht(befund) is False h["aktiv"] = False assert rueckfragen.befund_besteht(befund) is None def test_nachpruefen(umgebung, monkeypatch): monkeypatch.setattr(rueckfragen.time, "sleep", lambda s: None) befund = {"bereich": "box", "id": "dienst:x", "titel": "X ist abgestürzt"} rueckfragen.stellen("Neu starten?", "/api/hinweise/dienst:x/aktion/neustart", befund=befund, still=True, jetzt=T0) r = rueckfragen.liste(jetzt=T0)["offen"][0] umgebung["hinweise"]["hinweise"] = [{"id": "dienst:x", "zuletzt": T0}] rueckfragen._nachpruefen(r, dauer_s=0.05, takt_s=0.01) assert _telegram(umgebung)[-1]["text"].startswith("Nachgeprüft: „X ist abgestürzt“ steht 0 Minuten nach Rückfrage 1") umgebung["gemeldet"].clear() umgebung["hinweise"]["hinweise"] = [] rueckfragen._nachpruefen(r, dauer_s=5, takt_s=0.01) assert umgebung["gemeldet"] == [] assert json.loads(umgebung["pfad"].read_text(encoding="utf-8"))["fragen"][0]["nachgeprueft"] == "behoben" def test_gesammelt_haelt_meldungen_zurueck(umgebung): with rueckfragen.gesammelt() as liste: rueckfragen.stellen("Suchen?", "/api/radar/suche", jetzt=T0) rueckfragen.melden("[Rückfrage]", "dringend", dringend=True) assert [m["text"] for m in liste] == ["Suchen?\nAntworte „ja 1“ oder „nein 1“."] assert liste[0]["knoepfe"] == ["ja 1", "nein 1"] assert [m["text"] for m in _telegram(umgebung)] == ["dringend"] rueckfragen.melden("[Rückfrage]", "danach") assert _telegram(umgebung)[-1]["text"] == "danach" def test_mc2_nicht_erreichbar(umgebung, monkeypatch): def kaputt(*a, **k): raise httpx.ConnectError("weg") monkeypatch.setattr(rueckfragen.httpx, "post", kaputt) rueckfragen.stellen("Suchen?", "/api/radar/suche", jetzt=T0) ergebnis = rueckfragen.beantworten(1, True, jetzt=T0) assert not ergebnis["ok"] and "antwortete nicht (ConnectError)" in ergebnis["detail"] def test_schnittstelle(umgebung): from routers import lucy app = FastAPI() app.include_router(lucy.router) c = TestClient(app) r = c.post("/api/rueckfragen", json={"frage": "Soll ich suchen?", "pfad": "/api/radar/suche", "still": True}) assert r.status_code == 200 and r.json()["rueckfrage"]["nr"] == 1 assert c.post("/api/rueckfragen", json={"frage": "?", "pfad": "/api/maintenance/geheimnisse"}).status_code == 409 assert [f["nr"] for f in c.get("/api/rueckfragen").json()["offen"]] == [1] assert c.post("/api/rueckfragen/antwort", json={"nr": "eins", "ja": True}).status_code == 422 assert c.post("/api/rueckfragen/1/nein").json()["ok"] assert c.post("/api/rueckfragen/1/ja").status_code == 409 c.post("/api/rueckfragen", json={"frage": "Soll ich testen?", "pfad": "/api/radar/x/testen", "still": True}) assert c.post("/api/rueckfragen/antwort", json={"nr": None, "ja": True}).json()["ok"] assert "schluessel" not in c.get("/api/rueckfragen").json()["zuletzt"][0]