MoviePilot/tests/test_llm_thinking_options.py

"""
验证各模型提供方的思考选项与客户端参数映射。
"""

import asyncio
from types import SimpleNamespace
from unittest.mock import AsyncMock, Mock

import httpx
import pytest

from app.agent.llm import helper
from app.agent.llm.helper import LLMHelper


@pytest.mark.parametrize(
    ("provider", "model", "thinking_level", "expected"),
    [
        (
            "openai",
            "deepseek-v4.1-flash",
            "high",
            {"reasoning_effort": "high"},
        ),
        (
            "openai",
            "custom-reasoning-model",
            "max",
            {"reasoning_effort": "max"},
        ),
        (
            "chatgpt",
            "claude-sonnet-4-5",
            "medium",
            {"reasoning_effort": "medium"},
        ),
    ],
)
def test_openai_compatible_models_forward_reasoning_effort(
    provider,
    model,
    thinking_level,
    expected,
):
    """
    未知模型目录时，OpenAI-compatible 端点应透传用户选择的思考级别。
    """
    assert (
        LLMHelper._build_thinking_kwargs(
            provider=provider,
            model=model,
            thinking_level=thinking_level,
        )
        == expected
    )


def test_openai_reasoning_effort_respects_model_catalog():
    """
    模型目录声明的 effort 范围应约束超出范围的统一思考级别。
    """
    metadata = {
        "reasoning": True,
        "reasoning_options": [
            {"type": "effort", "values": ["low", "medium", "high"]}
        ],
    }

    assert LLMHelper._build_thinking_kwargs(
        provider="openai",
        model="catalog-model",
        thinking_level="max",
        model_metadata=metadata,
    ) == {"reasoning_effort": "high"}
    assert LLMHelper._build_thinking_kwargs(
        provider="openai",
        model="catalog-model",
        thinking_level="off",
        model_metadata=metadata,
    ) == {}


def test_grok_max_uses_highest_catalog_effort():
    """
    Grok 4.6 目录最高为 xhigh，选择 max 时按已知能力降级。
    """
    metadata = {
        "reasoning": True,
        "reasoning_options": [
            {"type": "effort", "values": ["low", "medium", "high", "xhigh"]}
        ],
    }

    assert LLMHelper._build_thinking_kwargs(
        provider="openai", model="grok-4.6", thinking_level="max", model_metadata=metadata,
    ) == {"reasoning_effort": "xhigh"}


@pytest.fixture
def offline_runtime(monkeypatch):
    """
    替换模型运行配置和客户端构造，记录思考选项且不连接外部服务。
    """
    runtime = SimpleNamespace(resolve_runtime=AsyncMock(return_value={
        "provider_id": "openai",
        "runtime": "openai_compatible",
        "model_id": "gpt-6-luna",
        "api_key": "offline-key",
        "base_url": "https://offline.example/v1",
        "model_metadata": None,
    }))
    monkeypatch.setattr(helper, "_patch_openai_responses_instructions_support", lambda: None)
    monkeypatch.setattr(helper, "_patch_tool_schema_request_support", lambda *_args, **_kwargs: None)
    clients = []

    def reject_request(_request):
        """
        拒绝任何意外的提供商网络调用。
        """
        raise AssertionError("思考级别序列化测试不得发起外部请求")

    def make_client(_proxy, *, async_client=False):
        """
        创建仅使用离线传输器的客户端，并登记退出时的释放责任。
        """
        client_type = httpx.AsyncClient if async_client else httpx.Client
        client = client_type(transport=httpx.MockTransport(reject_request))
        clients.append(client)
        return client

    monkeypatch.setattr(helper, "_build_httpx_client", make_client)
    yield runtime
    for client in clients:
        if isinstance(client, httpx.AsyncClient):
            asyncio.run(client.aclose())
        else:
            client.close()


