zhifu-platform/tests/test_model_parameters.py

import asyncio
import json
import secrets
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from queue import SimpleQueue
from threading import Thread

import httpx
import pytest
from langchain_openai import ChatOpenAI
from openai import APIConnectionError

from zhifu.core.model_parameters import chat_parameters


class Receiver(BaseHTTPRequestHandler):
    def do_POST(self):
        # 标准 HTTP 服务读取 SDK 实际发送的 JSON，然后关闭连接，不生成模型响应。
        body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        self.server.received.put((self.path, body))
        self.close_connection = True

    def log_message(self, format, *args):
        pass


@pytest.mark.parametrize("base_url,expected_field", [
    ("https://api.deepseek.com", "max_tokens"),
    ("https://api.deepseek.com/v1", "max_tokens"),
    ("https://api.openai.com/v1", "max_completion_tokens"),
    ("https://api.deepseek.com.example.org/v1", "max_completion_tokens"),
])
@pytest.mark.parametrize("with_tools", [False, True])
async def test_sdk_sends_bounded_parameters_to_real_local_receiver(
    base_url, expected_field, with_tools,
):
    server = ThreadingHTTPServer(("127.0.0.1", 0), Receiver)
    server.received = SimpleQueue()
    thread = Thread(target=server.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True)
    thread.start()
    # 参数按供应商 URL 构建，传输仅连接本地接收器，整个测试不访问模型供应商。
    client = ChatOpenAI(
        model="deepseek-flash", base_url=f"http://127.0.0.1:{server.server_port}/v1",
        api_key=secrets.token_urlsafe(32), max_retries=0, timeout=2, use_responses_api=False,
        **chat_parameters(base_url, 256, {"thinking": {"type": "disabled"}}),
        http_client=httpx.Client(timeout=2, trust_env=False),
        http_async_client=httpx.AsyncClient(timeout=2, trust_env=False),
    )
    try:
        runnable = client.bind_tools([{
            "type": "function", "function": {
                "name": "get_ticket", "description": "查询当前授权工单",
                "parameters": {"type": "object", "properties": {"ticket_id": {
                    "type": "string"}}, "required": ["ticket_id"]},
            },
        }]) if with_tools else client
        async with asyncio.timeout(5):
            with pytest.raises(APIConnectionError):
                await runnable.ainvoke("请说明当前工单状态。")
        assert server.received.qsize() == 1
        path, body = server.received.get_nowait()
        assert path == "/v1/chat/completions"
        assert body[expected_field] == 256
        alternative = "max_completion_tokens" if expected_field == "max_tokens" else "max_tokens"
        assert alternative not in body
        assert "max_output_tokens" not in body
        assert body["thinking"] == {"type": "disabled"}
        assert bool(body.get("tools")) == with_tools
    finally:
        await client.root_async_client.close()
        client.root_client.close()
        await asyncio.to_thread(server.shutdown)
        await asyncio.to_thread(thread.join, 2)
        server.server_close()


@pytest.mark.parametrize("name", ["max_tokens", "max_completion_tokens", "max_output_tokens"])
@pytest.mark.parametrize("value", [512, 0, None, True, "256"])
def test_extra_parameters_cannot_override_output_limit(name, value):
    with pytest.raises(ValueError, match="输出上限冲突"):
        chat_parameters("https://api.deepseek.com/v1", 256, {name: value})


def test_equal_limit_aliases_preserve_configuration_and_use_provider_field():
    body = {"max_completion_tokens": 256, "max_tokens": 256,
            "thinking": {"type": "disabled"}}
    assert chat_parameters("https://API.DEEPSEEK.COM/v1", 256, body) == {
        "extra_body": {"max_tokens": 256, "thinking": {"type": "disabled"}}}
    assert body["max_completion_tokens"] == 256
    assert chat_parameters("https://api.deepseek.com@other.example/v1", 256, body) == {
        "max_tokens": 256, "extra_body": {"thinking": {"type": "disabled"}}}