"""Einstellungen der Oberfläche (24.09.2026): Update-Zeitfenster, Radar-Schalter, Telegram-Test. Kein Test setzt echte systemctl-Befehle ab: services/einstellungen spricht systemd nur über _systemctl, und das ersetzt hier ein kleines Abbild von systemd — samt der Nachhol-Regel von Persistent=true, die die Box am 24.09. mitten am Tag neu startete. Der Telegram-Test fährt das echte notify.sh, aber mit einem Ersatz für `hermes send`, einer Tmp-Datei als Telegram-Zugang und einer lokalen Bot-API: Es geht nie eine echte Nachricht raus.""" import json import os import re import shutil import subprocess import sys import threading from datetime import datetime, timedelta from datetime import time as uhr from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs import pytest from fastapi import FastAPI from fastapi.testclient import TestClient from kern.zeit import LOCAL_TZ from routers import einstellungen as einstellungen_router from services import einstellungen, waechter UPDATE, RADAR = "mc2-autoupdate.timer", "mc2-radar.timer" class FakeSystemd: """Genug systemd für die Einstellungen: Zustände, Aufrufe — und die Nachhol-Regel eines Timers mit Persistent=true. Beim Start gilt der Stempel als letzter Lauf; liegt ein Termin des geladenen Plans zwischen ihm und jetzt, läuft der Dienst sofort. Beim daemon-reload rechnet ein laufender Timer mit seinem letzten Lauf im Speicher und dem NEUEN Plan (so verhält sich systemd beim Deserialisieren) — hat sich sein Plan nicht geändert, ist nichts nachzuholen, sonst wäre er schon gelaufen.""" def __init__(self, stempel: Path, dropin: Path, letzter_lauf: datetime): self.stempel, self.dropin = stempel, dropin self.aufrufe: list[tuple[str, ...]] = [] self.nachgeholt: list[str] = [] self.scheitert: set[str] = set() # Befehle, die beim nächsten Aufruf einmal scheitern self.aktiv = {UPDATE: True, RADAR: True, "mc2-autoupdate.service": False} self.eingetragen = {UPDATE: True, RADAR: True} self.im_speicher = {UPDATE: letzter_lauf, RADAR: letzter_lauf} self.geladen = {UPDATE: self._plan_aus_datei(UPDATE), RADAR: self._plan_aus_datei(RADAR)} stempel.mkdir(parents=True, exist_ok=True) for timer in (UPDATE, RADAR): datei = stempel / f"stamp-{timer}" datei.touch() os.utime(datei, (letzter_lauf.timestamp(), letzter_lauf.timestamp())) def _plan_aus_datei(self, timer: str) -> tuple[int | None, int, int]: if timer == RADAR: return None, 0, 30 # täglich 00:30 text = self.dropin.read_text(encoding="utf-8") if self.dropin.exists() else "" m = re.search(r"^OnCalendar=(\w{3}) \*-\*-\* (\d\d):(\d\d):00$", text, re.MULTILINE) if not m: return 7, 4, 30 # Plan der Repo-Unit: So 04:30 return einstellungen.SYSTEMD_TAGE.index(m.group(1)) + 1, int(m.group(2)), int(m.group(3)) @staticmethod def termin_dazwischen(plan: tuple[int | None, int, int], von: datetime, bis: datetime) -> bool: wochentag, stunde, minute = plan tag = von.date() while tag <= bis.date(): termin = datetime.combine(tag, uhr(stunde, minute), LOCAL_TZ) if (wochentag is None or termin.isoweekday() == wochentag) and von < termin <= bis: return True tag += timedelta(days=1) return False def __call__(self, *args: str) -> tuple[bool, str]: self.aufrufe.append(args) befehl, jetzt = args[0], datetime.now(LOCAL_TZ) if befehl in self.scheitert: self.scheitert.discard(befehl) return False, f"{befehl} ging schief" if befehl == "show": unit = args[1] eingetragen = self.eingetragen.get(unit) datei = "static" if eingetragen is None else ("enabled" if eingetragen else "disabled") return True, (f"LoadState=loaded\nActiveState={'active' if self.aktiv.get(unit) else 'inactive'}\n" f"UnitFileState={datei}") if befehl == "list-timers": unit = args[-1] naechster = int((jetzt + timedelta(days=1)).timestamp() * 1_000_000) if self.aktiv.get(unit) else None return True, json.dumps([{"unit": unit, "next": naechster, "last": None}]) if befehl == "stop": self.aktiv[args[1]] = False elif befehl == "daemon-reload": for timer in (UPDATE, RADAR): neu = self._plan_aus_datei(timer) if self.aktiv[timer] and neu != self.geladen[timer] \ and self.termin_dazwischen(neu, self.im_speicher[timer], jetzt): self.nachgeholt.append(timer) self.geladen[timer] = neu elif befehl == "start": timer = args[1] if not self.aktiv[timer]: datei = self.stempel / f"stamp-{timer}" letzter = datetime.fromtimestamp(datei.stat().st_mtime, LOCAL_TZ) if datei.exists() else None if letzter is not None and letzter < jetzt: self.im_speicher[timer] = letzter if self.termin_dazwischen(self.geladen[timer], letzter, jetzt): self.nachgeholt.append(timer) self.aktiv[timer] = True elif befehl == "enable": self.eingetragen[args[1]] = True elif befehl == "disable": self.eingetragen[args[-1]] = False if "--now" in args: self.aktiv[args[-1]] = False return True, "" def befehle(self, *arten: str) -> list[str]: return [a[0] for a in self.aufrufe if a[0] in arten] @pytest.fixture def box(tmp_path, monkeypatch): """Einstellungsdatei, Units und Stempel in tmp; systemd ist das Abbild oben (letzter Lauf vor vier Tagen).""" monkeypatch.setenv("MC_EINSTELLUNGEN", str(tmp_path / "mc2-einstellungen.json")) monkeypatch.setenv("MC_SYSTEMD_USER_DIR", str(tmp_path / "units")) monkeypatch.setenv("MC_TIMER_STEMPEL_DIR", str(tmp_path / "timers")) monkeypatch.setattr(einstellungen, "AUF_DER_BOX", True) fake = FakeSystemd(tmp_path / "timers", einstellungen.dropin(), datetime.now(LOCAL_TZ) - timedelta(days=4)) monkeypatch.setattr(einstellungen, "_systemctl", fake) return fake def _vorgestern() -> tuple[int, str]: """Ein Plan, dessen Termin vorgestern war — zwischen dem letzten Lauf (vor vier Tagen) und jetzt.""" d = datetime.now(LOCAL_TZ) - timedelta(days=2) if (d.isoweekday(), d.strftime("%H:%M")) == (7, "04:30"): # wäre der alte Plan — dann nichts „Neues“ d -= timedelta(minutes=1) return d.isoweekday(), d.strftime("%H:%M") # --- Datei und Regeln ------------------------------------------------------------------------------ def test_ohne_datei_gilt_der_standard(box): assert einstellungen.lesen() == {"update_fenster": {"tag": 7, "uhrzeit": "04:30"}, "radar": {"an": True}} @pytest.mark.parametrize("inhalt", [ "kein json", "[]", '{"update_fenster": {"tag": 8, "uhrzeit": "04:30"}, "radar": {"an": "nein"}}', '{"update_fenster": {"tag": true, "uhrzeit": "4:30"}}', '{"update_fenster": {"tag": 3, "uhrzeit": "24:00"}}', ]) def test_unsinn_in_der_datei_faellt_auf_den_standard(box, inhalt): einstellungen.datei().write_text(inhalt, encoding="utf-8") assert einstellungen.lesen() == {"update_fenster": {"tag": 7, "uhrzeit": "04:30"}, "radar": {"an": True}} @pytest.mark.parametrize("tag, uhrzeit, text", [ (7, "04:30", "Sonntag 04:00–06:59"), (6, "05:00", "Samstag 05:00–07:59"), (6, "23:30", "Samstag 23:00 – Sonntag 01:59"), (7, "22:00", "Sonntag 22:00 – Montag 00:59"), (1, "00:00", "Montag 00:00–02:59"), ]) def test_neustart_fenster(tag, uhrzeit, text): assert einstellungen.neustart_fenster(tag, uhrzeit) == text def test_dropin_ersetzt_den_plan_statt_einen_zweiten_daneben(): text = einstellungen.dropin_text(6, "05:00") assert "[Timer]\nOnCalendar=\nOnCalendar=Sat *-*-* 05:00:00\n" in text @pytest.mark.parametrize("zustand", [ {"ActiveState": "inactive", "UnitFileState": "disabled"}, {"ActiveState": "inactive", "UnitFileState": "enabled"}, {"ActiveState": "active", "UnitFileState": "disabled"}, {"ActiveState": "failed", "UnitFileState": "disabled"}, {}, ]) def test_schlaeft_wie_der_waechter(zustand): assert einstellungen.schlaeft(zustand) == waechter.schlaeft(zustand) # --- Update-Zeitfenster ------------------------------------------------------------------------------ def test_zeitfenster_holt_nicht_nach(box): """Die Falle vom 24.09.: Ein Plan, dessen Termin seit dem letzten Lauf schon vorbei ist, darf nicht sofort laufen.""" tag, uhrzeit = _vorgestern() ergebnis = einstellungen.update_fenster_setzen(tag, uhrzeit) assert ergebnis["ok"], ergebnis assert box.nachgeholt == [] # Erst anhalten, dann neu einlesen, dann (mit Stempel) starten — sonst rechnet systemd mit dem alten Lauf. assert box.befehle("stop", "daemon-reload", "start") == ["stop", "daemon-reload", "start"] assert einstellungen.dropin().read_text(encoding="utf-8") == einstellungen.dropin_text(tag, uhrzeit) assert einstellungen.lesen()["update_fenster"] == {"tag": tag, "uhrzeit": uhrzeit} assert einstellungen.neustart_fenster(tag, uhrzeit) in ergebnis["text"] def test_gegenprobe_das_abbild_erkennt_die_falle(box): """Ohne die Vorsicht holt das Abbild nach — sonst bewiese der Test oben nichts.""" tag, uhrzeit = _vorgestern() einstellungen._dropin_schreiben(einstellungen.dropin_text(tag, uhrzeit)) box("daemon-reload") # Reload bei laufendem Timer assert box.nachgeholt == [UPDATE] def test_ohne_stempel_wuerde_der_timer_nachholen(box, monkeypatch): monkeypatch.setattr(einstellungen, "_stempel_setzen", lambda timer: None) tag, uhrzeit = _vorgestern() assert einstellungen.update_fenster_setzen(tag, uhrzeit)["ok"] assert box.nachgeholt == [UPDATE] def test_zeitfenster_nicht_waehrend_eines_updates(box): box.aktiv["mc2-autoupdate.service"] = True ergebnis = einstellungen.update_fenster_setzen(6, "05:00") assert not ergebnis["ok"] and "läuft das Update" in ergebnis["detail"] assert box.befehle("stop", "daemon-reload", "start") == [] assert not einstellungen.dropin().exists() @pytest.mark.parametrize("tag, uhrzeit", [(0, "04:30"), (8, "04:30"), (7, "4:30"), (7, "24:00"), (7, "04:60")]) def test_zeitfenster_prueft_die_eingabe(box, tag, uhrzeit): ergebnis = einstellungen.update_fenster_setzen(tag, uhrzeit) assert not ergebnis["ok"] and ergebnis["detail"] assert box.aufrufe == [] def test_zeitfenster_rollt_bei_fehler_zurueck(box): einstellungen._dropin_schreiben(einstellungen.dropin_text(6, "05:00")) vorher = einstellungen.dropin().read_text(encoding="utf-8") box.scheitert.add("daemon-reload") ergebnis = einstellungen.update_fenster_setzen(3, "04:00") assert not ergebnis["ok"] and "nicht übernommen" in ergebnis["detail"] assert einstellungen.dropin().read_text(encoding="utf-8") == vorher assert box.aktiv[UPDATE] is True # ohne Timer gäbe es keine Updates mehr assert not einstellungen.datei().exists() assert box.nachgeholt == [] def test_gestoppter_timer_bleibt_aus(box): box.aktiv[UPDATE] = False assert einstellungen.update_fenster_setzen(6, "05:00")["ok"] assert box.befehle("stop", "start") == [] and box.aktiv[UPDATE] is False assert einstellungen.stand()["update_fenster"]["timer_aktiv"] is False def test_stand_zeigt_ob_der_timer_zum_wunsch_passt(box): stand = einstellungen.stand() assert stand["schaltbar"] is True assert stand["update_fenster"]["wirksam"] is True and stand["update_fenster"]["tag_name"] == "Sonntag" assert stand["update_fenster"]["naechster_lauf"] # Wunsch in der Datei, aber kein Drop-in (etwa nach einem Neuaufbau): nicht wirksam. einstellungen.datei().write_text('{"update_fenster": {"tag": 6, "uhrzeit": "05:00"}}', encoding="utf-8") assert einstellungen.stand()["update_fenster"]["wirksam"] is False assert einstellungen.update_fenster_setzen(6, "05:00")["ok"] assert einstellungen.stand()["update_fenster"]["wirksam"] is True def test_am_pc_wird_nichts_geschaltet(box, monkeypatch): monkeypatch.setattr(einstellungen, "AUF_DER_BOX", False) assert not einstellungen.update_fenster_setzen(6, "05:00")["ok"] assert not einstellungen.radar_schalten(False)["ok"] assert box.befehle("stop", "start", "enable", "disable", "daemon-reload") == [] # --- Radar-Schalter ----------------------------------------------------------------------------------- def test_radar_aus_und_wieder_an_ohne_nachholen(box): ergebnis = einstellungen.radar_schalten(False) assert ergebnis["ok"] and "schläft" in ergebnis["text"] assert ("disable", "--now", RADAR) in box.aufrufe assert einstellungen.lesen()["radar"] == {"an": False} radar = einstellungen.stand()["radar"] assert radar["zustand"] == "schlaeft" and radar["wirksam"] is True and radar["naechster_lauf"] is None ergebnis = einstellungen.radar_schalten(True) assert ergebnis["ok"] assert box.befehle("enable", "start") == ["enable", "start"] assert box.nachgeholt == [] # der letzte Lauf war vor vier Tagen, 00:30 lag dazwischen assert einstellungen.lesen()["radar"] == {"an": True} assert einstellungen.stand()["radar"]["zustand"] == "an" def test_radar_nur_gestoppt_ist_nicht_aus(box): """Gestoppt, aber noch im Autostart: nach einem Neustart der Box liefe das Radar wieder.""" einstellungen.datei().write_text('{"radar": {"an": false}}', encoding="utf-8") box.aktiv[RADAR] = False radar = einstellungen.stand()["radar"] assert radar["zustand"] == "aus" and radar["wirksam"] is False box.eingetragen[RADAR] = False assert einstellungen.stand()["radar"]["wirksam"] is True def test_radar_ohne_stempel_wuerde_nachholen(box, monkeypatch): monkeypatch.setattr(einstellungen, "_stempel_setzen", lambda timer: None) einstellungen.radar_schalten(False) einstellungen.radar_schalten(True) assert box.nachgeholt == [RADAR] def test_radar_fehler_wird_nicht_gespeichert(box): box.scheitert.add("disable") ergebnis = einstellungen.radar_schalten(False) assert not ergebnis["ok"] and "nicht ausgeschaltet" in ergebnis["detail"] assert not einstellungen.datei().exists() def test_waechter_laesst_den_schlafenden_radar_in_ruhe(monkeypatch): """Ein alter Fehlschlag des Radars ist kein Hinweis mehr, sobald sein Timer schläft (Radar aus).""" zustaende = { "mc2-radar": {"LoadState": "loaded", "ActiveState": "failed", "Result": "exit-code"}, "mc2-radar.timer": {"LoadState": "loaded", "ActiveState": "inactive", "UnitFileState": "disabled"}, } monkeypatch.setattr(waechter, "_systemctl_show", lambda name, system: zustaende.get( name, {"LoadState": "loaded", "ActiveState": "inactive", "Result": "success"})) monkeypatch.setattr(waechter, "_journal_fehlerzeile", lambda name, system=False: "") assert "timer:mc2-radar" not in {b.id for b in waechter.pruefe_timer_dienste()} zustaende["mc2-radar.timer"] = {"LoadState": "loaded", "ActiveState": "active", "UnitFileState": "enabled"} assert "timer:mc2-radar" in {b.id for b in waechter.pruefe_timer_dienste()} # --- Schnittstellen ----------------------------------------------------------------------------------- def _client() -> TestClient: app = FastAPI() app.include_router(einstellungen_router.router) app.include_router(einstellungen_router.homelab_router) return TestClient(app) def test_schnittstellen(box): c = _client() assert c.get("/api/einstellungen").json()["radar"]["an"] is True r = c.post("/api/einstellungen/update-fenster", json={"tag": 9, "uhrzeit": "04:30"}) assert r.status_code == 409 and "Wochentag" in r.json()["detail"] r = c.post("/api/einstellungen/update-fenster", json={"tag": 6, "uhrzeit": "05:00"}) assert r.status_code == 200 and r.json()["einstellungen"]["update_fenster"]["tag"] == 6 r = c.post("/api/einstellungen/radar", json={"an": False}) assert r.status_code == 200 and r.json()["einstellungen"]["radar"]["zustand"] == "schlaeft" BACKEND = Path(__file__).resolve().parent.parent @pytest.mark.parametrize("rolle, da, nicht_da", [ ("box", "/api/einstellungen/telegram-test", "/api/homelab/einstellungen/telegram-test"), ("homelab", "/api/homelab/einstellungen/telegram-test", "/api/einstellungen"), ]) def test_jede_rolle_hat_ihren_test_knopf(tmp_path, rolle, da, nicht_da): """Der Homelab-Teil bekommt nur den Telegram-Test (und lädt dafür nichts von der KI-Box, siehe test_partner).""" env = {**os.environ, "MC_ROLLE": rolle, "MC_MODELS_DIR": str(tmp_path), "MC_DATEN_DIR": str(tmp_path), "MC_PARTNER_URL": ""} lauf = subprocess.run([sys.executable, "-c", "import json, app; print(json.dumps(sorted(app.app.openapi()['paths'])))"], cwd=BACKEND, env=env, capture_output=True, text=True, timeout=120) assert lauf.returncode == 0, lauf.stderr[-2000:] pfade = json.loads(lauf.stdout.strip().splitlines()[-1]) assert da in pfade and nicht_da not in pfade # --- Telegram-Test ------------------------------------------------------------------------------------ BASH = shutil.which("bash") braucht_bash = 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 melden(tmp_path, monkeypatch): """notify.sh mit einem Ersatz für `hermes send`, der nur mitschreibt, und ohne echten Telegram-Zugang.""" protokoll = tmp_path / "gesendet.txt" stub = tmp_path / "hermes-stub" stub.write_text( "#!/usr/bin/env bash\n" 'printf "%s\\n" "$*" >> "$STUB_PROTOKOLL"\n' '[ -n "${STUB_AUSGABE:-}" ] && echo "$STUB_AUSGABE"\n' 'exit "${STUB_EXIT:-0}"\n', encoding="utf-8", newline="\n") stub.chmod(0o755) for name, wert in {"MC_NOTIFY_LOG": tmp_path / "notify.log", "MC_NIGHT_QUEUE": tmp_path / "night-queue.txt", "MC_NOTIFY_HERMES": stub, "MC_TELEGRAM_ENV": tmp_path / "keine.env", "STUB_PROTOKOLL": protokoll}.items(): monkeypatch.setenv(name, Path(wert).as_posix()) monkeypatch.setenv("MC_TELEGRAM_API", "http://127.0.0.1:9") monkeypatch.setenv("MC_NOTIFY_STUNDE", "03") # mitten in der Nacht: -d muss trotzdem sofort senden monkeypatch.delenv("STUB_EXIT", raising=False) monkeypatch.delenv("STUB_AUSGABE", raising=False) monkeypatch.setattr(einstellungen, "AUF_DER_BOX", True) monkeypatch.setitem(einstellungen._letzter_test, "ts", 0.0) return tmp_path, protokoll @pytest.fixture def bot_api(): """Ein Ersatz für api.telegram.org, der mitschreibt, was ankommt.""" 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() @braucht_bash def test_telegram_test_ueber_hermes_auch_nachts(melden): tmp, protokoll = melden ergebnis = einstellungen.telegram_test() assert ergebnis["ok"] and ergebnis["gesendet"] and ergebnis["weg"] == "hermes", ergebnis gesendet = protokoll.read_text(encoding="utf-8") assert "send --to telegram --subject [Test]" in gesendet and f"Kennung {ergebnis['kennung']}" in gesendet assert not (tmp / "night-queue.txt").exists() assert "OK telegram: Testmeldung aus den Einstellungen" in (tmp / "notify.log").read_text(encoding="utf-8") @braucht_bash def test_telegram_test_ueber_den_zweitweg(melden, bot_api, monkeypatch): tmp, _ = melden url, empfangen, _ = bot_api env = tmp / "telegram.env" env.write_text("TELEGRAM_BOT_TOKEN=1:test\nTELEGRAM_HOME_CHANNEL=42\n", encoding="utf-8", newline="\n") monkeypatch.setenv("MC_TELEGRAM_ENV", env.as_posix()) monkeypatch.setenv("MC_TELEGRAM_API", url) monkeypatch.setenv("STUB_EXIT", "1") # wie im Homelab-Teil: kein Hermes ergebnis = einstellungen.telegram_test() assert ergebnis["gesendet"] and ergebnis["weg"] == "direkt", ergebnis assert len(empfangen) == 1 and empfangen[0]["text"].startswith("[Test]\nTestmeldung aus den Einstellungen") @braucht_bash def test_telegram_test_nennt_den_grund(melden, bot_api, monkeypatch): tmp, _ = melden monkeypatch.setenv("STUB_EXIT", "1") monkeypatch.setenv("STUB_AUSGABE", "Fehler: Telegram-Plattform nicht verbunden") ergebnis = einstellungen.telegram_test() assert not ergebnis["ok"] and not ergebnis["gesendet"] assert ergebnis["grund"].startswith("Fehler: Telegram-Plattform nicht verbunden; Zweitweg: Zugangsdatei") assert "fehlt oder ist nicht lesbar" in ergebnis["grund"] and ergebnis["detail"].startswith("Die Testmeldung ging") # Telegram lehnt ab (etwa ein falsches Token): der Grund kommt aus curls Fehlercode. url, _, antwort = bot_api antwort["code"] = 401 env = tmp / "telegram.env" env.write_text("TELEGRAM_BOT_TOKEN=1:falsch\nTELEGRAM_HOME_CHANNEL=42\n", encoding="utf-8", newline="\n") monkeypatch.setenv("MC_TELEGRAM_ENV", env.as_posix()) monkeypatch.setenv("MC_TELEGRAM_API", url) monkeypatch.setitem(einstellungen._letzter_test, "ts", 0.0) assert "Zweitweg: Telegram lehnt ab" in einstellungen.telegram_test()["grund"] @braucht_bash def test_telegram_test_nicht_im_sekundentakt(melden): assert einstellungen.telegram_test()["gesendet"] zweiter = einstellungen.telegram_test() assert not zweiter["ok"] and "kurz warten" in zweiter["detail"] def test_telegram_test_am_pc(monkeypatch): monkeypatch.setattr(einstellungen, "AUF_DER_BOX", False) monkeypatch.setitem(einstellungen._letzter_test, "ts", 0.0) ergebnis = einstellungen.telegram_test() assert not ergebnis["ok"] and "nur auf der Box" in ergebnis["detail"] def test_melde_log_lesen(tmp_path): """Eine mehrzeilige Ausgabe von `hermes send` landet mitten im FALLBACK-Eintrag (notify.sh).""" text = "Testmeldung aus den Einstellungen (Box-Wart, 24.09. 21:00 Uhr, Kennung abc123). Kommt sie an …" log = tmp_path / "notify.log" log.write_text( "2026-09-24 20:00:00 OK telegram: Frühere Meldung\n" "2026-09-24 21:00:00 FALLBACK (telegram fehlgeschlagen: Traceback (most recent call last):\n" " ConnectionError: gateway down; Zweitweg: Telegram antwortet nicht rechtzeitig): " + text + "\n" "2026-09-24 21:00:05 OK telegram: Spätere Meldung\n", encoding="utf-8") eintrag = einstellungen.log_eintrag(log, "Kennung abc123") assert eintrag.startswith("2026-09-24 21:00:00 FALLBACK") and eintrag.endswith(text) gedeutet = einstellungen.deute_eintrag(eintrag, text) assert gedeutet == {"gesendet": False, "weg": None, "grund": "Traceback (most recent call last): ConnectionError: " "gateway down; Zweitweg: Telegram antwortet nicht rechtzeitig"} assert einstellungen.log_eintrag(log, "Kennung fehlt") is None assert einstellungen.log_eintrag(tmp_path / "gibt-es-nicht.log", "x") is None assert einstellungen.deute_eintrag("2026-09-24 21:00:00 OK telegram direkt: x", "x")["weg"] == "direkt" assert einstellungen.deute_eintrag("2026-09-24 03:00:00 QUEUED für Morgen-Digest: x", "x")["gesendet"] is False