File size: 3,298 Bytes
3a2b22f a5b5479 3a2b22f a5b5479 3a2b22f c6e9843 3a2b22f c6e9843 3a2b22f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 |
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from src.orchestrators.advanced import AdvancedOrchestrator
from src.orchestrators.factory import create_orchestrator
from src.utils.config import settings
@pytest.mark.asyncio
async def test_timeout_synthesizes_evidence():
"""Timeout should produce synthesis, not empty message."""
mock_client = MagicMock()
orchestrator = AdvancedOrchestrator(
max_rounds=1,
timeout_seconds=0.01,
chat_client=mock_client,
)
async def slow_stream(*args, **kwargs):
import asyncio
await asyncio.sleep(0.1)
yield MagicMock()
mock_workflow = MagicMock()
mock_workflow.run_stream = slow_stream
# Mock dependencies used inside the timeout block
# Note: get_magentic_state and create_report_agent are imported at module level in advanced.py
# so we must patch them in that module's namespace, not their original location
with (
patch.object(orchestrator, "_build_workflow", return_value=mock_workflow),
patch("src.orchestrators.advanced.init_magentic_state"),
patch("src.orchestrators.advanced.get_magentic_state") as mock_get_state,
patch("src.orchestrators.advanced.create_report_agent") as mock_create_agent,
):
# Setup mock state and memory
mock_memory = AsyncMock()
mock_memory.get_context_summary.return_value = "Mocked Evidence Summary"
mock_state = MagicMock()
mock_state.memory = mock_memory
mock_get_state.return_value = mock_state
# Setup mock ReportAgent
# ChatAgent.run() returns AgentRunResponse with .text property
mock_report_agent = AsyncMock()
mock_response = MagicMock()
mock_response.text = "Final Synthesized Report"
mock_report_agent.run.return_value = mock_response
mock_create_agent.return_value = mock_report_agent
events = []
async for e in orchestrator.run("test query"):
events.append(e)
complete_events = [e for e in events if e.type == "complete"]
assert len(complete_events) > 0
complete_event = complete_events[-1]
# Verify synthesis happened
assert complete_event.message == "Final Synthesized Report"
assert complete_event.data["reason"] == "timeout_synthesis"
# Verify mocks were called
mock_memory.get_context_summary.assert_called_once()
mock_create_agent.assert_called_once()
mock_report_agent.run.assert_awaited_once()
@pytest.mark.asyncio
async def test_factory_uses_advanced_max_rounds():
"""Factory should use settings.advanced_max_rounds for advanced mode."""
assert settings.advanced_max_rounds == 5
# Mock the internal helper that returns the class
with patch("src.orchestrators.factory._get_advanced_orchestrator_class") as mock_get_cls:
# Create a mock class that acts like AdvancedOrchestrator
mock_cls = MagicMock()
mock_get_cls.return_value = mock_cls
create_orchestrator(
mode="advanced",
api_key="sk-test",
)
# Verify the mock class was instantiated with correct max_rounds
_, kwargs = mock_cls.call_args
assert kwargs["max_rounds"] == 5
|