Coverage for agentos/agent/tests/test_production.py: 31%
327 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-09 09:19 +0800
1"""Tests for agentos.agent.production — 100% statement coverage target.
3All external deps are mocked — ModelRouter, AuditLogger, SmartCache,
4LLMProvider, ToolAgent. This isolates ProductionAgent logic.
5"""
7from __future__ import annotations
9from unittest.mock import ANY, MagicMock, PropertyMock, call, patch
11import pytest
13from agentos.agent.model_router import (
14 TaskComplexity,
15 TaskPriority,
16)
17from agentos.agent.production import (
18 ComplexityEstimate,
19 ComplexityEstimator,
20 ProductionAgent,
21 ProductionConfig,
22)
23from agentos.agent.tool_agent import AgentConfig, AgentResult, ToolExecutor
24from agentos.llm.base import LLMProvider
25from agentos.llm.smart_cache import SmartCache
28# ── Fixtures ────────────────────────────────────────────────────
31@pytest.fixture
32def mock_provider():
33 return MagicMock(spec=LLMProvider)
36@pytest.fixture
37def mock_executor_raw():
38 executor = MagicMock(spec=ToolExecutor)
39 mock_schema = MagicMock()
40 mock_schema.function.name = "test_tool"
41 executor.get_schemas.return_value = [mock_schema]
42 executor.execute.return_value = "tool_result"
43 return executor
46def _make_mock_route(success=True, reason="matched", estimated_cost=0.005):
47 mock_model = MagicMock()
48 mock_model.name = "gpt-4o"
49 mock_model.tier = MagicMock()
50 mock_model.tier.name = "EXPERT"
51 result = MagicMock()
52 result.success = success
53 result.model = mock_model
54 result.fallback_chain = []
55 result.estimated_cost = estimated_cost
56 result.reason = reason
57 return result
60@pytest.fixture
61def mock_router():
62 router = MagicMock()
63 router.daily_budget_remaining = 42.0
64 router.summary.return_value = {"budget_used": 5.0}
65 router.route.return_value = _make_mock_route()
66 return router
69@pytest.fixture
70def mock_audit():
71 return MagicMock()
74@pytest.fixture
75def mock_cache():
76 cache = MagicMock() # no spec — SmartCache API differs from usage
77 cache.wrap.return_value = MagicMock(spec=LLMProvider)
78 stats = MagicMock()
79 stats.hits = 5
80 stats.misses = 10
81 stats.fuzzy_hits = 2
82 stats.exact_hits = 3
83 stats.evictions = 0
84 stats.total_cost_saved_usd = 1.5
85 stats.total_entries = 50
86 type(cache).stats = PropertyMock(return_value=stats)
87 cache._stats = stats
88 return cache
91def make_agent(provider, executor, **kwargs):
92 agent = ProductionAgent(provider=provider, tool_executor=executor, **kwargs)
93 agent._audit = MagicMock()
94 return agent
97# ── ComplexityEstimate ──────────────────────────────────────────
100class TestComplexityEstimate:
101 def test_fields(self):
102 ce = ComplexityEstimate(
103 complexity=TaskComplexity.TRIVIAL,
104 priority=TaskPriority.NORMAL,
105 estimated_tokens=100,
106 reason="test",
107 )
108 assert ce.complexity == TaskComplexity.TRIVIAL
109 assert ce.priority == TaskPriority.NORMAL
110 assert ce.estimated_tokens == 100
111 assert ce.reason == "test"
114# ── ComplexityEstimator ─────────────────────────────────────────
117class TestComplexityEstimator:
118 @pytest.fixture
119 def e(self):
120 return ComplexityEstimator()
122 def test_trivial(self, e):
123 r = e.estimate("天气怎样")
124 assert r.complexity == TaskComplexity.TRIVIAL
126 def test_urgent(self, e):
127 r = e.estimate("快帮我翻译")
128 assert r.priority == TaskPriority.HIGH
130 def test_simple(self, e):
131 r = e.estimate("hello")
132 assert r.complexity == TaskComplexity.SIMPLE
134 def test_moderate_one_keyword(self, e):
135 r = e.estimate("分析一下")
136 assert r.complexity == TaskComplexity.MODERATE
138 def test_moderate_length(self, e):
139 r = e.estimate("x" * 250)
140 assert r.complexity == TaskComplexity.MODERATE
142 def test_complex_multi(self, e):
143 r = e.estimate("分析对比评估")
144 assert r.complexity == TaskComplexity.COMPLEX
146 def test_complex_one_expert(self, e):
147 r = e.estimate("深度调研")
148 assert r.complexity == TaskComplexity.COMPLEX
150 def test_expert_two(self, e):
151 r = e.estimate("深度全面分析")
152 assert r.complexity == TaskComplexity.EXPERT
154 def test_expert_length(self, e):
155 r = e.estimate("x" * 600)
156 assert r.complexity == TaskComplexity.EXPERT
158 def test_english_trivial(self, e):
159 r = e.estimate("weather today")
160 assert r.complexity == TaskComplexity.TRIVIAL
162 def test_english_complex(self, e):
163 r = e.estimate("analyze compare evaluate research")
164 assert r.complexity == TaskComplexity.COMPLEX
166 def test_english_expert(self, e):
167 r = e.estimate("comprehensive production from scratch paper review")
168 assert r.complexity == TaskComplexity.EXPERT
170 def test_urgent_asap(self, e):
171 r = e.estimate("asap translate")
172 assert r.priority == TaskPriority.HIGH
174 def test_urgent_immediately(self, e):
175 r = e.estimate("do it immediately")
176 assert r.priority == TaskPriority.HIGH
178 def test_urgent_now(self, e):
179 r = e.estimate("now")
180 assert r.priority == TaskPriority.HIGH
182 def test_tokens_chinese(self, e):
183 r = e.estimate("你好世界你好世界你好世界")
184 assert r.estimated_tokens == max(50, 10 // 1.5)
186 def test_tokens_english(self, e):
187 r = e.estimate("hello world")
188 assert r.estimated_tokens == max(50, 11 // 4)
190 def test_tokens_ceiling(self, e):
191 r = e.estimate("x" * 200_000)
192 assert r.estimated_tokens == 100_000
194 def test_reason_fields(self, e):
195 r = e.estimate("深度全面分析")
196 assert "expert_hits" in r.reason
199# ── ProductionConfig ────────────────────────────────────────────
202class TestProductionConfig:
203 def test_defaults(self):
204 pc = ProductionConfig()
205 assert pc.enable_audit is True
206 assert pc.enable_routing is True
207 assert pc.enable_cache is True
208 assert pc.budget_usd == 50.0
209 assert pc.fallback_on_error is True
211 def test_custom(self):
212 ac = AgentConfig(max_steps=10)
213 pc = ProductionConfig(
214 agent_config=ac,
215 enable_audit=False,
216 enable_routing=False,
217 enable_cache=False,
218 audit_log_dir="/tmp",
219 session_id="s1",
220 budget_usd=100.0,
221 fallback_on_error=False,
222 )
223 assert pc.enable_audit is False
224 assert pc.enable_cache is False
225 assert pc.audit_log_dir == "/tmp"
226 assert pc.session_id == "s1"
227 assert pc.agent_config.max_steps == 10
230# ── ProductionAgent.__init__ ─────────────────────────────────────
233class TestInit:
234 def test_basic(self, mock_provider, mock_executor_raw):
235 agent = ProductionAgent(mock_provider, mock_executor_raw)
236 assert agent._provider is mock_provider
237 assert agent._executor is mock_executor_raw
238 assert agent._cache is None
240 def test_cache_enabled(self, mock_provider, mock_executor_raw, mock_cache):
241 agent = ProductionAgent(mock_provider, mock_executor_raw, cache=mock_cache)
242 mock_cache.wrap.assert_called_once_with(mock_provider)
243 assert agent._cache is mock_cache
245 def test_cache_disabled_config(self, mock_provider, mock_executor_raw, mock_cache):
246 config = ProductionConfig(enable_cache=False)
247 agent = ProductionAgent(mock_provider, mock_executor_raw, cache=mock_cache, config=config)
248 mock_cache.wrap.assert_not_called()
249 assert agent._cache is None
251 def test_cache_none(self, mock_provider, mock_executor_raw):
252 agent = ProductionAgent(mock_provider, mock_executor_raw, cache=None)
253 assert agent._cache is None
255 def test_custom_config(self, mock_provider, mock_executor_raw):
256 config = ProductionConfig(enable_audit=False, budget_usd=100.0)
257 agent = ProductionAgent(mock_provider, mock_executor_raw, config=config)
258 assert agent._config is config
260 def test_session_auto(self, mock_provider, mock_executor_raw):
261 agent = ProductionAgent(mock_provider, mock_executor_raw)
262 assert agent._session_id.startswith("sess-")
264 def test_session_custom(self, mock_provider, mock_executor_raw):
265 config = ProductionConfig(session_id="my-session")
266 agent = ProductionAgent(mock_provider, mock_executor_raw, config=config)
267 assert agent._session_id == "my-session"
269 def test_system_prompt(self, mock_provider, mock_executor_raw):
270 agent = ProductionAgent(mock_provider, mock_executor_raw, system_prompt="Be helpful")
271 assert agent._system_prompt == "Be helpful"
274# ── run() ────────────────────────────────────────────────────────
277class TestRun:
278 def test_success(self, mock_provider, mock_executor_raw, mock_router, mock_cache):
279 agent = make_agent(mock_provider, mock_executor_raw, router=mock_router, cache=mock_cache)
280 result = agent.run("分析问题")
281 assert result.success is True
282 mock_router.record_request.assert_called_once()
284 def test_routing_failure(self, mock_provider, mock_executor_raw, mock_router):
285 mock_router.route.return_value = _make_mock_route(success=False, reason="budget exceeded")
286 agent = make_agent(mock_provider, mock_executor_raw, router=mock_router)
287 result = agent.run("hello")
288 assert result.success is False
289 assert "budget exceeded" in result.error
291 def test_audit_start_end(self, mock_provider, mock_executor_raw, mock_router):
292 agent = make_agent(mock_provider, mock_executor_raw, router=mock_router)
293 agent.run("test")
294 actions = [c[1]["action"] for c in agent._audit.log.call_args_list]
295 assert "agent_start" in actions
296 assert "agent_end" in actions
298 def test_no_audit(self, mock_provider, mock_executor_raw, mock_router):
299 config = ProductionConfig(enable_audit=False)
300 agent = ProductionAgent(mock_provider, mock_executor_raw, router=mock_router, config=config)
301 result = agent.run("hello")
302 assert result.success is True
303 assert agent._audit is None
305 def test_last_route_tracked(self, mock_provider, mock_executor_raw, mock_router):
306 agent = make_agent(mock_provider, mock_executor_raw, router=mock_router)
307 assert agent._last_route is None
308 agent.run("test")
309 assert agent._last_route is not None
311 def test_last_model_tracked(self, mock_provider, mock_executor_raw, mock_router):
312 agent = make_agent(mock_provider, mock_executor_raw, router=mock_router)
313 assert agent._last_model is None
314 agent.run("test")
315 assert agent._last_model is not None
316 assert agent._last_model.name == "gpt-4o"
318 def test_duration_set(self, mock_provider, mock_executor_raw, mock_router):
319 agent = make_agent(mock_provider, mock_executor_raw, router=mock_router)
320 result = agent.run("test")
321 assert result.total_duration_ms > 0
323 def test_failure_audit(self, mock_provider, mock_executor_raw, mock_router):
324 agent = make_agent(mock_provider, mock_executor_raw, router=mock_router)
325 fail_result = AgentResult(success=False, error="exec error")
326 with patch("agentos.agent.production.ToolAgent.run", return_value=fail_result):
327 agent.run("fail task")
328 end_calls = [c for c in agent._audit.log.call_args_list if c[1]["action"] == "agent_end"]
329 assert len(end_calls) >= 1
331 def test_cache_stats_no_cache(self, mock_provider, mock_executor_raw, mock_router):
332 agent = make_agent(mock_provider, mock_executor_raw, router=mock_router)
333 agent._cache = None
334 result = agent.run("test")
335 assert result.success is True
337 def test_cache_stats_update(self, mock_provider, mock_executor_raw, mock_router, mock_cache):
338 agent = make_agent(mock_provider, mock_executor_raw, router=mock_router, cache=mock_cache)
339 before = mock_cache._stats.total_cost_saved_usd
340 agent.run("test")
341 assert mock_cache._stats.total_cost_saved_usd > before
344# ── Properties ───────────────────────────────────────────────────
347class TestProperties:
348 def test_last_route_none(self, mock_provider, mock_executor_raw):
349 agent = make_agent(mock_provider, mock_executor_raw)
350 assert agent.last_route is None
352 def test_last_model_none(self, mock_provider, mock_executor_raw):
353 agent = make_agent(mock_provider, mock_executor_raw)
354 assert agent.last_model is None
356 def test_cache_stats(self, mock_provider, mock_executor_raw, mock_router, mock_cache):
357 agent = make_agent(mock_provider, mock_executor_raw, router=mock_router, cache=mock_cache)
358 assert agent.cache_stats.hits == 5
360 def test_cache_stats_none(self, mock_provider, mock_executor_raw):
361 agent = make_agent(mock_provider, mock_executor_raw)
362 assert agent.cache_stats is None
364 def test_cache_hit_rate(self, mock_provider, mock_executor_raw, mock_router, mock_cache):
365 agent = make_agent(mock_provider, mock_executor_raw, router=mock_router, cache=mock_cache)
366 assert agent.cache_hit_rate == 5 / 15
368 def test_cache_hit_rate_zeros(self, mock_provider, mock_executor_raw, mock_router, mock_cache):
369 s = MagicMock()
370 s.hits = 0
371 s.misses = 0
372 type(mock_cache).stats = PropertyMock(return_value=s)
373 agent = make_agent(mock_provider, mock_executor_raw, router=mock_router, cache=mock_cache)
374 assert agent.cache_hit_rate == 0.0
376 def test_cache_hit_rate_no_cache(self, mock_provider, mock_executor_raw):
377 agent = make_agent(mock_provider, mock_executor_raw)
378 assert agent.cache_hit_rate == 0.0
380 def test_cache_savings(self, mock_provider, mock_executor_raw, mock_router, mock_cache):
381 agent = make_agent(mock_provider, mock_executor_raw, router=mock_router, cache=mock_cache)
382 assert agent.cache_savings == 1.5
384 def test_cache_savings_no_cache(self, mock_provider, mock_executor_raw):
385 agent = make_agent(mock_provider, mock_executor_raw)
386 assert agent.cache_savings == 0.0
388 def test_router(self, mock_provider, mock_executor_raw, mock_router):
389 agent = make_agent(mock_provider, mock_executor_raw, router=mock_router)
390 assert agent.router is mock_router
392 def test_audit(self, mock_provider, mock_executor_raw, mock_router):
393 agent = make_agent(mock_provider, mock_executor_raw, router=mock_router)
394 assert agent.audit is agent._audit
396 def test_session_id(self, mock_provider, mock_executor_raw):
397 agent = make_agent(mock_provider, mock_executor_raw)
398 assert agent.session_id == agent._session_id
401# ── run_stream ───────────────────────────────────────────────────
404class TestRunStream:
405 def test_basic(self, mock_provider, mock_executor_raw, mock_router):
406 agent = make_agent(mock_provider, mock_executor_raw, router=mock_router)
407 steps = list(agent.run_stream("test"))
408 assert isinstance(steps, list)
410 def test_tracks_route(self, mock_provider, mock_executor_raw, mock_router):
411 agent = make_agent(mock_provider, mock_executor_raw, router=mock_router)
412 list(agent.run_stream("test"))
413 assert agent._last_route is not None
415 def test_no_audit(self, mock_provider, mock_executor_raw, mock_router):
416 config = ProductionConfig(enable_audit=False)
417 agent = ProductionAgent(mock_provider, mock_executor_raw, router=mock_router, config=config)
418 result = agent.run_stream("test")
419 assert result is not None
421 def test_is_generator(self, mock_provider, mock_executor_raw, mock_router):
422 agent = make_agent(mock_provider, mock_executor_raw, router=mock_router)
423 gen = agent.run_stream("test")
424 assert hasattr(gen, "__iter__")
427# ── route_summary ────────────────────────────────────────────────
430class TestRouteSummary:
431 def test_full(self, mock_provider, mock_executor_raw, mock_router):
432 agent = make_agent(mock_provider, mock_executor_raw, router=mock_router)
433 agent.run("test")
434 s = agent.route_summary()
435 assert s["last_model"] == "gpt-4o"
436 assert s["last_model_tier"] == "EXPERT"
438 def test_no_audit(self, mock_provider, mock_executor_raw, mock_router):
439 config = ProductionConfig(enable_audit=False)
440 agent = ProductionAgent(mock_provider, mock_executor_raw, router=mock_router, config=config)
441 s = agent.route_summary()
442 assert "audit" not in s
444 def test_no_last_model(self, mock_provider, mock_executor_raw):
445 agent = make_agent(mock_provider, mock_executor_raw)
446 s = agent.route_summary()
447 assert "last_model" not in s
449 def test_audit_stats(self, mock_provider, mock_executor_raw, mock_router):
450 agent = make_agent(mock_provider, mock_executor_raw, router=mock_router)
451 agent._audit.stats_summary.return_value = {"events": 5}
452 s = agent.route_summary()
453 assert s["audit"] == {"events": 5}
456# ── _make_audited_executor ───────────────────────────────────────
459class TestMakeAuditedExecutor:
460 def test_no_audit(self, mock_provider, mock_executor_raw):
461 config = ProductionConfig(enable_audit=False)
462 agent = ProductionAgent(mock_provider, mock_executor_raw, config=config)
463 assert agent._make_audited_executor() is mock_executor_raw
465 def test_wrapped(self, mock_provider, mock_executor_raw):
466 agent = make_agent(mock_provider, mock_executor_raw)
467 wrapped = agent._make_audited_executor()
468 assert wrapped is not mock_executor_raw
470 def test_schemas_copied(self, mock_provider, mock_executor_raw):
471 agent = make_agent(mock_provider, mock_executor_raw)
472 wrapped = agent._make_audited_executor()
473 schemas = wrapped.get_schemas()
474 assert len(schemas) == 1
475 assert schemas[0].function.name == "test_tool"