forked from Rockachopa/Timmy-time-dashboard
Alexander Whitestone
d05ce8bf86
Co-authored-by: Alexander Whitestone <alexander@alexanderwhitestone.com> Co-committed-by: Alexander Whitestone <alexander@alexanderwhitestone.com>
497 lines
19 KiB
Python
497 lines
19 KiB
Python
"""Comprehensive unit tests for timmy.tools._registry.
|
|
|
|
Covers:
|
|
- _register_* helpers (web_fetch, search, core, grok, memory, agentic_loop,
|
|
introspection, delegation, gematria, artifact, thinking)
|
|
- create_full_toolkit factory
|
|
- create_experiment_tools factory
|
|
- AGENT_TOOLKITS registry & get_tools_for_agent
|
|
- Backward-compat aliases
|
|
- Tool catalog functions (_core, _analysis, _ai, _introspection, _experiment)
|
|
- _import_creative_catalogs / _merge_catalog
|
|
- get_all_available_tools
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
# All functions under test
|
|
from timmy.tools._registry import (
|
|
AGENT_TOOLKITS,
|
|
PERSONA_TOOLKITS,
|
|
_core_tool_catalog,
|
|
_analysis_tool_catalog,
|
|
_ai_tool_catalog,
|
|
_create_stub_toolkit,
|
|
_experiment_tool_catalog,
|
|
_import_creative_catalogs,
|
|
_introspection_tool_catalog,
|
|
_merge_catalog,
|
|
_register_artifact_tools,
|
|
_register_core_tools,
|
|
_register_delegation_tools,
|
|
_register_gematria_tool,
|
|
_register_grok_tool,
|
|
_register_introspection_tools,
|
|
_register_memory_tools,
|
|
_register_search_tools,
|
|
_register_thinking_tools,
|
|
_register_web_fetch_tool,
|
|
create_experiment_tools,
|
|
create_full_toolkit,
|
|
get_all_available_tools,
|
|
get_tools_for_agent,
|
|
get_tools_for_persona,
|
|
)
|
|
|
|
# import_module is used inside _merge_catalog as a local import
|
|
from importlib import import_module as _real_import_module
|
|
|
|
# _register_agentic_loop_tool may fail to import if conftest stubs interfere
|
|
try:
|
|
from timmy.tools._registry import _register_agentic_loop_tool
|
|
except ImportError:
|
|
_register_agentic_loop_tool = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture()
|
|
def mock_toolkit():
|
|
"""A mock Toolkit with a register method that records calls."""
|
|
tk = MagicMock()
|
|
tk.name = "test"
|
|
tk.registered_tools = {}
|
|
|
|
def _register(func, name=None):
|
|
tk.registered_tools[name or func.__name__] = func
|
|
|
|
tk.register = MagicMock(side_effect=_register)
|
|
return tk
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _register_* helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRegisterWebFetchTool:
|
|
def test_registers_web_fetch(self, mock_toolkit):
|
|
_register_web_fetch_tool(mock_toolkit)
|
|
mock_toolkit.register.assert_called_once()
|
|
assert "web_fetch" in mock_toolkit.registered_tools
|
|
|
|
def test_raises_on_failure(self, mock_toolkit):
|
|
mock_toolkit.register.side_effect = RuntimeError("boom")
|
|
with pytest.raises(RuntimeError, match="boom"):
|
|
_register_web_fetch_tool(mock_toolkit)
|
|
|
|
|
|
class TestRegisterSearchTools:
|
|
def test_registers_both_tools(self, mock_toolkit):
|
|
_register_search_tools(mock_toolkit)
|
|
assert mock_toolkit.register.call_count == 2
|
|
assert "web_search" in mock_toolkit.registered_tools
|
|
assert "scrape_url" in mock_toolkit.registered_tools
|
|
|
|
def test_raises_on_failure(self, mock_toolkit):
|
|
mock_toolkit.register.side_effect = RuntimeError("fail")
|
|
with pytest.raises(RuntimeError):
|
|
_register_search_tools(mock_toolkit)
|
|
|
|
|
|
class TestRegisterCoreTools:
|
|
@patch("timmy.tools._registry.FileTools")
|
|
@patch("timmy.tools._registry.ShellTools")
|
|
@patch("timmy.tools._registry.PythonTools")
|
|
@patch("timmy.tools._registry._make_smart_read_file")
|
|
def test_registers_core_tools(self, mock_smart_read, mock_py, mock_sh, mock_ft, mock_toolkit):
|
|
mock_smart_read.return_value = lambda: "read"
|
|
_register_core_tools(mock_toolkit, Path("/tmp/test"))
|
|
# python, shell, read_file, write_file, list_files, calculator = 6
|
|
assert mock_toolkit.register.call_count == 6
|
|
names = set(mock_toolkit.registered_tools.keys())
|
|
assert {"python", "shell", "read_file", "write_file", "list_files", "calculator"} == names
|
|
|
|
|
|
class TestRegisterGrokTool:
|
|
@patch("timmy.tools._registry.consult_grok")
|
|
def test_registers_when_available(self, mock_grok, mock_toolkit):
|
|
with patch.dict("sys.modules", {"timmy.backends": MagicMock(grok_available=lambda: True)}):
|
|
_register_grok_tool(mock_toolkit)
|
|
assert "consult_grok" in mock_toolkit.registered_tools
|
|
|
|
@patch("timmy.tools._registry.consult_grok")
|
|
def test_skips_when_unavailable(self, mock_grok, mock_toolkit):
|
|
with patch.dict("sys.modules", {"timmy.backends": MagicMock(grok_available=lambda: False)}):
|
|
_register_grok_tool(mock_toolkit)
|
|
assert "consult_grok" not in mock_toolkit.registered_tools
|
|
|
|
def test_raises_on_import_error(self, mock_toolkit):
|
|
with patch.dict("sys.modules", {"timmy.backends": None}):
|
|
with pytest.raises((ImportError, AttributeError)):
|
|
_register_grok_tool(mock_toolkit)
|
|
|
|
|
|
class TestRegisterMemoryTools:
|
|
def test_registers_four_tools(self, mock_toolkit):
|
|
mock_mod = MagicMock()
|
|
with patch.dict("sys.modules", {"timmy.memory_system": mock_mod}):
|
|
_register_memory_tools(mock_toolkit)
|
|
assert mock_toolkit.register.call_count == 4
|
|
names = set(mock_toolkit.registered_tools.keys())
|
|
assert {"memory_search", "memory_write", "memory_read", "memory_forget"} == names
|
|
|
|
|
|
@pytest.mark.skipif(_register_agentic_loop_tool is None, reason="agentic_loop not importable")
|
|
class TestRegisterAgenticLoopTool:
|
|
def test_registers_plan_and_execute(self, mock_toolkit):
|
|
mock_mod = MagicMock()
|
|
with patch.dict("sys.modules", {"timmy.agentic_loop": mock_mod}):
|
|
_register_agentic_loop_tool(mock_toolkit)
|
|
assert "plan_and_execute" in mock_toolkit.registered_tools
|
|
|
|
def test_raises_on_import_error(self, mock_toolkit):
|
|
with patch.dict("sys.modules", {"timmy.agentic_loop": None}):
|
|
with pytest.raises((ImportError, AttributeError)):
|
|
_register_agentic_loop_tool(mock_toolkit)
|
|
|
|
|
|
class TestRegisterIntrospectionTools:
|
|
def test_registers_all_introspection(self, mock_toolkit):
|
|
mock_intro = MagicMock()
|
|
mock_mcp = MagicMock()
|
|
mock_session = MagicMock()
|
|
with patch.dict(
|
|
"sys.modules",
|
|
{
|
|
"timmy.tools_intro": mock_intro,
|
|
"timmy.mcp_tools": mock_mcp,
|
|
"timmy.session_logger": mock_session,
|
|
},
|
|
):
|
|
_register_introspection_tools(mock_toolkit)
|
|
# 4 intro + 1 avatar + 2 session = 7
|
|
assert mock_toolkit.register.call_count == 7
|
|
names = set(mock_toolkit.registered_tools.keys())
|
|
assert "get_system_info" in names
|
|
assert "check_ollama_health" in names
|
|
assert "update_gitea_avatar" in names
|
|
assert "session_history" in names
|
|
assert "self_reflect" in names
|
|
|
|
|
|
class TestRegisterDelegationTools:
|
|
def test_registers_three_tools(self, mock_toolkit):
|
|
mock_mod = MagicMock()
|
|
with patch.dict("sys.modules", {"timmy.tools_delegation": mock_mod}):
|
|
_register_delegation_tools(mock_toolkit)
|
|
assert mock_toolkit.register.call_count == 3
|
|
names = set(mock_toolkit.registered_tools.keys())
|
|
assert {"delegate_task", "delegate_to_kimi", "list_swarm_agents"} == names
|
|
|
|
def test_raises_on_failure(self, mock_toolkit):
|
|
with patch.dict("sys.modules", {"timmy.tools_delegation": None}):
|
|
with pytest.raises((ImportError, AttributeError)):
|
|
_register_delegation_tools(mock_toolkit)
|
|
|
|
|
|
class TestRegisterGematriaTool:
|
|
def test_registers_gematria(self, mock_toolkit):
|
|
mock_mod = MagicMock()
|
|
with patch.dict("sys.modules", {"timmy.gematria": mock_mod}):
|
|
_register_gematria_tool(mock_toolkit)
|
|
assert "gematria" in mock_toolkit.registered_tools
|
|
|
|
def test_raises_on_import_error(self, mock_toolkit):
|
|
with patch.dict("sys.modules", {"timmy.gematria": None}):
|
|
with pytest.raises((ImportError, AttributeError)):
|
|
_register_gematria_tool(mock_toolkit)
|
|
|
|
|
|
class TestRegisterArtifactTools:
|
|
def test_registers_jot_and_log(self, mock_toolkit):
|
|
mock_mod = MagicMock()
|
|
with patch.dict("sys.modules", {"timmy.memory_system": mock_mod}):
|
|
_register_artifact_tools(mock_toolkit)
|
|
assert mock_toolkit.register.call_count == 2
|
|
assert "jot_note" in mock_toolkit.registered_tools
|
|
assert "log_decision" in mock_toolkit.registered_tools
|
|
|
|
|
|
class TestRegisterThinkingTools:
|
|
def test_registers_thought_search(self, mock_toolkit):
|
|
mock_mod = MagicMock()
|
|
with patch.dict("sys.modules", {"timmy.thinking": mock_mod}):
|
|
_register_thinking_tools(mock_toolkit)
|
|
assert "thought_search" in mock_toolkit.registered_tools
|
|
|
|
def test_raises_on_import_error(self, mock_toolkit):
|
|
with patch.dict("sys.modules", {"timmy.thinking": None}):
|
|
with pytest.raises((ImportError, AttributeError)):
|
|
_register_thinking_tools(mock_toolkit)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Toolkit factories
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCreateFullToolkit:
|
|
@patch("timmy.tools._registry._AGNO_TOOLS_AVAILABLE", False)
|
|
def test_returns_none_without_agno(self):
|
|
result = create_full_toolkit()
|
|
assert result is None
|
|
|
|
@patch("timmy.tools._registry._register_thinking_tools")
|
|
@patch("timmy.tools._registry._register_artifact_tools")
|
|
@patch("timmy.tools._registry._register_gematria_tool")
|
|
@patch("timmy.tools._registry._register_delegation_tools")
|
|
@patch("timmy.tools._registry._register_introspection_tools")
|
|
@patch("timmy.tools._registry._register_agentic_loop_tool")
|
|
@patch("timmy.tools._registry._register_memory_tools")
|
|
@patch("timmy.tools._registry._register_grok_tool")
|
|
@patch("timmy.tools._registry._register_search_tools")
|
|
@patch("timmy.tools._registry._register_web_fetch_tool")
|
|
@patch("timmy.tools._registry._register_core_tools")
|
|
@patch("timmy.tools._registry._AGNO_TOOLS_AVAILABLE", True)
|
|
def test_calls_all_register_helpers(
|
|
self,
|
|
mock_core,
|
|
mock_web,
|
|
mock_search,
|
|
mock_grok,
|
|
mock_memory,
|
|
mock_agentic,
|
|
mock_intro,
|
|
mock_deleg,
|
|
mock_gematria,
|
|
mock_artifact,
|
|
mock_thinking,
|
|
):
|
|
mock_settings = MagicMock(repo_root="/tmp/test")
|
|
with patch.dict("sys.modules", {"config": MagicMock(settings=mock_settings)}):
|
|
with patch("timmy.tools._registry.Toolkit") as MockTK:
|
|
mock_tk_inst = MagicMock()
|
|
MockTK.return_value = mock_tk_inst
|
|
with patch.dict(
|
|
"sys.modules", {"timmy.tool_safety": MagicMock(DANGEROUS_TOOLS=["shell"])}
|
|
):
|
|
result = create_full_toolkit()
|
|
|
|
assert result is mock_tk_inst
|
|
mock_core.assert_called_once()
|
|
mock_web.assert_called_once()
|
|
mock_search.assert_called_once()
|
|
mock_grok.assert_called_once()
|
|
mock_memory.assert_called_once()
|
|
mock_agentic.assert_called_once()
|
|
mock_intro.assert_called_once()
|
|
mock_deleg.assert_called_once()
|
|
mock_gematria.assert_called_once()
|
|
mock_artifact.assert_called_once()
|
|
mock_thinking.assert_called_once()
|
|
|
|
|
|
class TestCreateExperimentTools:
|
|
@patch("timmy.tools._registry._AGNO_TOOLS_AVAILABLE", False)
|
|
def test_raises_without_agno(self):
|
|
with pytest.raises(ImportError, match="Agno tools not available"):
|
|
create_experiment_tools()
|
|
|
|
@patch("timmy.tools._registry._AGNO_TOOLS_AVAILABLE", True)
|
|
def test_creates_experiment_toolkit(self):
|
|
mock_settings = MagicMock(
|
|
repo_root="/tmp/test",
|
|
autoresearch_workspace="workspace",
|
|
autoresearch_time_budget=300,
|
|
autoresearch_metric="loss",
|
|
)
|
|
mock_autoresearch = MagicMock()
|
|
with (
|
|
patch.dict("sys.modules", {"config": MagicMock(settings=mock_settings)}),
|
|
patch.dict("sys.modules", {"timmy.autoresearch": mock_autoresearch}),
|
|
patch("timmy.tools._registry.Toolkit") as MockTK,
|
|
patch("timmy.tools._registry.ShellTools"),
|
|
patch("timmy.tools._registry.FileTools"),
|
|
patch("timmy.tools._registry._make_smart_read_file", return_value=lambda: None),
|
|
):
|
|
mock_tk = MagicMock()
|
|
MockTK.return_value = mock_tk
|
|
result = create_experiment_tools()
|
|
|
|
assert result is mock_tk
|
|
# prepare_experiment, run_experiment, evaluate_result, shell, read_file, write_file, list_files = 7
|
|
assert mock_tk.register.call_count == 7
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Agent toolkit registry
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAgentToolkitRegistry:
|
|
def test_agent_toolkits_has_expected_agents(self):
|
|
expected = {"echo", "mace", "helm", "seer", "forge", "quill", "lab", "pixel", "lyra", "reel"}
|
|
assert set(AGENT_TOOLKITS.keys()) == expected
|
|
|
|
def test_persona_toolkits_is_alias(self):
|
|
assert PERSONA_TOOLKITS is AGENT_TOOLKITS
|
|
|
|
def test_get_tools_for_persona_is_alias(self):
|
|
assert get_tools_for_persona is get_tools_for_agent
|
|
|
|
|
|
class TestGetToolsForAgent:
|
|
def test_unknown_agent_returns_none(self):
|
|
result = get_tools_for_agent("nonexistent_agent_xyz")
|
|
assert result is None
|
|
|
|
def test_stub_agents_return_toolkit(self):
|
|
"""Pixel, lyra, reel use stub toolkits."""
|
|
for agent_id in ("pixel", "lyra", "reel"):
|
|
result = get_tools_for_agent(agent_id)
|
|
# May be None if agno not available, or a Toolkit stub
|
|
# Just verify no exception is raised
|
|
assert result is None or hasattr(result, "name")
|
|
|
|
|
|
class TestCreateStubToolkit:
|
|
@patch("timmy.tools._registry._AGNO_TOOLS_AVAILABLE", False)
|
|
def test_returns_none_without_agno(self):
|
|
assert _create_stub_toolkit("test") is None
|
|
|
|
@patch("timmy.tools._registry._AGNO_TOOLS_AVAILABLE", True)
|
|
def test_creates_named_toolkit(self):
|
|
with patch("timmy.tools._registry.Toolkit") as MockTK:
|
|
mock_tk = MagicMock()
|
|
MockTK.return_value = mock_tk
|
|
result = _create_stub_toolkit("pixel")
|
|
MockTK.assert_called_once_with(name="pixel")
|
|
assert result is mock_tk
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool catalog functions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestToolCatalogs:
|
|
def test_core_catalog_has_expected_tools(self):
|
|
cat = _core_tool_catalog()
|
|
assert isinstance(cat, dict)
|
|
assert {"shell", "python", "read_file", "write_file", "list_files"} == set(cat.keys())
|
|
for tool_id, info in cat.items():
|
|
assert "name" in info
|
|
assert "description" in info
|
|
assert "available_in" in info
|
|
assert isinstance(info["available_in"], list)
|
|
|
|
def test_analysis_catalog(self):
|
|
cat = _analysis_tool_catalog()
|
|
assert {"calculator", "web_fetch", "web_search", "scrape_url"} == set(cat.keys())
|
|
|
|
def test_ai_catalog(self):
|
|
cat = _ai_tool_catalog()
|
|
assert "consult_grok" in cat
|
|
assert "aider" in cat
|
|
|
|
def test_introspection_catalog(self):
|
|
cat = _introspection_tool_catalog()
|
|
expected = {
|
|
"get_system_info",
|
|
"check_ollama_health",
|
|
"get_memory_status",
|
|
"session_history",
|
|
"thought_search",
|
|
"self_reflect",
|
|
"update_gitea_avatar",
|
|
}
|
|
assert expected == set(cat.keys())
|
|
|
|
def test_experiment_catalog(self):
|
|
cat = _experiment_tool_catalog()
|
|
assert {"prepare_experiment", "run_experiment", "evaluate_result"} == set(cat.keys())
|
|
|
|
def test_all_catalogs_have_consistent_schema(self):
|
|
"""Every catalog entry must have name, description, available_in."""
|
|
for fn in (
|
|
_core_tool_catalog,
|
|
_analysis_tool_catalog,
|
|
_ai_tool_catalog,
|
|
_introspection_tool_catalog,
|
|
_experiment_tool_catalog,
|
|
):
|
|
cat = fn()
|
|
for tool_id, info in cat.items():
|
|
assert isinstance(info.get("name"), str), f"{tool_id} missing 'name'"
|
|
assert isinstance(info.get("description"), str), f"{tool_id} missing 'description'"
|
|
assert isinstance(info.get("available_in"), list), f"{tool_id} missing 'available_in'"
|
|
|
|
|
|
class TestMergeCatalog:
|
|
def test_merges_catalog_entries(self):
|
|
catalog = {}
|
|
mock_mod = MagicMock()
|
|
mock_mod.TEST_CATALOG = {
|
|
"tool_a": {"name": "Tool A", "description": "Does A"},
|
|
"tool_b": {"name": "Tool B", "description": "Does B"},
|
|
}
|
|
with patch("importlib.import_module", return_value=mock_mod):
|
|
_merge_catalog(catalog, "fake.module", "TEST_CATALOG", ["pixel", "orchestrator"])
|
|
assert "tool_a" in catalog
|
|
assert catalog["tool_a"]["available_in"] == ["pixel", "orchestrator"]
|
|
assert catalog["tool_b"]["name"] == "Tool B"
|
|
|
|
def test_handles_import_error_gracefully(self):
|
|
catalog = {}
|
|
with patch("importlib.import_module", side_effect=ImportError("nope")):
|
|
# Should NOT raise — just logs and skips
|
|
_merge_catalog(catalog, "missing.module", "CATALOG", [])
|
|
assert catalog == {}
|
|
|
|
|
|
class TestImportCreativeCatalogs:
|
|
def test_calls_merge_for_each_source(self):
|
|
catalog = {}
|
|
with patch("timmy.tools._registry._merge_catalog") as mock_merge:
|
|
_import_creative_catalogs(catalog)
|
|
# Should be called once per _CREATIVE_CATALOG_SOURCES entry (6 sources)
|
|
assert mock_merge.call_count == 6
|
|
|
|
|
|
class TestGetAllAvailableTools:
|
|
def test_returns_merged_catalog(self):
|
|
catalog = get_all_available_tools()
|
|
assert isinstance(catalog, dict)
|
|
# Must contain core tools at minimum
|
|
assert "shell" in catalog
|
|
assert "calculator" in catalog
|
|
assert "web_search" in catalog
|
|
assert "consult_grok" in catalog
|
|
assert "get_system_info" in catalog
|
|
assert "prepare_experiment" in catalog
|
|
|
|
def test_no_duplicate_keys(self):
|
|
"""Each sub-catalog shouldn't override another's keys."""
|
|
catalog = get_all_available_tools()
|
|
# Count total keys from individual catalogs
|
|
individual = {}
|
|
for fn in (
|
|
_core_tool_catalog,
|
|
_analysis_tool_catalog,
|
|
_ai_tool_catalog,
|
|
_introspection_tool_catalog,
|
|
_experiment_tool_catalog,
|
|
):
|
|
for k in fn():
|
|
assert k not in individual, f"Duplicate key '{k}' across catalogs"
|
|
individual[k] = True
|