HitonabiandClaude Opus 5.5
c31ef827fc
User 25.09.: „Hermes' Persona ist Lucy – das muss sich in all ihren Meldungen widerspiegeln“; das angehängte „(Diese Meldung kam auch an Lucy.)“ tat so, als wäre sie jemand anderes als der Absender. - notify.sh: Kopfzeile aus dem Betreff in Lucys Ton (SOUL.md: „Commander“, knapp, sachlich), bei Dringendem „Commander, das ist dringend.“; keine zweite Anrede, wenn der Text schon mit ihr beginnt. Morgenmeldung mit kurzen Marken statt Betreffen in Klammern; die des Homelabs sagt, woher sie kommt. - Wächter: Zusatz „kam auch an Lucy“ entfernt (der Briefkasten bekommt die Meldung weiter). - Melde-Log und Auswertungen unverändert. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
246 lines
11 KiB
Python
246 lines
11 KiB
Python
"""Nachtruhe und Morgenmeldung (deploy/notify.sh, deploy/morgenmeldung.sh, 24.09.2026).
|
|
|
|
Bis zum 24.09. parkte notify.sh nachts alles in einer Warteschlange, die niemand mehr auslieferte.
|
|
Die Tests fahren die echten Skripte mit einem Ersatz für `hermes send`, der nur mitschreibt."""
|
|
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
DEPLOY = Path(__file__).resolve().parents[2] / "deploy"
|
|
BASH = shutil.which("bash")
|
|
|
|
# Unter Windows kann „bash“ die WSL-Hülle sein, die Windows-Pfade nicht versteht — dann überspringen.
|
|
pytestmark = pytest.mark.skipif(
|
|
BASH is None or (sys.platform == "win32" and "system32" in BASH.lower()),
|
|
reason="braucht eine bash (Git-Bash oder Linux)",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def umgebung(tmp_path):
|
|
protokoll = tmp_path / "gesendet.txt"
|
|
stub = tmp_path / "hermes-stub"
|
|
stub.write_text(
|
|
"#!/usr/bin/env bash\n"
|
|
'printf "%s\\n" "$*" >> "$STUB_PROTOKOLL"\n'
|
|
'exit "${STUB_EXIT:-0}"\n',
|
|
encoding="utf-8", newline="\n")
|
|
stub.chmod(0o755)
|
|
env = {
|
|
**os.environ,
|
|
"MC_NOTIFY_LOG": (tmp_path / "notify.log").as_posix(),
|
|
"MC_NIGHT_QUEUE": (tmp_path / "night-queue.txt").as_posix(),
|
|
"MC_NOTIFY_HERMES": stub.as_posix(),
|
|
"MC_NOTIFY_NO_ANNOUNCE": "1",
|
|
"MC_NOTIFY_OHNE_WALL": "1",
|
|
# Auf der Box läge unter ~/.hermes/.env das echte Bot-Token — der Zweitweg darf in Tests
|
|
# nie dorthin greifen, sonst schickt der Deploy-Testlauf echte Telegram-Nachrichten.
|
|
"MC_TELEGRAM_ENV": (tmp_path / "keine.env").as_posix(),
|
|
"MC_TELEGRAM_API": "http://127.0.0.1:9",
|
|
"STUB_PROTOKOLL": protokoll.as_posix(),
|
|
"MC_MORGENMELDUNG_PAUSE": "0",
|
|
}
|
|
return tmp_path, env, protokoll
|
|
|
|
|
|
def _notify(env, *args, stunde="03"):
|
|
return subprocess.run([BASH, (DEPLOY / "notify.sh").as_posix(), *args],
|
|
env={**env, "MC_NOTIFY_STUNDE": stunde}, capture_output=True, text=True, timeout=60)
|
|
|
|
|
|
def _gesendet(protokoll: Path) -> str:
|
|
return protokoll.read_text(encoding="utf-8") if protokoll.exists() else ""
|
|
|
|
|
|
def test_nachts_wird_gesammelt(umgebung):
|
|
tmp, env, protokoll = umgebung
|
|
assert _notify(env, "-s", "[Box-Update]", "Engine aktualisiert").returncode == 0
|
|
assert _gesendet(protokoll) == ""
|
|
# Die Zeile der Morgenmeldung trägt Lucys kurze Marke statt des Betreffs in Klammern (25.09.2026).
|
|
assert (tmp / "night-queue.txt").read_text(encoding="utf-8") == "- Box, Wartung: Engine aktualisiert\n"
|
|
assert "QUEUED für Morgen-Digest: Engine aktualisiert" in (tmp / "notify.log").read_text(encoding="utf-8")
|
|
|
|
|
|
@pytest.mark.parametrize("args", [
|
|
("-d", "-s", "[Box-Update]", "Wichtig"),
|
|
("-s", "[Alarm]", "Lucy antwortet nicht"),
|
|
("-s", "[Box-Update]", "KRITISCH: Update UND Rollback fehlgeschlagen"),
|
|
])
|
|
def test_dringendes_geht_nachts_sofort_raus(umgebung, args):
|
|
tmp, env, protokoll = umgebung
|
|
assert _notify(env, *args).returncode == 0
|
|
assert "send --to telegram" in _gesendet(protokoll)
|
|
assert not (tmp / "night-queue.txt").exists()
|
|
|
|
|
|
@pytest.mark.parametrize("stunde", ["07", "08", "09", "23"])
|
|
def test_tagsueber_sofort(umgebung, stunde):
|
|
tmp, env, protokoll = umgebung
|
|
assert _notify(env, "-s", "[Test]", "Hallo", stunde=stunde).returncode == 0
|
|
assert "Hallo" in _gesendet(protokoll)
|
|
assert not (tmp / "night-queue.txt").exists()
|
|
|
|
|
|
def _morgenmeldung(env, **extra):
|
|
return subprocess.run([BASH, (DEPLOY / "morgenmeldung.sh").as_posix()], env={**env, **extra},
|
|
capture_output=True, text=True, timeout=60)
|
|
|
|
|
|
def test_morgenmeldung_schickt_alles_als_eine_nachricht(umgebung):
|
|
tmp, env, protokoll = umgebung
|
|
_notify(env, "-s", "[Box-Update]", "Router aktualisiert")
|
|
_notify(env, "-s", "[Modell-Radar]", "Kandidat bestanden")
|
|
assert _morgenmeldung(env).returncode == 0
|
|
gesendet = _gesendet(protokoll)
|
|
assert gesendet.count("send --to telegram") == 1
|
|
assert "Guten Morgen, Commander. Heute Nacht gab es 2 Meldungen:" in gesendet
|
|
assert "--subject" not in gesendet # die Anrede steht schon im Text, keine zweite Kopfzeile
|
|
assert "Box, Wartung: Router aktualisiert" in gesendet and "Modell-Radar: Kandidat bestanden" in gesendet
|
|
assert not (tmp / "night-queue.txt").exists()
|
|
assert (tmp / "night-queue.txt.zuletzt-gesendet").exists()
|
|
|
|
|
|
def test_morgenmeldung_ohne_nachtmeldungen_schweigt(umgebung):
|
|
_, env, protokoll = umgebung
|
|
assert _morgenmeldung(env).returncode == 0
|
|
assert _gesendet(protokoll) == ""
|
|
|
|
|
|
def test_morgenmeldung_verliert_nichts_wenn_telegram_klemmt(umgebung):
|
|
tmp, env, protokoll = umgebung
|
|
_notify(env, "-s", "[Box-Update]", "Erste Nacht")
|
|
fehl = _morgenmeldung(env, STUB_EXIT="1", MC_MORGENMELDUNG_VERSUCHE="2")
|
|
assert fehl.returncode == 1
|
|
assert (tmp / "night-queue.txt.senden").exists()
|
|
|
|
_notify(env, "-s", "[Box-Update]", "Zweite Nacht")
|
|
protokoll.unlink()
|
|
assert _morgenmeldung(env).returncode == 0
|
|
gesendet = _gesendet(protokoll)
|
|
assert "2 Meldungen" in gesendet
|
|
assert gesendet.index("Erste Nacht") < gesendet.index("Zweite Nacht")
|
|
assert not (tmp / "night-queue.txt.senden").exists()
|
|
|
|
|
|
# ── Zweitweg direkt an die Bot-API (seit 24.09.2026) ──────────────────────────────────────────────
|
|
|
|
@pytest.fixture
|
|
def bot_api():
|
|
"""Ein Ersatz für api.telegram.org, der mitschreibt, was ankommt."""
|
|
import threading
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from urllib.parse import parse_qs
|
|
|
|
empfangen: list[dict] = []
|
|
antwort = {"code": 200}
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_POST(self):
|
|
laenge = int(self.headers.get("Content-Length", 0))
|
|
felder = parse_qs(self.rfile.read(laenge).decode("utf-8"))
|
|
empfangen.append({"pfad": self.path, **{k: v[0] for k, v in felder.items()}})
|
|
self.send_response(antwort["code"])
|
|
self.end_headers()
|
|
self.wfile.write(b'{"ok": true}')
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
threading.Thread(target=server.serve_forever, daemon=True).start()
|
|
yield f"http://127.0.0.1:{server.server_port}", empfangen, antwort
|
|
server.shutdown()
|
|
|
|
|
|
def _env_datei(tmp: Path, inhalt: str) -> str:
|
|
datei = tmp / "hermes.env"
|
|
datei.write_text(inhalt, encoding="utf-8", newline="\n")
|
|
return datei.as_posix()
|
|
|
|
|
|
def test_zweitweg_springt_ein_wenn_hermes_scheitert(umgebung, bot_api):
|
|
tmp, env, _ = umgebung
|
|
url, empfangen, _ = bot_api
|
|
env = {**env, "STUB_EXIT": "1", "MC_TELEGRAM_API": url, "MC_TELEGRAM_ENV": _env_datei(
|
|
tmp, 'OPENAI_API_KEY=egal\nexport TELEGRAM_BOT_TOKEN="123:abc"\nTELEGRAM_HOME_CHANNEL=4711\n'
|
|
"TELEGRAM_HOME_CHANNEL_THREAD_ID=\n")}
|
|
lauf = _notify(env, "-s", "[Alarm]", "Hermes steht", stunde="12")
|
|
assert lauf.returncode == 0
|
|
assert empfangen == [{"pfad": "/bot123:abc/sendMessage", "chat_id": "4711",
|
|
"text": "Commander, das ist dringend.\nHermes steht"}]
|
|
assert "OK telegram direkt: Hermes steht" in (tmp / "notify.log").read_text(encoding="utf-8")
|
|
|
|
|
|
def test_zweitweg_nimmt_den_thread_mit(umgebung, bot_api):
|
|
tmp, env, _ = umgebung
|
|
url, empfangen, _ = bot_api
|
|
env = {**env, "STUB_EXIT": "1", "MC_TELEGRAM_API": url, "MC_TELEGRAM_ENV": _env_datei(
|
|
tmp, "TELEGRAM_BOT_TOKEN='9:x'\r\nTELEGRAM_HOME_CHANNEL=-100\r\nTELEGRAM_HOME_CHANNEL_THREAD_ID=7\r\n")}
|
|
assert _notify(env, "Ohne Betreff", stunde="12").returncode == 0
|
|
assert empfangen == [{"pfad": "/bot9:x/sendMessage", "chat_id": "-100",
|
|
"text": "Commander, kurz gesagt:\nOhne Betreff",
|
|
"message_thread_id": "7"}]
|
|
|
|
|
|
def test_hermes_zuerst_zweitweg_bleibt_still(umgebung, bot_api):
|
|
tmp, env, protokoll = umgebung
|
|
url, empfangen, _ = bot_api
|
|
env = {**env, "MC_TELEGRAM_API": url,
|
|
"MC_TELEGRAM_ENV": _env_datei(tmp, "TELEGRAM_BOT_TOKEN=1:a\nTELEGRAM_HOME_CHANNEL=1\n")}
|
|
assert _notify(env, "Normal", stunde="12").returncode == 0
|
|
assert "Normal" in _gesendet(protokoll)
|
|
assert empfangen == []
|
|
|
|
|
|
@pytest.mark.parametrize("inhalt, code", [
|
|
("", 200), # keine Zugangsdaten
|
|
("TELEGRAM_BOT_TOKEN=1:a\n", 200), # Chat fehlt
|
|
("TELEGRAM_BOT_TOKEN=1:a\nTELEGRAM_HOME_CHANNEL=1\n", 500), # Telegram lehnt ab
|
|
])
|
|
def test_letzter_ausweg_wenn_auch_der_zweitweg_scheitert(umgebung, bot_api, inhalt, code):
|
|
tmp, env, _ = umgebung
|
|
url, _, antwort = bot_api
|
|
antwort["code"] = code
|
|
env = {**env, "STUB_EXIT": "1", "MC_NOTIFY_STRICT": "1", "MC_TELEGRAM_API": url,
|
|
"MC_TELEGRAM_ENV": _env_datei(tmp, inhalt)}
|
|
lauf = _notify(env, "-s", "[Test]", "Geht nirgends hin", stunde="12")
|
|
assert lauf.returncode == 1
|
|
assert "FALLBACK (telegram fehlgeschlagen" in (tmp / "notify.log").read_text(encoding="utf-8")
|
|
|
|
|
|
|
|
@pytest.mark.parametrize("args, kopf", [
|
|
(("-s", "[Homelab-Problem]", "Platte von AdGuard Home zu 94 % voll."), "Commander, im Homelab stimmt etwas nicht."),
|
|
(("-s", "[Homelab wieder ok]", "Erledigt: Gitea antwortet nicht."), "Commander, Entwarnung im Homelab."),
|
|
(("-s", "[Homelab-Update]", "Gitea: eingespielt, Prüfung grün."), "Commander, Bericht aus dem Homelab."),
|
|
(("-s", "[Box-Problem]", "llama-swap steht."), "Commander, auf der Box stimmt etwas nicht."),
|
|
(("-s", "[Alarm] Homelab-Update", "Gitea: Update gescheitert."), "Commander, das ist dringend."),
|
|
(("-d", "-s", "[Test]", "Der Meldeweg geht."), "Commander, ein Test des Meldewegs."),
|
|
(("-s", "[Modell-Radar]", "Kandidat bestanden."), "Commander, Bericht vom Modell-Radar."),
|
|
])
|
|
def test_lucy_spricht(umgebung, args, kopf):
|
|
"""25.09.2026: Hermes' Persona ist Lucy — jede Meldung kommt von ihrem Bot, also in ihrem Ton (SOUL.md)."""
|
|
_, env, protokoll = umgebung
|
|
assert _notify(env, *args, stunde="12").returncode == 0
|
|
assert f"--subject {kopf} -- {args[-1]}" in _gesendet(protokoll)
|
|
|
|
|
|
def test_lucy_formuliert_selbst_keine_zweite_anrede(umgebung):
|
|
_, env, protokoll = umgebung
|
|
assert _notify(env, "-s", "[Box-Update]", "Commander, die Wochenpflege der Box ist durch.", stunde="12").returncode == 0
|
|
assert _gesendet(protokoll) == "send --to telegram -- Commander, die Wochenpflege der Box ist durch.\n"
|
|
|
|
|
|
def test_morgenmeldung_des_homelabs_sagt_woher(umgebung):
|
|
_, env, protokoll = umgebung
|
|
_notify(env, "-s", "[Homelab-Problem]", "NetBird antwortet nicht.")
|
|
assert _morgenmeldung(env, MC_MORGENMELDUNG_BETREFF="[Morgenmeldung Homelab]").returncode == 0
|
|
gesendet = _gesendet(protokoll)
|
|
assert "Guten Morgen, Commander. Aus dem Homelab kamen heute Nacht 1 Meldung:" in gesendet
|
|
assert "Homelab, Störung: NetBird antwortet nicht." in gesendet
|