Coverage for agentos/agent/tests/test_agent_builder.py: 95%
255 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-09 10:28 +0800
1"""Tests for agentos.agent.agent_builder — 100% statement coverage target."""
3from __future__ import annotations
5import importlib
6from unittest.mock import MagicMock, patch
8from agentos.agent.agent_builder import (
9 _MockProvider,
10 build_agent,
11 create_provider,
12 discover_tools,
13)
14from agentos.llm.base import MessageRole
16# ── helpers ──────────────────────────────────────────────────────
19def _make_tool_cls(tool_name: str):
20 """Create a concrete BaseTool subclass with the given name."""
21 from agentos.tools.base import BaseTool
23 class _Tool(BaseTool):
24 _tool_name = tool_name
26 @property
27 def name(self) -> str:
28 return self._tool_name
30 @property
31 def parameters(self) -> dict:
32 return {}
34 async def execute(self, arguments: dict, sandbox=None):
35 from agentos.tools.base import ToolResult
37 return ToolResult(success=True, output=tool_name)
39 return _Tool
42class _ToolMod:
43 """Fake module-like object for inspect.getmembers."""
45 def __init__(self, **kw):
46 self.__dict__.update(kw)
49_real_import = importlib.import_module
52# ── _MockProvider ────────────────────────────────────────────────
55class TestMockProvider:
56 def test_provider_name(self):
57 mp = _MockProvider()
58 assert mp.provider_name == "mock-dev"
60 def test_chat_returns_completion(self):
61 mp = _MockProvider()
62 result = mp.chat([])
63 assert result.choices[0].message.content.startswith("Mock provider")
64 assert result.model == "mock"
66 def test_achat_same_as_chat(self):
67 import asyncio
69 mp = _MockProvider()
70 sync = mp.chat([])
71 async_result = asyncio.run(mp.achat([]))
72 assert async_result.choices[0].message.content == sync.choices[0].message.content
74 def test__make_usage_fields(self):
75 mp = _MockProvider()
76 result = mp._make("test")
77 assert result.usage.prompt_tokens == 0
78 assert result.usage.completion_tokens == 0
79 assert result.usage.total_tokens == 0
80 assert result.choices[0].message.role == MessageRole.ASSISTANT
81 assert result.choices[0].finish_reason == "stop"
84# ── discover_tools ───────────────────────────────────────────────
87class TestDiscoverTools:
88 def test_empty_package(self):
89 with patch("importlib.import_module") as mock_import, patch(
90 "pkgutil.iter_modules", return_value=[]
91 ):
92 mock_module = MagicMock()
93 mock_module.__file__ = "/fake/agentos/tools/__init__.py"
94 mock_import.return_value = mock_module
95 result = discover_tools("agentos.tools")
96 assert result == []
98 def test_skip_underscore_module(self):
99 """Use real temp package — coverage tracks _-prefixed module skip."""
100 import sys
102 _tmp_root = "/home/marvis/Marvis/User/oAN1i2Yfn4aIvXkoz-oN0h5oHcb4/workspace/conv_19f08962d4e_9d1241f0e39e/temp"
103 sys.path.insert(0, _tmp_root)
104 try:
105 result = discover_tools("_fake_pkg")
106 finally:
107 sys.path.remove(_tmp_root)
108 # _underscore module skipped; bad_import skipped; real_one from real_tool.py only
109 names = [t.name for t in result]
110 assert "real_one" in names
111 # _underscore directory is not a module, so it shouldn't appear
112 for t in result:
113 assert not t.name.startswith("_")
115 def test_import_error_skipped(self):
116 """Use real temp package with a module that raises on import."""
117 import sys
119 _tmp_root = "/home/marvis/Marvis/User/oAN1i2Yfn4aIvXkoz-oN0h5oHcb4/workspace/conv_19f08962d4e_9d1241f0e39e/temp"
120 sys.path.insert(0, _tmp_root)
121 try:
122 result = discover_tools("_fake_pkg")
123 finally:
124 sys.path.remove(_tmp_root)
125 # bad_import.py raises ImportError at module level → skipped gracefully
126 # real_tool.py provides real_one → should be present
127 names = [t.name for t in result]
128 assert "real_one" in names
130 def test_skip_base_class_itself(self):
131 from agentos.tools.base import BaseTool
133 fake_tool_cls = _make_tool_cls("fake_tool")
134 fake_mod = _ToolMod(FakeTool=fake_tool_cls, BaseTool=BaseTool)
135 pkg = MagicMock()
136 pkg.__file__ = "/fake/agentos/tools/__init__.py"
138 def side_effect(name):
139 if name == "agentos.tools":
140 return pkg
141 if name == "agentos.tools.fake_mod":
142 return fake_mod
143 if name.startswith("agentos.tools."):
144 raise ImportError(f"mocked: {name}")
145 return _real_import(name)
147 with patch("importlib.import_module", side_effect=side_effect), patch(
148 "pkgutil.iter_modules",
149 return_value=[("fake_mod", "fake_mod", False)],
150 ):
151 result = discover_tools("agentos.tools")
153 assert len(result) == 1
154 assert result[0].name == "fake_tool"
156 def test_skip_no_name_attr(self):
157 from agentos.tools.base import BaseTool
159 class NoNameTool(BaseTool):
160 name = None # triggers: getattr(obj, "name", None) is None
162 @property
163 def parameters(self) -> dict:
164 return {}
166 async def execute(self, arguments: dict, sandbox=None):
167 from agentos.tools.base import ToolResult
168 return ToolResult(success=True, output="")
170 mod = _ToolMod(NoNameTool=NoNameTool)
171 pkg = MagicMock()
172 pkg.__file__ = "/fake/agentos/tools/__init__.py"
174 def side_effect(name):
175 if name == "agentos.tools":
176 return pkg
177 if name == "agentos.tools.fake_mod":
178 return mod
179 if name.startswith("agentos.tools."):
180 raise ImportError(f"mocked: {name}")
181 return _real_import(name)
183 with patch("importlib.import_module", side_effect=side_effect), patch(
184 "pkgutil.iter_modules",
185 return_value=[("fake_mod", "fake_mod", False)],
186 ):
187 result = discover_tools("agentos.tools")
189 assert result == []
191 def test_skip_duplicate_class(self):
192 """Use real temp package — two modules export same tool name, deduped."""
193 import sys
195 _tmp_root = "/home/marvis/Marvis/User/oAN1i2Yfn4aIvXkoz-oN0h5oHcb4/workspace/conv_19f08962d4e_9d1241f0e39e/temp"
196 sys.path.insert(0, _tmp_root)
197 try:
198 result = discover_tools("_fake_pkg")
199 finally:
200 sys.path.remove(_tmp_root)
201 # real_tool.py and real_tool2.py both export RealTool(name='real_one')
202 # → only one instance returned
203 names = [t.name for t in result]
204 assert names.count("real_one") == 1
205 assert "real_one" in names
207 def test_instantiation_error_skipped(self):
208 from agentos.tools.base import BaseTool
210 class CrashTool(BaseTool):
211 _name = "crash"
213 @property
214 def name(self) -> str:
215 return self._name
217 @property
218 def parameters(self) -> dict:
219 return {}
221 def __init__(self):
222 raise RuntimeError("oops")
224 async def execute(self, arguments: dict, sandbox=None):
225 from agentos.tools.base import ToolResult
226 return ToolResult(success=True, output="")
228 mod = _ToolMod(CrashTool=CrashTool)
229 pkg = MagicMock()
230 pkg.__file__ = "/fake/agentos/tools/__init__.py"
232 def side_effect(name):
233 if name == "agentos.tools":
234 return pkg
235 if name == "agentos.tools.fake_mod":
236 return mod
237 if name.startswith("agentos.tools."):
238 raise ImportError(f"mocked: {name}")
239 return _real_import(name)
241 with patch("importlib.import_module", side_effect=side_effect), patch(
242 "pkgutil.iter_modules",
243 return_value=[("fake_mod", "fake_mod", False)],
244 ):
245 result = discover_tools("agentos.tools")
247 assert result == []
250# ── create_provider ──────────────────────────────────────────────
253class TestCreateProvider:
254 def test_deepseek_api_key(self):
255 with patch.dict("os.environ", {"DEEPSEEK_API_KEY": "sk-test"}, clear=False):
256 with patch(
257 "agentos.llm.providers.deepseek.DeepSeekProvider"
258 ) as mock_cls:
259 mock_cls.return_value = "deepseek_inst"
260 result = create_provider()
261 assert result == "deepseek_inst"
262 mock_cls.assert_called_once_with(model="deepseek-chat")
264 def test_openai_api_key(self):
265 with patch.dict("os.environ", {"OPENAI_API_KEY": "sk-test"}):
266 with patch.dict("os.environ", {"DEEPSEEK_API_KEY": ""}):
267 with patch(
268 "agentos.llm.providers.openai.OpenAIProvider"
269 ) as mock_cls:
270 mock_cls.return_value = "openai_inst"
271 result = create_provider()
272 assert result == "openai_inst"
273 mock_cls.assert_called_once_with(model="gpt-4o-mini")
275 def test_anthropic_api_key(self):
276 env = {"ANTHROPIC_API_KEY": "sk-test"}
277 with patch.dict("os.environ", env, clear=True):
278 with patch(
279 "agentos.llm.providers.anthropic.AnthropicProvider"
280 ) as mock_cls:
281 mock_cls.return_value = "anthropic_inst"
282 result = create_provider()
283 assert result == "anthropic_inst"
284 mock_cls.assert_called_once_with(model="claude-3-5-sonnet-20241022")
286 def test_mock_fallback(self):
287 with patch.dict("os.environ", {}, clear=True):
288 result = create_provider()
289 assert isinstance(result, _MockProvider)
292# ── build_agent ──────────────────────────────────────────────────
295class TestBuildAgent:
296 def test_minimal(self):
297 agent = build_agent(discover_all=False)
298 assert agent is not None
300 def test_custom_system_prompt(self):
301 agent = build_agent(
302 system_prompt="自定义提示词",
303 discover_all=False,
304 )
305 assert agent._system_prompt == "自定义提示词"
307 def test_auto_system_prompt_with_tools(self):
308 my_tool_cls = _make_tool_cls("my_tool")
309 agent = build_agent(
310 tools=[my_tool_cls()],
311 discover_all=False,
312 )
313 assert "my_tool" in agent._system_prompt
315 def test_auto_system_prompt_no_tools(self):
316 agent = build_agent(
317 tools=[],
318 discover_all=False,
319 include_skills=False,
320 )
321 assert "无" in agent._system_prompt
323 def test_manual_provider(self):
324 mp = _MockProvider()
325 agent = build_agent(
326 provider=mp,
327 discover_all=False,
328 )
329 assert agent._provider is mp
331 def test_manual_tools(self):
332 t1_cls = _make_tool_cls("t1")
333 t1 = t1_cls()
334 agent = build_agent(
335 tools=[t1],
336 discover_all=False,
337 )
338 schemas = agent._executor.get_schemas()
339 names = [s.function.name for s in schemas]
340 assert "t1" in names
342 def test_discover_all_true(self):
343 with patch(
344 "agentos.agent.agent_builder.discover_tools", return_value=[]
345 ) as mock_dt:
346 agent = build_agent(discover_all=True)
347 mock_dt.assert_called_once()
348 assert agent is not None
350 def test_discover_all_false_no_tools(self):
351 agent = build_agent(discover_all=False)
352 assert agent is not None
354 def test_include_skills_success(self):
355 with patch(
356 "agentos.tools.skill_tool.discover_skills",
357 return_value=[],
358 ) as mock_ds:
359 agent = build_agent(
360 discover_all=False,
361 include_skills=True,
362 )
363 mock_ds.assert_called_once()
364 assert agent is not None
366 def test_include_skills_error_swallowed(self):
367 with patch(
368 "agentos.tools.skill_tool.discover_skills",
369 side_effect=ImportError("no module"),
370 ):
371 agent = build_agent(
372 discover_all=False,
373 include_skills=True,
374 )
375 assert agent is not None
377 def test_include_skills_false(self):
378 with patch(
379 "agentos.tools.skill_tool.discover_skills"
380 ) as mock_ds:
381 agent = build_agent(
382 discover_all=False,
383 include_skills=False,
384 )
385 mock_ds.assert_not_called()
386 assert agent is not None
388 def test_skills_with_existing_tools(self):
389 t2_cls = _make_tool_cls("t2")
390 fake_skill = MagicMock()
391 fake_skill.name = "skill_a"
393 with patch(
394 "agentos.tools.skill_tool.discover_skills",
395 return_value=[fake_skill],
396 ):
397 agent = build_agent(
398 tools=[t2_cls()],
399 discover_all=False,
400 include_skills=True,
401 )
402 names = [s.function.name for s in agent._executor.get_schemas()]
403 assert "t2" in names
404 assert "skill_a" in names
406 def test_skills_with_no_existing_tools(self):
407 fake_skill = MagicMock()
408 fake_skill.name = "skill_b"
410 with patch(
411 "agentos.tools.skill_tool.discover_skills",
412 return_value=[fake_skill],
413 ):
414 agent = build_agent(
415 tools=None,
416 discover_all=False,
417 include_skills=True,
418 )
419 names = [s.function.name for s in agent._executor.get_schemas()]
420 assert "skill_b" in names
422 def test_verbose_flag(self):
423 agent = build_agent(
424 discover_all=False,
425 verbose=True,
426 )
427 assert agent._config.verbose is True
429 def test_max_steps_default(self):
430 agent = build_agent(discover_all=False)
431 assert agent._config.max_steps == 10
433 def test_max_steps_custom(self):
434 agent = build_agent(
435 discover_all=False,
436 max_steps=5,
437 )
438 assert agent._config.max_steps == 5