Coverage for agentos/llm/tests/test_providers.py: 0%

1"""LLM Provider 模块单元测试 — v1.3.36。 

2测试范围: factory, base types, Function Calling, DeepSeek, Anthropic (unit/mock)。 

3""" 

4 

5import os 

6from unittest.mock import patch 

7 

8import pytest 

9 

10from agentos.llm import ( 

11 AnthropicProvider, 

12 CompletionUsage, 

13 DeepSeekProvider, 

14 LLMProvider, 

15 Message, 

16 MessageRole, 

17 OpenAIProvider, 

18 StreamChunk, 

19 TokenUsage, 

20 Tool, 

21 ToolCall, 

22 ToolParameter, 

23 create_provider, 

24) 

25 

26# ── Base Types ─────────────────────────────────────────────────── 

27 

28 

29class TestTokenUsage: 

30 def test_defaults(self): 

31 u = TokenUsage() 

32 assert u.prompt_tokens == 0 

33 assert u.completion_tokens == 0 

34 assert u.total_tokens == 0 

35 

36 def test_values(self): 

37 u = TokenUsage(prompt_tokens=100, completion_tokens=50, total_tokens=150) 

38 assert u.prompt_tokens == 100 

39 

40 

41class TestCompletionUsage: 

42 def test_cost_default(self): 

43 u = CompletionUsage(prompt_tokens=500, completion_tokens=200, total_tokens=700) 

44 assert u.cost_usd == 0.0 

45 

46 

47class TestMessage: 

48 def test_basic(self): 

49 m = Message(role=MessageRole.USER, content="hi") 

50 d = m.as_dict() 

51 assert d["role"] == "user" 

52 assert d["content"] == "hi" 

53 

54 def test_with_tool_call_id(self): 

55 m = Message(role=MessageRole.TOOL, content="result", tool_call_id="call_123") 

56 d = m.as_dict() 

57 assert d["tool_call_id"] == "call_123" 

58 

59 def test_with_tool_calls(self): 

60 m = Message( 

61 role=MessageRole.ASSISTANT, 

62 content="", 

63 tool_calls=[ToolCall(id="tc1", name="get_weather", arguments='{"city":"NYC"}')], 

64 ) 

65 assert m.tool_calls[0].name == "get_weather" 

66 

67 

68# ── Tool / Function Calling ─────────────────────────────────────── 

69 

70 

71class TestToolParameter: 

72 def test_basic_schema(self): 

73 p = ToolParameter(type="string", description="City name", required=True) 

74 s = p.as_schema() 

75 assert s["type"] == "string" 

76 assert s["description"] == "City name" 

77 

78 def test_with_enum(self): 

79 p = ToolParameter(type="string", enum=["celsius", "fahrenheit"]) 

80 s = p.as_schema() 

81 assert s["enum"] == ["celsius", "fahrenheit"] 

82 

83 

84class TestTool: 

85 def test_from_function(self): 

86 t = Tool.from_function( 

87 "get_weather", 

88 "Get weather for a city", 

89 { 

90 "city": ToolParameter(type="string", description="City", required=True), 

91 "unit": ToolParameter(type="string", enum=["celsius", "fahrenheit"]), 

92 }, 

93 required=["city"], 

94 ) 

95 schema = t.as_schema() 

96 assert schema["type"] == "function" 

97 fn = schema["function"] 

98 assert fn["name"] == "get_weather" 

99 assert fn["parameters"]["required"] == ["city"] 

100 assert "city" in fn["parameters"]["properties"] 

101 

102 def test_to_openai_format(self): 

103 t = Tool.from_function( 

104 "search", 

105 "Web search", 

106 { 

107 "query": ToolParameter(type="string", description="Query", required=True), 

108 }, 

109 ) 

110 schema = t.as_schema() 

111 assert schema["function"]["name"] == "search" 

112 props = schema["function"]["parameters"]["properties"] 

113 assert props["query"]["type"] == "string" 

114 

115 

116class TestToolCall: 

117 def test_create_and_parse(self): 

118 tc = ToolCall(id="call_1", name="add", arguments='{"a":1,"b":2}') 

119 assert tc.id == "call_1" 

120 assert tc.name == "add" 

121 assert tc.parsed_arguments == {"a": 1, "b": 2} 

122 

123 def test_empty_arguments(self): 

124 tc = ToolCall(id="x", name="ping", arguments="{}") 

125 assert tc.parsed_arguments == {} 

126 

127 

128# ── StreamChunk ──────────────────────────────────────────────────── 

129 

130 

131class TestStreamChunk: 

132 def test_defaults(self): 

133 c = StreamChunk() 

134 assert c.content == "" 

135 assert c.finish_reason is None 

136 

137 def test_with_content(self): 

138 c = StreamChunk(content="hello") 

139 assert c.content == "hello" 

140 

141 

142# ── Factory ──────────────────────────────────────────────────────── 

143 

144 

145class TestCreateProvider: 

146 def test_openai_default(self): 

147 with patch.dict(os.environ, {"OPENAI_API_KEY": "sk-test"}, clear=True): 

148 p = create_provider("openai") 

149 assert p.provider_name == "openai" 

150 assert p.model == "gpt-4o-mini" 

151 

152 def test_deepseek_default(self): 

153 with patch.dict(os.environ, {"DEEPSEEK_API_KEY": "sk-ds"}, clear=True): 

154 p = create_provider("deepseek") 

155 assert p.provider_name == "deepseek" 

156 assert p.model == "deepseek-chat" 

157 assert "deepseek.com" in p.base_url 

158 

159 def test_anthropic_default(self): 

