script-build-and-framework-tmp/agent/tests/test_qwen_runtime_options.py

from __future__ import annotations

from types import SimpleNamespace
from typing import Any

import pytest

from agent.llm.qwen import create_qwen_llm_call


@pytest.mark.asyncio
async def test_qwen_factory_applies_timeout_and_non_thinking_mode(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    client_options: dict[str, Any] = {}
    request_options: dict[str, Any] = {}

    async def create(**kwargs: Any) -> Any:
        request_options.update(kwargs)
        return SimpleNamespace(
            choices=[
                SimpleNamespace(
                    message=SimpleNamespace(content="ok", tool_calls=None),
                    finish_reason="stop",
                )
            ],
            usage=SimpleNamespace(prompt_tokens=3, completion_tokens=2),
        )

    def fake_client(**kwargs: Any) -> Any:
        client_options.update(kwargs)
        return SimpleNamespace(chat=SimpleNamespace(completions=SimpleNamespace(create=create)))

    monkeypatch.setattr("agent.llm.qwen.AsyncOpenAI", fake_client)
    call = create_qwen_llm_call(
        api_key="secret",
        enable_thinking=False,
        request_timeout_seconds=45,
        max_retries=2,
    )

    result = await call(messages=[{"role": "user", "content": "hello"}])

    assert client_options["timeout"] == 45
    assert client_options["max_retries"] == 2
    assert request_options["extra_body"] == {"enable_thinking": False}
    assert result["content"] == "ok"


@pytest.mark.asyncio
async def test_qwen_factory_preserves_call_specific_extra_body(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    request_options: dict[str, Any] = {}

    async def create(**kwargs: Any) -> Any:
        request_options.update(kwargs)
        return SimpleNamespace(
            choices=[
                SimpleNamespace(
                    message=SimpleNamespace(content="ok", tool_calls=None),
                    finish_reason="stop",
                )
            ],
            usage=SimpleNamespace(prompt_tokens=1, completion_tokens=1),
        )

    monkeypatch.setattr(
        "agent.llm.qwen.AsyncOpenAI",
        lambda **_kwargs: SimpleNamespace(
            chat=SimpleNamespace(completions=SimpleNamespace(create=create))
        ),
    )
    call = create_qwen_llm_call(api_key="secret", enable_thinking=False)

    await call(
        messages=[{"role": "user", "content": "hello"}],
        extra_body={"enable_thinking": True, "custom": "kept"},
    )

    assert request_options["extra_body"] == {"enable_thinking": True, "custom": "kept"}