maboto/tests/test_produktcache.py
Jeuner 077a877480 Produkt->Kategorie-Cache: bekannte Produkte ohne LLM (SQLite, modellübergreifend)
Neuer produktcache.py (Stufe 2): speichert pro Produkt (Titel+Marke, mengen-
invariant) die einmal ermittelte Gruppe in SQLite, bulk-load ins dict -> O(1).
Schnitt gewahrt: kein LLM-Import, nur Gruppe (nie Angebotsdaten), Whitelist
beim Lesen+Schreiben, nur SICHERE Zuordnungen gecacht.

kategorisiere(cache=, statistik=): Lookup vor dem LLM, Dedup im Lauf (ein
Produkt = ein Posten), Write-Back danach. Parallel-/id-Logik unverändert.
als_struktur/web/cli verdrahtet (Statistik 'X aus Cache · Y neu', --no-cache).

Live verifiziert (1903 Angebote PLZ 60487): Lauf 1 (gemini) 1551 neu; Lauf 2
(deepseek, anderes Modell) nur 110 neu, 1765 aus Cache -> ~93% weniger LLM-Calls,
modellübergreifend. +12 Tests (Round-Trip, Whitelist, Hit-vermeidet-Call, Dedup,
nur-sichere, Schnitt). 70 Tests grün.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-03 18:37:12 +02:00

101 lines
3 KiB
Python

"""Tests für den Produkt→Kategorie-Cache -- offline, eigene DB pro Test."""
import subprocess
import sys
from pathlib import Path
from angebote.produktcache import ProduktCache, produkt_schluessel
SRC = Path(__file__).resolve().parents[1] / "src"
def _cache(tmp_path) -> ProduktCache:
return ProduktCache(db_pfad=tmp_path / "cache.sqlite")
# -- Schlüssel ----------------------------------------------------------------
def test_schluessel_ist_mengen_invariant():
# Titel + Marke bestimmen den Schlüssel; Menge spielt keine Rolle.
assert produkt_schluessel("Butter", "Meggle") == produkt_schluessel(
" butter ", "MEGGLE"
)
def test_schluessel_unterscheidet_marke():
assert produkt_schluessel("Butter", "Meggle") != produkt_schluessel(
"Butter", "Kerrygold"
)
# -- Round-Trip / Persistenz --------------------------------------------------
def test_round_trip_ueber_instanzgrenzen(tmp_path):
db = tmp_path / "c.sqlite"
c1 = ProduktCache(db_pfad=db)
s = produkt_schluessel("Toffifee", "Storck")
c1.schreibe_viele([(s, "Süßwaren & Snacks", "deepseek")])
# frische Instanz auf derselben DB
c2 = ProduktCache(db_pfad=db)
assert c2.hole(s) == "Süßwaren & Snacks"
assert c2.groesse() == 1
def test_unbekannter_schluessel_gibt_none(tmp_path):
c = _cache(tmp_path)
assert c.hole("gibtsnicht") is None
# -- Geschlossene Liste (Whitelist) ------------------------------------------
def test_off_list_gruppe_wird_nicht_geschrieben(tmp_path):
c = _cache(tmp_path)
n = c.schreibe_viele([("k1", "Weltraumzeug", None)])
assert n == 0
assert c.hole("k1") is None
def test_off_list_zeile_in_db_wird_beim_lesen_verworfen(tmp_path):
import sqlite3
db = tmp_path / "c.sqlite"
ProduktCache(db_pfad=db) # legt Tabelle an
# manipulierte Zeile direkt in die DB schreiben
with sqlite3.connect(str(db)) as con:
con.execute(
"INSERT INTO produkt_kategorie VALUES (?,?,?,?)",
("k1", "Quatschgruppe", None, "2026-01-01"),
)
c = ProduktCache(db_pfad=db)
assert c.hole("k1") is None # Whitelist filtert sie heraus
assert c.groesse() == 0
# -- Schnitt: kein LLM im Cache ----------------------------------------------
def test_cache_laedt_kein_anthropic():
code = (
"import sys; import angebote.produktcache; "
"assert 'anthropic' not in sys.modules, 'Cache hat anthropic geladen'; "
"print('ok')"
)
proc = subprocess.run(
[sys.executable, "-c", code], cwd=str(SRC), capture_output=True, text=True
)
assert proc.returncode == 0, proc.stderr
assert "ok" in proc.stdout
def test_cache_importiert_kein_llm():
quelltext = (SRC / "angebote" / "produktcache.py").read_text("utf-8")
import_zeilen = "\n".join(
z for z in quelltext.splitlines()
if z.strip().startswith(("import ", "from "))
).lower()
assert "anthropic" not in import_zeilen
assert "openai" not in import_zeilen
assert "kategorisieren" not in import_zeilen