sanguo_vnpy_v2/tests/api/test_llm_provider.py

# tests/api/test_llm_provider.py
"""LLM provider 表与 env 解析(C6:TradingAgents ProviderSpec 范式,glm_cn 默认行)."""
import pytest

from sanguo_api.llm.provider import (
    LLMConfig, LLMConfigError, PROVIDERS, chat_endpoint, chat_url,
    messages_url, resolve_config,
)


def _env(**kw):
    base = {
        "SANGUO_LLM_PROVIDER": "",
        "SANGUO_LLM_BASE_URL": "",
        "SANGUO_LLM_MODEL": "",
        "SANGUO_LLM_API_KEY": "sk-test",
    }
    base.update(kw)
    return base


class TestProviderTable:
    def test_glm_cn_default_row(self):
        # 10-08 勘误:glm_cn 指向 coding plan 接入点(anthropic 兼容)——
        # paas/v4 是按量计费端点,coding key 打它=1113 余额不足(VPS 实测)
        spec = PROVIDERS["glm_cn"]
        assert spec.base_url == "https://open.bigmodel.cn/api/anthropic"
        assert spec.default_model == "glm-5.3-flash"
        assert spec.api_key_env == "SANGUO_LLM_API_KEY"
        assert spec.protocol == "anthropic"

    def test_openai_compat_row_requires_env_fill(self):
        spec = PROVIDERS["openai_compat"]
        assert spec.base_url == "" and spec.default_model == ""


class TestResolveConfig:
    def test_defaults_to_glm_cn(self):
        cfg = resolve_config(_env())
        assert (cfg.provider, cfg.model) == ("glm_cn", "glm-5.3-flash")
        assert cfg.base_url == "https://open.bigmodel.cn/api/anthropic"
        assert cfg.protocol == "anthropic"
        assert cfg.api_key == "sk-test"

    def test_env_overrides_win(self):
        cfg = resolve_config(_env(SANGUO_LLM_BASE_URL="http://10.0.0.1:8000/v4/",
                                  SANGUO_LLM_MODEL="glm-5.3"))
        assert cfg.base_url == "http://10.0.0.1:8000/v4/"
        assert cfg.model == "glm-5.3"

    def test_openai_compat_requires_base_url_and_model(self):
        with pytest.raises(LLMConfigError) as e:
            resolve_config(_env(SANGUO_LLM_PROVIDER="openai_compat"))
        msg = str(e.value)
        assert "SANGUO_LLM_BASE_URL" in msg and "SANGUO_LLM_MODEL" in msg

    def test_missing_api_key_lists_env_name(self):
        with pytest.raises(LLMConfigError) as e:
            resolve_config(_env(SANGUO_LLM_API_KEY=""))
        assert "SANGUO_LLM_API_KEY" in str(e.value)


class TestChatUrl:
    def test_appends_chat_completions_without_v1_magic(self):
        # GLM 式 /api/paas/v4/ 不补 /v1(tickflow 归一化教训:补了就 404)
        assert chat_url("https://open.bigmodel.cn/api/paas/v4/") == \
            "https://open.bigmodel.cn/api/paas/v4/chat/completions"
        assert chat_url("http://10.0.0.1:8000/v4") == \
            "http://10.0.0.1:8000/v4/chat/completions"


class TestAnthropicProtocol:
    """10-08 coding plan 端点:glm_cn 默认走 anthropic 兼容协议(/v1/messages)."""

    def test_messages_url_appends_v1_messages(self):
        assert messages_url("https://open.bigmodel.cn/api/anthropic") == \
            "https://open.bigmodel.cn/api/anthropic/v1/messages"
        assert messages_url("https://open.bigmodel.cn/api/anthropic/") == \
            "https://open.bigmodel.cn/api/anthropic/v1/messages"

    def test_chat_endpoint_anthropic_uses_x_api_key(self):
        cfg = LLMConfig("glm_cn", "https://x/api/anthropic", "m", "sk",
                        protocol="anthropic")
        url, headers = chat_endpoint(cfg)
        assert url == "https://x/api/anthropic/v1/messages"
        assert headers == {"x-api-key": "sk",
                           "anthropic-version": "2023-06-01"}

    def test_chat_endpoint_openai_keeps_bearer(self):
        cfg = LLMConfig("openai_compat", "http://x/v4/", "m", "sk")
        url, headers = chat_endpoint(cfg)
        assert url == "http://x/v4/chat/completions"
        assert headers == {"Authorization": "Bearer sk"}

    def test_protocol_env_override_wins(self):
        cfg = resolve_config(_env(SANGUO_LLM_PROTOCOL="openai"))
        assert cfg.protocol == "openai"

    def test_openai_compat_row_defaults_openai_protocol(self):
        assert PROVIDERS["openai_compat"].protocol == "openai"


class TestResolveConfigOverrides:
    """P5.1 三层:DB overrides > env > 默认;overrides=None 零行为变化."""

    def test_override_beats_env_and_default(self):
        cfg = resolve_config(_env(), overrides={
            "provider": "openai_compat", "base_url": "http://x/v1/",
            "model": "m2", "api_key": "sk-db", "timeout": "30"})
        assert (cfg.provider, cfg.base_url, cfg.model, cfg.api_key) == \
            ("openai_compat", "http://x/v1/", "m2", "sk-db")
        assert cfg.timeout == 30.0

    def test_env_fills_when_override_absent(self):
        cfg = resolve_config(_env(SANGUO_LLM_MODEL="env-model"),
                             overrides={"api_key": "sk-db"})
        assert cfg.model == "env-model" and cfg.api_key == "sk-db"

    def test_empty_override_falls_through(self):
        # 空串 override=无此层(设置页"清空"语义由 DELETE 行实现,不靠空串)
        cfg = resolve_config(_env(), overrides={"model": "  "})
        assert cfg.model == "glm-5.3-flash"

    def test_disabled_override_raises(self):
        with pytest.raises(LLMConfigError, match="停用"):
            resolve_config(_env(), overrides={"enabled": "0"})

    def test_timeout_env_fallback_and_bounds(self):
        assert resolve_config(
            _env(SANGUO_LLM_TIMEOUT="15")).timeout == 15.0
        with pytest.raises(LLMConfigError, match="timeout"):
            resolve_config(_env(), overrides={"timeout": "900"})
        with pytest.raises(LLMConfigError, match="非数字"):
            resolve_config(_env(), overrides={"timeout": "abc"})

    def test_no_overrides_backward_compatible(self):
        assert resolve_config(_env()) == resolve_config(_env(), overrides=None)