"""Angriffsfaelle fuer die Abendkontrolle (abendkontrolle_wsl.py in WSL, abendkontrolle.py unter Windows). Koordinator-Werkzeug, nicht fuers Repo. Jeder Fall laeuft einzeln auf einer Wegwerfkopie, zuerst die Grundlinie; ohne Netz (--ohne-broker/--ohne-kerzen, Kalender und Journal aus Dateien, eine Ersatz- Auswertung fuer die Brokerzeile) und ohne Meldung (--ohne-meldung). Faelle an zwei Listenpositionen, wo die Reihenfolge zaehlen kann (AI/Regelwerk/pruefwerkzeuge.md, Abschnitt 3). Aufruf in WSL (Venv) und unter Windows: python abendkontrolle_angriffe.py Exit 0, wenn jeder Fall so ausgeht wie erwartet; sonst 1 mit der Liste der Abweichungen. """ import json import os import shutil import sqlite3 import subprocess import sys import tempfile from pathlib import Path HIER = Path(__file__).resolve().parent TAG = "2026-09-23" ECHTES_PROJEKT = Path("/mnt/d/Claude/Code Projekte/active - Trading Plattform") ECHTES_PROTOKOLL = ECHTES_PROJEKT / "artifacts" / "paper" / "2026-09-23T131023-paper.jsonl" KALENDER_TAG = [{"date": TAG, "open": "09:30", "close": "16:00"}] JOURNAL_GUT = ( "Ende: boersenschluss, Position 0, Orders 2, Killswitch aktiv, Exit 0\n" "trading-paper.service: Deactivated successfully.\n" ) DIENSTE_GUT = { "trading-dashboard.service": "active", "trading-paper.timer": "active", "trading-auswertung.timer": "active", } abweichungen: list[str] = [] # --- WSL-Teil ---------------------------------------------------------------------------------- def baue_projekt(wurzel: Path, killswitch: str | None = "aktiv") -> Path: projekt = wurzel / "projekt" (projekt / "artifacts" / "paper").mkdir(parents=True) (projekt / "data" / "paper" / "auswertungen").mkdir(parents=True) (projekt / "config").mkdir() # Die Auswertung importiert trading.geheimnisse aus dem Arbeitsverzeichnis. (projekt / "trading").symlink_to(ECHTES_PROJEKT / "trading") shutil.copyfile(ECHTES_PROTOKOLL, projekt / "artifacts" / "paper" / f"{TAG}T131023-paper.jsonl") (projekt / "data" / "paper" / "auswertungen" / f"{TAG}.md").write_text("# Bericht\n", encoding="utf-8") with sqlite3.connect(projekt / "data" / "registry.db") as db: db.execute("CREATE TABLE killswitch_zustand (id INTEGER PRIMARY KEY, zustand TEXT, seit TEXT, ausloeser TEXT)") if killswitch is not None: db.execute("INSERT INTO killswitch_zustand VALUES (1, ?, '2026-09-23T13:31:00+00:00', 'pascal')", (killswitch,)) (wurzel / "kalender.json").write_text(json.dumps(KALENDER_TAG), encoding="utf-8") (wurzel / "journal.txt").write_text(JOURNAL_GUT, encoding="utf-8") (wurzel / "dienste.json").write_text(json.dumps(DIENSTE_GUT), encoding="utf-8") return projekt def fahre(wurzel: Path, projekt: Path, extra_env: dict | None = None, tag: str = TAG, meldung: bool = False) -> dict: befehl = [ sys.executable, str(HIER / "abendkontrolle_wsl.py"), "--tag", tag, "--projekt", str(projekt), "--kalender", str(wurzel / "kalender.json"), "--journal", str(wurzel / "journal.txt"), "--dienste", str(wurzel / "dienste.json"), "--ohne-kerzen", ] if not meldung: befehl.append("--ohne-meldung") # Mit der Ersatz-Auswertung darf die Brokerzeile geprueft werden: Der Ersatz geht nicht ans Netz. if not (extra_env and "ABENDKONTROLLE_AUSWERTUNG" in extra_env): befehl.append("--ohne-broker") env = {k: v for k, v in os.environ.items() if k != "ABENDKONTROLLE_AUSWERTUNG"} env.update(extra_env or {}) lauf = subprocess.run(befehl, capture_output=True, text=True, env=env, timeout=300) try: ergebnis = json.loads(lauf.stdout.strip().splitlines()[-1]) except (IndexError, json.JSONDecodeError): return {"befunde": [f"KEINE JSON-AUSGABE exit {lauf.returncode}: {lauf.stderr[-300:]}"], "exit": lauf.returncode} ergebnis["_exit"] = lauf.returncode return ergebnis def erwarte(name: str, ergebnis: dict, *, befund: str | None, anzahl: int | None = None, exit_code: int = 0, auch: tuple = (), nicht: tuple = (), fakt=None) -> None: befunde = ergebnis.get("befunde", []) alles = " | ".join(befunde) ok = True if befund is None and befunde: ok = False if befund is not None and not any(befund in b for b in befunde): ok = False if any(teil not in alles for teil in auch) or any(teil in alles for teil in nicht): ok = False if anzahl is not None and len(befunde) != anzahl: ok = False if ergebnis.get("_exit", ergebnis.get("exit")) != exit_code: ok = False if fakt is not None: try: ok = ok and bool(fakt(ergebnis.get("fakten") or {}, ergebnis)) except Exception: ok = False zeile = f"{'OK ' if ok else 'FEHL'} {name}: {len(befunde)} Befund(e) {befunde[:3]} meldung={(ergebnis.get('fakten') or {}).get('meldung')}" print(zeile) if not ok: abweichungen.append(zeile) def wsl_faelle() -> None: def fall(name, vorbereiten, *, befund, anzahl=None, exit_code=0, env=None, tag=TAG, meldung=False, auch=(), nicht=(), fakt=None): with tempfile.TemporaryDirectory() as tmp: wurzel = Path(tmp) projekt = baue_projekt(wurzel) extra = vorbereiten(wurzel, projekt) umgebung = {**(env or {}), **(extra if isinstance(extra, dict) else {})} erwarte(name, fahre(wurzel, projekt, umgebung, tag, meldung), befund=befund, anzahl=anzahl, exit_code=exit_code, auch=auch, nicht=nicht, fakt=fakt) nichts = lambda w, p: None # noqa: E731 prot = lambda p: p / "artifacts" / "paper" / f"{TAG}T131023-paper.jsonl" # noqa: E731 def schreibe(pfad: Path, text: str) -> None: pfad.write_text(text, encoding="utf-8") def zeilen_aendern(p: Path, aendern) -> None: zeilen = prot(p).read_text(encoding="utf-8").splitlines() schreibe(prot(p), "\n".join(aendern(zeilen)) + "\n") fall("Grundlinie", nichts, befund=None, anzahl=0) # 1 Kalender # Ein Feiertag ohne Protokoll: kein Befund. Haelt die Kontrolle ihn faelschlich fuer einen Handelstag, # fehlt das Protokoll und es gibt einen -- so trennt der Fall "kein Handelstag" von "immer Handelstag". def feiertag(inhalt): def vorbereiten(w, p): schreibe(w / "kalender.json", inhalt) prot(p).unlink() return vorbereiten fall("Kalender leer (Feiertag, Wochenende)", feiertag("[]"), befund=None, anzahl=0) fall("Kalender nennt nur einen anderen Tag", feiertag(json.dumps([{"date": "2026-09-22"}])), befund=None, anzahl=0) fall("Kalender als Objekt", lambda w, p: schreibe(w / "kalender.json", "{}"), befund="Kalender in unerwarteter Form") fall("Kalender Eintrag ohne date, erste Position", lambda w, p: schreibe(w / "kalender.json", json.dumps([{"open": "09:30"}] + KALENDER_TAG)), befund="Kalender in unerwarteter Form") fall("Kalender Eintrag ohne date, letzte Position", lambda w, p: schreibe(w / "kalender.json", json.dumps(KALENDER_TAG + [{"date": 5}])), befund="Kalender in unerwarteter Form") fall("Kalender-Datei fehlt", lambda w, p: (w / "kalender.json").unlink(), befund="Kalender, und damit") fall("Kalender kein JSON", lambda w, p: schreibe(w / "kalender.json", ""), befund="Kalender, und damit") # 2 Protokoll fall("kein Protokoll", lambda w, p: prot(p).unlink(), befund="der Runner lief nicht") fall("nur Probeprotokoll", lambda w, p: prot(p).rename(prot(p).with_name(f"{TAG}T131023-probe.jsonl")), befund="der Runner lief nicht") # 6 Handel fall("nie aktiv", lambda w, p: zeilen_aendern(p, lambda z: [s.replace('"killswitch": "aktiv"', '"killswitch": "angehalten"') for s in z]), befund="Tag ohne Handel") fall("kaputte Zeile vorne", lambda w, p: zeilen_aendern(p, lambda z: [z[0], "{kaputt", *z[1:]]), befund="nicht lesbar") fall("kaputte Zeile hinten", lambda w, p: zeilen_aendern(p, lambda z: [*z, "[1, 2]"]), befund="nicht lesbar") # 3 Journal fall("Journal ohne Ende", lambda w, p: schreibe(w / "journal.txt", "gestartet\n"), befund="keine Zeile 'Ende") fall("Journal leer", lambda w, p: schreibe(w / "journal.txt", ""), befund="keine Zeile 'Ende") fall("Journal fehlt", lambda w, p: (w / "journal.txt").unlink(), befund="Journal nicht geprueft") fall("Exit 4", lambda w, p: schreibe(w / "journal.txt", JOURNAL_GUT.replace("Exit 0", "Exit 4")), befund="Exit 4") fall("Exit 4 im ersten von zwei Laeufen", lambda w, p: schreibe(w / "journal.txt", JOURNAL_GUT.replace("Exit 0", "Exit 4") + JOURNAL_GUT), befund="Lauf 1 von 2") fall("Exit 4 im letzten von zwei Laeufen", lambda w, p: schreibe(w / "journal.txt", JOURNAL_GUT + JOURNAL_GUT.replace("Exit 0", "Exit 4")), befund="Lauf 2 von 2") fall("Grund stoerung", lambda w, p: schreibe(w / "journal.txt", JOURNAL_GUT.replace("Ende: boersenschluss", "Ende: stoerung")), befund="nicht mit boersenschluss") fall("Position 10", lambda w, p: schreibe(w / "journal.txt", JOURNAL_GUT.replace("Position 0", "Position 10")), befund="Position 10") fall("Failed with result", lambda w, p: schreibe(w / "journal.txt", JOURNAL_GUT + "trading-paper.service: Failed with result 'exit-code'.\n"), befund="Failed with result") # 4 Auswertung fall("Orderzeile nicht gefuellt", lambda w, p: zeilen_aendern(p, lambda z: [s.replace('"status": "filled"', '"status": "canceled"', 1) if '"art": "order"' in s else s for s in z]), befund="nicht gefuellt") fall("Schlussposition 10 im Protokoll", lambda w, p: zeilen_aendern(p, lambda z: [s.replace('"position": 0', '"position": 10') if '"art": "ende"' in s else s for s in z]), befund="Schlussposition ungleich null") # 5 Broker (Ersatz-Auswertung, kein Netz) ersatz = HIER / "abendkontrolle_ersatz_auswertung.py" fall("Broker 2 offen", nichts, befund="2 offene Position", env={"ABENDKONTROLLE_AUSWERTUNG": str(ersatz), "ERSATZ_AUSGABE": "offene positionen beim broker jetzt: 2"}, ) fall("Broker 0 offen", nichts, befund=None, anzahl=0, env={"ABENDKONTROLLE_AUSWERTUNG": str(ersatz), "ERSATZ_AUSGABE": "offene positionen beim broker jetzt: 0"}) fall("Auswertung Exit 3", nichts, befund="Exit 3", env={"ABENDKONTROLLE_AUSWERTUNG": str(ersatz), "ERSATZ_AUSGABE": "VORAUSSETZUNG FEHLT: x", "ERSATZ_EXIT": "3"}) fall("Auswertung Kontowert in BEFUND wird maskiert", nichts, befund="kontowert [x]", env={"ABENDKONTROLLE_AUSWERTUNG": str(ersatz), "ERSATZ_AUSGABE": "BEFUND: kontowert 12345.67 zu tief"}) # 7 Bericht fall("Bericht fehlt", lambda w, p: (p / "data" / "paper" / "auswertungen" / f"{TAG}.md").unlink(), befund="Tagesbericht") # 8 Dienste fall("Dashboard inaktiv (erste Position)", lambda w, p: schreibe(w / "dienste.json", json.dumps({**DIENSTE_GUT, "trading-dashboard.service": "failed"})), befund="trading-dashboard.service ist nicht aktiv") fall("Auswertungs-Zeitgeber fehlt (letzte Position)", lambda w, p: schreibe(w / "dienste.json", json.dumps({k: v for k, v in DIENSTE_GUT.items() if "auswertung" not in k})), befund="trading-auswertung.timer ist nicht aktiv") fall("Dienste als Liste", lambda w, p: schreibe(w / "dienste.json", "[]"), befund="Dienste: Zustand nicht lesbar") # 9 Killswitch def killswitch(zustand): def vorbereiten(w, p): with sqlite3.connect(p / "data" / "registry.db") as db: if zustand is None: db.execute("DELETE FROM killswitch_zustand") else: db.execute("UPDATE killswitch_zustand SET zustand = ?", (zustand,)) return vorbereiten fall("Killswitch gesperrt", killswitch("gesperrt"), befund="gesperrt") fall("Killswitch ohne Zeile", killswitch(None), befund="kein Killswitch-Zustand") fall("Register fehlt", lambda w, p: (p / "data" / "registry.db").unlink(), befund="registry.db fehlt") # 3 Ende je Lauf: mittags abgebrochen, der zweite Lauf bleibt angehalten (Befund des Pruefers W2) def mittags_abgebrochen(w, p): zeilen = prot(p).read_text(encoding="utf-8").splitlines() takte = [z for z in zeilen if '"art": "takt"' in z] ende = json.loads(next(z for z in zeilen if '"art": "ende"' in z)) lauf1 = [zeilen[0], *takte[:30], json.dumps({**ende, "grund": "abbruch"})] lauf2 = [zeilen[0], *[z.replace('"killswitch": "aktiv"', '"killswitch": "angehalten"') for z in takte[30:]], json.dumps(ende)] schreibe(prot(p), "\n".join(lauf1) + "\n") schreibe(p / "artifacts" / "paper" / f"{TAG}T160000-paper.jsonl", "\n".join(lauf2) + "\n") schreibe(w / "journal.txt", JOURNAL_GUT.replace("Ende: boersenschluss", "Ende: abbruch") + JOURNAL_GUT) fall("mittags abgebrochen, danach angehalten", mittags_abgebrochen, befund="Lauf endete mit abbruch", auch=("Lauf 1 von 2) endete nicht mit boersenschluss",)) # Lauf nach Boersenschluss (WSL-Neustart abends): kein Befund (Befund des Pruefers W3) def nach_schluss(w, p): kopf = prot(p).read_text(encoding="utf-8").splitlines()[0] ende = {"art": "ende", "zeit": f"{TAG}T21:06:40+00:00", "grund": "wartezeit", "fehler": None, "position": 0, "orders": 0, "killswitch": "angehalten", "schritte": 0} schreibe(p / "artifacts" / "paper" / f"{TAG}T210500-paper.jsonl", kopf + "\n" + json.dumps(ende) + "\n") schreibe(w / "journal.txt", JOURNAL_GUT + "Ende: wartezeit, Position 0, Orders 0, Killswitch angehalten, Exit 3\n") fall("Lauf nach Boersenschluss (wartezeit, Exit 3)", nach_schluss, befund=None, anzahl=0) fall("nur ein Lauf mit wartezeit an einem Handelstag", lambda w, p: ( nach_schluss(w, p), prot(p).unlink(), schreibe(w / "journal.txt", "Ende: wartezeit, Position 0, Orders 0, Killswitch angehalten, Exit 3\n")), befund="keine Zeile 'Ende", auch=("kein Lauf des scharfen Runners in der Handelszeit",)) # 9 Killswitch: unbekannter Zustand, und eine Frage, die scheitert, haelt die anderen nicht auf fall("Killswitch unbekannter Zustand", killswitch("kaputt"), befund="unbekanntem Zustand") def register_ohne_tabelle(w, p): (p / "data" / "registry.db").unlink() sqlite3.connect(p / "data" / "registry.db").close() (p / "data" / "paper" / "auswertungen" / f"{TAG}.md").unlink() fall("Register ohne Tabelle, Bericht fehlt", register_ohne_tabelle, befund="Killswitch nicht geprueft", auch=("Tagesbericht",)) # Maske auch ohne Geldwort (Befund des Pruefers W4) fall("Maske ohne Geldwort", nichts, befund="Wert", env={"ABENDKONTROLLE_AUSWERTUNG": str(ersatz), "ERSATZ_AUSGABE": "BEFUND: Wert 100123.45 und 172.3456 und -4.56 USD\noffene positionen beim broker jetzt: 0"}, nicht=("100123", "172.3456", "4.56")) # Meldeweg (Befund des Pruefers B1): ein Fehlschlag darf nicht als "gesendet" stehen def mit_melder(port): def vorbereiten(w, p): text = (ECHTES_PROJEKT / "config" / "paper.toml").read_text(encoding="utf-8") anker = 'ntfy_url = "https://ntfy.sh"' if text.count(anker) != 1: raise RuntimeError("Anker ntfy_url fehlt in config/paper.toml") schreibe(p / "config" / "paper.toml", text.replace(anker, f'ntfy_url = "http://127.0.0.1:{port}"\nklartext_lokal = true')) schreibe(w / "secrets.toml", '[ntfy]\naktiv = true\nthema = "abendkontrolle-angriff"\n') (p / "data" / "paper" / "auswertungen" / f"{TAG}.md").unlink() # ein Befund, damit gemeldet wird return {"TRADING_GEHEIMNISSE": str(w / "secrets.toml")} return vorbereiten fall("Meldung scheitert (niemand hoert)", mit_melder(9), befund="Tagesbericht", meldung=True, fakt=lambda f, e: str(f.get("meldung", "")).startswith("nicht gesendet")) with tempfile.TemporaryDirectory() as tmp: mitschrift = Path(tmp) / "mitschrift.jsonl" port = 18765 server = subprocess.Popen([sys.executable, str(HIER / "ntfy_probe_server.py"), str(port), str(mitschrift)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) try: import socket import time for _ in range(50): try: socket.create_connection(("127.0.0.1", port), timeout=0.2).close() break except OSError: time.sleep(0.1) fall("Meldung kommt an", mit_melder(port), befund="Tagesbericht", meldung=True, fakt=lambda f, e: f.get("meldung") == "gesendet" and "Abendkontrolle" in mitschrift.read_text(encoding="utf-8")) finally: server.terminate() server.wait(timeout=10) # Bedienfehler fall("Tag falsch", nichts, befund="abgebrochen", exit_code=2, tag="2026-13-40") # --- Windows-Teil -------------------------------------------------------------------------------- def windows_faelle() -> None: sys.path.insert(0, str(HIER)) import abendkontrolle as ak from datetime import datetime gut = {"tag": TAG, "befunde": [], "hinweise": ["auto_fortsetzen: fortgesetzt"], "exit": 0, "fakten": {"handelstag": True, "ende": {"grund": "boersenschluss", "exit": "0", "orders": "79", "position": "0"}, "laeufe": [{"takte_aktiv": 375}], "broker_offen": 0, "bericht": True, "killswitch": {"zustand": "aktiv", "ausloeser": "automatik"}}} def lauf(name, notiz_vorher, ergebnis, *, jetzt="2026-09-23T22:45:00", pruefe): with tempfile.TemporaryDirectory() as tmp: vault = Path(tmp) (vault / "Inbox").mkdir() notiz = vault / "Inbox" / f"{TAG}.md" if notiz_vorher is not None: notiz.write_bytes(notiz_vorher if isinstance(notiz_vorher, bytes) else notiz_vorher.encode("utf-8")) eingabe = vault / "ergebnis.json" eingabe.write_text(ergebnis if isinstance(ergebnis, str) else json.dumps(ergebnis), encoding="utf-8") code = ak.main(["--vault", str(vault), "--tag", TAG, "--ohne-wsl", str(eingabe), "--jetzt", jetzt]) text = notiz.read_bytes().decode("utf-8", errors="replace") if notiz.exists() else "" marke = (vault / ak.ABLAGE_TEIL / f"{TAG}.json").exists() try: ok, grund = pruefe(code, text, marke) except Exception as ausnahme: # der Treiber selbst bricht nicht ab ok, grund = False, f"Pruefung warf {ausnahme!r}; Notiz {text[-120:]!r}" zeile = f"{'OK ' if ok else 'FEHL'} W {name}: {grund}" print(zeile) if not ok: abweichungen.append(zeile) kopf = "---\ntype: daily\n---\n\n# 2026-09-23\n\n" lauf("neue Notiz, ohne Befund", None, gut, pruefe=lambda c, t, m: ( c == 0 and m and "## Agentenprotokoll" in t and "ohne Befund" in t and "type: daily" in t, f"code {c} marke {m}")) lauf("Abschnitt vor einer weiteren Ueberschrift", kopf + "## Agentenprotokoll\n\n- alt\n\n## Danach\n\ntext\n", gut, pruefe=lambda c, t, m: (t.index("ohne Befund") < t.index("## Danach") and "- alt\n- 22:45" in t and "\n\n## Danach" in t and "\n\n\n" not in t, repr(t[-80:]))) lauf("Abschnitt am Ende", kopf + "## Agentenprotokoll\n\n- alt\n", gut, pruefe=lambda c, t, m: (t.endswith(".\n - Hinweis: auto_fortsetzen: fortgesetzt\n") and "- alt\n- 22:45" in t, repr(t[-60:]))) lauf("ohne Abschnitt", kopf + "## Anderes\n\nx\n", gut, pruefe=lambda c, t, m: (t.count("## Agentenprotokoll") == 1 and t.index("## Anderes") < t.index("## Agentenprotokoll"), repr(t[-60:]))) lauf("Abschnitt direkt vor naechster Ueberschrift", kopf + "## Agentenprotokoll\n## Danach\n", gut, pruefe=lambda c, t, m: ("## Agentenprotokoll\n\n- 22:45" in t and "\n\n## Danach" in t, repr(t))) lauf("CRLF-Notiz", (kopf + "## Agentenprotokoll\n\n- alt\n").replace("\n", "\r\n"), gut, pruefe=lambda c, t, m: ("\n" not in t.replace("\r\n", ""), "gemischte Zeilenenden" if "\n" in t.replace("\r\n", "") else "ok")) mit_befund = {**gut, "befunde": ["Runner endete mit Exit 4", "Beim Broker stehen 2 offene Position(en)"]} lauf("mit Befund", None, mit_befund, pruefe=lambda c, t, m: ( "**2 Befund(e)** #needs-review" in t and " - Runner endete mit Exit 4" in t and m, repr(t[-160:]))) lauf("Ergebnis ohne befunde", None, {"tag": TAG}, pruefe=lambda c, t, m: ( c == 1 and not m and "**gescheitert**" in t and "#needs-review" in t, f"code {c} marke {m}")) lauf("Ergebnis kein JSON", None, "nicht json", pruefe=lambda c, t, m: (c == 1 and not m and "**gescheitert**" in t, f"code {c}")) lauf("kein Handelstag", None, {"tag": TAG, "befunde": [], "hinweise": [], "fakten": {"handelstag": False}}, pruefe=lambda c, t, m: ("kein Handelstag" in t and m, repr(t[-60:]))) # Befunde des Pruefers: Ueberschrift mit Zusatz, Notiz ohne UTF-8, Abbruch mit JSON, Nachbardatei lauf("Ueberschrift mit Zusatz", kopf + "## Agentenprotokoll — Sitzung 9\n\n- alt\n\n## Offene Punkte\n\n- x\n", gut, pruefe=lambda c, t, m: (t.count("## Agentenprotokoll") == 1 and t.index("ohne Befund") < t.index("## Offene Punkte"), repr(t[-120:]))) kaputt = "---\ntype: daily\n---\n\n# Tag äöü\n".encode("cp1252") lauf("Notiz kein UTF-8", kaputt, mit_befund, pruefe=lambda c, t, m: (m and c == 1, f"code {c} marke {m}")) lauf("Abbruch mit JSON (exit 2) wird wiederholt", None, {"tag": TAG, "befunde": ["Abendkontrolle abgebrochen: X"], "hinweise": [], "fakten": {}, "exit": 2}, pruefe=lambda c, t, m: (not m and "**1 Befund(e)**" in t and c == 1, f"code {c} marke {m}")) with tempfile.TemporaryDirectory() as tmp: vault = Path(tmp) (vault / "Inbox").mkdir() eingabe = vault / "e.json" eingabe.write_text(json.dumps(gut), encoding="utf-8") ak.main(["--vault", str(vault), "--tag", TAG, "--ohne-wsl", str(eingabe), "--jetzt", "2026-09-23T22:45:00"]) rest = [p.name for p in (vault / "Inbox").iterdir() if not p.name.endswith(".md")] ok = not rest print(f"{'OK ' if ok else 'FEHL'} W keine Nachbardatei uebrig: {rest}") if not ok: abweichungen.append("Nachbardatei uebrig") # faellige_tage from datetime import date with tempfile.TemporaryDirectory() as tmp: ablage = Path(tmp) def tage(jetzt): return ak.faellige_tage(datetime.fromisoformat(jetzt), ablage) faelle = [ ("Do vor 22:30: heute nicht", "2026-09-24T21:00", ["2026-09-24"], False), ("Do nach 22:30: heute", "2026-09-24T22:45", ["2026-09-24"], True), ("Mo frueh: Fr nachholen, Wochenende nicht", "2026-09-28T08:00", ["2026-09-25"], True), ("Mo frueh: nichts vor ERSTER_TAG", "2026-09-28T08:00", ["2026-09-23"], False), ] for name, jetzt, erwartet, drin in faelle: ergebnis = tage(jetzt) ok = all((tag in ergebnis) == drin for tag in erwartet) and not any( date.fromisoformat(t).weekday() >= 5 for t in ergebnis) zeile = f"{'OK ' if ok else 'FEHL'} T {name}: {ergebnis}" print(zeile) if not ok: abweichungen.append(zeile) (ablage / "2026-09-25.json").write_text("{}", encoding="utf-8") ergebnis = tage("2026-09-28T08:00") ok = "2026-09-25" not in ergebnis print(f"{'OK ' if ok else 'FEHL'} T Ergebnisdatei da: nicht erneut: {ergebnis}") if not ok: abweichungen.append("Ergebnisdatei ignoriert") ergebnis = tage("2026-10-12T08:00") ok = min(ergebnis) >= "2026-10-05" print(f"{'OK ' if ok else 'FEHL'} T hoechstens sieben Tage zurueck: {ergebnis}") if not ok: abweichungen.append("Nachholfenster") if __name__ == "__main__": if sys.platform == "win32": windows_faelle() else: wsl_faelle() print(f"ERGEBNIS: {len(abweichungen)} Abweichung(en)") sys.exit(1 if abweichungen else 0)