jpmschweitzerandClaude Opus 4.5
3d11b7ae4f
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>
354 lines
11 KiB
Python
354 lines
11 KiB
Python
"""
|
|
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
|