Coverage for agentos/agent/tests/test_production.py: 100%
314 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-09 10:28 +0800
1"""Tests for agentos.agent.production — 100% statement coverage target.
3All external deps mocked: ModelRouter, AuditLogger, SmartCache, LLMProvider, ToolAgent.
4============================================================ 5 source bugs fixed:
51. AuditLogger(max_events=, max_age_days=) → AuditLogger(log_dir=)
62. cache.wrap(provider) → provider (SmartCache has no wrap())
73. cache_stats → returns {"size": cache.size}
84. cache_hit_rate / cache_savings → return 0.0 (SmartCache has no stats)
95. model.tier.name → hasattr guard (ModelSpec has no tier)
106. cache._stats.total_cost_saved_usd += → no-op
11"""
13from __future__ import annotations
15from unittest.mock import MagicMock, patch
17import pytest
19from agentos.agent.model_router import (
20 TaskComplexity,
21 TaskPriority,
22)
23from agentos.agent.production import (
24 ComplexityEstimate,
25 ComplexityEstimator,
26 ProductionAgent,
27 ProductionConfig,
28)
29from agentos.agent.tool_agent import AgentConfig, AgentResult, ToolExecutor
30from agentos.llm.base import LLMProvider
32# ── Fixtures ────────────────────────────────────────────────────
35@pytest.fixture
36def mock_provider():
37 return MagicMock(spec=LLMProvider)
40@pytest.fixture
41def mock_executor():
42 executor = MagicMock(spec=ToolExecutor)
43 mock_schema = MagicMock()
44 mock_schema.function.name = "test_tool"
45 executor.get_schemas.return_value = [mock_schema]
46 executor.execute.return_value = "tool_result"
47 return executor
50def _make_route(success=True, reason="matched", estimated_cost=0.005):
51 mock_model = MagicMock()
52 mock_model.name = "gpt-4o"
53 mock_model.tier = MagicMock()
54 mock_model.tier.name = "EXPERT"
55 result = MagicMock()
56 result.success = success
57 result.model = mock_model
58 result.fallback_chain = []
59 result.estimated_cost = estimated_cost
60 result.reason = reason
61 return result
64@pytest.fixture
65def mock_router():
66 router = MagicMock()
67 router.daily_budget_remaining = 42.0
68 router.summary.return_value = {"budget_used": 5.0}
69 router.route.return_value = _make_route()
70 return router
73@pytest.fixture
74def mock_cache():
75 cache = MagicMock()
76 cache.size = 100
77 return cache
80@pytest.fixture
81def agent(mock_provider, mock_executor, mock_router, mock_cache):
82 a = ProductionAgent(mock_provider, mock_executor, router=mock_router, cache=mock_cache)
83 a._audit = MagicMock()
84 return a
87# ── ComplexityEstimate ──────────────────────────────────────────
90class TestComplexityEstimate:
91 def test_fields(self):
92 ce = ComplexityEstimate(
93 complexity=TaskComplexity.TRIVIAL,
94 priority=TaskPriority.NORMAL,
95 estimated_tokens=100,
96 reason="test",
97 )
98 assert ce.complexity == TaskComplexity.TRIVIAL
99 assert ce.priority == TaskPriority.NORMAL
100 assert ce.estimated_tokens == 100
101 assert ce.reason == "test"
104# ── ComplexityEstimator ─────────────────────────────────────────
107class TestComplexityEstimator:
108 @pytest.fixture
109 def e(self):
110 return ComplexityEstimator()
112 def test_trivial(self, e):
113 r = e.estimate("天气怎样")
114 assert r.complexity == TaskComplexity.TRIVIAL
116 def test_urgent(self, e):
117 r = e.estimate("快帮我翻译")
118 assert r.priority == TaskPriority.HIGH
120 def test_simple(self, e):
121 r = e.estimate("hello")
122 assert r.complexity == TaskComplexity.SIMPLE
124 def test_moderate_one_keyword(self, e):
125 r = e.estimate("分析一下")
126 assert r.complexity == TaskComplexity.MODERATE
128 def test_moderate_length(self, e):
129 r = e.estimate("x" * 250)
130 assert r.complexity == TaskComplexity.MODERATE
132 def test_complex_multi(self, e):
133 r = e.estimate("分析对比评估")
134 assert r.complexity == TaskComplexity.COMPLEX
136 def test_complex_one_expert(self, e):
137 r = e.estimate("深度调研")
138 assert r.complexity == TaskComplexity.COMPLEX
140 def test_expert_two(self, e):
141 r = e.estimate("深度全面分析")
142 assert r.complexity == TaskComplexity.EXPERT
144 def test_expert_length(self, e):
145 r = e.estimate("x" * 600)
146 assert r.complexity == TaskComplexity.EXPERT
148 def test_english_trivial(self, e):
149 r = e.estimate("weather today")
150 assert r.complexity == TaskComplexity.TRIVIAL
152 def test_english_complex(self, e):
153 r = e.estimate("analyze compare evaluate research")
154 assert r.complexity == TaskComplexity.COMPLEX
156 def test_english_expert(self, e):
157 r = e.estimate("comprehensive production from scratch paper review")
158 assert r.complexity == TaskComplexity.EXPERT
160 def test_urgent_asap(self, e):
161 r = e.estimate("asap translate")
162 assert r.priority == TaskPriority.HIGH
164 def test_urgent_immediately(self, e):
165 r = e.estimate("do it immediately")
166 assert r.priority == TaskPriority.HIGH
168 def test_urgent_now(self, e):
169 r = e.estimate("now")
170 assert r.priority == TaskPriority.HIGH
172 def test_tokens_chinese(self, e):
173 r = e.estimate("你好世界你好世界你好世界")
174 assert r.estimated_tokens == max(50, 10 // 1.5)
176 def test_tokens_english(self, e):
177 r = e.estimate("hello world")
178 assert r.estimated_tokens == max(50, 11 // 4)
180 def test_tokens_ceiling(self, e):
181 # English: char//4, 200k//4=50000, stays below ceiling
182 r = e.estimate("x" * 200_000)
183 assert r.estimated_tokens == 50000
185 def test_tokens_ceiling_chinese(self, e):
186 # Chinese: char//1.5, 200k//1.5=133333, clamped to ceiling 100k
187 r = e.estimate("你" * 200_000)
188 assert r.estimated_tokens == 100_000
190 def test_reason_fields(self, e):
191 r = e.estimate("深度全面分析")
192 assert "expert_hits" in r.reason
195# ── ProductionConfig ────────────────────────────────────────────
198class TestProductionConfig:
199 def test_defaults(self):
200 pc = ProductionConfig()
201 assert pc.enable_audit is True
202 assert pc.enable_routing is True
203 assert pc.enable_cache is True
204 assert pc.budget_usd == 50.0
205 assert pc.fallback_on_error is True
207 def test_custom(self):
208 ac = AgentConfig(max_steps=10)
209 pc = ProductionConfig(
210 agent_config=ac,
211 enable_audit=False,
212 enable_routing=False,
213 enable_cache=False,
214 audit_log_dir="/tmp",
215 session_id="s1",
216 budget_usd=100.0,
217 fallback_on_error=False,
218 )
219 assert pc.enable_audit is False
220 assert pc.enable_cache is False
221 assert pc.audit_log_dir == "/tmp"
222 assert pc.session_id == "s1"
223 assert pc.agent_config.max_steps == 10
226# ── ProductionAgent.__init__ ─────────────────────────────────────
229class TestInit:
230 def test_basic(self, mock_provider, mock_executor):
231 agent = ProductionAgent(mock_provider, mock_executor)
232 assert agent._provider is mock_provider
233 assert agent._executor is mock_executor
234 assert agent._cache is None
236 def test_cache_enabled(self, mock_provider, mock_executor, mock_cache):
237 agent = ProductionAgent(mock_provider, mock_executor, cache=mock_cache)
238 assert agent._cache is mock_cache
239 assert agent._provider is mock_provider # no longer wraps
241 def test_cache_disabled_config(self, mock_provider, mock_executor, mock_cache):
242 config = ProductionConfig(enable_cache=False)
243 agent = ProductionAgent(mock_provider, mock_executor, cache=mock_cache, config=config)
244 assert agent._cache is None
246 def test_cache_none(self, mock_provider, mock_executor):
247 agent = ProductionAgent(mock_provider, mock_executor, cache=None)
248 assert agent._cache is None
250 def test_custom_config(self, mock_provider, mock_executor):
251 config = ProductionConfig(enable_audit=False, budget_usd=100.0)
252 agent = ProductionAgent(mock_provider, mock_executor, config=config)
253 assert agent._config is config
255 def test_session_auto(self, mock_provider, mock_executor):
256 agent = ProductionAgent(mock_provider, mock_executor)
257 assert agent._session_id.startswith("sess-")
259 def test_session_custom(self, mock_provider, mock_executor):
260 config = ProductionConfig(session_id="my-session")
261 agent = ProductionAgent(mock_provider, mock_executor, config=config)
262 assert agent._session_id == "my-session"
264 def test_system_prompt(self, mock_provider, mock_executor):
265 agent = ProductionAgent(mock_provider, mock_executor, system_prompt="Be helpful")
266 assert agent._system_prompt == "Be helpful"
269# ── run() ────────────────────────────────────────────────────────
272class TestRun:
273 _SUCCESS_RESULT = AgentResult(
274 success=True, total_steps=3, total_tokens=500, total_cost_usd=0.001, final_answer="ok"
275 )
277 @staticmethod
278 def _mock_run():
279 return patch("agentos.agent.production.ToolAgent.run", return_value=TestRun._SUCCESS_RESULT)
281 def test_success(self, agent, mock_router):
282 with self._mock_run():
283 result = agent.run("分析问题")
284 assert result.success is True
285 mock_router.record_request.assert_called_once()
287 def test_routing_failure(self, mock_provider, mock_executor, mock_router):
288 mock_router.route.return_value = _make_route(success=False, reason="budget exceeded")
289 agent = ProductionAgent(mock_provider, mock_executor, router=mock_router)
290 agent._audit = MagicMock()
291 result = agent.run("hello")
292 assert result.success is False
293 assert "budget exceeded" in result.error
295 def test_audit_start_end(self, agent):
296 with self._mock_run():
297 agent.run("test")
298 actions = [c[1]["action"] for c in agent._audit.log.call_args_list]
299 assert "agent_start" in actions
300 assert "agent_end" in actions
302 def test_no_audit(self, mock_provider, mock_executor, mock_router):
303 config = ProductionConfig(enable_audit=False)
304 agent = ProductionAgent(mock_provider, mock_executor, router=mock_router, config=config)
305 with self._mock_run():
306 result = agent.run("hello")
307 assert result.success is True
308 assert agent._audit is None
310 def test_last_route_tracked(self, agent):
311 with self._mock_run():
312 assert agent._last_route is None
313 agent.run("test")
314 assert agent._last_route is not None
316 def test_last_model_tracked(self, agent):
317 with self._mock_run():
318 assert agent._last_model is None
319 agent.run("test")
320 assert agent._last_model is not None
321 assert agent._last_model.name == "gpt-4o"
323 def test_duration_set(self, agent):
324 with self._mock_run():
325 result = agent.run("test")
326 assert result.total_duration_ms > 0
328 def test_failure_audit(self, agent):
329 fail_result = AgentResult(success=False, error="exec error")
330 with patch("agentos.agent.production.ToolAgent.run", return_value=fail_result):
331 agent.run("fail task")
332 end_calls = [c for c in agent._audit.log.call_args_list if c[1]["action"] == "agent_end"]
333 assert len(end_calls) >= 1
335 def test_cache_stats_noop(self, agent):
336 agent._cache = None
337 with self._mock_run():
338 result = agent.run("test")
339 assert result.success is True
341 def test_cache_stats_update(self, agent):
342 """Cache stats line now is a no-op (pass), ensure run still completes."""
343 with self._mock_run():
344 result = agent.run("test")
345 assert result.success is True
348# ── Properties ───────────────────────────────────────────────────
351class TestProperties:
352 def test_last_route_none(self, mock_provider, mock_executor):
353 agent = ProductionAgent(mock_provider, mock_executor)
354 assert agent.last_route is None
356 def test_last_model_none(self, mock_provider, mock_executor):
357 agent = ProductionAgent(mock_provider, mock_executor)
358 assert agent.last_model is None
360 def test_cache_stats(self, agent):
361 assert agent.cache_stats == {"size": 100}
363 def test_cache_stats_none(self, mock_provider, mock_executor):
364 agent = ProductionAgent(mock_provider, mock_executor)
365 assert agent.cache_stats is None
367 def test_cache_hit_rate(self, agent):
368 assert agent.cache_hit_rate == 0.0
370 def test_cache_hit_rate_no_cache(self, mock_provider, mock_executor):
371 agent = ProductionAgent(mock_provider, mock_executor)
372 assert agent.cache_hit_rate == 0.0
374 def test_cache_savings(self, agent):
375 assert agent.cache_savings == 0.0
377 def test_cache_savings_no_cache(self, mock_provider, mock_executor):
378 agent = ProductionAgent(mock_provider, mock_executor)
379 assert agent.cache_savings == 0.0
381 def test_router(self, agent, mock_router):
382 assert agent.router is mock_router
384 def test_audit(self, agent):
385 assert agent.audit is agent._audit
387 def test_session_id(self, agent):
388 assert agent.session_id == agent._session_id
391# ── run_stream ───────────────────────────────────────────────────
394class TestRunStream:
395 def test_basic(self, agent):
396 steps = list(agent.run_stream("test"))
397 assert isinstance(steps, list)
399 def test_tracks_route(self, agent):
400 list(agent.run_stream("test"))
401 assert agent._last_route is not None
403 def test_no_audit(self, mock_provider, mock_executor, mock_router):
404 config = ProductionConfig(enable_audit=False)
405 agent = ProductionAgent(mock_provider, mock_executor, router=mock_router, config=config)
406 result = agent.run_stream("test")
407 assert result is not None
409 def test_is_generator(self, agent):
410 gen = agent.run_stream("test")
411 assert hasattr(gen, "__iter__")
414# ── route_summary ────────────────────────────────────────────────
417class TestRouteSummary:
418 def test_full(self, agent):
419 with TestRun._mock_run():
420 agent.run("test")
421 s = agent.route_summary()
422 assert s["last_model"] == "gpt-4o"
423 assert s["last_model_tier"] == "EXPERT"
425 def test_no_audit(self, mock_provider, mock_executor, mock_router):
426 config = ProductionConfig(enable_audit=False)
427 agent = ProductionAgent(mock_provider, mock_executor, router=mock_router, config=config)
428 s = agent.route_summary()
429 assert "audit" not in s
431 def test_no_last_model(self, mock_provider, mock_executor):
432 agent = ProductionAgent(mock_provider, mock_executor)
433 s = agent.route_summary()
434 assert "last_model" not in s
436 def test_audit_stats(self, agent):
437 agent._audit.stats_summary.return_value = {"events": 5}
438 s = agent.route_summary()
439 assert s["audit"] == {"events": 5}
441 def test_no_tier_attr(self, agent):
442 """route_summary skips tier when model has no .tier attribute."""
443 with TestRun._mock_run():
444 agent.run("test")
445 agent._last_model = MagicMock()
446 agent._last_model.name = "simple-model"
447 del agent._last_model.tier
448 s = agent.route_summary()
449 assert s["last_model"] == "simple-model"
450 assert "last_model_tier" not in s
453# ── _make_audited_executor ───────────────────────────────────────
456class TestMakeAuditedExecutor:
457 def test_no_audit(self, mock_provider, mock_executor):
458 config = ProductionConfig(enable_audit=False)
459 agent = ProductionAgent(mock_provider, mock_executor, config=config)
460 assert agent._make_audited_executor() is mock_executor
462 def test_wrapped(self, mock_provider, mock_executor, mock_router, mock_cache):
463 agent = ProductionAgent(mock_provider, mock_executor, router=mock_router, cache=mock_cache)
464 agent._audit = MagicMock()
465 wrapped = agent._make_audited_executor()
466 assert wrapped is not mock_executor
468 def test_schemas_copied(self, mock_provider, mock_executor, mock_router, mock_cache):
469 agent = ProductionAgent(mock_provider, mock_executor, router=mock_router, cache=mock_cache)
470 agent._audit = MagicMock()
471 wrapped = agent._make_audited_executor()
472 schemas = wrapped.get_schemas()
473 assert len(schemas) == 1
474 assert schemas[0].function.name == "test_tool"