tests/test_reranker_embedder_scope.py
Rama main · réplica de lectura del repositorio original.
"""
Tests for the reranker, embedder and search scope.
Focus: the FAIL-OPEN behaviour. These building blocks are quality
improvements, not hard dependencies — missing configuration or a dead
endpoint must NEVER break the pipeline.
"""
import asyncio
import os
import unittest
import tests.conftest # noqa: F401
from src.llm.reranker import Reranker, build_reranker_from_pipeline_config
from src.llm.embedder import Embedder, cosine
from src.pipeline.classifiers.search_scope import classify_search_scope
from src.config import PipelineConfig
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
def run(coro):
return asyncio.run(coro)
class TestRerankerFailOpen(unittest.TestCase):
def test_no_api_key_is_unavailable(self):
rr = Reranker(base_url="https://x/rerank", model="m", api_key="")
self.assertFalse(rr.available)
def test_unavailable_rank_returns_identity(self):
rr = Reranker("", "m", "")
out = run(rr.rank("frage", ["a", "b", "c"]))
self.assertEqual([i for i, _ in out], [0, 1, 2])
def test_unavailable_order_keeps_items(self):
rr = Reranker("", "m", "")
items = ["x", "y", "z"]
self.assertEqual(run(rr.order("q", list(items))), items)
def test_single_doc(self):
rr = Reranker("https://x", "m", "key")
self.assertEqual(run(rr.rank("q", ["only"])), [(0, 1.0)])
def test_dead_endpoint_fails_open(self):
# key set → available, but the endpoint does not exist.
# httpx is stubbed in the test and raises → identity expected.
rr = Reranker("https://nonexistent.invalid/rerank", "m", "key")
out = run(rr.rank("q", ["a", "b"]))
self.assertEqual([i for i, _ in out], [0, 1])
def test_factory_from_pipeline_config(self):
cfg = PipelineConfig()
cfg.reranker_base_url = ""
cfg.reranker_api_key = ""
rr = build_reranker_from_pipeline_config(cfg)
self.assertIsInstance(rr, Reranker)
self.assertFalse(rr.available)
class TestEmbedderFailOpen(unittest.TestCase):
def test_unavailable_returns_none(self):
emb = Embedder("", "m", "")
self.assertFalse(emb.available)
self.assertIsNone(run(emb.embed(["a", "b"])))
def test_dead_endpoint_returns_none(self):
emb = Embedder("https://nonexistent.invalid", "m", "key")
self.assertIsNone(run(emb.embed(["a"])))
def test_cosine_basics(self):
self.assertAlmostEqual(cosine([1, 0], [1, 0]), 1.0, places=6)
self.assertAlmostEqual(cosine([1, 0], [0, 1]), 0.0, places=6)
self.assertEqual(cosine([], [1]), 0.0)
self.assertEqual(cosine([0, 0], [0, 0]), 0.0)
self.assertAlmostEqual(cosine([1, 1], [1, 1]), 1.0, places=6)
class TestSearchScopeClassifier(unittest.TestCase):
def test_fallback_keeps_both_passes(self):
"""Empty LLM answer ⇒ fail-open: both additional passes active."""
from tests._helpers import MockLLM
llm = MockLLM() # empty queue → default ""
scope, call = run(classify_search_scope("irgendwas", "", llm))
self.assertTrue(scope.time_sensitive)
self.assertTrue(scope.academic)
self.assertTrue(scope.fallback_used)
def test_classifies_not_timesensitive_not_academic(self):
from tests._helpers import MockLLM
llm = MockLLM()
llm.queue_json({
"time_sensitive": False, "academic": False,
"confidence": 0.9, "reasoning": "zeitlose Sachfrage",
})
scope, _ = run(classify_search_scope(
"Was ist ein Mutex?", "", llm))
self.assertFalse(scope.time_sensitive)
self.assertFalse(scope.academic)
self.assertFalse(scope.fallback_used)
def test_classifies_timesensitive(self):
from tests._helpers import MockLLM
llm = MockLLM()
llm.queue_json({
"time_sensitive": True, "academic": False,
"confidence": 0.95, "reasoning": "aktuelle Preise",
})
scope, _ = run(classify_search_scope(
"aktueller Preis der DGX B300", "", llm))
self.assertTrue(scope.time_sensitive)
self.assertFalse(scope.academic)
def test_language_whitelist_parsed(self):
from tests._helpers import MockLLM
llm = MockLLM()
llm.queue_json({
"time_sensitive": False, "academic": True,
"languages": ["de", "en", "zh"], "recency": "",
"confidence": 0.9, "reasoning": "KI-Thema",
})
scope, _ = run(classify_search_scope("KI-Modelle", "", llm))
self.assertEqual(scope.languages, ["de", "en", "zh"])
def test_german_only_topic(self):
from tests._helpers import MockLLM
llm = MockLLM()
llm.queue_json({
"time_sensitive": False, "academic": False,
"languages": ["de"], "confidence": 0.92,
"reasoning": "rein deutsches Verwaltungsthema",
})
scope, _ = run(classify_search_scope(
"Promotionsordnung der Beispiel-Uni", "", llm))
self.assertEqual(scope.languages, ["de"])
def test_recency_validated(self):
from tests._helpers import MockLLM
llm = MockLLM()
llm.queue_json({
"time_sensitive": True, "academic": False,
"recency": "week", "languages": [],
"confidence": 0.9, "reasoning": "Eilmeldung",
})
scope, _ = run(classify_search_scope("news heute", "", llm))
self.assertEqual(scope.recency, "week")
def test_bogus_recency_falls_back_to_empty(self):
from tests._helpers import MockLLM
llm = MockLLM()
llm.queue_json({
"time_sensitive": True, "academic": False,
"recency": "gestern", "languages": ["xx", "", "deu"],
"confidence": 0.5, "reasoning": "x",
})
scope, _ = run(classify_search_scope("x", "", llm))
self.assertEqual(scope.recency, "") # invalid → ""
# The language filter only discards empty/too long/non-alphabetic
# codes. "xx" is structurally a valid 2-character code and stays;
# "" is discarded.
self.assertEqual(scope.languages, ["xx", "deu"])
def test_fallback_languages_empty_means_no_restriction(self):
from tests._helpers import MockLLM
scope, _ = run(classify_search_scope("x", "", MockLLM()))
self.assertTrue(scope.fallback_used)
self.assertEqual(scope.languages, [])
class TestWiringRegression(unittest.TestCase):
"""Structural regression check of the wiring."""
def setUp(self):
self.orch = open(
os.path.join(REPO, "src/pipeline/orchestrator.py")
).read()
def test_reranker_wired_into_prefetch(self):
self.assertIn("build_reranker_from_pipeline_config", self.orch)
self.assertIn("reranked", self.orch)
def test_embedder_wired_into_dedup(self):
self.assertIn("build_embedder_from_pipeline_config", self.orch)
self.assertIn("extract_dedup_threshold", self.orch)
def test_search_scope_gates_fanout(self):
self.assertIn("classify_search_scope", self.orch)
self.assertIn("_do_recent", self.orch)
self.assertIn("_do_science", self.orch)
# Additional passes are gated to round 0 and by the scope.
# The recent pass is a single line, the science pass is bracketed
# over several lines — hence check the substrings that really occur.
self.assertIn("round_num == 0 and _do_recent:", self.orch)
self.assertIn("and _do_science):", self.orch)
# language allow list wired
self.assertIn("_effective_langs", self.orch)
self.assertIn("_scope_langs", self.orch)
# tunable web-search cap
self.assertIn("max_web_searches", self.orch)
def test_config_has_new_fields(self):
cfg = PipelineConfig.from_env()
for f in ("reranker_base_url", "reranker_model",
"reranker_api_key", "embedder_base_url",
"embedder_model", "embedder_api_key",
"extract_dedup_threshold"):
self.assertTrue(hasattr(cfg, f), f"PipelineConfig.{f} missing")
self.assertEqual(cfg.extract_dedup_threshold, 0.93)
def test_map_call_disables_thinking(self):
"""The map call must have a budget for reasoning: enable_thinking=False +
a retry before the raw-extract fallback."""
src = self.orch
# The map call must disable thinking
self.assertIn("enable_thinking=False", src)
# There must be a retry path before raw extracts are used
# (LLM gateways return transient 504s)
self.assertIn("one retry", src)
self.assertIn("failed permanently", src)
# token budget raised (from 2048)
self.assertIn("max_tokens=4096", src)
def test_link_following_has_relevance_gate(self):
"""Link following: rerank against the research question before
fetching (keeps dictionary sites and the like out)."""
src = self.orch
self.assertIn("relevance_query", src)
self.assertIn("Relevance gate", src)
self.assertIn("_link_rel_q", src)
# other_primary (web-search sources) MUST be gated
self.assertIn("relevance_query=_link_rel_q", src)
if __name__ == "__main__":
unittest.main()