tests/test_dag_nodes.py
Rama main · réplica de lectura del repositorio original.
"""
Integration tests for the pipeline nodes of the classifier layer.
Acceptance:
- every node can be tested in isolation (without the full orchestrator).
- nodes can be chained into a pipeline.
- a query that a heuristic would misread as a person runs through the
DAG with the same behaviour as the direct classifier call.
"""
import unittest
from unittest.mock import MagicMock
from src.pipeline.classifiers.query_anchor import QueryAnchor
from src.pipeline.dag import PipelineDAG, NodeStatus
from src.pipeline.dag_nodes import (
QueryAnchorNode,
OffTopicFilterNode,
CoverageNode,
ContinueDecisionNode,
DiagnosisNode,
FactoidVerificationNode,
)
from src.pipeline.models import (
HarvestContext,
ResearchPlan,
ResearchQuestion,
SourceDocument,
SourceExtract,
SourceType,
)
from tests._helpers import MockLLM, async_test
def _patch_adapter_with(llm: MockLLM):
from src.llm import classifier_adapter
class _Pass:
def __init__(self, _dual):
pass
async def complete(self, messages, max_tokens=None):
return await llm.complete(messages, max_tokens=max_tokens)
original = classifier_adapter.HarvestModelAdapter
classifier_adapter.HarvestModelAdapter = _Pass
return original
def _restore_adapter(original):
from src.llm import classifier_adapter
classifier_adapter.HarvestModelAdapter = original
def _make_ctx(query: str = "Test", with_plan: bool = True):
ctx = HarvestContext(query=query)
ctx.classifier_calls = []
if with_plan:
ctx.research_plan = ResearchPlan(
questions=[ResearchQuestion(id="F1", question="Frage 1")],
summary="",
)
return ctx
# ───────────────────────────────────────────────────────────────────
# QueryAnchorNode
# ───────────────────────────────────────────────────────────────────
class TestQueryAnchorNode(unittest.TestCase):
@async_test
async def test_dgx_query_classifies_none(self):
"""CRITICAL: a product-comparison query → anchor.type='none'."""
llm = MockLLM().queue_json({
"anchor_type": "none", "target": "", "confidence": 0.94,
"reasoning": "Vergleichsanfrage zu Hardware-Spezifikationen ohne Personenbezug.",
})
original = _patch_adapter_with(llm)
try:
node = QueryAnchorNode(llm=MagicMock())
ctx = _make_ctx("Vergleich DGX B300 vs B200 — Preis-Leistungs-Verhältnis")
metadata = await node.run(ctx)
finally:
_restore_adapter(original)
self.assertEqual(ctx.query_anchor.type, "none")
self.assertFalse(ctx.query_anchor.is_person())
self.assertEqual(metadata["anchor_type"], "none")
self.assertEqual(len(ctx.classifier_calls), 1)
@async_test
async def test_skipped_if_already_classified(self):
"""If ctx.query_anchor is already set: not again."""
node = QueryAnchorNode(llm=MagicMock())
ctx = _make_ctx()
ctx.query_anchor = QueryAnchor(type="person", target="X",
confidence=0.9)
applies, reason = node.applies_to(ctx)
self.assertFalse(applies)
self.assertIn("already set", reason)
# ───────────────────────────────────────────────────────────────────
# OffTopicFilterNode
# ───────────────────────────────────────────────────────────────────
class TestOffTopicFilterNode(unittest.TestCase):
@async_test
async def test_filters_off_topic_sources(self):
llm = MockLLM().queue_json([
{"index": 1, "relevance": "high", "confidence": 0.9,
"reasoning": "Direkter Bezug"},
{"index": 2, "relevance": "off_topic", "confidence": 0.95,
"reasoning": "Komplett unverwandtes Thema"},
])
original = _patch_adapter_with(llm)
try:
node = OffTopicFilterNode(llm=MagicMock())
ctx = _make_ctx()
ctx.sources = [
SourceDocument(SourceType.WEB_PAGE,
"https://relevant.com", "ok",
"x" * 600),
SourceDocument(SourceType.WEB_PAGE,
"https://cooking.com", "irrelevant",
"y" * 600),
]
await node.run(ctx)
finally:
_restore_adapter(original)
self.assertEqual(len(ctx.sources), 1)
self.assertEqual(ctx.sources[0].url, "https://relevant.com")
@async_test
async def test_skipped_without_plan(self):
node = OffTopicFilterNode(llm=MagicMock())
ctx = _make_ctx(with_plan=False)
applies, _ = node.applies_to(ctx)
self.assertFalse(applies)
# ───────────────────────────────────────────────────────────────────
# CoverageNode + ContinueDecisionNode (together)
# ───────────────────────────────────────────────────────────────────
class TestCoverageAndContinueNodes(unittest.TestCase):
@async_test
async def test_coverage_then_continue_stops(self):
"""Coverage answered → Continue stop_done → Pipeline-Stop."""
llm = MockLLM()
# coverage answer
llm.queue_json({
"coverage": "answered", "confidence": 0.9,
"missing_aspects": [], "supporting_extract_ids": [],
"reasoning": "Frage ist beantwortet.",
})
# continue answer
llm.queue_json({
"decision": "stop_done", "confidence": 0.88,
"next_round_focus": "",
"reasoning": "Coverage hoch, weitere Runde brächte nichts.",
})
original = _patch_adapter_with(llm)
try:
ctx = _make_ctx()
ctx.extracts = [
SourceExtract("u", "t", "F1", "fact", polarity="positive"),
]
ctx.sources = [
SourceDocument(SourceType.WEB_PAGE, "u", "t", "c"),
]
cov_node = CoverageNode(llm=MagicMock(), round_number=1)
cont_node = ContinueDecisionNode(
llm=MagicMock(), round_number=1, max_rounds=3,
new_extracts=1, new_sources=1,
)
pipeline = PipelineDAG([cov_node, cont_node])
await pipeline.run(ctx)
finally:
_restore_adapter(original)
# coverage delivered a result
self.assertEqual(len(ctx.coverage_per_round), 1)
self.assertEqual(
ctx.coverage_per_round[0]["results"]["F1"]["coverage"],
"answered",
)
# continue decided stop_done
self.assertEqual(ctx.continue_decisions[0]["decision"], "stop_done")
# the engine processed the STOP_PIPELINE signal
cont_result = ctx.node_results[1]
self.assertEqual(cont_result.status, NodeStatus.STOP_PIPELINE)
# ───────────────────────────────────────────────────────────────────
# DiagnosisNode — a run in which a filter discarded everything
# ───────────────────────────────────────────────────────────────────
class TestDiagnosisNode(unittest.TestCase):
@async_test
async def test_dgx_diagnosis_via_dag(self):
"""CRITICAL: final state of such a run → diagnosis='filter_too_strict'.
This time via the DAG pipeline — verifies that the classifier
behaviour is preserved in the DAG.
"""
llm = MockLLM().queue_json({
"diagnosis": "filter_too_strict", "confidence": 0.95,
"remediation": "Recherche ohne Personen-Filter neu starten",
"user_message": "528 Extrakte vom Filter verworfen",
"reasoning": "Filter-Verlustrate 100%, alle Fragen filter_blocked.",
})
original = _patch_adapter_with(llm)
try:
ctx = _make_ctx("DGX B300 vs B200")
ctx.research_plan = ResearchPlan(
questions=[
ResearchQuestion(id=f"F{i}", question="x", priority="hoch")
for i in range(1, 5)
],
summary="",
)
ctx.rounds_completed = 2
ctx.coverage_per_round = [{
"round": 2,
"results": {
f"F{i}": {"coverage": "filter_blocked", "confidence": 0.9,
"reasoning": ""}
for i in range(1, 5)
},
}]
ctx.continue_decisions = [{
"round": 2, "decision": "stop_filter_problem",
"confidence": 0.91,
}]
ctx.filter_stats_per_round = [{
"person_hallucination": {
"activated": True, "loss_rate": 1.0,
"rejected": 528, "total": 528,
},
}]
ctx.query_anchor = QueryAnchor(type="person",
target="Preis Leistungs",
confidence=0.55)
node = DiagnosisNode(llm=MagicMock(), max_rounds=3)
await node.run(ctx)
finally:
_restore_adapter(original)
self.assertEqual(ctx.final_diagnosis["diagnosis"], "filter_too_strict")
self.assertTrue(ctx.final_diagnosis["is_problematic"])
self.assertIn("Personen-Filter", ctx.final_diagnosis["remediation"])
# ───────────────────────────────────────────────────────────────────
# FactoidVerificationNode
# ───────────────────────────────────────────────────────────────────
class TestFactoidVerificationNode(unittest.TestCase):
@async_test
async def test_skipped_if_no_report(self):
node = FactoidVerificationNode(llm=MagicMock())
ctx = _make_ctx()
ctx.final_report = ""
applies, reason = node.applies_to(ctx)
self.assertFalse(applies)
@async_test
async def test_full_flow_with_unverified(self):
llm = MockLLM()
# extraction
llm.queue_json([
{"factoid": "DGX kam 1999", "type": "date",
"report_position": "..."},
])
# verification
llm.queue_json([
{"factoid_index": 1, "verified": "false", "confidence": 0.95,
"supporting_extract_id": "E0",
"supporting_quote": "Released 17.10.2024",
"reasoning": "Widerspruch"},
])
original = _patch_adapter_with(llm)
try:
node = FactoidVerificationNode(llm=MagicMock())
ctx = _make_ctx()
ctx.final_report = "Die DGX kam 1999 raus."
ctx.extracts = [
SourceExtract("u", "t", "F1", "Released 17.10.2024",
polarity="positive"),
]
metadata = await node.run(ctx)
finally:
_restore_adapter(original)
self.assertEqual(metadata["unverified"], 1)
# FactoidVerificationNode does NOT handle high-confidence cases (≥0.9)
# itself — they go to ReportRevisionNode.
# The report stays unchanged after FactoidVerificationNode;
# only ctx.factoid_verifications is set.
self.assertEqual(ctx.final_report, "Die DGX kam 1999 raus.",
"high-confidence case: FactoidVerificationNode "
"must not annotate the report itself.")
self.assertEqual(len(ctx.factoid_verifications), 1)
self.assertEqual(ctx.factoid_verifications[0]["verified"], "false")
self.assertEqual(ctx.factoid_verifications[0]["confidence"], 0.95)
# ───────────────────────────────────────────────────────────────────
# Run a complete mini pipeline
# ───────────────────────────────────────────────────────────────────
class TestMultiNodePipeline(unittest.TestCase):
@async_test
async def test_full_post_synthesis_pipeline(self):
"""End to end: diagnosis + factoid verification as a DAG."""
llm = MockLLM()
# Diagnose
llm.queue_json({
"diagnosis": "successful", "confidence": 0.9,
"remediation": "", "user_message": "ok",
"reasoning": "Solide Quellenlage.",
})
# Extract
llm.queue_json([
{"factoid": "Behauptung", "type": "claim",
"report_position": "..."},
])
# Verify
llm.queue_json([
{"factoid_index": 1, "verified": "true", "confidence": 0.85,
"supporting_extract_id": "E0", "supporting_quote": "X",
"reasoning": "match"},
])
original = _patch_adapter_with(llm)
try:
ctx = _make_ctx()
ctx.rounds_completed = 1
ctx.final_report = "## Bericht\n\nBehauptung."
ctx.extracts = [
SourceExtract("u", "t", "F1", "X", polarity="positive"),
]
ctx.coverage_per_round = [{
"round": 1,
"results": {"F1": {"coverage": "answered", "confidence": 0.9,
"reasoning": ""}},
}]
ctx.filter_stats_per_round = []
ctx.continue_decisions = []
ctx.query_anchor = QueryAnchor()
pipeline = PipelineDAG([
DiagnosisNode(llm=MagicMock(), max_rounds=3),
FactoidVerificationNode(llm=MagicMock()),
])
await pipeline.run(ctx)
finally:
_restore_adapter(original)
# both nodes ran successfully
statuses = [r.status for r in ctx.node_results]
self.assertEqual(statuses, [NodeStatus.OK, NodeStatus.OK])
self.assertEqual(ctx.final_diagnosis["diagnosis"], "successful")
self.assertEqual(len(ctx.factoid_verifications), 1)
self.assertEqual(ctx.factoid_verifications[0]["verified"], "true")
if __name__ == "__main__":
unittest.main()