phase2a: Kern fuer zwei Instanzen (Rolle, Partner, Telegram-Zweitweg)
- kern/einstellungen.py: MC_ROLLE (box|homelab), Instanzname, Datenordner, Partner-URL - kern/zeit.py: eine Zeitzone statt sechs Kopien - app.py/steward.py: Box-Router und Warm-Set nur in der Rolle box; der Homelab-Teil laedt nichts von der Box - /api/health nennt Rolle und Instanz; Engine/Gateway/Hirn nur auf der Box - /api/partner (Lebenszeichen) und /api/partner/<pfad> (Durchreiche, ohne Schleifen und SSE) - Waechter prueft die andere Instanz (rot erst nach 5 Takten, ein Neustart ist kein Alarm) - notify.sh: Zweitweg direkt an die Telegram-Bot-API, wenn hermes send scheitert oder fehlt; Token nicht in der Prozessliste, Tests greifen nie auf die echte .env - update_verlauf erkennt "OK telegram direkt" Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
99ba6a78ab
commit
75611be9a9
@@ -37,6 +37,11 @@ def umgebung(tmp_path):
|
||||
"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",
|
||||
}
|
||||
@@ -118,3 +123,88 @@ def test_morgenmeldung_verliert_nichts_wenn_telegram_klemmt(umgebung):
|
||||
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": "[Alarm]\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": "Ohne 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")
|
||||
|
||||
Reference in New Issue
Block a user