"""Mutationsproben für die TypeScript-Prüfungen des Plugin-Patches (Phase E). Je Probe genau eine Ersetzung auf der COMMITTETEN Basis: 1. Worktree muss sauber sein (git status --porcelain leer). 2. Der Suchtext muss genau einmal vorkommen; Ersetzung schreiben. 3. run-tests.sh mit den zugeordneten Testdateien ausführen (Logs behalten). 4. Rote Tests aus den Logs lesen; jede erwartete Testbezeichnung muss rot sein (sonst ist das Rot ggf. "aus dem falschen Grund") und jede rote Datei muss geladen haben (pass > 0 oder fail > 0, kein Ladefehler). 5. git checkout -- ; Worktree wieder sauber; Blob-Hash = HEAD. Aufruf: python3 -I telegram/plugin/test/mutationsproben.py [ID ...] """ import json import os import re import shutil import subprocess import sys import time REPO = os.path.realpath(os.path.join(os.path.dirname(os.path.realpath(__file__)), "..", "..", "..")) RUNNER = os.path.join(REPO, "telegram", "plugin", "test", "run-tests.sh") GW = "telegram/plugin/aria-gateway.ts" SRV = "telegram/plugin/server.ts" PROBES = [ {"id": "T01", "check": "sender_id-Quelle (Weiterleitungsfeld)", "file": GW, "old": " const senderId = String(from.id)\n", "new": " const senderId = String(message.forward_origin?.sender_user?.id ?? from.id)\n", "tests": ["gateway-unit.test.ts", "modes.test.ts"], "expect": ["gefälschte IDs in Text, Caption, Weiterleitung", "Modus active > gefälschte sender_id", "Modus shadow > gefälschte sender_id"]}, {"id": "T02", "check": "sender_id-Quelle (Text)", "file": GW, "old": " const senderId = String(from.id)\n", "new": " const senderId = String(/sender_id\\D{0,4}(\\d{1,20})/.exec(text)?.[1] ?? from.id)\n", "tests": ["gateway-unit.test.ts", "modes.test.ts"], "expect": ["gefälschte IDs in Text, Caption, Weiterleitung", "Modus active > gefälschte sender_id"]}, {"id": "T03", "check": "Blocked nicht weiterleiten", "file": GW, "old": " return { forward: false, reply: r.reply_text ?? TEXTS.blocked, notice: null, outcome: 'BLOCKED' }\n", "new": " return { forward: true, reply: r.reply_text ?? TEXTS.blocked, notice: null, outcome: 'BLOCKED' }\n", "tests": ["gateway-unit.test.ts", "modes.test.ts", "e2e-python.test.ts"], "expect": ["reply/blocked/failed/drop/not_intercepted: unter BEIDEN", "Modus active > Blocked", "verweigerte Freigaben kippen nie", "Modusbefehle: /codex setzt den Chat-Modus"]}, {"id": "T04", "check": "Verweigerung aus ungültiger Antwort respektieren", "file": GW, "old": " if ((outcome.rawHandling && NON_FORWARD_HANDLINGS.has(outcome.rawHandling))\n", "new": " if (false && (outcome.rawHandling && NON_FORWARD_HANDLINGS.has(outcome.rawHandling))\n", "tests": ["gateway-unit.test.ts", "modes.test.ts"], "expect": ["Verweigerung mit beigelegter Weiterleitung", "verweigerte Freigaben kippen nie"]}, {"id": "T05", "check": "Modus-Allowlist (Einhängepunkt erst nach gate())", "file": SRV, "old": " const result = gate(ctx)\n", "new": " if (ariaGateway.enabled && /^\\/(auto|codex|claude)\\b/.test(text.trim()) && !(await ariaGateway.inbound({ bot, ctx, text, imagePath: undefined, attachment }))) return\n const result = gate(ctx)\n", "tests": ["modes.test.ts", "e2e-python.test.ts"], "expect": ["Modus active > Modusbefehl von Fremdem", "Modus shadow > Modusbefehl von Fremdem", "Fremder erreicht das Python-Gateway nie"]}, {"id": "T06", "check": "Fehler-Default claude_status_quo", "file": GW, "old": " let onFailure: FailurePolicy = 'claude_status_quo'\n", "new": " let onFailure: FailurePolicy = 'block'\n", "tests": ["gateway-unit.test.ts", "modes.test.ts", "e2e-python.test.ts"], "expect": ["Ausfallregel: Default claude_status_quo", "Modus active > Router-Timeout", "Gateway-Fehler (Schlüssel fehlt, Konfiguration fehlt)"]}, {"id": "T07", "check": "Ausfallregel block wird beachtet", "file": GW, "old": " if (onFailure === 'claude_status_quo') return { forward: true, reply: null, notice: null, outcome: `ROUTER_FAILURE:${code}` }\n", "new": " if (onFailure || true) return { forward: true, reply: null, notice: null, outcome: `ROUTER_FAILURE:${code}` }\n", "tests": ["gateway-unit.test.ts", "modes.test.ts"], "expect": ["Router-Ausfall: claude_status_quo leitet weiter, block nicht", "Modus active > Router-Absturz"]}, {"id": "T08", "check": "Ein Poller: kein bot.start() im Gateway (Laufzeit + statisch)", "file": GW, "old": " if (settings.mode === 'off') return Promise.resolve(true)\n", "new": " if (settings.mode === 'off') return Promise.resolve(true)\n void (input.bot as any).start?.({})\n", "tests": ["static.test.ts", "modes.test.ts"], "expect": ["aria-gateway.ts: nur erlaubte Importe, kein grammy, kein Bot, kein Polling", "genau ein Bot-Objekt, ein bot.start()"]}, {"id": "T09", "check": "Ein Poller: kein getUpdates im Gateway (Laufzeit + statisch)", "file": GW, "old": " if (settings.mode === 'off') return Promise.resolve(true)\n", "new": " if (settings.mode === 'off') return Promise.resolve(true)\n void (input.bot as any).api?.raw?.getUpdates?.({ offset: -1 })?.catch?.(() => {})\n", "tests": ["static.test.ts", "modes.test.ts"], "expect": ["aria-gateway.ts: nur erlaubte Importe, kein grammy, kein Bot, kein Polling", "genau ein Bot-Objekt, ein bot.start()"]}, {"id": "T10", "check": "Ein Poller: Patch fügt Token-/PID-Logik hinzu (statisch)", "file": SRV, "old": " const ariaForward = ariaGateway.enabled", "new": " try { writeFileSync(PID_FILE, String(process.pid)) } catch {}\n const ariaForward = ariaGateway.enabled", "tests": ["static.test.ts"], "expect": ["hinzugefügte Patch-Zeilen enthalten keine Poller-, Token- oder PID-Logik", "server.ts hat dieselbe Anzahl Poller-relevanter Stellen wie Prod"]}, {"id": "T11", "check": "Dedup je (chat_id, message_id)", "file": GW, "old": " if (seen.has(key)) {\n", "new": " if (false && seen.has(key)) {\n", "tests": ["gateway-unit.test.ts", "modes.test.ts"], "expect": ["active: doppelte message_id wird nur einmal verarbeitet", "Modus active > doppelte message_id"]}, {"id": "T12", "check": "Payload-Bindung (content)", "file": GW, "old": " if (content !== expected.content && content !== oneShotRemainder(expected.content)) return 'FORWARD_CONTENT_MISMATCH'\n", "new": "", "tests": ["gateway-unit.test.ts", "modes.test.ts"], "expect": ["Weiterleitung mit falscher Bindung ist ein Router-Ausfall", "Modus active > ungültige Bridge-Antwort"]}, {"id": "T13", "check": "Shadow wartet nie auf die Bridge", "file": GW, "old": " void track(shadow(settings, input))\n return Promise.resolve(true) // Weiterleitung unverändert, ohne auf die Bridge zu warten\n", "new": " return track(shadow(settings, input)).then(() => true)\n", "tests": ["gateway-unit.test.ts", "modes.test.ts"], "expect": ["shadow: liefert sofort true", "Modus shadow > Router-Timeout", "off/shadow: Permission-Antwort sofort"]}, {"id": "T14", "check": "Timeout tötet die Prozessgruppe", "file": GW, "old": " failure = code\n killGroup()\n", "new": " failure = code\n", "tests": ["gateway-unit.test.ts"], "expect": ["Timeout: Prozessgruppe wird getötet"]}, {"id": "T15", "check": "Kind-Umgebung nur PATH (kein Token-Leck)", "file": GW, "old": " env: { PATH: '/usr/bin:/bin' },\n", "new": " env: { ...process.env, PATH: '/usr/bin:/bin' },\n", "tests": ["gateway-unit.test.ts", "static.test.ts"], "expect": ["Kindprozess: nur PATH in der Umgebung", "aria-gateway.ts: nur erlaubte Importe"]}, # T16 überlebt unter bun 1.3.13 erwartbar: bun meldet hier kein stdin-'error' (Befund, MUTATIONSPROBEN.md). {"id": "T16", "check": "stdin-Fehler nicht fatal (gültige Antwort zählt)", "file": GW, "old": " child.stdin?.on('error', () => { stdinFailed = true })\n", "new": " child.stdin?.on('error', () => { stdinFailed = true; fail('BRIDGE_STDIN_FAILED') })\n", "tests": ["gateway-unit.test.ts"], "expect": ["Bridge antwortet ohne stdin zu lesen"]}, {"id": "T17", "check": "Supabase-Log unabhängig von der Weiterleitung", "file": SRV, "old": " if (SB_URL && SB_KEY) {\n", "new": " if (ariaForward && SB_URL && SB_KEY) {\n", "tests": ["static.test.ts", "modes.test.ts"], "expect": ["der Supabase-Log-Block ist unverändert", "Modus active > Codex-Route"]}, {"id": "T18", "check": "Schemaprüfung: unbekanntes handling", "file": GW, "old": " if (typeof value.handling !== 'string' || !(HANDLINGS as readonly string[]).includes(value.handling)) return null\n", "new": "", "tests": ["gateway-unit.test.ts"], "expect": ["jede Abweichung vom Schema wird abgelehnt"]}, ] def git(*args): return subprocess.run(["/usr/bin/git", *args], cwd=REPO, capture_output=True, text=True, check=True).stdout def clean(): return git("status", "--porcelain", "-uall") == "" def run_tests(files, tmpdir): env = {"PATH": "/usr/bin:/bin", "HOME": tmpdir, "ARIA_TEST_TMPDIR": tmpdir, "ARIA_TEST_KEEP": "1", "LANG": "C.UTF-8"} proc = subprocess.run(["/bin/sh", RUNNER, *files], cwd=REPO, env=env, capture_output=True, text=True, timeout=900) out = proc.stdout + proc.stderr match = re.search(r"^Logs: (\S+)/out$", out, re.M) work = match.group(1) if match else None failing, loaded = [], {} if work and os.path.isdir(os.path.join(work, "out")): for name in sorted(os.listdir(os.path.join(work, "out"))): text = open(os.path.join(work, "out", name), encoding="utf-8", errors="replace").read() failing += [line[len("(fail) "):].strip() for line in text.splitlines() if line.startswith("(fail) ")] p = re.findall(r"^\s*(\d+) pass$", text, re.M) f = re.findall(r"^\s*(\d+) fail$", text, re.M) loaded[name] = (int(p[-1]) if p else 0, int(f[-1]) if f else 0) shutil.rmtree(work, ignore_errors=True) summary = re.search(r"^SUMME: .*$", out, re.M) return proc.returncode, sorted(set(failing)), loaded, summary.group(0) if summary else out[-400:] def main(): result_path = sys.argv[1] wanted = set(sys.argv[2:]) tmpdir = os.environ.get("ARIA_MUTATION_TMPDIR") or "/tmp" results = json.load(open(result_path)) if os.path.exists(result_path) else {} head = git("rev-parse", "HEAD").strip() for probe in PROBES: if wanted and probe["id"] not in wanted: continue if not clean(): raise SystemExit("Worktree nicht sauber - Abbruch vor " + probe["id"]) path = os.path.join(REPO, probe["file"]) original = open(path, encoding="utf-8").read() occurrences = original.count(probe["old"]) if occurrences != 1: raise SystemExit(f"{probe['id']}: Suchtext {occurrences}x statt genau 1x") blob_before = git("hash-object", probe["file"]).strip() started = time.time() try: with open(path, "w", encoding="utf-8") as handle: handle.write(original.replace(probe["old"], probe["new"], 1)) rc, failing, loaded, summary = run_tests(probe["tests"], tmpdir) finally: git("checkout", "--", probe["file"]) if not clean() or git("hash-object", probe["file"]).strip() != blob_before: raise SystemExit(probe["id"] + ": Zurücksetzen fehlgeschlagen") expected_hits = {e: [t for t in failing if e in t] for e in probe["expect"]} load_errors = [name for name, (p, f) in loaded.items() if p + f == 0] results[probe["id"]] = { "check": probe["check"], "file": probe["file"], "base": head, "tests": probe["tests"], "red": len(failing), "failing": failing, "expected": probe["expect"], "expected_all_red": all(expected_hits.values()), "missing_expected": [e for e, v in expected_hits.items() if not v], "load_errors": load_errors, "summary": summary, "seconds": round(time.time() - started, 1), } print(f"{probe['id']} {probe['check']}: rot={len(failing)} erwartet_rot={results[probe['id']]['expected_all_red']} " f"fehlend={results[probe['id']]['missing_expected']} ladefehler={load_errors}", flush=True) with open(result_path, "w", encoding="utf-8") as handle: json.dump(results, handle, ensure_ascii=False, indent=1) print("sauber:", clean(), "HEAD:", head) if __name__ == "__main__": main()