""" Tests for the routing of `run()` to the DAG paths. Acceptance: - `run()` with standard arguments delegates to `run_via_dag()` (= ctx.node_results is filled). - `run()` with `plan_only=True` goes through `run_via_dag_plan_only()`. - `run()` with `existing_ctx + skip_decompose` goes through the DAG path; FormatAgent/Analysis are skipped. - a crash in the DAG path propagates (no silent fallback). """ import json import unittest from src.connectors.base import ConnectorRegistry from src.pipeline.models import ( HarvestContext, OutputSchema, ResearchPlan, ResearchQuestion, SearchResult, SourceType, ) from tests._helpers import async_test from tests.test_pipeline_smoke import ( _MockSearchConnector, _MockWebScraper, _ScriptedLLM, ) def _build_orchestrator(): """Standard setup: LLM + connectors with mock answers.""" primary = { "Beantwortet folgende Anfrage": json.dumps({ "summary": "x", "questions": [{ "id": "F1", "question": "?", "search_terms": ["x"], "priority": "hoch", }], "direct_urls": [], "git_repos": [], "directory_queries": [], }), "Erstelle einen strukturierten Bericht": "Bericht.", } harvest = { "PERSONEN- oder ORGANISATIONS-ANKER": json.dumps({ "anchor_type": "none", "target": "", "confidence": 0.9, "reasoning": "Produktanfrage ohne Personenbezug.", }), "Suchergebnisse, ob es zur Beantwortung": json.dumps([ {"index": i+1, "relevance": "high", "confidence": 0.9, "reasoning": "ja"} for i in range(10) ]), "Identifiziere alle FAKTOIDE": "[]", "Recherche-Frage durch die vorliegenden Extrakte": json.dumps({ "coverage": "answered", "confidence": 0.88, "missing_aspects": [], "reasoning": "ok.", }), "Du steuerst eine autonome Recherche": json.dumps({ "decision": "stop_done", "confidence": 0.85, "reasoning": "fertig.", }), "Diagnostiziere den Zustand": json.dumps({ "diagnosis": "successful", "confidence": 0.88, "user_message": "ok", "remediation": "", "reasoning": "ok.", }), "Extrahiere relevante Informationen": ( "[F1] [POSITIV] Fakt: x\n Verlässlichkeit: hoch" ), } llm = _ScriptedLLM(primary, harvest) reg = ConnectorRegistry() reg.register(_MockSearchConnector([ SearchResult( "Test", "https://test.com/1", "snippet long enough for filter passage", SourceType.WEB_SEARCH, ), ]), is_search_engine=True) reg.register(_MockWebScraper({}), is_web_scraper=True) from src.config import PipelineConfig from src.pipeline.orchestrator import ResearchOrchestrator config = PipelineConfig() config.max_rounds = 1 config.data_dir = "/tmp/rh_switchover_test" return ResearchOrchestrator(llm=llm, connectors=reg, config=config) async def _noop_cb(*args, **kwargs): pass class TestSwitchover(unittest.TestCase): @async_test async def test_standard_run_delegates_to_dag(self): """run() with standard arguments → DAG path.""" orch = _build_orchestrator() ctx = await orch.run( query="x", chat_history=[], context_docs="", template_name="", progress_callback=_noop_cb, mode="web", ) self.assertEqual(ctx.status, "done") # node_results is specific to the DAG path self.assertGreater( len(ctx.node_results), 0, "expected filled node_results — did the DAG path not run?", ) @async_test async def test_plan_only_uses_dag(self): """run() with plan_only=True → DAG plan-only path.""" orch = _build_orchestrator() ctx = await orch.run( query="x", chat_history=[], context_docs="", template_name="", progress_callback=_noop_cb, mode="web", plan_only=True, ) # Plan-Preview-Status self.assertEqual(ctx.status, "plan_ready") # DAG path: node_results filled (with FormatAgent + Analysis) self.assertGreater( len(getattr(ctx, "node_results", [])), 0, "plan_only should go through the DAG plan-only path", ) # NOT 13+ nodes as in the full pipeline — only the plan phase self.assertLessEqual( len(ctx.node_results), 5, f"plan_only should run only plan nodes, " f"but ran {len(ctx.node_results)}", ) @async_test async def test_existing_ctx_uses_dag_skips_plan_phase(self): """run() with existing_ctx + skip_decompose → DAG path, FormatAgent/Analysis are skipped.""" orch = _build_orchestrator() # Create a prepared context (as after plan_only) existing = HarvestContext(query="x", chat_history=[]) existing.output_schema = OutputSchema( title="X", format_type="report", sections=["A"], ) existing.research_plan = ResearchPlan( summary="schon da mit Text", questions=[ResearchQuestion( id="F1", question="?", search_terms=["x"], priority="hoch", )], ) ctx = await orch.run( query="x", chat_history=[], context_docs="", template_name="", progress_callback=_noop_cb, mode="web", existing_ctx=existing, skip_decompose=True, ) # Resume mode goes through the DAG → node_results filled self.assertGreater( len(getattr(ctx, "node_results", [])), 0, "resume mode should go through the DAG path", ) # FormatAgent and Analysis should appear as SKIPPED from src.pipeline.dag import NodeStatus skipped_names = [ r.node_name for r in ctx.node_results if r.status == NodeStatus.SKIPPED ] self.assertIn( "format_agent", skipped_names, f"format_agent expected skipped, found: {skipped_names}", ) self.assertIn( "analysis", skipped_names, f"analysis expected skipped, found: {skipped_names}", ) @async_test async def test_dag_failure_propagates(self): """A crash in the DAG path propagates — no silent fallback. No fallback is wanted: a crash in the DAG path must become visible as a bug, not be covered up by faulty masking. """ orch = _build_orchestrator() # Replace run_via_dag by a broken version async def broken(**kw): raise RuntimeError("DAG broken") orch.run_via_dag = broken # type: ignore with self.assertRaises(RuntimeError) as raised: await orch.run( query="x", chat_history=[], context_docs="", template_name="", progress_callback=_noop_cb, mode="web", ) self.assertIn("DAG broken", str(raised.exception)) if __name__ == "__main__": unittest.main()