8e5f65d864
Regelbasierte Dedup-Prüfung bleibt maßgeblich; nur Grenzfälle gehen an ein LLM: Titel mit Wort-Überschneidung unter der Fuzzy-Schwelle (Editionen, Big Boxes, Übersetzungen wie 'El Grande Big Box' vs. 'El Grande: 25 Jahre') und ein ergänzender Call zum Verlags-/Titel-Hinweis bei Verlags-Konflikt (Verlags-Aliase). Klare Treffer und Non-Treffer bleiben regelbasiert. - Neu: ki_pruefung.py — Env-Konfiguration (SPIELE_DEDUP_LLM_*), robuster JSON-Parser (Code-Fences, Zusatztext), OpenAI-kompatibler Client via httpx mit Timeout und genau einem Retry; liefert bei jedem Fehler None. - pruefung.py: bestätigte Grenzfälle laufen in die üblichen Prüfungen a)-d) zurück; deutscher Titel füllt die Titel-Empfehlung, Verlags-Empfehlung landet im Konflikt-Detail. Neues Feld PruefErgebnis.ki_befunde. - __init__.py: Aktivierungsprüfung vor jeder Fabrik — aus/unvollständig heißt nie ein LLM-Aufruf; KI-Befunde werden über die audit-log-API protokolliert (Modell, Konfidenz, Entscheidung), best effort. - Tests: gemockte HTTP-Antworten (MockTransport/Fakes), Fallback-Fälle (aktiv=0, kein Key, Timeout, kaputtes JSON), Konfidenz-Schwelle, Parsing-Robustheit; kein echter LLM-Call in CI. - README.md um die neuen Env-Variablen und den Zweitprüfungs-Abschnitt ergänzt.
528 lines
20 KiB
Python
528 lines
20 KiB
Python
"""Tests: KI-Zweitprüfung des Plugins „dedup“.
|
||
|
||
Alle LLM-Antworten werden über httpx.MockTransport bzw. Fake-Clients gestellt —
|
||
in CI läuft niemals ein echter LLM-Call. Getestet werden: Konfiguration,
|
||
JSON-Parsing-Robustheit (Code-Fences, Zusatztext), Retry-/Fallback-Verhalten,
|
||
Konfidenz-Schwelle und die Rückführung bestätigter Treffer in die Regelprüfung.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
|
||
import httpx
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy import select
|
||
|
||
from redaktionskern.app import create_app
|
||
|
||
from plugins.dedup.ki_pruefung import (
|
||
VERSUCHE,
|
||
KiKonfiguration,
|
||
KiPruefClient,
|
||
extrahiere_json,
|
||
lade_konfiguration,
|
||
validiere_antwort,
|
||
)
|
||
|
||
from tests.conftest import lege_benutzer_an
|
||
from tests.test_dedup import benutzer_objekt, neuheit_anlegen, planung_anlegen, plugin_modul
|
||
|
||
|
||
# ---------------- Hilfen ----------------
|
||
|
||
def gueltige_llm_antwort(inhalt: str, status: int = 200) -> httpx.Response:
|
||
"""Wrappt einen Assistant-Text in eine Chat-Completions-Antwort."""
|
||
body = {"choices": [{"message": {"role": "assistant", "content": inhalt}}]}
|
||
return httpx.Response(status, content=json.dumps(body).encode())
|
||
|
||
|
||
def mock_transport(antworten: list) -> tuple[httpx.MockTransport, dict]:
|
||
"""Transport mit gestaffelten Antworten: str = Assistant-Inhalt,
|
||
int = HTTP-Status, Exception = wird geworfen. Zählt die Aufrufe mit."""
|
||
zaehler = {"aufrufe": 0}
|
||
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
index = min(zaehler["aufrufe"], len(antworten) - 1)
|
||
zaehler["aufrufe"] += 1
|
||
spezifikation = antworten[index]
|
||
if isinstance(spezifikation, Exception):
|
||
raise spezifikation
|
||
if isinstance(spezifikation, int):
|
||
return httpx.Response(spezifikation, json={"error": "kaputt"})
|
||
return gueltige_llm_antwort(spezifikation)
|
||
|
||
return httpx.MockTransport(handler), zaehler
|
||
|
||
|
||
class FakeKiClient:
|
||
"""Ersetzt den KI-Client komplett: feste Antwort, zählt Aufrufe."""
|
||
|
||
modell = "fake-modell"
|
||
|
||
def __init__(self, antwort=None):
|
||
self.antwort = antwort
|
||
self.aufrufe: list[dict] = []
|
||
|
||
def bewerte(self, **kwargs):
|
||
self.aufrufe.append(kwargs)
|
||
return self.antwort
|
||
|
||
def schliessen(self):
|
||
pass
|
||
|
||
|
||
GLEICHES_SPIEL_HOCH = {
|
||
"ist_gleiches_spiel": True, "konfidenz": 0.9,
|
||
"begruendung": "Beides El Grande-Jubiläumsausgaben.",
|
||
"empfohlener_verlag": None, "deutscher_titel": None,
|
||
}
|
||
|
||
|
||
def app_mit_ki(settings, monkeypatch, *, konfidenz_min=None):
|
||
"""App, deren Plugin-Umgebung die KI-Zweitprüfung aktiviert."""
|
||
monkeypatch.setenv("SPIELE_DEDUP_LLM_AKTIV", "1")
|
||
monkeypatch.setenv("SPIELE_DEDUP_LLM_BASIS_URL", "https://llm.example.test/v1")
|
||
monkeypatch.setenv("SPIELE_DEDUP_LLM_API_KEY", "test-key")
|
||
monkeypatch.setenv("SPIELE_DEDUP_LLM_MODELL", "test-modell")
|
||
if konfidenz_min is not None:
|
||
monkeypatch.setenv("SPIELE_DEDUP_LLM_KONFIDENZ_MIN", str(konfidenz_min))
|
||
return create_app(settings)
|
||
|
||
|
||
def pruefe_mit_ki(app, antworten, *args, **kwargs):
|
||
"""check_titel mit per Fabrik eingehängtem Mock-LLM (Antwort-Stafette)."""
|
||
transport, zaehler = mock_transport(antworten)
|
||
dedup = app.state.registry.get("dedup")
|
||
dedup.ki_client_fabrik = lambda: KiPruefClient(
|
||
KiKonfiguration(
|
||
aktiv=True, basis_url="https://llm.example.test/v1",
|
||
api_key="test-key", modell="test-modell",
|
||
konfidenz_min=dedup.ki_konfiguration.konfidenz_min,
|
||
),
|
||
transport=transport,
|
||
)
|
||
try:
|
||
ergebnis = asyncio.run(dedup.check_titel(*args, **kwargs))
|
||
finally:
|
||
dedup.ki_client_fabrik = None
|
||
return ergebnis, zaehler
|
||
|
||
|
||
# ---------------- Konfiguration ----------------
|
||
|
||
def test_konfiguration_standard_aus():
|
||
k = lade_konfiguration({})
|
||
assert not k.aktiv
|
||
assert not k.vollstaendig
|
||
assert k.konfidenz_min == 0.7
|
||
assert k.timeout >= 1.0
|
||
|
||
|
||
def test_konfiguration_aus_env_und_fehlertolerant():
|
||
k = lade_konfiguration({
|
||
"SPIELE_DEDUP_LLM_AKTIV": "1",
|
||
"SPIELE_DEDUP_LLM_BASIS_URL": "https://api.example.test/v1/",
|
||
"SPIELE_DEDUP_LLM_API_KEY": " sk-123 ",
|
||
"SPIELE_DEDUP_LLM_MODELL": " gpt-test ",
|
||
# Kaputte Zahlen fallen auf die Standards zurück:
|
||
"SPIELE_DEDUP_LLM_KONFIDENZ_MIN": "abc",
|
||
"SPIELE_DEDUP_LLM_TIMEOUT_SEKUNDEN": "-5",
|
||
})
|
||
assert k.vollstaendig
|
||
assert k.basis_url == "https://api.example.test/v1"
|
||
assert k.api_key == "sk-123"
|
||
assert k.modell == "gpt-test"
|
||
assert k.konfidenz_min == 0.7 # Standard bei Unsinn
|
||
assert k.timeout >= 1.0
|
||
|
||
|
||
def test_konfiguration_schwellenwerte_werden_geklemmt():
|
||
k = lade_konfiguration({"SPIELE_DEDUP_LLM_KONFIDENZ_MIN": "1.5"})
|
||
assert k.konfidenz_min == 1.0
|
||
k = lade_konfiguration({"SPIELE_DEDUP_LLM_TIMEOUT_SEKUNDEN": "3.5"})
|
||
assert k.timeout == 3.5
|
||
|
||
|
||
# ---------------- JSON-Parsing-Robustheit ----------------
|
||
|
||
def test_extrahiere_json_plain():
|
||
assert extrahiere_json('{"ist_gleiches_spiel": true}') == {"ist_gleiches_spiel": True}
|
||
|
||
|
||
def test_extrahiere_json_code_fence():
|
||
text = 'Hier die Bewertung:\n\n```json\n{"ist_gleiches_spiel": false}\n```\n\nViele Grüße'
|
||
assert extrahiere_json(text) == {"ist_gleiches_spiel": False}
|
||
|
||
|
||
def test_extrahiere_json_text_um_json():
|
||
text = 'Klar! {"ist_gleiches_spiel": true, "konfidenz": 0.8} — hoffe das hilft.'
|
||
assert extrahiere_json(text)["konfidenz"] == 0.8
|
||
|
||
|
||
def test_extrahiere_json_unbrauchbar():
|
||
assert extrahiere_json("Ich weiß es leider nicht.") is None
|
||
assert extrahiere_json("[1, 2, 3]") is None # kein Objekt
|
||
assert extrahiere_json("") is None
|
||
assert extrahiere_json(None) is None
|
||
|
||
|
||
def test_validiere_antwort_normalisiert():
|
||
roh = {
|
||
"ist_gleiches_spiel": "ja", # Text statt Bool
|
||
"konfidenz": 92, # Prozent-Skala
|
||
"begruendung": " Gleiche Edition. ",
|
||
"empfohlener_verlag": " Kosmos ",
|
||
"deutscher_titel": "",
|
||
}
|
||
sauber = validiere_antwort(roh)
|
||
assert sauber == {
|
||
"ist_gleiches_spiel": True,
|
||
"konfidenz": 0.92,
|
||
"begruendung": "Gleiche Edition.",
|
||
"empfohlener_verlag": "Kosmos",
|
||
"deutscher_titel": None,
|
||
}
|
||
|
||
|
||
@pytest.mark.parametrize("roh", [
|
||
{"ist_gleiches_spiel": True}, # konfidenz fehlt
|
||
{"ist_gleiches_spiel": "vielleicht", "konfidenz": 0.9},
|
||
{"ist_gleiches_spiel": True, "konfidenz": "hoch"},
|
||
{"ist_gleiches_spiel": True, "konfidenz": 7}, # weder 0–1 noch 0–100
|
||
["kein", "objekt"],
|
||
])
|
||
def test_validiere_antwort_lehnt_unbrauchbares_ab(roh):
|
||
assert validiere_antwort(roh) is None
|
||
|
||
|
||
# ---------------- Client: Retry, Fehler, Header ----------------
|
||
|
||
KONFIG = KiKonfiguration(
|
||
aktiv=True, basis_url="https://llm.example.test/v1",
|
||
api_key="geheim", modell="test-modell",
|
||
)
|
||
|
||
ANTWORT_JSON = '{"ist_gleiches_spiel": true, "konfidenz": 0.9, "begruendung": "b", ' \
|
||
'"empfohlener_verlag": null, "deutscher_titel": null}'
|
||
|
||
|
||
def test_client_sendet_modell_und_key():
|
||
gesehen = {}
|
||
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
gesehen["url"] = str(request.url)
|
||
gesehen["auth"] = request.headers.get("Authorization")
|
||
gesehen["body"] = json.loads(request.content.decode())
|
||
return gueltige_llm_antwort(ANTWORT_JSON)
|
||
|
||
client = KiPruefClient(KONFIG, transport=httpx.MockTransport(handler))
|
||
try:
|
||
ergebnis = client.bewerte(titel_a="A", titel_b="B")
|
||
finally:
|
||
client.schliessen()
|
||
assert ergebnis["ist_gleiches_spiel"] is True
|
||
assert gesehen["url"].endswith("/v1/chat/completions")
|
||
assert gesehen["auth"] == "Bearer geheim"
|
||
assert gesehen["body"]["model"] == "test-modell"
|
||
|
||
|
||
def test_client_retryt_genau_einmal_bei_serverstoerung():
|
||
transport, zaehler = mock_transport([500, ANTWORT_JSON])
|
||
client = KiPruefClient(KONFIG, transport=transport)
|
||
try:
|
||
ergebnis = client.bewerte(titel_a="A", titel_b="B")
|
||
finally:
|
||
client.schliessen()
|
||
assert ergebnis is not None
|
||
assert zaehler["aufrufe"] == 2 # erster Versuch + genau ein Retry
|
||
|
||
|
||
def test_client_gibt_nach_zweitem_fehler_auf():
|
||
transport, zaehler = mock_transport([503, 503])
|
||
client = KiPruefClient(KONFIG, transport=transport)
|
||
try:
|
||
assert client.bewerte(titel_a="A", titel_b="B") is None
|
||
finally:
|
||
client.schliessen()
|
||
assert zaehler["aufrufe"] == VERSUCHE
|
||
|
||
|
||
def test_client_timeout_wird_genau_einmal_wiederholt():
|
||
transport, zaehler = mock_transport([
|
||
httpx.ReadTimeout("zu langsam"), httpx.ReadTimeout("weiter zu langsam"),
|
||
])
|
||
client = KiPruefClient(KONFIG, transport=transport)
|
||
try:
|
||
assert client.bewerte(titel_a="A", titel_b="B") is None
|
||
finally:
|
||
client.schliessen()
|
||
assert zaehler["aufrufe"] == 2
|
||
|
||
|
||
def test_client_kein_retry_bei_clientfehler_oder_kaputtem_json():
|
||
# 404 ist nicht wiederholbar:
|
||
transport, zaehler = mock_transport([404, 404])
|
||
client = KiPruefClient(KONFIG, transport=transport)
|
||
try:
|
||
assert client.bewerte(titel_a="A", titel_b="B") is None
|
||
finally:
|
||
client.schliessen()
|
||
assert zaehler["aufrufe"] == 1
|
||
|
||
# Kaputtes JSON hilft kein Retry:
|
||
transport, zaehler = mock_transport(["Leider weiß ich das nicht."])
|
||
client = KiPruefClient(KONFIG, transport=transport)
|
||
try:
|
||
assert client.bewerte(titel_a="A", titel_b="B") is None
|
||
finally:
|
||
client.schliessen()
|
||
assert zaehler["aufrufe"] == 1
|
||
|
||
|
||
def test_client_ohne_vollstaendige_konfig_ruft_nichts():
|
||
transport, zaehler = mock_transport([ANTWORT_JSON])
|
||
leer = KiKonfiguration(aktiv=True, basis_url="", api_key="", modell="")
|
||
client = KiPruefClient(leer, transport=transport)
|
||
try:
|
||
assert client.bewerte(titel_a="A", titel_b="B") is None
|
||
finally:
|
||
client.schliessen()
|
||
assert zaehler["aufrufe"] == 0
|
||
|
||
|
||
# ---------------- Integration: Fallback & Verhalten ----------------
|
||
|
||
def test_ohne_ki_verhaelt_sich_alles_wie_bisher(settings, monkeypatch):
|
||
"""Standard (aktiv=0): Grenzfälle bleiben regelbasiert, kein LLM-Aufruf."""
|
||
app = create_app(settings)
|
||
lege_benutzer_an(app, "rez1", "rezensent")
|
||
rez1 = benutzer_objekt(app, "rez1")
|
||
planung_anlegen(app, "El Grande Big Box", rez1.id)
|
||
|
||
dedup = app.state.registry.get("dedup")
|
||
|
||
def aufschrei(): # dürfte nie aufgerufen werden
|
||
raise AssertionError("KI war deaktiviert, wurde aber aufgerufen!")
|
||
|
||
dedup.ki_client_fabrik = lambda: aufschrei()
|
||
try:
|
||
ergebnis = asyncio.run(dedup.check_titel("El Grande: 25 Jahre"))
|
||
finally:
|
||
dedup.ki_client_fabrik = None
|
||
assert not ergebnis.hat_konflikte
|
||
assert ergebnis.ki_befunde == []
|
||
# Empfehlung bleibt beim eigenen Titel (Heuristik ohne Varianten):
|
||
assert ergebnis.titel_empfehlung == "El Grande: 25 Jahre"
|
||
|
||
|
||
def test_aktiv_ohne_key_faellt_auf_regeln_zurueck(settings, monkeypatch):
|
||
"""AKTIV=1, aber kein API-Key → Zweitprüfung bleibt aus, kein Fehler."""
|
||
monkeypatch.setenv("SPIELE_DEDUP_LLM_AKTIV", "1")
|
||
monkeypatch.setenv("SPIELE_DEDUP_LLM_BASIS_URL", "https://llm.example.test/v1")
|
||
# bewusst KEIN SPIELE_DEDUP_LLM_API_KEY / MODELL:
|
||
app = create_app(settings)
|
||
lege_benutzer_an(app, "rez1", "rezensent")
|
||
rez1 = benutzer_objekt(app, "rez1")
|
||
planung_anlegen(app, "El Grande Big Box", rez1.id)
|
||
|
||
dedup = app.state.registry.get("dedup")
|
||
dedup.ki_client_fabrik = lambda: (_ for _ in ()).throw(
|
||
AssertionError("Ohne Key darf kein KI-Client gebaut werden.")
|
||
)
|
||
try:
|
||
ergebnis = asyncio.run(dedup.check_titel("El Grande: 25 Jahre"))
|
||
finally:
|
||
dedup.ki_client_fabrik = None
|
||
assert not ergebnis.hat_konflikte
|
||
assert ergebnis.ki_befunde == []
|
||
|
||
|
||
def test_llm_fehler_blockiert_nie(settings, monkeypatch):
|
||
"""Netzwerk-/Server-Probleme: Ergebnis identisch zur reinen Regelprüfung."""
|
||
app = app_mit_ki(settings, monkeypatch)
|
||
lege_benutzer_an(app, "rez1", "rezensent")
|
||
rez1 = benutzer_objekt(app, "rez1")
|
||
planung_anlegen(app, "El Grande Big Box", rez1.id, verlag="Rio Grande Games")
|
||
|
||
ergebnis, zaehler = pruefe_mit_ki(
|
||
app,
|
||
[httpx.ConnectError("offline")] * 6,
|
||
"El Grande: 25 Jahre", "Devir", None,
|
||
)
|
||
assert zaehler["aufrufe"] >= 1
|
||
# Ohne funktionierende Zweitprüfung bleibt es beim Regelbefund (hier leer —
|
||
# die Grenzfall-Paarung matchen die Regeln ja gerade nicht):
|
||
assert not ergebnis.hat_konflikte
|
||
assert all(b["entscheidung"] == "fehler_verworfen" for b in ergebnis.ki_befunde)
|
||
|
||
# Exakt dasselbe Ergebnis wie ohne KI:
|
||
dedup = app.state.registry.get("dedup")
|
||
referenz = asyncio.run(dedup.check_titel("El Grande: 25 Jahre", "Devir", None))
|
||
assert referenz.als_liste() == ergebnis.als_liste()
|
||
assert referenz.titel_empfehlung == ergebnis.titel_empfehlung
|
||
assert referenz.verlags_optionen == ergebnis.verlags_optionen
|
||
|
||
|
||
# ---------------- Integration: Grenzfälle & Schwelle ----------------
|
||
|
||
def test_grenzfall_wird_durch_ki_bestaetigt(settings, monkeypatch):
|
||
"""Big Box vs. Jubiläumsausgabe: Regeln finden nichts, das LLM schon."""
|
||
app = app_mit_ki(settings, monkeypatch)
|
||
lege_benutzer_an(app, "rez1", "rezensent")
|
||
rez1 = benutzer_objekt(app, "rez1")
|
||
planung_anlegen(app, "El Grande Big Box", rez1.id)
|
||
|
||
antwort = dict(GLEICHES_SPIEL_HOCH)
|
||
antwort["begruendung"] = "Beides sind El Grande-Ausgaben der Jubiläumsreihe."
|
||
ergebnis, _ = pruefe_mit_ki(app, [json.dumps(antwort)], "El Grande: 25 Jahre")
|
||
|
||
arten = [k.art for k in ergebnis.konflikte]
|
||
assert "planung" in arten
|
||
planungs_konflikt = next(k for k in ergebnis.konflikte if k.art == "planung")
|
||
assert planungs_konflikt.details["ki_bestaetigt"] is True
|
||
assert planungs_konflikt.details["ki_konfidenz"] == 0.9
|
||
|
||
assert ergebnis.ki_befunde[0]["entscheidung"] == "gleiches_spiel"
|
||
assert ergebnis.ki_befunde[0]["modell"] == "test-modell"
|
||
assert ergebnis.ki_befunde[0]["konfidenz"] == 0.9
|
||
|
||
|
||
def test_grenzfall_unter_konfidenzschwelle_wird_verworfen(settings, monkeypatch):
|
||
"""Gleicher Befund, aber Konfidenz 0.5 < 0.7 → nur der Regelbefund zählt."""
|
||
app = app_mit_ki(settings, monkeypatch)
|
||
lege_benutzer_an(app, "rez1", "rezensent")
|
||
rez1 = benutzer_objekt(app, "rez1")
|
||
planung_anlegen(app, "El Grande Big Box", rez1.id)
|
||
|
||
antwort = dict(GLEICHES_SPIEL_HOCH, konfidenz=0.5)
|
||
ergebnis, _ = pruefe_mit_ki(app, [json.dumps(antwort)], "El Grande: 25 Jahre")
|
||
assert not ergebnis.hat_konflikte
|
||
assert ergebnis.ki_befunde[0]["entscheidung"] == "verworfen_unter_schwelle"
|
||
|
||
|
||
def test_konfidenzschwelle_laesst_sich_per_env_verschieben(settings, monkeypatch):
|
||
app = app_mit_ki(settings, monkeypatch, konfidenz_min=0.95)
|
||
lege_benutzer_an(app, "rez1", "rezensent")
|
||
rez1 = benutzer_objekt(app, "rez1")
|
||
planung_anlegen(app, "El Grande Big Box", rez1.id)
|
||
|
||
antwort = dict(GLEICHES_SPIEL_HOCH) # 0.9 reicht jetzt nicht mehr
|
||
ergebnis, _ = pruefe_mit_ki(app, [json.dumps(antwort)], "El Grande: 25 Jahre")
|
||
assert not ergebnis.hat_konflikte
|
||
assert ergebnis.ki_befunde[0]["entscheidung"] == "verworfen_unter_schwelle"
|
||
|
||
|
||
def test_llm_sagt_verschiedene_spiele(settings, monkeypatch):
|
||
app = app_mit_ki(settings, monkeypatch)
|
||
lege_benutzer_an(app, "rez1", "rezensent")
|
||
rez1 = benutzer_objekt(app, "rez1")
|
||
planung_anlegen(app, "El Grande Big Box", rez1.id)
|
||
|
||
antwort = dict(GLEICHES_SPIEL_HOCH, ist_gleiches_spiel=False, begruendung="Big Box ≠ Jubiläum.")
|
||
ergebnis, _ = pruefe_mit_ki(app, [json.dumps(antwort)], "El Grande: 25 Jahre")
|
||
assert not ergebnis.hat_konflikte
|
||
assert ergebnis.ki_befunde[0]["entscheidung"] == "verschiedene_spiele"
|
||
|
||
|
||
def test_uebersetzungsgrenzfall_kriegt_deutschen_titel(settings, monkeypatch):
|
||
"""Catan Fast Edition vs. Catan – Das schnelle Spiel: KI liefert den deutschen Titel."""
|
||
app = app_mit_ki(settings, monkeypatch)
|
||
neuheit_anlegen(app, "El Grande: 25 Jahre", verlag="Hans im Glück", bgg_id=111)
|
||
|
||
antwort = dict(GLEICHES_SPIEL_HOCH, deutscher_titel="El Grande: 25 Jahre")
|
||
ergebnis, _ = pruefe_mit_ki(app, [json.dumps(antwort)], "El Grande Big Box")
|
||
|
||
# Ohne KI wäre hier nichts gefunden worden (beide Titel heuristisch neutral);
|
||
# mit KI gibt es eine Titel-Variante samt deutscher Empfehlung:
|
||
assert "titel" in [k.art for k in ergebnis.konflikte]
|
||
assert ergebnis.titel_empfehlung == "El Grande: 25 Jahre"
|
||
|
||
dedup = app.state.registry.get("dedup")
|
||
referenz = asyncio.run(dedup.check_titel("El Grande Big Box"))
|
||
assert not referenz.hat_konflikte
|
||
# Ohne KI bleibt die Empfehlung beim eigenen Titel:
|
||
assert referenz.titel_empfehlung == "El Grande Big Box"
|
||
|
||
|
||
def test_verlagskonflikt_kriegt_ki_empfehlung(settings, monkeypatch):
|
||
"""Verlags-Alias: Regelprüfung findet den Konflikt, das LLM den richtigen Verlag."""
|
||
app = app_mit_ki(settings, monkeypatch)
|
||
neuheit_anlegen(app, "El Grande", verlag="Hans im Glück Verlag", bgg_id=21)
|
||
|
||
antwort = dict(GLEICHES_SPIEL_HOCH, empfohlener_verlag="Hans im Glück Verlag")
|
||
ergebnis, _ = pruefe_mit_ki(app, [json.dumps(antwort)], "El Grande", "Hans im Glück", 21)
|
||
|
||
assert ergebnis.hat_verlagskonflikt
|
||
verlags_konflikt = next(k for k in ergebnis.konflikte if k.art == "verlag")
|
||
assert verlags_konflikt.details["ki_empfohlener_verlag"] == "Hans im Glück Verlag"
|
||
assert ergebnis.ki_befunde[-1]["typ"] == "verlag"
|
||
|
||
|
||
# ---------------- Integration: Budget & Audit ----------------
|
||
|
||
def test_ki_budget_begrenzt_anzahl_aufrufe(settings, monkeypatch):
|
||
from plugins.dedup.pruefung import MAX_KI_PAARUNGEN
|
||
|
||
app = app_mit_ki(settings, monkeypatch)
|
||
for nr in range(MAX_KI_PAARUNGEN + 3):
|
||
neuheit_anlegen(app, f"Catan Ausgabe Nummer {nr}", bgg_id=200 + nr)
|
||
|
||
fake = FakeKiClient(dict(GLEICHES_SPIEL_HOCH))
|
||
dedup = app.state.registry.get("dedup")
|
||
dedup.ki_client_fabrik = lambda: fake
|
||
try:
|
||
ergebnis = asyncio.run(dedup.check_titel("Catan Neuauflage"))
|
||
finally:
|
||
dedup.ki_client_fabrik = None
|
||
|
||
assert len(fake.aufrufe) <= MAX_KI_PAARUNGEN
|
||
assert len(ergebnis.ki_befunde) == len(fake.aufrufe)
|
||
|
||
|
||
def test_ki_befunde_landen_im_audit_log(settings, monkeypatch):
|
||
app = app_mit_ki(settings, monkeypatch)
|
||
lege_benutzer_an(app, "rez1", "rezensent") # fremder Eintrag → echter Konflikt
|
||
lege_benutzer_an(app, "anna", "redakteur") # prüfende Person
|
||
rez1 = benutzer_objekt(app, "rez1")
|
||
anna = benutzer_objekt(app, "anna")
|
||
planung_anlegen(app, "El Grande Big Box", rez1.id)
|
||
|
||
fake = FakeKiClient(dict(GLEICHES_SPIEL_HOCH))
|
||
dedup = app.state.registry.get("dedup")
|
||
dedup.ki_client_fabrik = lambda: fake
|
||
try:
|
||
ergebnis = asyncio.run(dedup.check_titel("El Grande: 25 Jahre", user=anna))
|
||
finally:
|
||
dedup.ki_client_fabrik = None
|
||
assert ergebnis.hat_konflikte
|
||
|
||
AuditEintrag = plugin_modul(app, "audit-log").AuditEintrag
|
||
with app.state.session_factory() as db:
|
||
eintraege = db.scalars(
|
||
select(AuditEintrag).where(AuditEintrag.objekt_typ == "dedup_ki")
|
||
).all()
|
||
assert len(eintraege) == 1
|
||
eintrag = eintraege[0]
|
||
assert eintrag.action == "geprüft"
|
||
assert eintrag.actor_name == "anna"
|
||
assert eintrag.details["modell"] == "fake-modell"
|
||
befaelle = eintrag.details["befunde"]
|
||
assert befaelle[0]["entscheidung"] == "gleiches_spiel"
|
||
assert befaelle[0]["konfidenz"] == 0.9
|
||
assert befaelle[0]["titel_a"] == "El Grande: 25 Jahre"
|
||
|
||
|
||
def test_keine_ki_befunde_heissen_kein_audit_eintrag(settings, monkeypatch):
|
||
"""Rein regelbasierte Prüfung (KI aus) schreibt keinen dedup_ki-Eintrag."""
|
||
app = create_app(settings)
|
||
asyncio.run(app.state.registry.get("dedup").check_titel("Beliebiges Spiel"))
|
||
|
||
AuditEintrag = plugin_modul(app, "audit-log").AuditEintrag
|
||
with app.state.session_factory() as db:
|
||
eintraege = db.scalars(
|
||
select(AuditEintrag).where(AuditEintrag.objekt_typ == "dedup_ki")
|
||
).all()
|
||
assert eintraege == []
|