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

"""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) == ""
    assert (tmp / "night-queue.txt").read_text(encoding="utf-8") == "- [Box-Update]: 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 "[Morgenmeldung]" in gesendet and "2 Meldungen" in gesendet
    assert "Router aktualisiert" in gesendet and "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": "[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")