"""Provider 层测试：httpx MockTransport，零网络。 覆盖：OpenAI 兼容请求/响应解析、图片回喂、错误处理；Claude（anthropic SDK） 的消息/图片/tool_result 转换与 tool_use 解析；注册表可用性与无 key 降级。 """ import base64 import json import httpx import pytest from quercus_core.config import Config from quercus_core.providers import create_provider, provider_status from quercus_core.providers.base import ( LLMProvider, Message, ProviderError, ProviderUnavailable, Role, ToolCall, ToolSchema, ) from quercus_core.providers.claude import ClaudeProvider from quercus_core.providers.openai_compat import OpenAICompatProvider from quercus_core.types import PngBytes PNG = PngBytes(b"\x89PNG\r\n\x1a\n" + bytes(range(64))) GET_FRAME_SCHEMA = ToolSchema( name="get_frame", description="取帧", parameters={"type": "object", "properties": {"time": {"type": "object"}}, "required": ["time"]}, read_only=True, ) CLAUDE_MESSAGE = { "id": "msg_01", "type": "message", "role": "assistant", "model": "claude-sonnet-4-5", "content": [], "stop_reason": "end_turn", "stop_sequence": None, "usage": {"input_tokens": 10, "output_tokens": 5}, } # ---- OpenAI 兼容 ---- def test_openai_compat_request_and_parse(): captured: dict = {} def handler(request: httpx.Request) -> httpx.Response: captured["url"] = str(request.url) captured["authorization"] = request.headers.get("authorization") captured["body"] = json.loads(request.content) return httpx.Response( 200, json={ "id": "chatcmpl-1", "object": "chat.completion", "created": 1, "model": "gpt-4o", "choices": [ { "index": 0, "message": { "role": "assistant", "content": "我来调用工具", "tool_calls": [ { "id": "call_1", "type": "function", "function": { "name": "get_frame", "arguments": json.dumps({"time": {"num": 1, "den": 1}}), }, } ], }, "finish_reason": "tool_calls", } ], "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}, }, ) client = httpx.Client(transport=httpx.MockTransport(handler)) provider = OpenAICompatProvider( api_key="sk-test", base_url="https://gw.example.com/v1", model="local-vl", client=client, ) turn = provider.generate( [Message(role=Role.USER, text="hi")], tools=[GET_FRAME_SCHEMA] ) assert captured["url"] == "https://gw.example.com/v1/chat/completions" assert captured["authorization"] == "Bearer sk-test" assert captured["body"]["model"] == "local-vl" assert captured["body"]["tools"][0]["function"]["name"] == "get_frame" assert turn.text == "我来调用工具" assert turn.tool_calls[0].name == "get_frame" assert turn.tool_calls[0].arguments == {"time": {"num": 1, "den": 1}} def test_openai_compat_image_message(): captured: dict = {} def handler(request: httpx.Request) -> httpx.Response: captured["body"] = json.loads(request.content) return httpx.Response(200, json={"choices": [{"message": {"content": "ok"}}]}) client = httpx.Client(transport=httpx.MockTransport(handler)) provider = OpenAICompatProvider(api_key="k", base_url="https://x/v1", client=client) provider.generate([Message(role=Role.USER, text="看图", images=(PNG,))]) content = captured["body"]["messages"][0]["content"] assert content[0] == {"type": "text", "text": "看图"} url = content[1]["image_url"]["url"] assert url.startswith("data:image/png;base64,") assert base64.b64decode(url.split(",", 1)[1]) == PNG def test_openai_compat_tool_result_image_follow_up_user_message(): captured: dict = {} def handler(request: httpx.Request) -> httpx.Response: captured["body"] = json.loads(request.content) return httpx.Response(200, json={"choices": [{"message": {"content": "ok"}}]}) client = httpx.Client(transport=httpx.MockTransport(handler)) provider = OpenAICompatProvider(api_key="k", base_url="https://x/v1", client=client) provider.generate( [ Message(role=Role.TOOL, text="帧", images=(PNG,), tool_call_id="c1", name="get_frame"), ] ) msgs = captured["body"]["messages"] assert msgs[0]["role"] == "tool" assert msgs[0]["tool_call_id"] == "c1" assert msgs[1]["role"] == "user" # 帧图以随后的 user 消息回喂 assert msgs[1]["content"][0]["type"] == "text" assert msgs[1]["content"][1]["image_url"]["url"].startswith("data:image/png;base64,") def test_openai_compat_http_error(): client = httpx.Client( transport=httpx.MockTransport(lambda req: httpx.Response(500, text="boom")) ) provider = OpenAICompatProvider(api_key="k", base_url="https://x/v1", client=client) with pytest.raises(ProviderError): provider.generate([Message(role=Role.USER, text="hi")]) # ---- Claude ---- def test_claude_tool_use_parse(): captured: dict = {} def handler(request: httpx.Request) -> httpx.Response: captured["body"] = json.loads(request.content) return httpx.Response( 200, headers={"content-type": "application/json"}, json={ "id": "msg_01", "type": "message", "role": "assistant", "model": "claude-sonnet-4-5", "content": [ {"type": "text", "text": "我来取帧"}, { "type": "tool_use", "id": "toolu_01", "name": "get_frame", "input": {"time": {"num": 1, "den": 1}, "max_size": {"width": 320, "height": 180}}, }, ], "stop_reason": "tool_use", "stop_sequence": None, "usage": {"input_tokens": 10, "output_tokens": 5}, }, ) client = httpx.Client(transport=httpx.MockTransport(handler)) provider = ClaudeProvider(api_key="sk-test", http_client=client) turn = provider.generate([Message(role=Role.USER, text="hi")]) assert turn.text == "我来取帧" assert len(turn.tool_calls) == 1 assert turn.tool_calls[0].id == "toolu_01" assert turn.tool_calls[0].name == "get_frame" assert turn.tool_calls[0].arguments["max_size"]["width"] == 320 def test_claude_image_message_becomes_image_block(): captured: dict = {} def handler(request: httpx.Request) -> httpx.Response: captured["body"] = json.loads(request.content) return httpx.Response( 200, headers={"content-type": "application/json"}, json=CLAUDE_MESSAGE ) client = httpx.Client(transport=httpx.MockTransport(handler)) provider = ClaudeProvider(api_key="sk-test", http_client=client) provider.generate([Message(role=Role.USER, text="看图", images=(PNG,))]) content = captured["body"]["messages"][0]["content"] assert content[0]["type"] == "text" # 文本块在前 img_block = content[1] assert img_block["type"] == "image" assert img_block["source"]["media_type"] == "image/png" assert base64.b64decode(img_block["source"]["data"]) == PNG def test_claude_tool_result_with_image(): captured: dict = {} def handler(request: httpx.Request) -> httpx.Response: captured["body"] = json.loads(request.content) return httpx.Response( 200, headers={"content-type": "application/json"}, json=CLAUDE_MESSAGE ) client = httpx.Client(transport=httpx.MockTransport(handler)) provider = ClaudeProvider(api_key="sk-test", http_client=client) provider.generate( [ Message( role=Role.ASSISTANT, tool_calls=(ToolCall(id="toolu_01", name="get_frame", arguments={}),), ), Message( role=Role.TOOL, text="帧", images=(PNG,), tool_call_id="toolu_01", name="get_frame", ), ] ) msgs = captured["body"]["messages"] assert msgs[0]["role"] == "assistant" assert msgs[0]["content"][0]["type"] == "tool_use" assert msgs[1]["role"] == "user" # 连续 tool 消息合并为一条带 tool_result 的 user tool_result = msgs[1]["content"][0] assert tool_result["type"] == "tool_result" assert tool_result["tool_use_id"] == "toolu_01" assert tool_result["content"][0]["type"] == "text" assert tool_result["content"][1]["type"] == "image" def test_claude_multiple_tool_results_merged_into_one_user_message(): """agent loop 的典型回喂形态：多条 tool 消息合并成一条 user(tool_result...)。""" captured: dict = {} def handler(request: httpx.Request) -> httpx.Response: captured["body"] = json.loads(request.content) return httpx.Response( 200, headers={"content-type": "application/json"}, json=CLAUDE_MESSAGE ) client = httpx.Client(transport=httpx.MockTransport(handler)) provider = ClaudeProvider(api_key="sk-test", http_client=client) provider.generate( [ Message( role=Role.ASSISTANT, tool_calls=( ToolCall(id="t1", name="place_clip", arguments={"x": 1}), ToolCall(id="t2", name="get_frame", arguments={"t": 1}), ), ), Message(role=Role.TOOL, text="placed", tool_call_id="t1", name="place_clip"), Message(role=Role.TOOL, text="frame", images=(PNG,), tool_call_id="t2", name="get_frame"), ] ) msgs = captured["body"]["messages"] assert len(msgs) == 2 assert msgs[0]["role"] == "assistant" assert [b["type"] for b in msgs[0]["content"]] == ["tool_use", "tool_use"] # 两条 tool_result 合并进同一条 user 消息（保持 assistant 之后紧跟 user） user_content = msgs[1]["content"] assert [b["type"] for b in user_content] == ["tool_result", "tool_result"] assert [b["tool_use_id"] for b in user_content] == ["t1", "t2"] # ---- 注册表 / 无 key 降级 ---- def test_registry_unavailable_without_keys(tmp_path, monkeypatch): monkeypatch.delenv("QUERCUS_ANTHROPIC_API_KEY", raising=False) monkeypatch.delenv("QUERCUS_OPENAI_API_KEY", raising=False) cfg = Config() status = provider_status(cfg) assert status["claude"].available is False assert status["openai_compat"].available is False assert status["claude"].reason with pytest.raises(ProviderUnavailable): create_provider("claude", cfg) with pytest.raises(ProviderUnavailable): create_provider("openai_compat", cfg) def test_registry_create_with_keys(): cfg = Config(anthropic_api_key="sk-ant", openai_api_key="sk-o") status = provider_status(cfg) assert status["claude"].available is True assert status["claude"].model assert status["openai_compat"].available is True mock_client = httpx.Client( transport=httpx.MockTransport(lambda req: httpx.Response(200, json={"choices": [{"message": {"content": "ok"}}]})) ) claude = create_provider("claude", cfg, http_client=mock_client) assert isinstance(claude, ClaudeProvider) openai = create_provider("openai_compat", cfg, http_client=mock_client) assert isinstance(openai, OpenAICompatProvider) assert isinstance(claude, LLMProvider) def test_registry_unknown_provider(): with pytest.raises(ValueError): create_provider("llamacpp", Config())