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 create_provider, 

12 OpenAIProvider, 

13 DeepSeekProvider, 

14 AnthropicProvider, 

15 LLMProvider, 

16 CompletionUsage, 

17 TokenUsage, 

18 Message, 

19 MessageRole, 

20 StreamChunk, 

21 Tool, 

22 ToolCall, 

23 ToolParameter, 

24) 

25 

26 

27# ── Base Types ─────────────────────────────────────────────────── 

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 

70class TestToolParameter: 

71 def test_basic_schema(self): 

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

73 s = p.as_schema() 

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

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

76 

77 def test_with_enum(self): 

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

79 s = p.as_schema() 

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

81 

82 

83class TestTool: 

84 def test_from_function(self): 

85 t = Tool.from_function( 

86 "get_weather", 

87 "Get weather for a city", 

88 { 

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

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

91 }, 

92 required=["city"], 

93 ) 

94 schema = t.as_schema() 

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

96 fn = schema["function"] 

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

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

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

100 

101 def test_to_openai_format(self): 

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

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

104 }) 

105 schema = t.as_schema() 

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

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

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

109 

110 

111class TestToolCall: 

112 def test_create_and_parse(self): 

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

114 assert tc.id == "call_1" 

115 assert tc.name == "add" 

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

117 

118 def test_empty_arguments(self): 

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

120 assert tc.parsed_arguments == {} 

121 

122 

123# ── StreamChunk ──────────────────────────────────────────────────── 

124 

125class TestStreamChunk: 

126 def test_defaults(self): 

127 c = StreamChunk() 

128 assert c.content == "" 

129 assert c.finish_reason is None 

130 

131 def test_with_content(self): 

132 c = StreamChunk(content="hello") 

133 assert c.content == "hello" 

134 

135 

136# ── Factory ──────────────────────────────────────────────────────── 

137 

138class TestCreateProvider: 

139 def test_openai_default(self): 

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

141 p = create_provider("openai") 

142 assert p.provider_name == "openai" 

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

144 

145 def test_deepseek_default(self): 

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

147 p = create_provider("deepseek") 

148 assert p.provider_name == "deepseek" 

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

150 assert "deepseek.com" in p.base_url 

151 

152 def test_anthropic_default(self): 

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

154 p = create_provider("anthropic") 

155 assert p.provider_name == "anthropic" 

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

157 

158 def test_unknown_provider(self): 

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

160 create_provider("nonexistent") 

161 

162 def test_api_key_env_openai(self): 

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

164 p = create_provider("openai") 

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

166 

167 def test_custom_model(self): 

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

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

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

171 

172 

173# ── DeepSeek Provider ────────────────────────────────────────────── 

174 

175class TestDeepSeekProvider: 

176 def test_is_openai_subclass(self): 

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

178 assert isinstance(p, OpenAIProvider) 

179 assert isinstance(p, LLMProvider) 

180 

181 def test_provider_name(self): 

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

183 assert p.provider_name == "deepseek" 

184 

185 def test_default_base_url(self): 

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

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

188 

189 def test_custom_base_url(self): 

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

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

192 

193 def test_default_model(self): 

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

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

196 

197 def test_factory_creates(self): 

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

199 p = create_provider("deepseek") 

200 assert isinstance(p, DeepSeekProvider) 

201 assert p.provider_name == "deepseek" 

202 

203 

204# ── Anthropic Provider ───────────────────────────────────────────── 

205 

206class TestAnthropicProvider: 

207 def test_provider_name(self): 

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

209 assert p.provider_name == "anthropic" 

210 

211 def test_default_model(self): 

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

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

214 

215 def test_default_base_url(self): 

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

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

218 

219 def test_custom_base_url(self): 

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

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

222 

223 def test_headers(self): 

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

225 h = p._headers() 

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

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

228 

229 def test_tools_conversion(self): 

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

231 tools = [ 

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

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

234 }), 

235 ] 

236 from agentos.llm.anthropic_provider import _tools_to_anthropic 

237 result = _tools_to_anthropic(tools) 

238 assert len(result) == 1 

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

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

241 

242 def test_message_conversion_simple(self): 

243 from agentos.llm.anthropic_provider import _messages_to_anthropic 

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

245 system, api_msgs = _messages_to_anthropic(msgs) 

246 assert system is None 

247 assert len(api_msgs) == 1 

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

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

250 

251 def test_message_conversion_with_system(self): 

252 from agentos.llm.anthropic_provider import _messages_to_anthropic 

253 msgs = [ 

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

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

256 ] 

257 system, api_msgs = _messages_to_anthropic(msgs) 

258 assert system == "You are helpful." 

259 assert len(api_msgs) == 1 

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

261 

262 def test_build_body_includes_tools(self): 

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

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

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

266 })] 

267 body = p._build_body( 

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

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

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

271 ) 

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

273 assert "tools" in body 

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