@pytest.mark.parametrize("model_name", ["gemini-3.8-flash", "models/gemini-3.8-flash", "gemini-3-flash-preview"])
@pytest.mark.parametrize("level", ["off", "minimal", "low", "medium", "high", "max", "xhigh", "auto"])
def test_google_thinking_payload_respects_model_minimum(monkeypatch, offline_runtime, model_name, level) -> None:
    """
    原生 Google 请求对 Gemini 3.8 Flash 使用最低可用的 low，并保留旧模型的 minimal。
    """
    from langchain_google_genai import chat_models

    offline_runtime.resolve_runtime.return_value.update(
        provider_id="google", runtime="google", model_id=model_name, base_url=None,
    )
    monkeypatch.setattr(helper, "_patch_gemini_thought_signature", lambda: None)
    monkeypatch.setattr(chat_models, "Client", Mock(return_value=Mock(aio=None)))

    model = asyncio.run(LLMHelper.get_llm(
        provider="google", model=model_name, api_key="offline-key", use_proxy=False,
        web_search_mode="disabled", thinking_level=level, provider_runtime=offline_runtime,
    ))
    config = model._prepare_params(None).thinking_config

    if level == "auto":
        assert config is None or config.thinking_level is None
        return
    expected = {"off": "minimal", "max": "high", "xhigh": "high"}.get(level, level)
    if "gemini-3.8-flash" in model_name and expected == "minimal":
        expected = "low"
    assert config.thinking_level.value.lower() == expected
    assert config.include_thoughts is False


@pytest.mark.parametrize("protocol", ["chat_completions", "responses"])
@pytest.mark.parametrize("effort", ["max", "xhigh"])
@pytest.mark.parametrize("has_catalog", [False, True])
def test_openai_payload_preserves_selected_reasoning_effort(
    offline_runtime, protocol, effort, has_catalog,
) -> None:
    """
    SDK 实际请求须区分 max 和 xhigh，目录未知或支持两者时均保留用户选择。
    """
    if has_catalog:
        offline_runtime.resolve_runtime.return_value["model_metadata"] = {
            "reasoning": True,
            "reasoning_options": [
                {"type": "effort", "values": ["none", "low", "medium", "high", "xhigh", "max"]}
            ],
        }

    model = asyncio.run(LLMHelper.get_llm(
        provider="openai", model="gpt-6-luna", api_key="offline-key",
        base_url="https://offline.example/v1", use_proxy=False,
        api_protocol=protocol, web_search_mode="disabled", thinking_level=effort,
        provider_runtime=offline_runtime,
    ))
    payload = model._get_request_payload("只回复 OK")

    if protocol == "responses":
        assert payload["reasoning"]["effort"] == effort
    else:
        assert payload["reasoning_effort"] == effort


@pytest.mark.parametrize("protocol", ["chat_completions", "responses"])
def test_grok_payload_uses_highest_catalog_effort(offline_runtime, protocol) -> None:
    """
    两种协议均按 Grok 4.6 的已知能力把 max 降级为 xhigh。
    """
    metadata = {
        "reasoning": True,
        "reasoning_options": [
            {"type": "effort", "values": ["low", "medium", "high", "xhigh"]}
        ],
    }
    offline_runtime.resolve_runtime.return_value.update(
        model_id="grok-4.6", model_metadata=metadata,
    )

    model = asyncio.run(LLMHelper.get_llm(
        provider="openai", model="grok-4.6", api_key="offline-key",
        base_url="https://offline.example/v1", use_proxy=False,
        api_protocol=protocol, web_search_mode="disabled", thinking_level="max",
        provider_runtime=offline_runtime,
    ))
    payload = model._get_request_payload("只回复 OK")

    if protocol == "responses":
        assert payload["reasoning"]["effort"] == "xhigh"
    else:
        assert payload["reasoning_effort"] == "xhigh"


def test_openai_reasoning_effort_skips_known_unsupported_model(monkeypatch):
    """
    模型目录明确不支持思考时应跳过参数并记录原因。
    """
    logger = Mock()
    monkeypatch.setattr("app.agent.llm.helper.logger", logger)

    result = LLMHelper._build_thinking_kwargs(
        provider="openai",
        model="chat-model",
        thinking_level="high",
        model_metadata={"reasoning": False, "reasoning_options": None},
    )

    assert result == {}
    logger.warning.assert_called_once()