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>
101 lines
3 KiB
Python
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
|