gyxx-flow/tests/modules/product_commerce/test_direct_llm.py

from __future__ import annotations

import json

import pytest

from gyxx_flow.modules.product_commerce.direct_llm import (
    DirectLLMConfig,
    DirectLLMConfigurationError,
    DirectLLMResponseError,
    DirectLLMTransientError,
    chat_completion,
    load_direct_llm_config,
)


def _config() -> DirectLLMConfig:
    return DirectLLMConfig(
        endpoint="https://api.minimaxi.com/v1/chat/completions",
        credential="test-only-secret",
        title_model="MiniMax-M3",
        vision_model="MiniMax-M3",
    )


def _response(content: str) -> bytes:
    return json.dumps(
        {
            "choices": [{"message": {"content": content}}],
            "usage": {"prompt_tokens": 3, "completion_tokens": 4},
        }
    ).encode()


def test_direct_chat_uses_provider_endpoint_and_model_without_echoing_secret() -> None:
    observed: dict[str, object] = {}

    def transport(url, headers, body, timeout):
        payload = json.loads(body)
        observed.update(
            url=url,
            model=payload["model"],
            messages=payload["messages"],
            reasoning_split=payload["reasoning_split"],
            timeout=timeout,
            authorization=headers["Authorization"],
        )
        return _response('{"ok":true}')

    content, usage = chat_completion(
        [{"role": "user", "content": "hello"}],
        model="MiniMax-M3",
        timeout=17,
        transport=transport,
        config=_config(),
    )

    assert content == '{"ok":true}'
    assert usage == {"prompt_tokens": 3, "completion_tokens": 4}
    assert observed["url"] == "https://api.minimaxi.com/v1/chat/completions"
    assert observed["model"] == "MiniMax-M3"
    assert observed["reasoning_split"] is True
    assert observed["timeout"] == 17.0
    assert observed["authorization"] == "Bearer test-only-secret"


def test_direct_chat_can_disable_minimax_thinking_for_structured_video_tasks() -> None:
    observed: dict[str, object] = {}

    def transport(_url, _headers, body, _timeout):
        observed.update(json.loads(body))
        return _response("ok")

    chat_completion(
        [{"role": "user", "content": "return one token"}],
        thinking_mode="disabled",
        transport=transport,
        config=_config(),
    )

    assert observed["thinking"] == {"type": "disabled"}


def test_direct_chat_rejects_unknown_thinking_mode() -> None:
    with pytest.raises(DirectLLMConfigurationError, match="thinking mode"):
        chat_completion(
            [{"role": "user", "content": "hello"}],
            thinking_mode="sometimes",
            transport=lambda *_args: _response("unused"),
            config=_config(),
        )


def test_config_reads_shared_minimax_environment(monkeypatch) -> None:
    monkeypatch.delenv("GYXX_DIRECT_LLM_BASE_URL", raising=False)
    monkeypatch.delenv("GYXX_DIRECT_LLM_API_KEY", raising=False)
    monkeypatch.delenv("GYXX_DIRECT_LLM_MODEL", raising=False)
    monkeypatch.setenv("STYLE_ANALYSIS_LLM_BASE_URL", "https://api.minimaxi.com/v1")
    monkeypatch.setenv("STYLE_ANALYSIS_LLM_API_KEY", "test-only-secret")
    monkeypatch.setenv("STYLE_ANALYSIS_LLM_MODEL", "MiniMax-M3")

    config = load_direct_llm_config()

    assert config.endpoint == "https://api.minimaxi.com/v1/chat/completions"
    assert config.title_model == "MiniMax-M3"
    assert config.vision_model == "MiniMax-M3"
    assert "test-only-secret" not in repr(config)


def test_config_rejects_legacy_local_gateway(monkeypatch) -> None:
    monkeypatch.setenv(
        "GYXX_DIRECT_LLM_BASE_URL",
        "http://127.0.0.1:8642/v1",
    )
    monkeypatch.setenv("GYXX_DIRECT_LLM_API_KEY", "test-only-secret")

    with pytest.raises(DirectLLMConfigurationError, match="local gateway"):
        load_direct_llm_config()


def test_invalid_response_is_fail_closed() -> None:
    with pytest.raises(DirectLLMResponseError, match="invalid JSON"):
        chat_completion(
            [{"role": "user", "content": "hello"}],
            transport=lambda *_args: b"not-json",
            config=_config(),
        )


def test_direct_chat_retries_transient_provider_overload(monkeypatch) -> None:
    calls = 0

    def transport(_url, _headers, _body, _timeout):
        nonlocal calls
        calls += 1
        if calls < 3:
            raise DirectLLMTransientError(
                "direct model request failed with HTTP 529"
            )
        return _response("recovered")

    delays: list[float] = []
    monkeypatch.setattr(
        "gyxx_flow.modules.product_commerce.direct_llm.time.sleep",
        delays.append,
    )

    content, _usage = chat_completion(
        [{"role": "user", "content": "hello"}],
        transport=transport,
        config=_config(),
    )

    assert content == "recovered"
    assert calls == 3
    assert delays == [1.0, 3.0]


def test_direct_chat_preserves_transient_error_after_bounded_retries(
    monkeypatch,
) -> None:
    calls = 0

    def transport(_url, _headers, _body, _timeout):
        nonlocal calls
        calls += 1
        raise DirectLLMTransientError(
            "direct model request failed with HTTP 529"
        )

    monkeypatch.setattr(
        "gyxx_flow.modules.product_commerce.direct_llm.time.sleep",
        lambda _seconds: None,
    )

    with pytest.raises(DirectLLMTransientError, match="HTTP 529"):
        chat_completion(
            [{"role": "user", "content": "hello"}],
            transport=transport,
            config=_config(),
        )

    assert calls == 3