tests/test_pipeline_wiring.py
Rama main · réplica de lectura del repositorio original.
"""
Regression tests for the research pipeline's wiring.
Passing unit tests do not guarantee that the parts are connected. These
tests guard three such connections:
1. `create_app()` must boot — every UI module it imports must exist.
2. The polarity mechanism must be wired into the production harvest,
not only tested in isolation.
3. The off-topic filter runs exactly once per round (in
`_run_search_and_fetch`), not a second time as a DAG node over the
whole corpus.
"""
import os
import subprocess
import sys
import unittest
import tests.conftest # noqa: F401 (httpx/openai stubs)
from src.pipeline.harvest_parser import parse_harvest_response_with_polarity
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
class TestCreateAppBoots(unittest.TestCase):
"""create_app() must run through. Executed in a subprocess with a Gradio
stub, so that NO sys.modules mocks leak into other tests."""
SCRIPT = r"""
import sys, types
g = types.ModuleType("gradio")
class _C:
def __init__(s,*a,**k): s.value=k.get("value")
def __enter__(s): return s
def __exit__(s,*a): return False
def __getattr__(s,n): return lambda *a,**k: s
class _B(_C):
def queue(s,*a,**k): return s
def launch(s,*a,**k): return s
for n in ["Accordion","BrowserState","Button","Chatbot","Checkbox","Column","Dropdown",
"File","Group","HTML","Markdown","MultimodalTextbox","Row",
"State","TabItem","Tabs","Textbox"]:
setattr(g,n,type(n,(_C,),{}))
g.Blocks=_B
g.Info=g.Warning=lambda *a,**k: None
g.Error=type("Error",(Exception,),{})
g.update=lambda *a,**k: {"__u__":True}
th=types.ModuleType("gradio.themes")
th.Default=th.Base=type("T",(),{"__init__":lambda s,*a,**k: None})
g.themes=th
sys.modules["gradio"]=g; sys.modules["gradio.themes"]=th
import tests.conftest # httpx/openai stubs
from src.config import AppConfig
from src.ui.gradio_app import create_app
app = create_app(AppConfig())
assert app is not None
print("CREATE_APP_OK")
"""
def test_create_app_does_not_crash_on_boot(self):
res = subprocess.run(
[sys.executable, "-c", self.SCRIPT],
cwd=REPO, capture_output=True, text=True, timeout=120,
)
self.assertIn(
"CREATE_APP_OK", res.stdout,
msg=f"create_app() crashes on boot.\n"
f"STDOUT:\n{res.stdout}\nSTDERR:\n{res.stderr}",
)
def test_pipeline_run_module_exists_and_renders(self):
from src.ui.pipeline_run import render_pipeline_run
from src.pipeline.models import HarvestContext
self.assertEqual(
render_pipeline_run(None), "*No research started yet.*"
)
ctx = HarvestContext(query="Konsistenz der Web-Recherche")
ctx.status = "running"
out = render_pipeline_run(ctx)
self.assertIn("Pipeline run", out)
self.assertIn("Konsistenz der Web-Recherche", out)
def test_import_is_defensive(self):
"""The import in gradio_app is protected against a missing module
(try/except)."""
src = open(os.path.join(REPO, "src/ui/gradio_app.py")).read()
i = src.find("from src.ui.pipeline_run import render_pipeline_run")
self.assertGreater(i, 0)
# There must be a try: within the 200-character window before it
self.assertIn("try:", src[max(0, i - 200):i])
class TestPolarityWiredIntoProductionHarvest(unittest.TestCase):
"""The polarity parser is wired into the production path and the prompt
asks for the markers."""
def test_orchestrator_uses_polarity_parser(self):
src = open(os.path.join(REPO, "src/pipeline/orchestrator.py")).read()
self.assertIn("parse_harvest_response_with_polarity", src,
"the harvest must use the polarity-aware parser")
# A parser that ignores the marker must not be called in the
# harvest_one path.
self.assertNotIn("self._parse_harvest_response(\n", src)
def test_harvest_prompt_requests_polarity_markers(self):
src = open(os.path.join(REPO, "src/prompts/research.py")).read()
for marker in ("[POSITIVE]", "[NEGATIVE]", "[META]"):
self.assertIn(marker, src,
f"HARVEST_PROMPT must ask for {marker}")
def test_negative_meta_do_not_leak_into_fact_and_are_filtered(self):
response = (
"[F1] [POSITIV] Fakt: Das Modell hat 7 Mrd. Parameter\n"
" Kontext: Aus dem Datenblatt\n"
" Verlässlichkeit: hoch\n"
"[F2] [NEGATIV] Fakt: Die Quelle nennt keine Trainingsdaten\n"
" Kontext: Abschnitt fehlt\n"
"[F3] [META] Fakt: Die Seite ist Marketing, kein Datenblatt\n"
)
extracts = parse_harvest_response_with_polarity(
response, "http://x", "X"
)
by_pol = {e.question_id: e.polarity for e in extracts}
self.assertEqual(by_pol.get("F1"), "positive")
self.assertEqual(by_pol.get("F2"), "negative")
self.assertEqual(by_pol.get("F3"), "meta")
# the marker must NOT end up in the fact text
for e in extracts:
self.assertNotIn("[POSITIV]", e.fact)
self.assertNotIn("[NEGATIV]", e.fact)
self.assertNotIn("[META]", e.fact)
# the positive-only filter (as in HarvestNode) lets exactly F1 through
positive = [e for e in extracts
if getattr(e, "polarity", "positive") == "positive"]
self.assertEqual([e.question_id for e in positive], ["F1"])
class TestOffTopicFilterRunsOnce(unittest.TestCase):
"""Exactly one off-topic pass per round, over the new sources; no second
pass over the whole corpus, no double ctx.sources.extend."""
def setUp(self):
self.orch = open(
os.path.join(REPO, "src/pipeline/orchestrator.py")
).read()
self.nodes = open(
os.path.join(REPO, "src/pipeline/dag_nodes.py")
).read()
def test_no_offtopic_node_in_round_loop(self):
# run_via_dag builds the round sub-pipeline; OffTopicFilterNode
# must not be constructed there.
self.assertNotIn("OffTopicFilterNode(", self.orch,
"OffTopicFilterNode runs again in run_via_dag "
"over the whole corpus (regression)")
def test_single_offtopic_call_in_search_and_fetch(self):
n = self.orch.count("self._filter_off_topic_sources(")
self.assertEqual(
n, 1,
f"_filter_off_topic_sources may be called exactly once, "
f"found: {n}",
)
def test_no_double_ctx_sources_extend(self):
# _run_search_and_fetch must NOT extend ctx.sources itself
# (the SearchAndFetchNode does that — otherwise everything ends up twice).
seg_start = self.orch.find("async def _run_search_and_fetch")
seg_end = self.orch.find("\n def _enrich_zis_markdown")
segment = self.orch[seg_start:seg_end]
self.assertNotIn("ctx.sources.extend(new_sources)", segment,
"duplicate ctx.sources.extend (regression)")
self.assertIn("ctx.sources.extend(new_sources)", self.nodes,
"SearchAndFetchNode must fill ctx.sources")
class TestGitRepoLoopNotNested(unittest.TestCase):
"""The git-repository README loop must not be nested inside the
direct-URL loop."""
def test_git_repo_loop_is_sibling_of_direct_loop(self):
src = open(
os.path.join(REPO, "src/pipeline/orchestrator.py")
).read()
# position of the two loops
i_direct = src.find("for direct in ordered_direct:")
i_repo = src.find("for repo in plan.git_repos:", i_direct)
self.assertGreater(i_direct, 0)
self.assertGreater(i_repo, i_direct)
# The line with the repository loop must be at the 8-space level
# (method body), not at 12 (inside the direct loop).
line_start = src.rfind("\n", 0, i_repo) + 1
indent = len(src[line_start:i_repo])
self.assertEqual(
indent, 8,
f"git repo loop at indentation {indent} instead of 8 — "
f"it would be nested in the direct loop again (regression)",
)
if __name__ == "__main__":
unittest.main()