test: add tests for streaming orchestration
Comprehensive tests for orchestration module: - Delegation parsing from Steward's note - Context extraction (reason, complexity, context fields) - Delegation execution routing - Think update emission (before/after delegation) - Expert output yielding - Error handling for failed delegations - Pre-parsed task handling 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.5
@@ -0,0 +1,353 @@
|
||||
"""
|
||||
Tests for orchestration module.
|
||||
|
||||
Tests the multi-expert coordination infrastructure including
|
||||
delegation parsing, think updates, and result handling.
|
||||
"""
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
from src.agents.orchestration import (
|
||||
OrchestrationContext,
|
||||
parse_delegation_from_steward_note,
|
||||
execute_delegation,
|
||||
orchestrate_with_think_updates,
|
||||
extract_delegation_context,
|
||||
)
|
||||
from src.agents.delegation import DelegationTask, DelegationResult
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestParseDelegation:
|
||||
"""Tests for parsing delegation from Steward's note."""
|
||||
|
||||
def test_parse_librarian_create(self):
|
||||
"""Test parsing librarian create delegation."""
|
||||
note = """DELEGATE: librarian to create a wiki page about CI/CD pipelines
|
||||
REASON: User wants to document CI/CD concepts
|
||||
COMPLEXITY: moderate
|
||||
CONTEXT: none"""
|
||||
|
||||
task = parse_delegation_from_steward_note(note)
|
||||
|
||||
assert task is not None
|
||||
assert task.expert_name == "librarian"
|
||||
assert "create a wiki page about CI/CD pipelines" in task.task
|
||||
|
||||
def test_parse_librarian_search(self):
|
||||
"""Test parsing librarian search delegation."""
|
||||
note = """DELEGATE: librarian to search for information about Docker networking
|
||||
REASON: User needs Docker documentation
|
||||
COMPLEXITY: simple"""
|
||||
|
||||
task = parse_delegation_from_steward_note(note)
|
||||
|
||||
assert task is not None
|
||||
assert task.expert_name == "librarian"
|
||||
assert "search for information about Docker networking" in task.task
|
||||
|
||||
def test_parse_no_delegation(self):
|
||||
"""Test parsing when no delegation needed."""
|
||||
note = """DELEGATE: none (conversational response only)
|
||||
REASON: Simple greeting requires no tools
|
||||
COMPLEXITY: simple"""
|
||||
|
||||
task = parse_delegation_from_steward_note(note)
|
||||
|
||||
assert task is None
|
||||
|
||||
def test_parse_tatlock_core(self):
|
||||
"""Test parsing tatlock_core delegation."""
|
||||
note = """DELEGATE: tatlock_core to calculate the result
|
||||
REASON: Math calculation needed
|
||||
COMPLEXITY: simple"""
|
||||
|
||||
task = parse_delegation_from_steward_note(note)
|
||||
|
||||
assert task is not None
|
||||
assert task.expert_name == "tatlock_core"
|
||||
assert "calculate the result" in task.task
|
||||
|
||||
def test_parse_case_insensitive(self):
|
||||
"""Test parsing is case insensitive."""
|
||||
note = """delegate: LIBRARIAN to search docs
|
||||
reason: Research query"""
|
||||
|
||||
task = parse_delegation_from_steward_note(note)
|
||||
|
||||
assert task is not None
|
||||
assert task.expert_name == "librarian"
|
||||
|
||||
def test_parse_missing_delegate(self):
|
||||
"""Test parsing when DELEGATE line is missing."""
|
||||
note = """REASON: This has no delegation
|
||||
COMPLEXITY: simple"""
|
||||
|
||||
task = parse_delegation_from_steward_note(note)
|
||||
|
||||
assert task is None
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestExtractDelegationContext:
|
||||
"""Tests for extracting context from Steward's note."""
|
||||
|
||||
def test_extract_all_fields(self):
|
||||
"""Test extracting all context fields."""
|
||||
note = """DELEGATE: librarian to create wiki page
|
||||
REASON: User wants documentation
|
||||
COMPLEXITY: moderate
|
||||
CONTEXT: Related to previous discussion about DevOps"""
|
||||
|
||||
context = extract_delegation_context(note)
|
||||
|
||||
assert context["reason"] == "User wants documentation"
|
||||
assert context["complexity"] == "moderate"
|
||||
assert "Related to previous discussion" in context["context"]
|
||||
|
||||
def test_extract_partial_fields(self):
|
||||
"""Test extracting when some fields missing."""
|
||||
note = """DELEGATE: librarian to search
|
||||
REASON: Research query
|
||||
COMPLEXITY: simple"""
|
||||
|
||||
context = extract_delegation_context(note)
|
||||
|
||||
assert context["reason"] == "Research query"
|
||||
assert context["complexity"] == "simple"
|
||||
assert context["context"] == ""
|
||||
|
||||
def test_extract_empty_note(self):
|
||||
"""Test extracting from empty note."""
|
||||
context = extract_delegation_context("")
|
||||
|
||||
assert context["reason"] == ""
|
||||
assert context["complexity"] == ""
|
||||
assert context["context"] == ""
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestExecuteDelegation:
|
||||
"""Tests for executing delegation tasks."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_librarian_delegation(self):
|
||||
"""Test executing delegation to librarian."""
|
||||
task = DelegationTask(
|
||||
expert_name="librarian",
|
||||
task="search for Docker docs",
|
||||
context="User learning Docker",
|
||||
)
|
||||
|
||||
mock_result = DelegationResult(
|
||||
expert_name="librarian",
|
||||
task="search for Docker docs",
|
||||
success=True,
|
||||
output="Found Docker documentation...",
|
||||
)
|
||||
|
||||
with patch(
|
||||
"src.agents.orchestration.delegate_to_librarian",
|
||||
new_callable=AsyncMock,
|
||||
return_value=mock_result,
|
||||
) as mock_delegate:
|
||||
result = await execute_delegation(task)
|
||||
|
||||
mock_delegate.assert_called_once_with(
|
||||
task="search for Docker docs",
|
||||
context="User learning Docker",
|
||||
)
|
||||
|
||||
assert result.success is True
|
||||
assert "Docker" in result.output
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_unknown_expert(self):
|
||||
"""Test executing delegation to unknown expert."""
|
||||
task = DelegationTask(
|
||||
expert_name="unknown_expert",
|
||||
task="do something",
|
||||
)
|
||||
|
||||
result = await execute_delegation(task)
|
||||
|
||||
assert result.success is False
|
||||
assert "Unknown expert" in result.error
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestOrchestrateWithThinkUpdates:
|
||||
"""Tests for orchestration with think updates."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_orchestrate_emits_think_before_delegation(self):
|
||||
"""Test that think update is emitted before delegation."""
|
||||
mock_result = DelegationResult(
|
||||
expert_name="librarian",
|
||||
task="search docs",
|
||||
success=True,
|
||||
output="Found results",
|
||||
)
|
||||
|
||||
with patch(
|
||||
"src.agents.orchestration.delegate_to_librarian",
|
||||
new_callable=AsyncMock,
|
||||
return_value=mock_result,
|
||||
):
|
||||
updates = []
|
||||
async for update in orchestrate_with_think_updates(
|
||||
user_message="Search for Docker info",
|
||||
steward_note="DELEGATE: librarian to search for Docker info",
|
||||
):
|
||||
updates.append(update)
|
||||
|
||||
# First update should be think tag about consulting
|
||||
assert any("<think>" in u and "Consulting" in u for u in updates)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_orchestrate_emits_think_after_delegation(self):
|
||||
"""Test that think update is emitted after delegation."""
|
||||
mock_result = DelegationResult(
|
||||
expert_name="librarian",
|
||||
task="search docs",
|
||||
success=True,
|
||||
output="Found results",
|
||||
)
|
||||
|
||||
with patch(
|
||||
"src.agents.orchestration.delegate_to_librarian",
|
||||
new_callable=AsyncMock,
|
||||
return_value=mock_result,
|
||||
):
|
||||
updates = []
|
||||
async for update in orchestrate_with_think_updates(
|
||||
user_message="Search for Docker info",
|
||||
steward_note="DELEGATE: librarian to search for Docker info",
|
||||
):
|
||||
updates.append(update)
|
||||
|
||||
# Should have think tag about completion
|
||||
assert any("<think>" in u and "completed" in u for u in updates)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_orchestrate_yields_expert_output(self):
|
||||
"""Test that expert output is yielded."""
|
||||
mock_result = DelegationResult(
|
||||
expert_name="librarian",
|
||||
task="search docs",
|
||||
success=True,
|
||||
output="Found Docker documentation with networking details",
|
||||
)
|
||||
|
||||
with patch(
|
||||
"src.agents.orchestration.delegate_to_librarian",
|
||||
new_callable=AsyncMock,
|
||||
return_value=mock_result,
|
||||
):
|
||||
updates = []
|
||||
async for update in orchestrate_with_think_updates(
|
||||
user_message="Search for Docker info",
|
||||
steward_note="DELEGATE: librarian to search for Docker info",
|
||||
):
|
||||
updates.append(update)
|
||||
|
||||
# Should include expert output
|
||||
all_output = "".join(updates)
|
||||
assert "Docker documentation" in all_output
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_orchestrate_handles_delegation_failure(self):
|
||||
"""Test that delegation failure emits warning think update."""
|
||||
mock_result = DelegationResult(
|
||||
expert_name="librarian",
|
||||
task="search docs",
|
||||
success=False,
|
||||
output="",
|
||||
error="Connection timeout",
|
||||
)
|
||||
|
||||
with patch(
|
||||
"src.agents.orchestration.delegate_to_librarian",
|
||||
new_callable=AsyncMock,
|
||||
return_value=mock_result,
|
||||
):
|
||||
updates = []
|
||||
async for update in orchestrate_with_think_updates(
|
||||
user_message="Search for info",
|
||||
steward_note="DELEGATE: librarian to search",
|
||||
):
|
||||
updates.append(update)
|
||||
|
||||
# Should have warning think update
|
||||
all_output = "".join(updates)
|
||||
assert "⚠️" in all_output or "issue" in all_output.lower()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_orchestrate_no_delegation_returns_empty(self):
|
||||
"""Test that no delegation yields nothing."""
|
||||
updates = []
|
||||
async for update in orchestrate_with_think_updates(
|
||||
user_message="Hello",
|
||||
steward_note="DELEGATE: none (conversational)",
|
||||
):
|
||||
updates.append(update)
|
||||
|
||||
assert len(updates) == 0
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_orchestrate_with_preparsed_task(self):
|
||||
"""Test orchestration with pre-parsed delegation task."""
|
||||
task = DelegationTask(
|
||||
expert_name="librarian",
|
||||
task="create wiki page",
|
||||
)
|
||||
|
||||
mock_result = DelegationResult(
|
||||
expert_name="librarian",
|
||||
task="create wiki page",
|
||||
success=True,
|
||||
output="Wiki page created",
|
||||
)
|
||||
|
||||
with patch(
|
||||
"src.agents.orchestration.delegate_to_librarian",
|
||||
new_callable=AsyncMock,
|
||||
return_value=mock_result,
|
||||
):
|
||||
updates = []
|
||||
async for update in orchestrate_with_think_updates(
|
||||
user_message="Create wiki page",
|
||||
steward_note="", # Empty note since task is pre-parsed
|
||||
delegation_task=task,
|
||||
):
|
||||
updates.append(update)
|
||||
|
||||
assert len(updates) > 0
|
||||
all_output = "".join(updates)
|
||||
assert "Wiki page created" in all_output
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestOrchestrationContext:
|
||||
"""Tests for OrchestrationContext dataclass."""
|
||||
|
||||
def test_context_creation(self):
|
||||
"""Test creating orchestration context."""
|
||||
ctx = OrchestrationContext(
|
||||
user_message="Test message",
|
||||
steward_note="Test note",
|
||||
conversation_id="conv_123",
|
||||
)
|
||||
|
||||
assert ctx.user_message == "Test message"
|
||||
assert ctx.steward_note == "Test note"
|
||||
assert ctx.conversation_id == "conv_123"
|
||||
|
||||
def test_context_defaults(self):
|
||||
"""Test orchestration context default values."""
|
||||
ctx = OrchestrationContext(
|
||||
user_message="Test",
|
||||
steward_note="Note",
|
||||
)
|
||||
|
||||
assert ctx.conversation_id is None
|
||||
Reference in New Issue
Block a user