test: add unit tests for delegation infrastructure
Tests for DelegationTask, DelegationResult, delegate_to_librarian: - Task creation with auto-generated IDs - Task dependencies and custom IDs - Successful delegation with result - Error handling in delegation - Result preservation Tests for get_delegation_tools(): - Returns wrapper for members with agent - Returns raw tools for members without agent - Handles mixed member types correctly - Graceful handling of non-existent members 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,195 @@
|
||||
"""
|
||||
Tests for delegation infrastructure.
|
||||
|
||||
Tests the DelegationTask dataclass and delegation wrapper functions
|
||||
that implement the agent-as-tool pattern.
|
||||
"""
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock, patch, MagicMock
|
||||
|
||||
from src.agents.delegation import (
|
||||
DelegationTask,
|
||||
DelegationResult,
|
||||
delegate_to_librarian,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestDelegationTask:
|
||||
"""Tests for the DelegationTask dataclass."""
|
||||
|
||||
def test_delegation_task_creation(self):
|
||||
"""Test basic DelegationTask creation."""
|
||||
task = DelegationTask(
|
||||
expert_name="librarian",
|
||||
task="Create a wiki page about CI/CD",
|
||||
context="User is setting up a homelab",
|
||||
action="create",
|
||||
)
|
||||
|
||||
assert task.expert_name == "librarian"
|
||||
assert task.task == "Create a wiki page about CI/CD"
|
||||
assert task.context == "User is setting up a homelab"
|
||||
assert task.action == "create"
|
||||
|
||||
def test_delegation_task_default_values(self):
|
||||
"""Test DelegationTask default values."""
|
||||
task = DelegationTask(
|
||||
expert_name="librarian",
|
||||
task="Search for Docker info",
|
||||
)
|
||||
|
||||
assert task.context == ""
|
||||
assert task.action == ""
|
||||
assert task.priority == 0
|
||||
assert task.depends_on == []
|
||||
assert task.result is None
|
||||
|
||||
def test_delegation_task_auto_generates_id(self):
|
||||
"""Test DelegationTask auto-generates unique IDs."""
|
||||
task1 = DelegationTask(expert_name="librarian", task="Task 1")
|
||||
task2 = DelegationTask(expert_name="librarian", task="Task 2")
|
||||
|
||||
assert task1.task_id.startswith("librarian_")
|
||||
assert task2.task_id.startswith("librarian_")
|
||||
assert task1.task_id != task2.task_id
|
||||
|
||||
def test_delegation_task_preserves_custom_id(self):
|
||||
"""Test DelegationTask preserves custom ID if provided."""
|
||||
task = DelegationTask(
|
||||
expert_name="librarian",
|
||||
task="Custom task",
|
||||
task_id="custom_id_123",
|
||||
)
|
||||
|
||||
assert task.task_id == "custom_id_123"
|
||||
|
||||
def test_delegation_task_with_dependencies(self):
|
||||
"""Test DelegationTask with dependencies."""
|
||||
task = DelegationTask(
|
||||
expert_name="librarian",
|
||||
task="Update wiki page",
|
||||
depends_on=["memory_abc123", "search_def456"],
|
||||
)
|
||||
|
||||
assert len(task.depends_on) == 2
|
||||
assert "memory_abc123" in task.depends_on
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestDelegationResult:
|
||||
"""Tests for the DelegationResult dataclass."""
|
||||
|
||||
def test_delegation_result_success(self):
|
||||
"""Test successful DelegationResult."""
|
||||
result = DelegationResult(
|
||||
expert_name="librarian",
|
||||
task="Search for Docker info",
|
||||
success=True,
|
||||
output="Found 5 relevant documents about Docker...",
|
||||
)
|
||||
|
||||
assert result.expert_name == "librarian"
|
||||
assert result.success is True
|
||||
assert result.output.startswith("Found")
|
||||
assert result.error is None
|
||||
|
||||
def test_delegation_result_failure(self):
|
||||
"""Test failed DelegationResult."""
|
||||
result = DelegationResult(
|
||||
expert_name="librarian",
|
||||
task="Search for Docker info",
|
||||
success=False,
|
||||
output="",
|
||||
error="Connection timeout to library-desk API",
|
||||
)
|
||||
|
||||
assert result.success is False
|
||||
assert result.output == ""
|
||||
assert result.error == "Connection timeout to library-desk API"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestDelegateToLibrarian:
|
||||
"""Tests for the delegate_to_librarian wrapper."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delegate_to_librarian_success(self):
|
||||
"""Test successful delegation to Librarian."""
|
||||
mock_output = "Successfully created wiki page about CI/CD pipelines..."
|
||||
|
||||
with patch(
|
||||
"src.agents.librarian.agent.run_librarian",
|
||||
new_callable=AsyncMock,
|
||||
return_value=mock_output,
|
||||
) as mock_run:
|
||||
result = await delegate_to_librarian(
|
||||
task="Create a wiki page about CI/CD pipelines",
|
||||
context="User is setting up a homelab",
|
||||
)
|
||||
|
||||
# Verify run_librarian was called correctly
|
||||
mock_run.assert_called_once_with(
|
||||
task="Create a wiki page about CI/CD pipelines",
|
||||
context="User is setting up a homelab",
|
||||
)
|
||||
|
||||
# Verify result
|
||||
assert isinstance(result, DelegationResult)
|
||||
assert result.expert_name == "librarian"
|
||||
assert result.success is True
|
||||
assert result.output == mock_output
|
||||
assert result.error is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delegate_to_librarian_without_context(self):
|
||||
"""Test delegation to Librarian without context."""
|
||||
mock_output = "Found information about Docker networking..."
|
||||
|
||||
with patch(
|
||||
"src.agents.librarian.agent.run_librarian",
|
||||
new_callable=AsyncMock,
|
||||
return_value=mock_output,
|
||||
) as mock_run:
|
||||
result = await delegate_to_librarian(
|
||||
task="Search for information about Docker networking",
|
||||
)
|
||||
|
||||
mock_run.assert_called_once_with(
|
||||
task="Search for information about Docker networking",
|
||||
context="",
|
||||
)
|
||||
|
||||
assert result.success is True
|
||||
assert result.output == mock_output
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delegate_to_librarian_handles_error(self):
|
||||
"""Test delegation handles Librarian errors gracefully."""
|
||||
with patch(
|
||||
"src.agents.librarian.agent.run_librarian",
|
||||
new_callable=AsyncMock,
|
||||
side_effect=Exception("Connection refused"),
|
||||
):
|
||||
result = await delegate_to_librarian(
|
||||
task="Search for information",
|
||||
)
|
||||
|
||||
assert isinstance(result, DelegationResult)
|
||||
assert result.success is False
|
||||
assert result.output == ""
|
||||
assert result.error == "Connection refused"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delegate_to_librarian_preserves_task(self):
|
||||
"""Test delegation result preserves original task."""
|
||||
original_task = "Create a wiki page about Kubernetes deployments"
|
||||
|
||||
with patch(
|
||||
"src.agents.librarian.agent.run_librarian",
|
||||
new_callable=AsyncMock,
|
||||
return_value="Page created",
|
||||
):
|
||||
result = await delegate_to_librarian(task=original_task)
|
||||
|
||||
assert result.task == original_task
|
||||
Reference in New Issue
Block a user