jpmschweitzerandClaude Opus 4.5
7a1d94ca78
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>
409 lines
14 KiB
Python
409 lines
14 KiB
Python
"""
|
|
Tests for household registry.
|
|
|
|
Tests capability registration, toolset scoping, and coordination features.
|
|
"""
|
|
import pytest
|
|
from pydantic_ai.tools import Tool
|
|
|
|
from src.core.household_registry import (
|
|
HouseholdCapability,
|
|
HouseholdMember,
|
|
HouseholdRegistry,
|
|
household_registry,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def registry():
|
|
"""Create a fresh registry for each test."""
|
|
reg = HouseholdRegistry()
|
|
return reg
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_capability():
|
|
"""Sample household capability."""
|
|
return HouseholdCapability(
|
|
name="test_tools",
|
|
role="Test Tools",
|
|
category="testing",
|
|
description="Tools for testing purposes",
|
|
domains=["testing", "validation"],
|
|
cost="low",
|
|
requires_network=False,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_tools():
|
|
"""Sample tool definitions."""
|
|
def test_function_1(x: int) -> int:
|
|
"""Test function 1."""
|
|
return x * 2
|
|
|
|
def test_function_2(x: str) -> str:
|
|
"""Test function 2."""
|
|
return x.upper()
|
|
|
|
return [
|
|
Tool(function=test_function_1, name="test_tool_1"),
|
|
Tool(function=test_function_2, name="test_tool_2"),
|
|
]
|
|
|
|
|
|
class TestHouseholdCapability:
|
|
"""Test HouseholdCapability model."""
|
|
|
|
def test_capability_creation(self, sample_capability):
|
|
"""Test creating a capability."""
|
|
assert sample_capability.name == "test_tools"
|
|
assert sample_capability.role == "Test Tools"
|
|
assert sample_capability.category == "testing"
|
|
assert "testing" in sample_capability.domains
|
|
assert sample_capability.cost == "low"
|
|
assert sample_capability.requires_network is False
|
|
|
|
def test_capability_validation(self):
|
|
"""Test capability field validation."""
|
|
# Should succeed with valid data
|
|
cap = HouseholdCapability(
|
|
name="valid",
|
|
role="Valid Role",
|
|
category="test",
|
|
description="Test description",
|
|
domains=["test"],
|
|
cost="medium",
|
|
requires_network=True,
|
|
)
|
|
assert cap.name == "valid"
|
|
|
|
|
|
class TestHouseholdMember:
|
|
"""Test HouseholdMember model."""
|
|
|
|
def test_member_creation(self, sample_capability, sample_tools):
|
|
"""Test creating a household member."""
|
|
member = HouseholdMember(
|
|
capability=sample_capability,
|
|
tools=sample_tools,
|
|
agent=None,
|
|
)
|
|
assert member.capability.name == "test_tools"
|
|
assert len(member.tools) == 2
|
|
assert member.agent is None
|
|
|
|
def test_member_with_agent(self, sample_capability, sample_tools):
|
|
"""Test member can include an agent."""
|
|
from unittest.mock import Mock
|
|
mock_agent = Mock()
|
|
|
|
member = HouseholdMember(
|
|
capability=sample_capability,
|
|
tools=sample_tools,
|
|
agent=mock_agent,
|
|
)
|
|
assert member.agent is mock_agent
|
|
|
|
|
|
class TestHouseholdRegistry:
|
|
"""Test HouseholdRegistry functionality."""
|
|
|
|
def test_registry_initialization(self, registry):
|
|
"""Test registry initializes empty."""
|
|
assert len(registry) == 0
|
|
assert registry.list_members() == []
|
|
|
|
def test_register_member(self, registry, sample_capability, sample_tools):
|
|
"""Test registering a household member."""
|
|
registry.register(
|
|
name="test_tools",
|
|
capability=sample_capability,
|
|
tools=sample_tools,
|
|
)
|
|
|
|
assert len(registry) == 1
|
|
assert "test_tools" in registry
|
|
assert "test_tools" in registry.list_members()
|
|
|
|
def test_register_name_mismatch(self, registry, sample_capability, sample_tools):
|
|
"""Test registration fails with name mismatch."""
|
|
with pytest.raises(ValueError, match="Name mismatch"):
|
|
registry.register(
|
|
name="wrong_name",
|
|
capability=sample_capability,
|
|
tools=sample_tools,
|
|
)
|
|
|
|
def test_unregister_member(self, registry, sample_capability, sample_tools):
|
|
"""Test unregistering a member."""
|
|
registry.register("test_tools", sample_capability, sample_tools)
|
|
assert "test_tools" in registry
|
|
|
|
registry.unregister("test_tools")
|
|
assert "test_tools" not in registry
|
|
assert len(registry) == 0
|
|
|
|
def test_get_member(self, registry, sample_capability, sample_tools):
|
|
"""Test retrieving a member."""
|
|
registry.register("test_tools", sample_capability, sample_tools)
|
|
|
|
member = registry.get_member("test_tools")
|
|
assert member is not None
|
|
assert member.capability.name == "test_tools"
|
|
assert len(member.tools) == 2
|
|
|
|
def test_get_nonexistent_member(self, registry):
|
|
"""Test retrieving non-existent member returns None."""
|
|
member = registry.get_member("nonexistent")
|
|
assert member is None
|
|
|
|
def test_get_all_capabilities(self, registry, sample_capability, sample_tools):
|
|
"""Test retrieving all capability summaries."""
|
|
# Register multiple members
|
|
cap1 = sample_capability
|
|
cap2 = HouseholdCapability(
|
|
name="other_tools",
|
|
role="Other Tools",
|
|
category="utility",
|
|
description="Other test tools",
|
|
domains=["utility"],
|
|
cost="medium",
|
|
requires_network=True,
|
|
)
|
|
|
|
registry.register("test_tools", cap1, sample_tools)
|
|
registry.register("other_tools", cap2, sample_tools[:1])
|
|
|
|
capabilities = registry.get_all_capabilities()
|
|
assert len(capabilities) == 2
|
|
assert any(cap.name == "test_tools" for cap in capabilities)
|
|
assert any(cap.name == "other_tools" for cap in capabilities)
|
|
|
|
def test_get_scoped_tools(self, registry, sample_capability, sample_tools):
|
|
"""Test creating scoped toolsets."""
|
|
registry.register("test_tools", sample_capability, sample_tools)
|
|
|
|
# Get scoped tools
|
|
tools = registry.get_scoped_tools(["test_tools"])
|
|
assert len(tools) == 2
|
|
assert tools[0].name == "test_tool_1"
|
|
assert tools[1].name == "test_tool_2"
|
|
|
|
def test_get_scoped_tools_multiple_members(self, registry, sample_tools):
|
|
"""Test scoping with multiple members."""
|
|
cap1 = HouseholdCapability(
|
|
name="member1",
|
|
role="Member 1",
|
|
category="test",
|
|
description="First member",
|
|
domains=["test"],
|
|
cost="low",
|
|
requires_network=False,
|
|
)
|
|
cap2 = HouseholdCapability(
|
|
name="member2",
|
|
role="Member 2",
|
|
category="test",
|
|
description="Second member",
|
|
domains=["test"],
|
|
cost="low",
|
|
requires_network=False,
|
|
)
|
|
|
|
registry.register("member1", cap1, sample_tools[:1])
|
|
registry.register("member2", cap2, sample_tools[1:])
|
|
|
|
# Get combined tools
|
|
tools = registry.get_scoped_tools(["member1", "member2"])
|
|
assert len(tools) == 2
|
|
|
|
def test_get_scoped_tools_nonexistent_member(self, registry, sample_capability, sample_tools):
|
|
"""Test scoping with non-existent member logs warning."""
|
|
registry.register("test_tools", sample_capability, sample_tools)
|
|
|
|
# Request includes non-existent member
|
|
tools = registry.get_scoped_tools(["test_tools", "nonexistent"])
|
|
# Should return only existing member's tools
|
|
assert len(tools) == 2
|
|
|
|
def test_get_members_by_domain(self, registry, sample_tools):
|
|
"""Test filtering members by domain."""
|
|
cap1 = HouseholdCapability(
|
|
name="research_tools",
|
|
role="Research Tools",
|
|
category="research",
|
|
description="Research tools",
|
|
domains=["research", "analysis"],
|
|
cost="medium",
|
|
requires_network=True,
|
|
)
|
|
cap2 = HouseholdCapability(
|
|
name="compute_tools",
|
|
role="Compute Tools",
|
|
category="computation",
|
|
description="Computation tools",
|
|
domains=["computation", "math"],
|
|
cost="low",
|
|
requires_network=False,
|
|
)
|
|
|
|
registry.register("research_tools", cap1, sample_tools)
|
|
registry.register("compute_tools", cap2, sample_tools)
|
|
|
|
# Filter by domain
|
|
research_caps = registry.get_members_by_domain("research")
|
|
assert len(research_caps) == 1
|
|
assert research_caps[0].name == "research_tools"
|
|
|
|
compute_caps = registry.get_members_by_domain("computation")
|
|
assert len(compute_caps) == 1
|
|
assert compute_caps[0].name == "compute_tools"
|
|
|
|
def test_get_members_by_category(self, registry, sample_tools):
|
|
"""Test filtering members by category."""
|
|
cap1 = HouseholdCapability(
|
|
name="core_tools",
|
|
role="Core Tools",
|
|
category="core",
|
|
description="Core tools",
|
|
domains=["general"],
|
|
cost="low",
|
|
requires_network=False,
|
|
)
|
|
cap2 = HouseholdCapability(
|
|
name="research_tools",
|
|
role="Research Tools",
|
|
category="research",
|
|
description="Research tools",
|
|
domains=["research"],
|
|
cost="medium",
|
|
requires_network=True,
|
|
)
|
|
|
|
registry.register("core_tools", cap1, sample_tools)
|
|
registry.register("research_tools", cap2, sample_tools)
|
|
|
|
# Filter by category
|
|
core_caps = registry.get_members_by_category("core")
|
|
assert len(core_caps) == 1
|
|
assert core_caps[0].name == "core_tools"
|
|
|
|
research_caps = registry.get_members_by_category("research")
|
|
assert len(research_caps) == 1
|
|
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."""
|
|
|
|
def test_global_registry_exists(self):
|
|
"""Test global registry is available."""
|
|
from src.core.household_registry import get_household_registry
|
|
|
|
registry = get_household_registry()
|
|
assert isinstance(registry, HouseholdRegistry)
|
|
|
|
def test_global_registry_singleton(self):
|
|
"""Test get_household_registry returns same instance."""
|
|
from src.core.household_registry import get_household_registry
|
|
|
|
reg1 = get_household_registry()
|
|
reg2 = get_household_registry()
|
|
assert reg1 is reg2
|