tests/test_plan_gate_split.py
Rama main · réplica de lectura del repositorio original.
"""
Tests for splitting the runner for the plan gate:
- decompose_only() returns the plan without executing it
- run(skip_decompose=True) skips the decomposer phase
- a two-phase run (decompose_only → run(skip_decompose=True)) gives the
same final result as a normal run().
This interruption is the basis of the plan preview gate: the UI can stop
between decompose and execute, show the plan and let the user confirm it.
"""
import asyncio
import sys
from pathlib import Path
from unittest.mock import MagicMock
# Dependency mocks as in test_all_use_cases
if "httpx" not in sys.modules:
sys.modules["httpx"] = MagicMock()
if "src.connectors.rate_limiter" not in sys.modules:
_rl = MagicMock()
_rl.get_rate_limiter = MagicMock(return_value=MagicMock())
sys.modules["src.connectors.rate_limiter"] = _rl
sys.path.insert(0, str(Path(__file__).parent.parent))
sys.path.insert(0, str(Path(__file__).parent))
from src.pipeline.analysis_pipeline import ( # noqa: E402
AnalysisPipelineRunner,
)
from src.pipeline.models import HarvestContext, TaskState # noqa: E402
# Reuse the shared SmartMockLLM — it covers all analysis use cases
# (peer review, decision, explainer, ...)
from tests._mocks import SmartMockLLM # noqa: E402
# ─── Mock-LLM ──────────────────────────────────────────────────────
class MockSubClient:
def __init__(self, name: str):
self.model_name = name
self.total_prompt_tokens = 0
self.total_completion_tokens = 0
self.total_requests = 0
class SimpleMockLLM:
"""Minimal mock for the peer-review pipeline."""
def __init__(self):
self.primary = MockSubClient("mock-primary")
self.harvest = MockSubClient("mock-harvest")
def _acc(self, client):
client.total_requests += 1
client.total_prompt_tokens += 100
client.total_completion_tokens += 50
async def primary_complete(self, messages, max_tokens=None,
enable_thinking=None) -> str:
self._acc(self.primary)
return self._respond(messages[0]["content"])
async def primary_complete_json(self, messages, max_tokens=None) -> dict:
self._acc(self.primary)
return self._respond_json(messages[0]["content"])
async def harvest_complete(self, messages, max_tokens=None) -> str:
self._acc(self.harvest)
return self._respond(messages[0]["content"])
async def harvest_complete_json(self, messages, max_tokens=None) -> dict:
self._acc(self.harvest)
return self._respond_json(messages[0]["content"])
def get_usage_stats(self) -> dict:
return {
"primary": {
"model": self.primary.model_name,
"requests": self.primary.total_requests,
"prompt_tokens": self.primary.total_prompt_tokens,
"completion_tokens": self.primary.total_completion_tokens,
"total_tokens": (self.primary.total_prompt_tokens
+ self.primary.total_completion_tokens),
},
"harvest": {
"model": self.harvest.model_name,
"requests": self.harvest.total_requests,
"prompt_tokens": self.harvest.total_prompt_tokens,
"completion_tokens": self.harvest.total_completion_tokens,
"total_tokens": (self.harvest.total_prompt_tokens
+ self.harvest.total_completion_tokens),
},
}
def _respond_json(self, prompt: str) -> dict:
# Peer-Review Planning
if "Peer Review" in prompt and "criteria" in prompt:
return {
"paper_type": "article",
"criteria": ["Methodik", "Relevanz", "Klarheit"],
"target_audience": "reviewer",
}
# criterion evaluation
if "score" in prompt.lower() or "bewertung" in prompt.lower():
return {
"score": 4,
"rationale": "Mock rationale",
"strengths": ["ok"],
"weaknesses": [],
}
return {}
def _respond(self, prompt: str) -> str:
if "finaler" in prompt.lower() or "synthese" in prompt.lower():
return "# Peer Review Mock\n\nMock report body."
return "Mock response text."
# ─── Test-Helpers ──────────────────────────────────────────────────
def _preflight_inputs_peer_review():
# Peer review expects at least 2000 characters of paper_text + discipline + role
paper_text = (
"Abstract\n\n" + "Einleitung zur Studie und Hintergrund. " * 50 +
"\n\nMethoden\n\n" + "Methodische Beschreibung der Studie. " * 50 +
"\n\nErgebnisse\n\n" + "Ergebnisbeschreibung und Analyse. " * 50 +
"\n\nDiskussion\n\n" + "Diskussion der Befunde im Kontext. " * 50
)
return {
"paper_text": paper_text,
"discipline": "Sozialwissenschaften",
"role": "Journal-Gutachten",
"focus": "",
}
async def _new_runner():
mock_llm = SmartMockLLM()
return AnalysisPipelineRunner(
progress_callback=None,
llm_client=mock_llm, max_parallel=3, rate_limit=100,
)
async def _new_context(use_case: str, inputs: dict) -> HarvestContext:
ctx = HarvestContext(query=inputs.get("paper_text", "test")[:80])
ctx.use_case = use_case
ctx.preflight_data = inputs
return ctx
# ─── Tests ─────────────────────────────────────────────────────────
async def test_decompose_only_returns_plan():
"""decompose_only returns a plan without executing it."""
print("Test: decompose_only returns a plan ...", end=" ")
runner = await _new_runner()
ctx = await _new_context("peer_review", _preflight_inputs_peer_review())
ctx._llm_client = runner.llm_client
plan = (await runner.decompose_only(ctx)).task_plan
assert plan is not None, f"plan is None: {ctx.error_message}"
assert plan.use_case == "peer_review"
assert len(plan.tasks) > 0
# task_plan must be in ctx too
assert ctx.task_plan is plan
# The pipeline status is not "done" yet — nothing was executed
assert ctx.status != "done"
# no task results
assert not ctx.task_results or len(ctx.task_results) == 0
print(f"✓ ({len(plan.tasks)} Tasks, {plan.estimated_calls} Calls)")
async def test_decompose_only_then_run_skip_decompose():
"""Two-phase flow: decompose_only → run(skip_decompose=True) returns
a complete result."""
print("Test: decompose_only + run(skip_decompose) ...", end=" ")
runner = await _new_runner()
ctx = await _new_context("peer_review", _preflight_inputs_peer_review())
ctx._llm_client = runner.llm_client
# Phase 1: only build the plan
plan = (await runner.decompose_only(ctx)).task_plan
assert plan is not None
# Phase 2: run the plan
await runner.run(ctx, skip_decompose=True)
assert ctx.status == "done", (
f"status: {ctx.status}, error: {ctx.error_message}"
)
assert ctx.final_report, "final_report is empty"
# the task results must be there now
assert len(ctx.task_results) > 0
# the task plan is the same as after decompose_only
assert ctx.task_plan is plan
print(f"✓ ({len(ctx.task_results)} Tasks done)")
async def test_skip_decompose_without_plan_fails():
"""run(skip_decompose=True) without a previous ctx.task_plan returns
an error."""
print("Test: skip_decompose without a plan → error ...", end=" ")
runner = await _new_runner()
ctx = await _new_context("peer_review", _preflight_inputs_peer_review())
ctx._llm_client = runner.llm_client
# no decompose_only() — ctx.task_plan is None
await runner.run(ctx, skip_decompose=True)
assert ctx.status == "error"
assert "TaskPlan" in (ctx.error_message or "")
print("✓")
async def test_normal_run_still_works():
"""A normal run() without skip_decompose works unchanged."""
print("Test: normal run() unchanged ...", end=" ")
runner = await _new_runner()
ctx = await _new_context("peer_review", _preflight_inputs_peer_review())
ctx._llm_client = runner.llm_client
await runner.run(ctx)
assert ctx.status == "done", f"Status: {ctx.status}: {ctx.error_message}"
assert ctx.final_report
assert len(ctx.task_results) > 0
print("✓")
async def test_decompose_only_result_matches_full_run():
"""The plan from decompose_only is structurally identical to the one a
normal run() creates."""
print("Test: decompose_only Plan == full run Plan ...", end=" ")
runner_a = await _new_runner()
ctx_a = await _new_context(
"peer_review", _preflight_inputs_peer_review()
)
ctx_a._llm_client = runner_a.llm_client
plan_a = (await runner_a.decompose_only(ctx_a)).task_plan
runner_b = await _new_runner()
ctx_b = await _new_context(
"peer_review", _preflight_inputs_peer_review()
)
ctx_b._llm_client = runner_b.llm_client
await runner_b.run(ctx_b)
# same number of tasks and same use case
assert len(plan_a.tasks) == len(ctx_b.task_plan.tasks)
assert plan_a.use_case == ctx_b.task_plan.use_case
# same task IDs
ids_a = sorted(t.id for t in plan_a.tasks)
ids_b = sorted(t.id for t in ctx_b.task_plan.tasks)
assert ids_a == ids_b, f"task IDs differ: {ids_a} vs {ids_b}"
print("✓")
async def test_two_phase_vs_single_phase_same_result():
"""The two-phase flow produces the same number of completed tasks as
the single-phase flow."""
print("Test: two vs. one phase, same result ...", end=" ")
# single-phase
runner_single = await _new_runner()
ctx_single = await _new_context(
"peer_review", _preflight_inputs_peer_review()
)
ctx_single._llm_client = runner_single.llm_client
await runner_single.run(ctx_single)
# two-phase
runner_two = await _new_runner()
ctx_two = await _new_context(
"peer_review", _preflight_inputs_peer_review()
)
ctx_two._llm_client = runner_two.llm_client
await runner_two.decompose_only(ctx_two)
await runner_two.run(ctx_two, skip_decompose=True)
assert ctx_single.status == "done"
assert ctx_two.status == "done"
# same number of done tasks
done_single = sum(
1 for r in ctx_single.task_results.values()
if r.state == TaskState.DONE
)
done_two = sum(
1 for r in ctx_two.task_results.values()
if r.state == TaskState.DONE
)
assert done_single == done_two, (
f"single-phase: {done_single} done, two-phase: {done_two} done"
)
# same number of tasks in the plan
assert len(ctx_single.task_plan.tasks) == len(ctx_two.task_plan.tasks)
print(f"✓ (both {done_single} done tasks)")
async def main():
print("=" * 60)
print("Plan-Gate: Runner-Split Tests")
print("=" * 60)
tests = [
test_decompose_only_returns_plan,
test_decompose_only_then_run_skip_decompose,
test_skip_decompose_without_plan_fails,
test_normal_run_still_works,
test_decompose_only_result_matches_full_run,
test_two_phase_vs_single_phase_same_result,
]
failed = 0
for t in tests:
try:
await t()
except AssertionError as e:
print(f"❌ {t.__name__}: {e}")
failed += 1
except Exception as e:
print(f"💥 {t.__name__}: {type(e).__name__}: {e}")
import traceback
traceback.print_exc()
failed += 1
print("=" * 60)
if failed:
print(f"❌ {failed}/{len(tests)} tests failed")
sys.exit(1)
else:
print(f"✅ All {len(tests)} tests passed")
if __name__ == "__main__":
asyncio.run(main())