"""Lucy untersucht rote Hinweise (Ausbauplan Welle 3, Punkt 14, 25.09.2026): Der Wächter meldet den Hinweis, bekommt die Antwortzeile für seinen Knopf zurück, und je nach Freiheit „ermitteln“ sieht Lucy selbst nach, fragt vorher oder bleibt still. Rückfragen, die sie beim Nachsehen stellt, hängen an ihrer einen Nachricht.""" import pytest from kern import einstellungen as einstellungen_mod from services import ermittlung, lucy_autonomie, rueckfragen T0 = 1_790_000_000.0 UNTERSUCHEN = ermittlung._untersuchen # das echte, bevor die Vorrichtung es ersetzt GESTOPPT = {"bereich": "box", "id": "dienst:lucy-stimme", "titel": "Lucys Stimme ist gestoppt", "stufe": "rot", "text": "Die Unit lucy-stimme läuft nicht.", "aktionen": [{"id": "neustart", "label": "Starten", "dienst": "lucy-stimme"}, {"id": "protokoll", "label": "Protokoll", "dienst": "lucy-stimme"}]} GAST = {"bereich": "homelab", "id": "gast:104", "titel": "Gitea antwortet nicht", "stufe": "rot", "text": "Seit 12 Minuten keine Antwort auf Port 3000.", "aktionen": []} @pytest.fixture def umgebung(tmp_path, monkeypatch): monkeypatch.setenv("MC_DATEN_DIR", str(tmp_path)) einstellungen_mod.einstellungen.cache_clear() gemeldet: list[dict] = [] gestartet: list[dict] = [] monkeypatch.setattr(rueckfragen.announce, "add", lambda text, **k: None) monkeypatch.setattr(rueckfragen.announce, "notify_telegram", lambda betreff, text, dringend=False, knoepfe=None: gemeldet.append( {"betreff": betreff, "text": text, "knoepfe": knoepfe})) monkeypatch.setattr(ermittlung, "_zuletzt", {}) monkeypatch.setattr(ermittlung, "_heute", []) monkeypatch.setattr(ermittlung, "HERMES_API_KEY", "schluessel") monkeypatch.setattr(ermittlung, "_untersuchen", lambda befund: gestartet.append(befund)) # Den Faden sofort laufen lassen, damit die Tests nicht warten müssen. monkeypatch.setattr(ermittlung.threading, "Thread", lambda target, args, name=None, daemon=None: type("F", (), {"start": lambda s: target(*args)})()) yield {"gemeldet": gemeldet, "gestartet": gestartet} einstellungen_mod.einstellungen.cache_clear() def test_knopf_wird_rueckfrage_und_lucy_sieht_nach(umgebung): antwort = ermittlung.befund_gemeldet(GESTOPPT, jetzt=T0) assert antwort == {"ok": True, "zeilen": ["Antworte „ja 1“, dann starte ich lucy-stimme.", "Ich sehe nach und melde mich mit Ursache und Weg."], "knoepfe": ["ja 1", "nein 1"]} frage = rueckfragen.liste(jetzt=T0)["offen"][0] assert frage["pfad"] == "/api/hinweise/dienst:lucy-stimme/aktion/neustart" and frage["von"] == "waechter" assert frage["befund"] == {"bereich": "box", "id": "dienst:lucy-stimme", "titel": "Lucys Stimme ist gestoppt"} assert umgebung["gestartet"] == [{"bereich": "box", "id": "dienst:lucy-stimme", "titel": "Lucys Stimme ist gestoppt", "text": "Die Unit lucy-stimme läuft nicht."}] assert umgebung["gemeldet"] == [] # die Zeilen schickt der Wächter mit seiner eigenen Nachricht def test_erinnerung_bietet_nur_den_knopf_wieder_an(umgebung): ermittlung.befund_gemeldet(GESTOPPT, jetzt=T0) antwort = ermittlung.befund_gemeldet({**GESTOPPT, "erinnerung": True}, jetzt=T0 + 6 * 3600) assert antwort["zeilen"] == ["Antworte „ja 1“, dann starte ich lucy-stimme."] and len(umgebung["gestartet"]) == 1 def test_erst_fragen_ohne_knopf(umgebung): lucy_autonomie.setzen({"stufen": {"ermitteln": "fragen"}}) antwort = ermittlung.befund_gemeldet(GAST, jetzt=T0) assert antwort["zeilen"] == ["Antworte „ja 1“, dann sehe ich nach, was dahintersteckt."] frage = rueckfragen.liste(jetzt=T0)["offen"][0] assert frage["pfad"] == "/api/lucy/ermitteln" and frage["daten"]["id"] == "gast:104" assert umgebung["gestartet"] == [] # Mit Knopf reicht die eine Frage — keine zweite zum Nachsehen. assert len(ermittlung.befund_gemeldet(GESTOPPT, jetzt=T0)["zeilen"]) == 1 def test_nur_lesen(umgebung): lucy_autonomie.setzen({"stufen": {"ermitteln": "lesen"}}) assert ermittlung.befund_gemeldet(GAST, jetzt=T0) == {"ok": True, "zeilen": [], "knoepfe": []} assert not ermittlung.befund_gemeldet({"id": "x"})["ok"] def test_bremsen(umgebung, monkeypatch): assert ermittlung.starten(GAST, jetzt=T0)["ok"] zweimal = ermittlung.starten(GAST, jetzt=T0 + 3600) assert not zweimal["ok"] and "vor Kurzem" in zweimal["detail"] assert ermittlung.starten(GAST, jetzt=T0 + 6 * 3600 + 1)["ok"] for i in range(ermittlung.JE_TAG - 2): assert ermittlung.starten({**GAST, "id": f"gast:{200 + i}"}, jetzt=T0 + 7 * 3600)["ok"] genug = ermittlung.starten({**GAST, "id": "gast:999"}, jetzt=T0 + 7 * 3600) assert not genug["ok"] and "8 Hinweise" in genug["detail"] monkeypatch.setattr(ermittlung, "HERMES_API_KEY", "") assert "Zugang" in ermittlung.starten({**GAST, "id": "gast:1"}, jetzt=T0)["detail"] def test_eine_nachricht_mit_allem(umgebung, monkeypatch): """Was Lucy beim Nachsehen auslöst (hier: eine Rückfrage über ihr Werkzeug), hängt an ihrer Nachricht.""" def lucy(text): assert "„Gitea antwortet nicht“" in text and "geraet_logs()" in text rueckfragen.stellen("Soll ich Gitea auf mc2-20260925-031000 zurücksetzen?", "/api/homelab/snapshots/ct-104/mc2-20260925-031000/zuruecksetzen", jetzt=T0) return "Commander, Gitea hängt seit dem Update um 03:10 beim Start der Datenbank." monkeypatch.setattr(ermittlung, "_frag_lucy", lucy) UNTERSUCHEN({"bereich": "homelab", "id": "gast:104", "titel": "Gitea antwortet nicht", "text": "…"}) assert len(umgebung["gemeldet"]) == 1 m = umgebung["gemeldet"][0] assert m["betreff"] == "[Lucy ermittelt]" and m["knoepfe"] == ["ja 1", "nein 1"] assert m["text"] == ("Commander, Gitea hängt seit dem Update um 03:10 beim Start der Datenbank.\n\n" "Soll ich Gitea auf mc2-20260925-031000 zurücksetzen?\nAntworte „ja 1“ oder „nein 1“.") def test_ohne_antwort_ein_fester_satz(umgebung): text, knoepfe = ermittlung.nachricht({"titel": "Gitea antwortet nicht"}, None, []) assert text.startswith("„Gitea antwortet nicht“: Ich konnte gerade nicht nachsehen") and knoepfe == [] @pytest.mark.parametrize("aktion, satz", [ ({"id": "neustart", "label": "Neu starten", "dienst": "llama-swap"}, "starte ich llama-swap neu"), ({"id": "neustart", "label": "Jetzt erneut laufen lassen", "dienst": "mc2-backup"}, "lasse ich mc2-backup jetzt noch einmal laufen"), ({"id": "job-wiederholen", "label": "Erneut ausführen", "job": "abc"}, "führe ich den Job noch einmal aus"), ({"id": "freigeben", "label": "Freigeben", "baustein": "hermes"}, "gebe ich hermes wieder für Updates frei"), ({"id": "protokoll", "label": "Protokoll"}, None), ]) def test_angebot(aktion, satz): assert ermittlung.angebot(aktion) == satz