File size: 2,333 Bytes
0257d2f e82d9c9 0257d2f e82d9c9 0257d2f e82d9c9 0257d2f |
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 |
from unittest.mock import MagicMock, patch
import pytest
# Skip entire module if agent_framework is not installed
agent_framework = pytest.importorskip("agent_framework")
from agent_framework import MagenticAgentMessageEvent, MagenticFinalResultEvent
from src.orchestrators.advanced import AdvancedOrchestrator as MagenticOrchestrator
class MockChatMessage:
def __init__(self, content):
self.content = content
@property
def text(self):
return self.content
@pytest.mark.asyncio
@pytest.mark.e2e
async def test_advanced_mode_completes_mocked():
"""Verify Advanced mode runs without crashing (mocked workflow)."""
# Initialize orchestrator (mocking requirements check)
with patch("src.orchestrators.advanced.check_magentic_requirements"):
orchestrator = MagenticOrchestrator(max_rounds=5)
# Mock the workflow
mock_workflow = MagicMock()
# Create fake events
# 1. Search Agent runs
mock_msg_1 = MockChatMessage("Found 5 papers on PubMed")
event1 = MagenticAgentMessageEvent(agent_id="SearchAgent", message=mock_msg_1)
# 2. Report Agent finishes
mock_result_msg = MockChatMessage("# Final Report\n\nFindings...")
event2 = MagenticFinalResultEvent(message=mock_result_msg)
async def mock_stream(task):
yield event1
yield event2
mock_workflow.run_stream = mock_stream
# Patch dependencies:
# _build_workflow: Returns our mock
# init_magentic_state: Avoids DB calls
# _init_embedding_service: Avoids loading embeddings
with (
patch.object(orchestrator, "_build_workflow", return_value=mock_workflow),
patch("src.orchestrators.advanced.init_magentic_state"),
patch.object(orchestrator, "_init_embedding_service", return_value=None),
):
events = []
async for event in orchestrator.run("test query"):
events.append(event)
# Check events
types = [e.type for e in events]
assert "started" in types
assert "thinking" in types
assert "search_complete" in types # Mapped from SearchAgent
assert "progress" in types # Added in SPEC_01
assert "complete" in types
complete_event = next(e for e in events if e.type == "complete")
assert "Final Report" in complete_event.message
|