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 json 

6import os 

7from unittest.mock import patch, MagicMock 

8 

9import pytest 

10 

11from agentos.llm import ( 

12 create_provider, 

13 OpenAIProvider, 

14 DeepSeekProvider, 

15 AnthropicProvider, 

16 LLMProvider, 

17 CompletionResult, 

18 CompletionChoice, 

19 CompletionUsage, 

20 TokenUsage, 

21 Message, 

22 MessageRole, 

23 StreamChunk, 

24 Tool, 

25 ToolCall, 

26 ToolFunction, 

27 ToolParameter, 

28) 

29 

30 

31# ── Base Types ─────────────────────────────────────────────────── 

32 

33class TestTokenUsage: 

34 def test_defaults(self): 

35 u = TokenUsage() 

36 assert u.prompt_tokens == 0 

37 assert u.completion_tokens == 0 

38 assert u.total_tokens == 0 

39 

40 def test_values(self): 

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

42 assert u.prompt_tokens == 100 

43 

44 

45class TestCompletionUsage: 

46 def test_cost_default(self): 

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

48 assert u.cost_usd == 0.0 

49 

50 

51class TestMessage: 

52 def test_basic(self): 

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

54 d = m.as_dict() 

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

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

57 

58 def test_with_tool_call_id(self): 

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

60 d = m.as_dict() 

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

62 

63 def test_with_tool_calls(self): 

64 m = Message( 

65 role=MessageRole.ASSISTANT, 

66 content="", 

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

68 ) 

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

70 

71 

72# ── Tool / Function Calling ─────────────────────────────────────── 

73 

74class TestToolParameter: 

75 def test_basic_schema(self): 

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

77 s = p.as_schema() 

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

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

80 

81 def test_with_enum(self): 

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

83 s = p.as_schema() 

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

85 

86 

87class TestTool: 

88 def test_from_function(self): 

89 t = Tool.from_function( 

90 "get_weather", 

91 "Get weather for a city", 

92 { 

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

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

95 }, 

96 required=["city"], 

97 ) 

98 schema = t.as_schema() 

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

100 fn = schema["function"] 

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

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

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

104 

105 def test_to_openai_format(self): 

106 t = Tool.from_function("search", "Web search", { 

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

108 }) 

109 schema = t.as_schema() 

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

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

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

113 

114 

115class TestToolCall: 

116 def test_create_and_parse(self): 

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

118 assert tc.id == "call_1" 

119 assert tc.name == "add" 

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

121 

122 def test_empty_arguments(self): 

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

124 assert tc.parsed_arguments == {} 

125 

126 

127# ── StreamChunk ──────────────────────────────────────────────────── 

128 

129class TestStreamChunk: 

130 def test_defaults(self): 

131 c = StreamChunk() 

132 assert c.content == "" 

133 assert c.finish_reason is None 

134 

135 def test_with_content(self): 

136 c = StreamChunk(content="hello") 

137 assert c.content == "hello" 

138 

139 

140# ── Factory ──────────────────────────────────────────────────────── 

141 

142class TestCreateProvider: 

143 def test_openai_default(self): 

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

145 p = create_provider("openai") 

146 assert p.provider_name == "openai" 

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

148 

149 def test_deepseek_default(self): 

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

151 p = create_provider("deepseek") 

152 assert p.provider_name == "deepseek" 

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

154 assert "deepseek.com" in p.base_url 

155 

156 def test_anthropic_default(self): 

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

158 p = create_provider("anthropic") 

159 assert p.provider_name == "anthropic" 

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

161 

162 def test_unknown_provider(self): 

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

164 create_provider("nonexistent") 

165 

166 def test_api_key_env_openai(self): 

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

168 p = create_provider("openai") 

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

170 

171 def test_custom_model(self): 

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

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

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

175 

176 

177# ── DeepSeek Provider ────────────────────────────────────────────── 

178 

179class TestDeepSeekProvider: 

180 def test_is_openai_subclass(self): 

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

182 assert isinstance(p, OpenAIProvider) 

183 assert isinstance(p, LLMProvider) 

184 

185 def test_provider_name(self): 

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

187 assert p.provider_name == "deepseek" 

188 

189 def test_default_base_url(self): 

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

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

192 

193 def test_custom_base_url(self): 

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

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

196 

197 def test_default_model(self): 

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

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

200 

201 def test_factory_creates(self): 

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

203 p = create_provider("deepseek") 

204 assert isinstance(p, DeepSeekProvider) 

205 assert p.provider_name == "deepseek" 

206 

207 

208# ── Anthropic Provider ───────────────────────────────────────────── 

209 

210class TestAnthropicProvider: 

211 def test_provider_name(self): 

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

213 assert p.provider_name == "anthropic" 

214 

215 def test_default_model(self): 

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

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

218 

219 def test_default_base_url(self): 

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

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

222 

223 def test_custom_base_url(self): 

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

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

226 

227 def test_headers(self): 

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

229 h = p._headers() 

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

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

232 

233 def test_tools_conversion(self): 

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

235 tools = [ 

236 Tool.from_function("get_weather", "Get weather", { 

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

238 }), 

239 ] 

240 from agentos.llm.anthropic_provider import _tools_to_anthropic 

241 result = _tools_to_anthropic(tools) 

242 assert len(result) == 1 

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

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

245 

246 def test_message_conversion_simple(self): 

247 from agentos.llm.anthropic_provider import _messages_to_anthropic 

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

249 system, api_msgs = _messages_to_anthropic(msgs) 

250 assert system is None 

251 assert len(api_msgs) == 1 

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

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

254 

255 def test_message_conversion_with_system(self): 

256 from agentos.llm.anthropic_provider import _messages_to_anthropic 

257 msgs = [ 

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

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

260 ] 

261 system, api_msgs = _messages_to_anthropic(msgs) 

262 assert system == "You are helpful." 

263 assert len(api_msgs) == 1 

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

265 

266 def test_build_body_includes_tools(self): 

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

268 tools = [Tool.from_function("search", "search the web", { 

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

270 })] 

271 body = p._build_body( 

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

273 temperature=0.5, max_tokens=100, top_p=0.9, 

274 stop=None, tools=tools, tool_choice="auto", 

275 ) 

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

277 assert "tools" in body 

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