160 with patch.dict(os.environ, {"ANTHROPIC_API_KEY": "sk-ant"}, clear=True): 

161 p = create_provider("anthropic") 

162 assert p.provider_name == "anthropic" 

163 assert "sonnet" in p.model.lower() 

164 

165 def test_unknown_provider(self): 

166 with pytest.raises(ValueError, match="Unknown provider"): 

167 create_provider("nonexistent") 

168 

169 def test_api_key_env_openai(self): 

170 with patch.dict(os.environ, {"OPENAI_API_KEY": "sk-env-test"}, clear=True): 

171 p = create_provider("openai") 

172 assert p.api_key == "sk-env-test" 

173 

174 def test_custom_model(self): 

175 with patch.dict(os.environ, {"OPENAI_API_KEY": "sk-test"}, clear=True): 

176 p = create_provider("openai", model="gpt-4o") 

177 assert p.model == "gpt-4o" 

178 

179 

180# ── DeepSeek Provider ────────────────────────────────────────────── 

181 

182 

183class TestDeepSeekProvider: 

184 def test_is_openai_subclass(self): 

185 p = DeepSeekProvider(api_key="sk-ds") 

186 assert isinstance(p, OpenAIProvider) 

187 assert isinstance(p, LLMProvider) 

188 

189 def test_provider_name(self): 

190 p = DeepSeekProvider(api_key="sk-ds") 

191 assert p.provider_name == "deepseek" 

192 

193 def test_default_base_url(self): 

194 p = DeepSeekProvider(api_key="sk-ds") 

195 assert p.base_url == "https://api.deepseek.com/v1" 

196 

197 def test_custom_base_url(self): 

198 p = DeepSeekProvider(api_key="sk-ds", base_url="http://localhost:8080/v1") 

199 assert p.base_url == "http://localhost:8080/v1" 

200 

201 def test_default_model(self): 

202 p = DeepSeekProvider(api_key="sk-ds") 

203 assert p.model == "deepseek-chat" 

204 

205 def test_factory_creates(self): 

206 with patch.dict(os.environ, {"DEEPSEEK_API_KEY": "sk-ds"}, clear=True): 

207 p = create_provider("deepseek") 

208 assert isinstance(p, DeepSeekProvider) 

209 assert p.provider_name == "deepseek" 

210 

211 

212# ── Anthropic Provider ───────────────────────────────────────────── 

213 

214 

215class TestAnthropicProvider: 

216 def test_provider_name(self): 

217 p = AnthropicProvider(api_key="sk-ant") 

218 assert p.provider_name == "anthropic" 

219 

220 def test_default_model(self): 

221 p = AnthropicProvider(api_key="sk-ant") 

222 assert p.model == "claude-sonnet-4-20250514" 

223 

224 def test_default_base_url(self): 

225 p = AnthropicProvider(api_key="sk-ant") 

226 assert "api.anthropic.com" in p.base_url 

227 

228 def test_custom_base_url(self): 

229 p = AnthropicProvider(api_key="sk-ant", base_url="http://localhost:9999") 

230 assert p.base_url == "http://localhost:9999" 

231 

232 def test_headers(self): 

233 p = AnthropicProvider(api_key="sk-ant-test") 

234 h = p._headers() 

235 assert h["x-api-key"] == "sk-ant-test" 

236 assert h["anthropic-version"] == "2023-06-01" 

237 

238 def test_tools_conversion(self): 

239 AnthropicProvider(api_key="sk-ant") 

240 tools = [ 

241 Tool.from_function( 

242 "get_weather", 

243 "Get weather", 

244 { 

245 "city": ToolParameter(type="string", description="City", required=True), 

246 }, 

247 ), 

248 ] 

249 from agentos.llm.anthropic_provider import _tools_to_anthropic 

250 

251 result = _tools_to_anthropic(tools) 

252 assert len(result) == 1 

253 assert result[0]["name"] == "get_weather" 

254 assert result[0]["input_schema"]["type"] == "object" 

255 

256 def test_message_conversion_simple(self): 

257 from agentos.llm.anthropic_provider import _messages_to_anthropic 

258 

259 msgs = [Message(role=MessageRole.USER, content="Hello")] 

260 system, api_msgs = _messages_to_anthropic(msgs) 

261 assert system is None 

262 assert len(api_msgs) == 1 

263 assert api_msgs[0]["role"] == "user" 

264 assert api_msgs[0]["content"] == "Hello" 

265 

266 def test_message_conversion_with_system(self): 

267 from agentos.llm.anthropic_provider import _messages_to_anthropic 

268 

269 msgs = [ 

270 Message(role=MessageRole.SYSTEM, content="You are helpful."), 

271 Message(role=MessageRole.USER, content="Hi"), 

272 ] 

273 system, api_msgs = _messages_to_anthropic(msgs) 

274 assert system == "You are helpful." 

275 assert len(api_msgs) == 1 

276 assert api_msgs[0]["role"] == "user" 

277 

278 def test_build_body_includes_tools(self): 

279 p = AnthropicProvider(api_key="sk-ant") 

280 tools = [ 

281 Tool.from_function( 

282 "search", 

283 "search the web", 

284 { 

285 "q": ToolParameter(type="string", description="query", required=True), 

286 }, 

287 ) 

288 ] 

289 body = p._build_body( 

290 [Message(role=MessageRole.USER, content="test")], 

291 temperature=0.5, 

292 max_tokens=100, 

293 top_p=0.9, 

294 stop=None, 

295 tools=tools, 

296 tool_choice="auto", 

297 ) 

298 assert body["model"] == "claude-sonnet-4-20250514" 

299 assert "tools" in body 

300 assert body["tools"][0]["name"] == "search"