1
0

test: add unit tests for agents/loader.py (#349)

Co-authored-by: Kimi Agent <kimi@timmy.local>
Co-committed-by: Kimi Agent <kimi@timmy.local>
This commit is contained in:
2026-03-18 21:13:01 -04:00
committed by hermes
parent 4aa86ff1cb
commit 4afc5daffb
3 changed files with 811 additions and 0 deletions

View File

View File

@@ -0,0 +1,393 @@
"""Unit tests for timmy.agents.loader — YAML-driven agent factory."""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import pytest
import timmy.agents.loader as loader
# ── Fixtures ──────────────────────────────────────────────────────────────────
MINIMAL_YAML = """
defaults:
model: test-model
prompt_tier: lite
max_history: 5
tools: []
routing:
method: pattern
patterns:
coder:
- code
- fix bug
writer:
- write
- draft
agents:
helper:
name: Helper
role: general
prompt: "You are a helpful agent."
coder:
name: Forge
role: code
model: big-model
prompt_tier: full
max_history: 15
tools:
- python
- shell
prompt: "You are a coding agent."
"""
@pytest.fixture(autouse=True)
def _reset_loader_cache():
"""Reset module-level caches before each test."""
loader._agents = None
loader._config = None
yield
loader._agents = None
loader._config = None
@pytest.fixture()
def mock_yaml_config(tmp_path):
"""Write a minimal agents.yaml and patch settings.repo_root to point at it."""
config_dir = tmp_path / "config"
config_dir.mkdir()
(config_dir / "agents.yaml").write_text(MINIMAL_YAML)
with patch.object(loader.settings, "repo_root", str(tmp_path)):
yield tmp_path
# ── _find_config_path ─────────────────────────────────────────────────────────
def test_find_config_path_returns_path(mock_yaml_config):
path = loader._find_config_path()
assert path.exists()
assert path.name == "agents.yaml"
def test_find_config_path_raises_when_missing(tmp_path):
with patch.object(loader.settings, "repo_root", str(tmp_path)):
with pytest.raises(FileNotFoundError, match="Agent config not found"):
loader._find_config_path()
# ── _load_config ──────────────────────────────────────────────────────────────
def test_load_config_parses_yaml(mock_yaml_config):
config = loader._load_config()
assert "defaults" in config
assert "agents" in config
assert "routing" in config
def test_load_config_caches(mock_yaml_config):
cfg1 = loader._load_config()
cfg2 = loader._load_config()
assert cfg1 is cfg2
def test_load_config_force_reload(mock_yaml_config):
cfg1 = loader._load_config()
cfg2 = loader._load_config(force_reload=True)
assert cfg1 is not cfg2
assert cfg1 == cfg2
# ── _resolve_model ────────────────────────────────────────────────────────────
def test_resolve_model_agent_specific():
assert loader._resolve_model("custom-model", {"model": "default-model"}) == "custom-model"
def test_resolve_model_defaults_fallback():
assert loader._resolve_model(None, {"model": "default-model"}) == "default-model"
def test_resolve_model_settings_fallback():
with patch.object(loader.settings, "ollama_model", "settings-model"):
assert loader._resolve_model(None, {}) == "settings-model"
# ── _resolve_prompt_tier ──────────────────────────────────────────────────────
def test_resolve_prompt_tier_agent_specific():
assert loader._resolve_prompt_tier("full", {"prompt_tier": "lite"}) == "full"
def test_resolve_prompt_tier_defaults_fallback():
assert loader._resolve_prompt_tier(None, {"prompt_tier": "full"}) == "full"
def test_resolve_prompt_tier_default_is_lite():
assert loader._resolve_prompt_tier(None, {}) == "lite"
# ── _build_system_prompt ──────────────────────────────────────────────────────
def test_build_system_prompt_full_tier():
with patch("timmy.prompts.get_system_prompt", return_value="BASE") as mock_gsp:
result = loader._build_system_prompt({"prompt": "Custom."}, "full")
mock_gsp.assert_called_once_with(tools_enabled=True)
assert result == "Custom.\n\nBASE"
def test_build_system_prompt_lite_tier():
with patch("timmy.prompts.get_system_prompt", return_value="BASE") as mock_gsp:
result = loader._build_system_prompt({"prompt": "Custom."}, "lite")
mock_gsp.assert_called_once_with(tools_enabled=False)
assert result == "Custom.\n\nBASE"
def test_build_system_prompt_no_custom():
with patch("timmy.prompts.get_system_prompt", return_value="BASE"):
result = loader._build_system_prompt({}, "lite")
assert result == "BASE"
def test_build_system_prompt_empty_custom():
with patch("timmy.prompts.get_system_prompt", return_value="BASE"):
result = loader._build_system_prompt({"prompt": " "}, "lite")
assert result == "BASE"
# ── load_agents ───────────────────────────────────────────────────────────────
def test_load_agents_creates_subagents(mock_yaml_config):
with (
patch("timmy.agents.base.SubAgent") as MockSubAgent,
patch("timmy.prompts.get_system_prompt", return_value="BASE"),
):
MockSubAgent.side_effect = lambda **kw: MagicMock(**kw)
agents = loader.load_agents()
assert len(agents) == 2
assert "helper" in agents
assert "coder" in agents
def test_load_agents_passes_correct_params(mock_yaml_config):
with (
patch("timmy.agents.base.SubAgent") as MockSubAgent,
patch("timmy.prompts.get_system_prompt", return_value="BASE"),
):
MockSubAgent.side_effect = lambda **kw: MagicMock(**kw)
loader.load_agents()
calls = {c.kwargs["agent_id"]: c.kwargs for c in MockSubAgent.call_args_list}
coder_kw = calls["coder"]
assert coder_kw["name"] == "Forge"
assert coder_kw["role"] == "code"
assert coder_kw["model"] == "big-model"
assert coder_kw["max_history"] == 15
assert coder_kw["tools"] == ["python", "shell"]
def test_load_agents_uses_defaults(mock_yaml_config):
with (
patch("timmy.agents.base.SubAgent") as MockSubAgent,
patch("timmy.prompts.get_system_prompt", return_value="BASE"),
):
MockSubAgent.side_effect = lambda **kw: MagicMock(**kw)
loader.load_agents()
calls = {c.kwargs["agent_id"]: c.kwargs for c in MockSubAgent.call_args_list}
helper_kw = calls["helper"]
assert helper_kw["model"] == "test-model"
assert helper_kw["max_history"] == 5
assert helper_kw["tools"] == []
def test_load_agents_caches(mock_yaml_config):
with (
patch("timmy.agents.base.SubAgent") as MockSubAgent,
patch("timmy.prompts.get_system_prompt", return_value="BASE"),
):
MockSubAgent.side_effect = lambda **kw: MagicMock(**kw)
a1 = loader.load_agents()
a2 = loader.load_agents()
assert a1 is a2
def test_load_agents_force_reload(mock_yaml_config):
with (
patch("timmy.agents.base.SubAgent") as MockSubAgent,
patch("timmy.prompts.get_system_prompt", return_value="BASE"),
):
MockSubAgent.side_effect = lambda **kw: MagicMock(**kw)
a1 = loader.load_agents()
a2 = loader.load_agents(force_reload=True)
assert a1 is not a2
# ── get_agent ─────────────────────────────────────────────────────────────────
def test_get_agent_returns_agent(mock_yaml_config):
with (
patch("timmy.agents.base.SubAgent") as MockSubAgent,
patch("timmy.prompts.get_system_prompt", return_value="BASE"),
):
MockSubAgent.side_effect = lambda **kw: MagicMock(agent_id=kw["agent_id"])
agent = loader.get_agent("helper")
assert agent.agent_id == "helper"
def test_get_agent_raises_for_unknown(mock_yaml_config):
with (
patch("timmy.agents.base.SubAgent") as MockSubAgent,
patch("timmy.prompts.get_system_prompt", return_value="BASE"),
):
MockSubAgent.side_effect = lambda **kw: MagicMock(**kw)
with pytest.raises(KeyError, match="Unknown agent.*nope"):
loader.get_agent("nope")
# ── list_agents ───────────────────────────────────────────────────────────────
def test_list_agents_returns_metadata(mock_yaml_config):
result = loader.list_agents()
assert len(result) == 2
ids = {a["id"] for a in result}
assert ids == {"helper", "coder"}
def test_list_agents_includes_model_and_tools(mock_yaml_config):
result = loader.list_agents()
coder = next(a for a in result if a["id"] == "coder")
assert coder["model"] == "big-model"
assert coder["tools"] == ["python", "shell"]
assert coder["status"] == "available"
def test_list_agents_uses_defaults_for_name_and_role(mock_yaml_config):
result = loader.list_agents()
helper = next(a for a in result if a["id"] == "helper")
assert helper["name"] == "Helper"
assert helper["role"] == "general"
# ── get_routing_config ────────────────────────────────────────────────────────
def test_get_routing_config(mock_yaml_config):
routing = loader.get_routing_config()
assert routing["method"] == "pattern"
assert "coder" in routing["patterns"]
def test_get_routing_config_default_when_missing(tmp_path):
"""When no routing section exists, returns a sensible default."""
config_dir = tmp_path / "config"
config_dir.mkdir()
(config_dir / "agents.yaml").write_text("defaults: {}\nagents: {}\n")
with patch.object(loader.settings, "repo_root", str(tmp_path)):
routing = loader.get_routing_config()
assert routing == {"method": "pattern", "patterns": {}}
# ── _matches_pattern ──────────────────────────────────────────────────────────
class TestMatchesPattern:
def test_single_word_match(self):
assert loader._matches_pattern("code", "please code this")
def test_single_word_no_partial(self):
assert not loader._matches_pattern("code", "barcode scanner")
def test_multi_word_all_present(self):
assert loader._matches_pattern("fix bug", "can you fix this bug?")
def test_multi_word_any_order(self):
assert loader._matches_pattern("fix bug", "there is a bug, please fix it")
def test_multi_word_missing_one(self):
assert not loader._matches_pattern("fix bug", "fix the typo")
def test_case_insensitive(self):
assert loader._matches_pattern("Code", "CODE this")
def test_word_boundary(self):
assert not loader._matches_pattern("test", "testing in progress")
# ── route_request ─────────────────────────────────────────────────────────────
def test_route_request_matches_coder(mock_yaml_config):
assert loader.route_request("please code this feature") == "coder"
def test_route_request_matches_writer(mock_yaml_config):
assert loader.route_request("write a summary") == "writer"
def test_route_request_returns_none_when_no_match(mock_yaml_config):
assert loader.route_request("hello there") is None
def test_route_request_non_pattern_method(mock_yaml_config):
"""When routing method is not 'pattern', always returns None."""
loader._load_config()
loader._config["routing"]["method"] = "llm"
assert loader.route_request("code this") is None
# ── route_request_with_match ──────────────────────────────────────────────────
def test_route_request_with_match_returns_tuple(mock_yaml_config):
agent_id, pattern = loader.route_request_with_match("fix this bug please")
assert agent_id == "coder"
assert pattern == "fix bug"
def test_route_request_with_match_no_match(mock_yaml_config):
agent_id, pattern = loader.route_request_with_match("hello")
assert agent_id is None
assert pattern is None
def test_route_request_with_match_non_pattern_method(mock_yaml_config):
loader._load_config()
loader._config["routing"]["method"] = "llm"
agent_id, pattern = loader.route_request_with_match("code this")
assert agent_id is None
assert pattern is None
# ── reload_agents ─────────────────────────────────────────────────────────────
def test_reload_agents_clears_caches(mock_yaml_config):
with (
patch("timmy.agents.base.SubAgent") as MockSubAgent,
patch("timmy.prompts.get_system_prompt", return_value="BASE"),
):
MockSubAgent.side_effect = lambda **kw: MagicMock(**kw)
loader.load_agents()
assert loader._agents is not None
assert loader._config is not None
loader.reload_agents()
assert loader._agents is not None
assert MockSubAgent.call_count == 4 # 2 agents * 2 loads

View File

@@ -0,0 +1,418 @@
"""Unit tests for timmy.agents.loader — YAML-driven agent factory."""
from __future__ import annotations
from pathlib import Path
from unittest.mock import mock_open, patch
import pytest
from timmy.agents import loader
# ── Fixtures ──────────────────────────────────────────────────────────────
MINIMAL_YAML = """\
defaults:
model: test-model
prompt_tier: lite
max_history: 5
tools: []
agents:
helper:
name: Helper
role: general
prompt: "You are a helpful agent."
routing:
method: pattern
patterns:
helper:
- help
- assist
"""
TWO_AGENT_YAML = """\
defaults:
model: default-model
prompt_tier: lite
max_history: 8
tools:
- default_tool
agents:
alpha:
name: Alpha
role: research
model: alpha-model
prompt_tier: full
max_history: 15
tools:
- search
prompt: "You are Alpha."
beta:
name: Beta
role: code
prompt: "You are Beta."
routing:
method: pattern
patterns:
alpha:
- research
- find out
beta:
- code
- fix bug
"""
NO_ROUTING_YAML = """\
defaults:
model: m1
agents:
solo:
name: Solo
role: general
prompt: "Solo agent."
"""
@pytest.fixture(autouse=True)
def _clear_loader_cache():
"""Reset module-level caches between tests."""
loader._agents = None
loader._config = None
yield
loader._agents = None
loader._config = None
def _mock_load_config(yaml_text: str):
"""Return a patcher that makes _load_config return parsed yaml_text."""
import yaml
parsed = yaml.safe_load(yaml_text)
return patch.object(loader, "_load_config", return_value=parsed)
def _mock_full_load(yaml_text: str):
"""Patch _find_config_path + open so _load_config reads yaml_text."""
fake_path = Path("/fake/config/agents.yaml")
return [
patch.object(loader, "_find_config_path", return_value=fake_path),
patch("builtins.open", mock_open(read_data=yaml_text)),
]
# ── _find_config_path ─────────────────────────────────────────────────────
class TestFindConfigPath:
def test_returns_path_when_exists(self, tmp_path):
cfg = tmp_path / "config" / "agents.yaml"
cfg.parent.mkdir()
cfg.write_text("agents: {}")
with patch.object(loader.settings, "repo_root", str(tmp_path)):
result = loader._find_config_path()
assert result == cfg
def test_raises_when_missing(self, tmp_path):
with patch.object(loader.settings, "repo_root", str(tmp_path)):
with pytest.raises(FileNotFoundError, match="Agent config not found"):
loader._find_config_path()
# ── _load_config ──────────────────────────────────────────────────────────
class TestLoadConfig:
def test_loads_and_caches(self):
patches = _mock_full_load(MINIMAL_YAML)
for p in patches:
p.start()
try:
cfg = loader._load_config()
assert "agents" in cfg
assert "helper" in cfg["agents"]
# Second call uses cache (no file read)
cfg2 = loader._load_config()
assert cfg2 is cfg
finally:
for p in patches:
p.stop()
def test_force_reload(self):
patches = _mock_full_load(MINIMAL_YAML)
for p in patches:
p.start()
try:
cfg1 = loader._load_config()
cfg2 = loader._load_config(force_reload=True)
# force_reload creates a new dict
assert cfg2 is not cfg1
finally:
for p in patches:
p.stop()
# ── _resolve_model ────────────────────────────────────────────────────────
class TestResolveModel:
def test_agent_model_wins(self):
assert loader._resolve_model("custom", {"model": "default"}) == "custom"
def test_falls_back_to_defaults(self):
assert loader._resolve_model(None, {"model": "default"}) == "default"
def test_falls_back_to_settings(self):
with patch.object(loader.settings, "ollama_model", "settings-model"):
assert loader._resolve_model(None, {}) == "settings-model"
# ── _resolve_prompt_tier ──────────────────────────────────────────────────
class TestResolvePromptTier:
def test_agent_tier_wins(self):
assert loader._resolve_prompt_tier("full", {"prompt_tier": "lite"}) == "full"
def test_falls_back_to_defaults(self):
assert loader._resolve_prompt_tier(None, {"prompt_tier": "full"}) == "full"
def test_falls_back_to_lite(self):
assert loader._resolve_prompt_tier(None, {}) == "lite"
# ── _build_system_prompt ──────────────────────────────────────────────────
class TestBuildSystemPrompt:
@patch("timmy.prompts.get_system_prompt", return_value="BASE")
def test_prepends_custom_prompt(self, mock_gsp):
result = loader._build_system_prompt({"prompt": "Custom."}, "full")
assert result == "Custom.\n\nBASE"
mock_gsp.assert_called_once_with(tools_enabled=True)
@patch("timmy.prompts.get_system_prompt", return_value="BASE")
def test_no_custom_prompt(self, mock_gsp):
result = loader._build_system_prompt({}, "lite")
assert result == "BASE"
mock_gsp.assert_called_once_with(tools_enabled=False)
# ── load_agents ───────────────────────────────────────────────────────────
class TestLoadAgents:
@patch("timmy.prompts.get_system_prompt", return_value="BASE")
@patch("timmy.agents.base.Agent")
def test_loads_agents_from_yaml(self, mock_agno, mock_gsp):
with _mock_load_config(TWO_AGENT_YAML):
agents = loader.load_agents(force_reload=True)
assert "alpha" in agents
assert "beta" in agents
assert agents["alpha"].name == "Alpha"
assert agents["alpha"].model == "alpha-model"
assert agents["alpha"].role == "research"
assert agents["alpha"].max_history == 15
assert agents["alpha"].tools == ["search"]
# beta inherits defaults
assert agents["beta"].model == "default-model"
assert agents["beta"].tools == ["default_tool"]
assert agents["beta"].max_history == 8
@patch("timmy.prompts.get_system_prompt", return_value="BASE")
@patch("timmy.agents.base.Agent")
def test_caching(self, mock_agno, mock_gsp):
with _mock_load_config(MINIMAL_YAML):
a1 = loader.load_agents(force_reload=True)
# Second call uses cache — doesn't need the patch
a2 = loader.load_agents()
assert a2 is a1
@patch("timmy.prompts.get_system_prompt", return_value="BASE")
@patch("timmy.agents.base.Agent")
def test_force_reload_rebuilds(self, mock_agno, mock_gsp):
with _mock_load_config(MINIMAL_YAML):
a1 = loader.load_agents(force_reload=True)
with _mock_load_config(MINIMAL_YAML):
a2 = loader.load_agents(force_reload=True)
assert a2 is not a1
@patch("timmy.prompts.get_system_prompt", return_value="BASE")
@patch("timmy.agents.base.Agent")
def test_default_name_is_titlecased_id(self, mock_agno, mock_gsp):
yaml_text = """\
defaults:
model: m
agents:
myagent:
role: general
"""
import yaml
with patch.object(loader, "_load_config", return_value=yaml.safe_load(yaml_text)):
agents = loader.load_agents(force_reload=True)
assert agents["myagent"].name == "Myagent"
# ── get_agent ─────────────────────────────────────────────────────────────
class TestGetAgent:
@patch("timmy.prompts.get_system_prompt", return_value="BASE")
@patch("timmy.agents.base.Agent")
def test_returns_agent(self, mock_agno, mock_gsp):
with _mock_load_config(MINIMAL_YAML):
agent = loader.get_agent("helper")
assert agent.agent_id == "helper"
@patch("timmy.prompts.get_system_prompt", return_value="BASE")
@patch("timmy.agents.base.Agent")
def test_raises_on_unknown(self, mock_agno, mock_gsp):
with _mock_load_config(MINIMAL_YAML):
loader.load_agents(force_reload=True)
with pytest.raises(KeyError, match="Unknown agent.*nope"):
loader.get_agent("nope")
# ── list_agents ───────────────────────────────────────────────────────────
class TestListAgents:
def test_returns_metadata(self):
with _mock_load_config(TWO_AGENT_YAML):
result = loader.list_agents()
ids = [a["id"] for a in result]
assert "alpha" in ids
assert "beta" in ids
alpha = next(a for a in result if a["id"] == "alpha")
assert alpha["name"] == "Alpha"
assert alpha["role"] == "research"
assert alpha["model"] == "alpha-model"
assert alpha["tools"] == ["search"]
assert alpha["status"] == "available"
def test_beta_inherits_defaults(self):
with _mock_load_config(TWO_AGENT_YAML):
result = loader.list_agents()
beta = next(a for a in result if a["id"] == "beta")
assert beta["model"] == "default-model"
assert beta["tools"] == ["default_tool"]
# ── get_routing_config ────────────────────────────────────────────────────
class TestGetRoutingConfig:
def test_returns_routing_section(self):
with _mock_load_config(MINIMAL_YAML):
routing = loader.get_routing_config()
assert routing["method"] == "pattern"
assert "helper" in routing["patterns"]
def test_fallback_when_no_routing(self):
with _mock_load_config(NO_ROUTING_YAML):
routing = loader.get_routing_config()
assert routing["method"] == "pattern"
assert routing["patterns"] == {}
# ── _matches_pattern ──────────────────────────────────────────────────────
class TestMatchesPattern:
def test_single_word_match(self):
assert loader._matches_pattern("search", "please search for it")
def test_single_word_no_match(self):
assert not loader._matches_pattern("search", "researching stuff")
def test_multi_word_match(self):
assert loader._matches_pattern("fix bug", "can you fix the bug")
def test_multi_word_order_independent(self):
assert loader._matches_pattern("fix bug", "there's a bug to fix")
def test_multi_word_partial_no_match(self):
assert not loader._matches_pattern("fix bug", "can you fix it")
def test_case_insensitive(self):
assert loader._matches_pattern("Search", "SEARCH for me")
def test_word_boundary(self):
# "test" should not match "testing" due to word boundary
assert not loader._matches_pattern("test", "testing things")
def test_word_boundary_positive(self):
assert loader._matches_pattern("test", "run the test now")
# ── route_request ─────────────────────────────────────────────────────────
class TestRouteRequest:
def test_matches_first_agent(self):
with _mock_load_config(TWO_AGENT_YAML):
assert loader.route_request("research this topic") == "alpha"
def test_matches_second_agent(self):
with _mock_load_config(TWO_AGENT_YAML):
assert loader.route_request("fix bug in module") == "beta"
def test_returns_none_on_no_match(self):
with _mock_load_config(TWO_AGENT_YAML):
assert loader.route_request("hello world") is None
def test_returns_none_for_non_pattern_method(self):
import yaml
cfg = yaml.safe_load(TWO_AGENT_YAML)
cfg["routing"]["method"] = "llm"
with patch.object(loader, "_load_config", return_value=cfg):
assert loader.route_request("research this") is None
# ── route_request_with_match ──────────────────────────────────────────────
class TestRouteRequestWithMatch:
def test_returns_agent_and_pattern(self):
with _mock_load_config(TWO_AGENT_YAML):
agent_id, pattern = loader.route_request_with_match("research this")
assert agent_id == "alpha"
assert pattern == "research"
def test_returns_none_tuple_on_no_match(self):
with _mock_load_config(TWO_AGENT_YAML):
agent_id, pattern = loader.route_request_with_match("hello world")
assert agent_id is None
assert pattern is None
def test_returns_none_tuple_for_non_pattern_method(self):
import yaml
cfg = yaml.safe_load(TWO_AGENT_YAML)
cfg["routing"]["method"] = "llm"
with patch.object(loader, "_load_config", return_value=cfg):
agent_id, pattern = loader.route_request_with_match("research")
assert agent_id is None
assert pattern is None
# ── reload_agents ─────────────────────────────────────────────────────────
class TestReloadAgents:
@patch("timmy.prompts.get_system_prompt", return_value="BASE")
@patch("timmy.agents.base.Agent")
def test_clears_cache_and_reloads(self, mock_agno, mock_gsp):
with _mock_load_config(MINIMAL_YAML):
first = loader.load_agents(force_reload=True)
with _mock_load_config(MINIMAL_YAML):
reloaded = loader.reload_agents()
assert reloaded is not first
assert "helper" in reloaded