diff --git a/tests/agents/test_delegation.py b/tests/agents/test_delegation.py new file mode 100644 index 0000000..0ad87e2 --- /dev/null +++ b/tests/agents/test_delegation.py @@ -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 diff --git a/tests/core/test_household_registry.py b/tests/core/test_household_registry.py index 00303c5..4783fc8 100644 --- a/tests/core/test_household_registry.py +++ b/tests/core/test_household_registry.py @@ -294,6 +294,101 @@ class TestHouseholdRegistry: assert research_caps[0].name == "research_tools" +class TestGetDelegationTools: + """Test get_delegation_tools() method for agent-as-tool pattern.""" + + def test_delegation_tools_returns_wrapper_for_member_with_agent(self, registry, sample_tools): + """Test delegation tools returns wrapper when member has an agent.""" + from unittest.mock import Mock + + cap = HouseholdCapability( + name="librarian", + role="The Librarian", + category="research", + description="Research and wiki management", + domains=["research", "wiki"], + cost="medium", + requires_network=True, + ) + + mock_agent = Mock() + registry.register("librarian", cap, sample_tools, agent=mock_agent) + + tools = registry.get_delegation_tools(["librarian"]) + + # Should return delegation wrapper, not raw tools + assert len(tools) == 1 + # The wrapper should be the delegate_to_librarian function + assert callable(tools[0]) + assert tools[0].__name__ == "delegate_to_librarian" + + def test_delegation_tools_returns_raw_tools_for_member_without_agent(self, registry, sample_capability, sample_tools): + """Test delegation tools returns raw tools when member has no agent.""" + registry.register("test_tools", sample_capability, sample_tools) + + tools = registry.get_delegation_tools(["test_tools"]) + + # Should return raw tools since no agent + assert len(tools) == 2 + assert tools[0].name == "test_tool_1" + assert tools[1].name == "test_tool_2" + + def test_delegation_tools_mixed_members(self, registry, sample_tools): + """Test delegation tools handles mix of agent and non-agent members.""" + from unittest.mock import Mock + + # Member with agent (librarian) + librarian_cap = HouseholdCapability( + name="librarian", + role="The Librarian", + category="research", + description="Research and wiki", + domains=["research"], + cost="medium", + requires_network=True, + ) + mock_agent = Mock() + registry.register("librarian", librarian_cap, sample_tools, agent=mock_agent) + + # Member without agent (tatlock_core) + core_cap = HouseholdCapability( + name="tatlock_core", + role="Butler's Core Tools", + category="core", + description="Basic tools", + domains=["computation"], + cost="low", + requires_network=False, + ) + registry.register("tatlock_core", core_cap, sample_tools) + + # Request both + tools = registry.get_delegation_tools(["librarian", "tatlock_core"]) + + # Should get 1 delegation wrapper + 2 raw tools = 3 total + assert len(tools) == 3 + + # First should be delegation wrapper + assert callable(tools[0]) + assert tools[0].__name__ == "delegate_to_librarian" + + # Rest should be raw tools + assert hasattr(tools[1], 'name') + assert hasattr(tools[2], 'name') + + def test_delegation_tools_nonexistent_member(self, registry): + """Test delegation tools handles non-existent member gracefully.""" + tools = registry.get_delegation_tools(["nonexistent"]) + + assert tools == [] + + def test_delegation_tools_empty_list(self, registry): + """Test delegation tools handles empty list.""" + tools = registry.get_delegation_tools([]) + + assert tools == [] + + class TestGlobalRegistry: """Test the global registry instance."""