179 lines
7.8 KiB
Python
179 lines
7.8 KiB
Python
"""Suche: Fallback-Kette, Circuit Breaker, URL-Dedupe. Laden: Snapshot stabil."""
|
|
import pytest
|
|
|
|
from backend import config, db, laden, suche
|
|
from .conftest import topic_anlegen
|
|
|
|
|
|
def test_url_normieren():
|
|
assert suche.url_normieren("HTTPS://Beispiel.de/Pfad/?utm_source=x&a=1") == \
|
|
"https://beispiel.de/Pfad?a=1"
|
|
assert suche.url_normieren("https://beispiel.de/seite/") == \
|
|
suche.url_normieren("https://beispiel.de/seite")
|
|
|
|
|
|
async def test_fallback_kette(monkeypatch):
|
|
monkeypatch.setattr(config, "FAKE", False)
|
|
suche._ausfaelle.clear()
|
|
|
|
async def leer_wiki(q):
|
|
return []
|
|
|
|
async def kaputt(q):
|
|
raise RuntimeError("down")
|
|
|
|
async def liefert(q):
|
|
return [{"titel": "T", "url": "https://x.de/a", "backend": "b2"}]
|
|
|
|
monkeypatch.setattr(suche, "_wikipedia", leer_wiki)
|
|
monkeypatch.setattr(suche, "_KETTE", (("b1", kaputt), ("b2", liefert)))
|
|
monkeypatch.setattr(config, "SUCHE_RETRIES", 0)
|
|
treffer, fehler = await suche.web_suchen("query")
|
|
assert [t["backend"] for t in treffer] == ["b2"]
|
|
assert fehler == ["b1: RuntimeError"]
|
|
|
|
|
|
async def test_circuit_breaker(monkeypatch):
|
|
monkeypatch.setattr(config, "FAKE", False)
|
|
monkeypatch.setattr(config, "SUCHE_RETRIES", 0)
|
|
suche._ausfaelle.clear()
|
|
aufrufe = {"n": 0}
|
|
|
|
async def leer_wiki(q):
|
|
return []
|
|
|
|
async def kaputt(q):
|
|
aufrufe["n"] += 1
|
|
raise RuntimeError("down")
|
|
|
|
monkeypatch.setattr(suche, "_wikipedia", leer_wiki)
|
|
monkeypatch.setattr(suche, "_KETTE", (("b1", kaputt),))
|
|
for _ in range(config.CIRCUIT_BREAKER_N + 3):
|
|
await suche.web_suchen("q")
|
|
assert aufrufe["n"] == config.CIRCUIT_BREAKER_N # danach übersprungen
|
|
|
|
|
|
async def test_suche_dedupliziert_urls():
|
|
topic = topic_anlegen()
|
|
run_id = db.insert("runs", topic=topic, status="running")
|
|
t = {"run_id": run_id, "topic": topic, "runde": 1,
|
|
"payload": db.j({"query": "thema ueberblick r1"})}
|
|
erg1 = await suche.suchen({**t})
|
|
erg2 = await suche.suchen({**t})
|
|
assert erg1.daten["neu"] == 1
|
|
assert erg2.daten["neu"] == 0 # gleiche URL → kein zweiter Laden-Task
|
|
|
|
|
|
async def test_inhalts_urteil_sortiert_nach_dem_laden_aus():
|
|
"""Kein Titel-Filter mehr — das Urteil fällt auf dem geladenen Inhalt
|
|
(Lektion 102). Fake-Handler: Titel „Werbeliste" → nein."""
|
|
topic = topic_anlegen()
|
|
run_id = db.insert("runs", topic=topic, status="running")
|
|
qid = db.insert("quellen", topic=topic, titel="Werbeliste Tools 2026",
|
|
url="https://beispiel.de/grundlagen",
|
|
url_norm="w1", backend="ddgs", zweck="korpus")
|
|
erg = await laden.laden({"run_id": run_id, "topic": topic, "runde": 1,
|
|
"knoten": "laden", "payload": db.j({"quelle_id": qid})})
|
|
assert erg.daten.get("aussortiert")
|
|
q = db.one("SELECT status, grund FROM quellen WHERE id=?", qid)
|
|
assert q["status"] == "aussortiert" and q["grund"]
|
|
assert erg.neue_tasks == [] # keine Soll-Extraktion für Aussortierte
|
|
|
|
|
|
async def test_wikipedia_seed_umgeht_deckel(monkeypatch):
|
|
monkeypatch.setattr(config, "QUELLEN_MAX", 1)
|
|
topic = topic_anlegen(name="seedtest", titel="Pomodoro Technik")
|
|
run_id = db.insert("runs", topic=topic, status="running")
|
|
erg = await suche.suchen({"run_id": run_id, "topic": topic, "runde": 1,
|
|
"item": "r1:wikiseed",
|
|
"payload": db.j({"seed": True})})
|
|
assert erg.daten["neu"] == 2 # de+en trotz Deckel 1
|
|
urls = [q["url"] for q in db.query(
|
|
"SELECT url FROM quellen WHERE topic=?", topic)]
|
|
assert any("de.wikipedia.org/wiki/Pomodoro_Technik" in u for u in urls)
|
|
assert any("en.wikipedia.org" in u for u in urls)
|
|
# idempotent: zweiter Lauf legt nichts Neues an
|
|
erg2 = await suche.suchen({"run_id": run_id, "topic": topic, "runde": 1,
|
|
"item": "r1:wikiseed",
|
|
"payload": db.j({"seed": True})})
|
|
assert erg2.daten["neu"] == 0
|
|
|
|
|
|
async def test_quellen_deckel(monkeypatch):
|
|
monkeypatch.setattr(config, "QUELLEN_MAX", 1)
|
|
topic = topic_anlegen()
|
|
run_id = db.insert("runs", topic=topic, status="running")
|
|
basis = {"run_id": run_id, "topic": topic, "runde": 1, "item": "t"}
|
|
erg1 = await suche.suchen({**basis,
|
|
"payload": db.j({"query": "thema ueberblick r1"})})
|
|
erg2 = await suche.suchen({**basis,
|
|
"payload": db.j({"query": "thema tutorial r1"})})
|
|
assert erg1.daten["neu"] == 1
|
|
assert erg2.daten["neu"] == 0 # Deckel erreicht — keine neue Quelle
|
|
|
|
|
|
async def test_quellen_deckel_pro_topic(monkeypatch):
|
|
monkeypatch.setattr(config, "QUELLEN_MAX", 1)
|
|
topic = topic_anlegen()
|
|
db.update("topics", "name=?", (topic,), quellen_max=0) # 0 = unbegrenzt
|
|
run_id = db.insert("runs", topic=topic, status="running")
|
|
basis = {"run_id": run_id, "topic": topic, "runde": 1, "item": "t"}
|
|
erg1 = await suche.suchen({**basis,
|
|
"payload": db.j({"query": "thema ueberblick r1"})})
|
|
erg2 = await suche.suchen({**basis,
|
|
"payload": db.j({"query": "thema tutorial r1"})})
|
|
assert erg1.daten["neu"] == 1
|
|
assert erg2.daten["neu"] == 1 # Topic-Einstellung schlägt globalen Deckel
|
|
|
|
|
|
async def test_laden_snapshot_stabil():
|
|
topic = topic_anlegen()
|
|
run_id = db.insert("runs", topic=topic, status="running")
|
|
qid = db.insert("quellen", topic=topic, url="https://beispiel.de/grundlagen",
|
|
url_norm="https://beispiel.de/grundlagen", zweck="korpus")
|
|
t = {"run_id": run_id, "topic": topic, "runde": 1,
|
|
"knoten": "laden", "payload": db.j({"quelle_id": qid})}
|
|
erg1 = await laden.laden({**t})
|
|
q = db.one("SELECT * FROM quellen WHERE id=?", qid)
|
|
assert q["status"] == "geladen" and q["snapshot"]
|
|
inhalt1 = laden.snapshot_lesen(q)
|
|
erg2 = await laden.laden({**t}) # Resume: kein Neuschreiben
|
|
assert erg2.daten.get("skip")
|
|
assert laden.snapshot_lesen(q) == inhalt1
|
|
assert len(erg1.neue_tasks) >= 1 # Soll-Extraktions-Chunks
|
|
|
|
|
|
async def test_laden_leere_quelle_befund():
|
|
topic = topic_anlegen()
|
|
run_id = db.insert("runs", topic=topic, status="running")
|
|
qid = db.insert("quellen", topic=topic, url="https://unbekannt.example/x",
|
|
url_norm="https://unbekannt.example/x")
|
|
erg = await laden.laden({"run_id": run_id, "topic": topic, "runde": 1,
|
|
"knoten": "laden", "payload": db.j({"quelle_id": qid})})
|
|
assert erg.neue_tasks == []
|
|
assert db.one("SELECT status FROM quellen WHERE id=?", qid)["status"] == "leer"
|
|
assert db.one("SELECT id FROM befunde WHERE run_id=? AND art='leer'", run_id)
|
|
|
|
|
|
async def test_seed_loest_wikipedia_titel_auf(monkeypatch):
|
|
"""Audit: Seed-Titel „AlpineJS" existiert nicht — Artikel heißt
|
|
„Alpine.js". Erst Wiki-Suche, direkter URL-Bau nur als Fallback."""
|
|
monkeypatch.setattr(config, "FAKE", False)
|
|
|
|
async def wiki(q):
|
|
assert q == "AlpineJS"
|
|
return [{"titel": "Alpine.js", "backend": "wikipedia_de",
|
|
"url": "https://de.wikipedia.org/wiki/Alpine.js"},
|
|
{"titel": "Nebentreffer", "backend": "wikipedia_de",
|
|
"url": "https://de.wikipedia.org/wiki/Nebentreffer"}]
|
|
|
|
monkeypatch.setattr(suche, "_wikipedia", wiki)
|
|
topic = topic_anlegen(name="seedaufl", titel="AlpineJS")
|
|
run_id = db.insert("runs", topic=topic, status="running")
|
|
await suche.suchen({"run_id": run_id, "topic": topic, "runde": 1,
|
|
"item": "r1:wikiseed", "payload": db.j({"seed": True})})
|
|
urls = [q["url"] for q in db.query("SELECT url FROM quellen WHERE topic=?",
|
|
topic)]
|
|
assert "https://de.wikipedia.org/wiki/Alpine.js" in urls # 1. Treffer zählt
|
|
assert any("en.wikipedia.org/wiki/AlpineJS" in u for u in urls) # Fallback
|