test: add comprehensive unit tests for agents/loader.py
All checks were successful
Tests / lint (pull_request) Successful in 4s
Tests / test (pull_request) Successful in 1m11s

Covers all public and private functions in the YAML-driven agent factory:
config loading/caching, model/tier resolution, system prompt building,
agent loading with defaults inheritance, get/list agents, pattern-based
routing, route_request_with_match, and reload_agents.

Fixes #347

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
kimi
2026-03-18 21:07:34 -04:00
parent db8cf5e13b
commit 60440ca035

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