fix: KI-Robustheit — harter Wanduhr-Deadline + persistenter Ledger
Ein echter Durchlauf auf bredelar.info deckte zwei Bugs auf, die mit gemockten Tests unsichtbar waren: 1. Hang: attempt_timeout wurde nur als requests-Inaktivitäts-Timeout übergeben. Ein Free-Modell, das die Verbindung offen hält oder Tokens langsam tröpfeln lässt, umging ihn → der ganze Scan hing unbegrenzt (real auf einem Bild >5 min). Fix: _post_with_deadline führt den Call in einem Worker-Thread mit future.result(timeout=...) aus — hartes Wanduhr-Limit, bei Überschreitung Eskalation statt Hang. 2. Verlust bei Abbruch: Das Ledger wurde nur ganz am Ende gespeichert, ein Hang/Crash verwarf alle berechneten Verdikte. Fix: Zwischensicherung nach der Textphase + im finally — Verdikte überleben einen Abbruch. Zusätzlich attempt_timeout 30 → 20 s (flottere Eskalation). Im erneuten Lauf griff der Deadline 4× sauber, der Scan lief vollständig durch (Gelb: ein off_topic_commercial-Fund auf einer Gewerbe-Seite). Tests: 239 grün (+2: Deadline kappt Hang <2s, Textverdikte überleben Bildphasen-Crash). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
parent
a8dcaabf7c
commit
31730403b3
4 changed files with 75 additions and 15 deletions
|
|
@ -4,11 +4,14 @@ from unittest.mock import patch
|
|||
|
||||
import pytest
|
||||
|
||||
import time
|
||||
|
||||
from scanner.ai_analyzer import (
|
||||
run_ai_analysis,
|
||||
score_ai_findings,
|
||||
_text_fingerprint,
|
||||
_models_for,
|
||||
_openrouter_chat,
|
||||
)
|
||||
from scanner.baseline import BaselineManager
|
||||
from scanner.config import DEFAULT_CONFIG
|
||||
|
|
@ -304,6 +307,39 @@ class TestRedGate:
|
|||
assert out["level"] == "yellow"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Robustness: hard wall-clock deadline + resilient ledger save
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestRobustness:
|
||||
def test_hard_deadline_bounds_a_hanging_call(self):
|
||||
"""Ein hängender requests.post wird durch den Wanduhr-Deadline gekappt."""
|
||||
ai_cfg = {"attempt_timeout": 0.2}
|
||||
|
||||
def slow_post(*a, **k):
|
||||
time.sleep(3) # simuliert ein hängendes/tröpfelndes Modell
|
||||
|
||||
start = time.time()
|
||||
with patch("scanner.ai_analyzer.requests.post", side_effect=slow_post):
|
||||
out = _openrouter_chat("m", [{"role": "user", "content": "x"}], ai_cfg, "key", timeout=0.2)
|
||||
elapsed = time.time() - start
|
||||
assert out is None # graceful → None → Eskalation
|
||||
assert elapsed < 2.0 # NICHT die vollen 3s — Deadline hat gegriffen
|
||||
|
||||
def test_text_verdicts_survive_image_crash(self, tmp_path, monkeypatch):
|
||||
"""Textphase wird zwischengesichert → ein Crash der Bildphase verliert sie nicht."""
|
||||
monkeypatch.setenv("OPENROUTER_API_KEY", "test")
|
||||
bm = BaselineManager(tmp_path)
|
||||
cfg = _cfg()
|
||||
cfg["ai_analysis"]["image"]["enabled"] = True
|
||||
with patch("scanner.ai_analyzer._openrouter_chat", return_value=_verdict()), \
|
||||
patch("scanner.ai_analyzer._analyze_images", side_effect=RuntimeError("boom")):
|
||||
res = run_ai_analysis(cfg, bm, _snap(_LONG), {})
|
||||
# Bildphase krachte, aber das Text-Verdikt liegt persistent im Ledger
|
||||
assert res["skipped"] and "interner Fehler" in res["skipped"]
|
||||
assert len(bm.load_ai_ledger()["entries"]) == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Model escalation chain
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue