diff --git a/tests/timmy/agents/__init__.py b/tests/timmy/agents/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/timmy/agents/test_loader.py b/tests/timmy/agents/test_loader.py new file mode 100644 index 0000000..795f41b --- /dev/null +++ b/tests/timmy/agents/test_loader.py @@ -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 diff --git a/tests/timmy/test_agents_loader.py b/tests/timmy/test_agents_loader.py new file mode 100644 index 0000000..bc1b7fd --- /dev/null +++ b/tests/timmy/test_agents_loader.py @@ -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