mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-09-06 16:07:01 +08:00
refactor(api): LLM 提供商管理端点收敛为通用 manage 接口，端点层零特色
- schemas 新增 LlmProviderAction 公共动作词汇表
- LLMProviderManager 新增 provider_manage 统一入口，默认值填充、
API Key 豁免判断、密钥脱敏与错误归因改写全部下沉封闭，
不再硬编码 chatgpt/github-copilot 等产品名
- endpoint 收敛为 POST /llm/manage（ManageRequest 透传）；
OAuth 回跳地址由端点按具名回调路由统一构造后注入动作参数，
/provider-auth/callback/{provider_id} 因浏览器回跳协议约束保留具名路由
- 新增 14 项 provider_manage 契约守护测试，
替换原针对端点函数的两个旧测试文件
This commit is contained in:
jxxghp
@@ -1,77 +0,0 @@
|
||||
import asyncio
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
from app.api.endpoints import llm as llm_endpoint
|
||||
|
||||
|
||||
def test_llm_test_maps_internal_model_dump_error_to_base_url_hint():
|
||||
"""LLM 测试遇到 SDK 内部响应解析错误时应提示检查基础地址。"""
|
||||
with patch.object(llm_endpoint.settings, "AI_AGENT_ENABLE", True), patch.object(
|
||||
llm_endpoint.settings, "LLM_PROVIDER", "openai"
|
||||
), patch.object(llm_endpoint.settings, "LLM_MODEL", "gpt-4o-mini"), patch.object(
|
||||
llm_endpoint.settings, "LLM_API_KEY", "sk-test"
|
||||
), patch.object(
|
||||
llm_endpoint.settings, "LLM_BASE_URL", "https://example.com/not-api"
|
||||
), patch.object(
|
||||
llm_endpoint.LLMHelper,
|
||||
"test_current_settings",
|
||||
AsyncMock(side_effect=RuntimeError("'str' object has no attribute 'model_dump'")),
|
||||
create=True,
|
||||
):
|
||||
resp = asyncio.run(llm_endpoint.llm_test(_="token"))
|
||||
|
||||
assert not resp.success
|
||||
assert "基础地址" in resp.message
|
||||
assert "API Base URL" in resp.message
|
||||
assert "model_dump" not in resp.message
|
||||
|
||||
|
||||
def test_llm_test_maps_internal_private_attribute_error_to_base_url_hint():
|
||||
"""LLM 测试遇到 SDK 内部属性错误时应提示检查基础地址。"""
|
||||
with patch.object(llm_endpoint.settings, "AI_AGENT_ENABLE", True), patch.object(
|
||||
llm_endpoint.settings, "LLM_PROVIDER", "openai"
|
||||
), patch.object(llm_endpoint.settings, "LLM_MODEL", "gpt-4o-mini"), patch.object(
|
||||
llm_endpoint.settings, "LLM_API_KEY", "sk-test"
|
||||
), patch.object(
|
||||
llm_endpoint.LLMHelper,
|
||||
"test_current_settings",
|
||||
AsyncMock(
|
||||
side_effect=AttributeError(
|
||||
"'str' object has no attribute '_set_private_attributes'"
|
||||
)
|
||||
),
|
||||
create=True,
|
||||
):
|
||||
resp = asyncio.run(llm_endpoint.llm_test(_="token"))
|
||||
|
||||
assert not resp.success
|
||||
assert "基础地址" in resp.message
|
||||
assert "API Base URL" in resp.message
|
||||
assert "_set_private_attributes" not in resp.message
|
||||
|
||||
|
||||
def test_llm_models_maps_internal_private_attribute_error_to_base_url_hint():
|
||||
"""LLM 模型列表遇到 SDK 内部属性错误时应提示检查基础地址。"""
|
||||
with patch.object(
|
||||
llm_endpoint.LLMHelper,
|
||||
"get_models",
|
||||
AsyncMock(
|
||||
side_effect=AttributeError(
|
||||
"'str' object has no attribute '_set_private_attributes'"
|
||||
)
|
||||
),
|
||||
create=True,
|
||||
):
|
||||
resp = asyncio.run(
|
||||
llm_endpoint.get_llm_models(
|
||||
provider="openai",
|
||||
api_key="sk-test",
|
||||
base_url="https://example.com",
|
||||
_="token",
|
||||
)
|
||||
)
|
||||
|
||||
assert not resp.success
|
||||
assert "基础地址" in resp.message
|
||||
assert "API Base URL" in resp.message
|
||||
assert "_set_private_attributes" not in resp.message
|
||||
@@ -0,0 +1,252 @@
|
||||
"""
|
||||
LLM 提供商通用管理契约（provider_manage）守护测试
|
||||
|
||||
验证通用模式的三条核心性质：
|
||||
1. 动作词汇表由 schemas 契约层统一定义，未支持动作返回统一错误结构
|
||||
2. 动作标识兼容枚举与原始字符串，统一返回 {"success", "message", "data"}
|
||||
3. 端点层零提供商特色：ManageRequest 原样透传，默认值填充/校验/脱敏封闭在 Manager 内
|
||||
"""
|
||||
import asyncio
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from app import schemas
|
||||
from app.agent.llm.helper import LLMHelper
|
||||
from app.agent.llm.provider import LLMProviderManager
|
||||
from app.runtime.config import settings
|
||||
from app.schemas.types import LlmProviderAction
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def manager():
|
||||
return LLMProviderManager()
|
||||
|
||||
|
||||
def test_provider_manage_rejects_unknown_action(manager):
|
||||
"""动作词汇表之外的请求返回统一错误结构。"""
|
||||
result = asyncio.run(manager.provider_manage("openai", "not_an_action"))
|
||||
assert result["success"] is False
|
||||
assert "不支持" in result["message"]
|
||||
assert "data" in result
|
||||
|
||||
|
||||
def test_provider_manage_accepts_enum_and_string_action(manager, monkeypatch):
|
||||
"""动作标识兼容枚举对象与原始字符串，两种形式等价。"""
|
||||
clear_mock = AsyncMock()
|
||||
monkeypatch.setattr(manager, "clear_auth", clear_mock)
|
||||
for action in (LlmProviderAction.DISCONNECT, "disconnect"):
|
||||
result = asyncio.run(manager.provider_manage("openai", action))
|
||||
assert result["success"] is True
|
||||
assert result["data"] is None
|
||||
assert clear_mock.await_count == 2
|
||||
|
||||
|
||||
def test_provider_manage_test_requires_ai_agent_enabled(manager):
|
||||
"""智能助手未启用时测试动作返回提示。"""
|
||||
with patch.object(settings, "AI_AGENT_ENABLE", False):
|
||||
result = asyncio.run(manager.provider_manage("deepseek", "test"))
|
||||
assert result["success"] is False
|
||||
assert result["message"] == "请先启用智能助手"
|
||||
|
||||
|
||||
def test_provider_manage_test_requires_model(manager):
|
||||
"""未配置模型时测试动作返回提示。"""
|
||||
with patch.object(settings, "AI_AGENT_ENABLE", True), patch.object(
|
||||
settings, "LLM_API_KEY", "sk-test"
|
||||
), patch.object(settings, "LLM_MODEL", ""):
|
||||
result = asyncio.run(manager.provider_manage("deepseek", "test"))
|
||||
assert result["success"] is False
|
||||
assert result["message"] == "请先配置 LLM 模型"
|
||||
|
||||
|
||||
def test_provider_manage_test_requires_api_key(manager, monkeypatch):
|
||||
"""无 OAuth 授权方式且无已保存凭据的提供商必须配置 API Key。"""
|
||||
monkeypatch.setattr(manager, "get_saved_auth", lambda provider_id: None)
|
||||
with patch.object(settings, "AI_AGENT_ENABLE", True), patch.object(
|
||||
settings, "LLM_API_KEY", None
|
||||
), patch.object(settings, "LLM_MODEL", "deepseek-chat"):
|
||||
result = asyncio.run(manager.provider_manage("deepseek", "test"))
|
||||
assert result["success"] is False
|
||||
assert result["message"] == "请先配置 LLM API Key"
|
||||
assert result["data"]["model"] == "deepseek-chat"
|
||||
|
||||
|
||||
def test_provider_manage_test_exempts_oauth_providers_from_api_key(manager, monkeypatch):
|
||||
"""支持 OAuth 授权的提供商无需 API Key，端点与 Manager 均不硬编码提供商名。"""
|
||||
monkeypatch.setattr(manager, "get_saved_auth", lambda provider_id: None)
|
||||
monkeypatch.setattr(
|
||||
manager, "get_provider", lambda provider_id: SimpleNamespace(oauth_methods=("browser_oauth",))
|
||||
)
|
||||
test_mock = AsyncMock(return_value={"provider": "chatgpt", "model": "gpt-4o", "reply_preview": "OK"})
|
||||
with patch.object(settings, "AI_AGENT_ENABLE", True), patch.object(
|
||||
settings, "LLM_API_KEY", None
|
||||
), patch.object(LLMHelper, "test_current_settings", test_mock):
|
||||
result = asyncio.run(
|
||||
manager.provider_manage("chatgpt", "test", model="gpt-4o")
|
||||
)
|
||||
assert result["success"] is True
|
||||
test_mock.assert_awaited_once()
|
||||
|
||||
|
||||
def test_provider_manage_test_returns_reply_preview(manager, monkeypatch):
|
||||
"""测试成功时返回模型响应预览，显式参数优先于已保存配置。"""
|
||||
monkeypatch.setattr(manager, "get_saved_auth", lambda provider_id: None)
|
||||
test_mock = AsyncMock(
|
||||
return_value={"provider": "openai", "model": "gpt-4.1-mini", "duration_ms": 123, "reply_preview": "OK"}
|
||||
)
|
||||
with patch.object(settings, "AI_AGENT_ENABLE", False), patch.object(
|
||||
LLMHelper, "test_current_settings", test_mock
|
||||
):
|
||||
result = asyncio.run(
|
||||
manager.provider_manage(
|
||||
"openai",
|
||||
"test",
|
||||
enabled=True,
|
||||
model="gpt-4.1-mini",
|
||||
thinking_level="high",
|
||||
api_key="sk-live",
|
||||
base_url="https://example.com/v1",
|
||||
use_proxy=False,
|
||||
)
|
||||
)
|
||||
test_mock.assert_awaited_once_with(
|
||||
provider="openai",
|
||||
model="gpt-4.1-mini",
|
||||
thinking_level="high",
|
||||
api_key="sk-live",
|
||||
base_url="https://example.com/v1",
|
||||
base_url_preset=None,
|
||||
user_agent=None,
|
||||
use_proxy=False,
|
||||
api_protocol=None,
|
||||
web_search_mode=None,
|
||||
)
|
||||
assert result["success"] is True
|
||||
assert result["data"]["reply_preview"] == "OK"
|
||||
|
||||
|
||||
def test_provider_manage_test_rejects_empty_reply(manager, monkeypatch):
|
||||
"""模型响应为空时返回失败但保留结果详情。"""
|
||||
monkeypatch.setattr(manager, "get_saved_auth", lambda provider_id: None)
|
||||
with patch.object(settings, "AI_AGENT_ENABLE", True), patch.object(
|
||||
LLMHelper,
|
||||
"test_current_settings",
|
||||
AsyncMock(return_value={"provider": "deepseek", "model": "deepseek-chat", "duration_ms": 12}),
|
||||
):
|
||||
result = asyncio.run(
|
||||
manager.provider_manage("deepseek", "test", model="deepseek-chat", api_key="sk-test")
|
||||
)
|
||||
assert result["success"] is False
|
||||
assert result["message"] == "模型响应为空"
|
||||
assert result["data"]["duration_ms"] == 12
|
||||
|
||||
|
||||
def test_provider_manage_test_maps_timeout_error(manager, monkeypatch):
|
||||
"""调用超时返回统一的超时提示。"""
|
||||
monkeypatch.setattr(manager, "get_saved_auth", lambda provider_id: None)
|
||||
with patch.object(settings, "AI_AGENT_ENABLE", True), patch.object(
|
||||
LLMHelper,
|
||||
"test_current_settings",
|
||||
AsyncMock(side_effect=TimeoutError("request timed out")),
|
||||
):
|
||||
result = asyncio.run(
|
||||
manager.provider_manage("deepseek", "test", model="deepseek-chat", api_key="sk-test")
|
||||
)
|
||||
assert result["success"] is False
|
||||
assert result["message"] == "LLM 调用超时"
|
||||
|
||||
|
||||
def test_provider_manage_test_sanitizes_error_message(manager, monkeypatch):
|
||||
"""错误信息中的密钥与授权头必须脱敏。"""
|
||||
monkeypatch.setattr(manager, "get_saved_auth", lambda provider_id: None)
|
||||
raw_error = (
|
||||
"request failed api_key=sk-secret "
|
||||
"Authorization: Bearer sk-secret "
|
||||
"base error sk-secret"
|
||||
)
|
||||
with patch.object(settings, "AI_AGENT_ENABLE", True), patch.object(
|
||||
LLMHelper, "test_current_settings", AsyncMock(side_effect=RuntimeError(raw_error))
|
||||
):
|
||||
result = asyncio.run(
|
||||
manager.provider_manage("deepseek", "test", model="deepseek-chat", api_key="sk-secret")
|
||||
)
|
||||
assert result["success"] is False
|
||||
assert "sk-secret" not in result["message"]
|
||||
assert "Authorization: Bearer" not in result["message"]
|
||||
assert "***" in result["message"]
|
||||
|
||||
|
||||
def test_provider_manage_test_maps_internal_error_to_base_url_hint(manager, monkeypatch):
|
||||
"""SDK 内部响应解析错误应改写为可定位的基础地址提示。"""
|
||||
monkeypatch.setattr(manager, "get_saved_auth", lambda provider_id: None)
|
||||
with patch.object(settings, "AI_AGENT_ENABLE", True), patch.object(
|
||||
LLMHelper,
|
||||
"test_current_settings",
|
||||
AsyncMock(side_effect=RuntimeError("'str' object has no attribute 'model_dump'")),
|
||||
):
|
||||
result = asyncio.run(
|
||||
manager.provider_manage("openai", "test", model="gpt-4o-mini", api_key="sk-test")
|
||||
)
|
||||
assert result["success"] is False
|
||||
assert "基础地址" in result["message"]
|
||||
assert "API Base URL" in result["message"]
|
||||
assert "model_dump" not in result["message"]
|
||||
|
||||
|
||||
def test_provider_manage_list_models_sanitizes_base_url_hint(manager):
|
||||
"""模型列表查询遇到 SDK 内部错误时同样给出基础地址提示。"""
|
||||
with patch.object(
|
||||
LLMHelper,
|
||||
"get_models",
|
||||
AsyncMock(side_effect=AttributeError("'str' object has no attribute '_set_private_attributes'")),
|
||||
):
|
||||
result = asyncio.run(
|
||||
manager.provider_manage("openai", "list_models", api_key="sk-test", base_url="https://example.com")
|
||||
)
|
||||
assert result["success"] is False
|
||||
assert "基础地址" in result["message"]
|
||||
assert "API Base URL" in result["message"]
|
||||
assert "_set_private_attributes" not in result["message"]
|
||||
|
||||
|
||||
def test_provider_manage_list_models_returns_catalog_with_auth_status(manager):
|
||||
"""模型目录查询成功时附带授权状态摘要。"""
|
||||
models = [{"id": "gpt-4o"}]
|
||||
with patch.object(LLMHelper, "get_models", AsyncMock(return_value=models)), patch.object(
|
||||
manager, "get_auth_status", lambda provider_id: {"connected": False}
|
||||
):
|
||||
result = asyncio.run(manager.provider_manage("openai", "list_models"))
|
||||
assert result["success"] is True
|
||||
assert result["data"]["provider"] == "openai"
|
||||
assert result["data"]["models"] == models
|
||||
assert result["data"]["auth_status"] == {"connected": False}
|
||||
|
||||
|
||||
def test_llm_manage_endpoint_passes_through_manage_request(monkeypatch):
|
||||
"""端点仅透传 ManageRequest，并按具名回调路由注入 OAuth 回跳地址。"""
|
||||
from app.api.endpoints import llm as llm_endpoint
|
||||
|
||||
captured = {}
|
||||
|
||||
async def fake_manage(self, provider, action, **params):
|
||||
captured["provider"] = provider
|
||||
captured["action"] = action
|
||||
captured["params"] = params
|
||||
return {"success": True, "message": "", "data": {"ok": True}}
|
||||
|
||||
monkeypatch.setattr(LLMProviderManager, "provider_manage", fake_manage)
|
||||
request = SimpleNamespace(
|
||||
url_for=lambda name, **kwargs: f"https://host/api/v1/llm/provider-auth/callback/{kwargs['provider_id']}"
|
||||
)
|
||||
payload = schemas.ManageRequest(target="chatgpt", action="start_auth", params={"method": "browser_oauth"})
|
||||
|
||||
resp = asyncio.run(llm_endpoint.manage_provider(request, payload, _="token"))
|
||||
|
||||
assert resp.success is True
|
||||
assert resp.data == {"ok": True}
|
||||
assert captured["provider"] == "chatgpt"
|
||||
assert captured["action"] == "start_auth"
|
||||
assert captured["params"]["method"] == "browser_oauth"
|
||||
assert captured["params"]["callback_url"].endswith("/callback/chatgpt")
|
||||
@@ -1,295 +0,0 @@
|
||||
import asyncio
|
||||
import unittest
|
||||
from types import ModuleType
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
from app.testing import stub_modules
|
||||
|
||||
|
||||
def _stub(name: str, **attrs) -> tuple:
|
||||
"""构造带指定属性的占位模块，返回 ``(模块名, 模块)`` 供 :func:`stub_modules` 使用。"""
|
||||
module = ModuleType(name)
|
||||
for key, value in attrs.items():
|
||||
setattr(module, key, value)
|
||||
return name, module
|
||||
|
||||
|
||||
class _Dummy:
|
||||
def __init__(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
def __getattr__(self, _name):
|
||||
return lambda *args, **kwargs: None
|
||||
|
||||
|
||||
class _DummyError(Exception):
|
||||
def __init__(self, message="", duration_ms=None):
|
||||
super().__init__(message)
|
||||
self.duration_ms = duration_ms
|
||||
|
||||
|
||||
# 在 import 期用占位模块替换重依赖/外部模块，import 完由 stub_modules 精确还原，避免污染其它用例
|
||||
_STUB_MODULES = dict([
|
||||
_stub("pillow_avif"),
|
||||
_stub("aiofiles"),
|
||||
_stub("psutil"),
|
||||
_stub("app.application.site.sites", SitesHelper=_Dummy),
|
||||
_stub("app.chain.mediaserver", MediaServerChain=_Dummy),
|
||||
_stub("app.chain.search", SearchChain=_Dummy),
|
||||
_stub("app.chain.system", SystemChain=_Dummy),
|
||||
_stub("app.agent.llm", LLMHelper=_Dummy, LLMProviderManager=_Dummy,
|
||||
LLMTestError=_DummyError, LLMTestTimeout=_DummyError,
|
||||
render_auth_result_html=lambda success, message: message),
|
||||
_stub("app.runtime.events", eventmanager=_Dummy(), Event=_Dummy, EventManager=_Dummy),
|
||||
_stub("app.domain.metainfo", MetaInfo=_Dummy),
|
||||
_stub("app.runtime.extensions.module_manager", ModuleManager=_Dummy),
|
||||
_stub("app.application.security.access", verify_apitoken=_Dummy, verify_resource_token=_Dummy, verify_token=_Dummy),
|
||||
_stub("app.db.models", User=_Dummy),
|
||||
_stub("app.db.oper.systemconfig", SystemConfigOper=_Dummy),
|
||||
_stub("app.api.deps", get_current_active_superuser=_Dummy,
|
||||
get_current_active_superuser_async=_Dummy, get_current_active_user_async=_Dummy),
|
||||
_stub("app.application.mediaserver", MediaServerHelper=_Dummy),
|
||||
_stub("app.application.messaging.message", MessageHelper=_Dummy),
|
||||
_stub("app.runtime.progress", ProgressHelper=_Dummy),
|
||||
_stub("app.application.filter", RuleHelper=_Dummy),
|
||||
_stub("app.adapters.external.server", MoviePilotServerHelper=_Dummy),
|
||||
_stub("app.runtime.state", SystemHelper=_Dummy),
|
||||
_stub("app.application.image", ImageHelper=_Dummy),
|
||||
_stub("app.scheduler", Scheduler=_Dummy),
|
||||
_stub("app.runtime.log", logger=_Dummy(), log_settings=_Dummy(),
|
||||
LogConfigModel=type("LogConfigModel", (), {})),
|
||||
_stub("app.foundation.crypto", HashUtils=_Dummy),
|
||||
_stub("app.adapters.network.http", RequestUtils=_Dummy, AsyncRequestUtils=_Dummy),
|
||||
_stub("version", APP_VERSION="test"),
|
||||
])
|
||||
|
||||
with stub_modules(_STUB_MODULES):
|
||||
from app.api.endpoints import llm as system_endpoint
|
||||
|
||||
|
||||
class LlmTestEndpointTest(unittest.TestCase):
|
||||
def test_llm_test_requires_ai_agent_enabled(self):
|
||||
with patch.object(system_endpoint.settings, "AI_AGENT_ENABLE", False):
|
||||
resp = asyncio.run(system_endpoint.llm_test(_="token"))
|
||||
|
||||
self.assertFalse(resp.success)
|
||||
self.assertEqual(resp.message, "请先启用智能助手")
|
||||
|
||||
def test_llm_test_requires_api_key(self):
|
||||
with patch.object(system_endpoint.settings, "AI_AGENT_ENABLE", True), patch.object(
|
||||
system_endpoint.settings, "LLM_API_KEY", None
|
||||
), patch.object(system_endpoint.settings, "LLM_MODEL", "deepseek-chat"):
|
||||
resp = asyncio.run(system_endpoint.llm_test(_="token"))
|
||||
|
||||
self.assertFalse(resp.success)
|
||||
self.assertEqual(resp.message, "请先配置 LLM API Key")
|
||||
self.assertEqual(resp.data["model"], "deepseek-chat")
|
||||
|
||||
def test_llm_test_requires_model(self):
|
||||
with patch.object(system_endpoint.settings, "AI_AGENT_ENABLE", True), patch.object(
|
||||
system_endpoint.settings, "LLM_API_KEY", "sk-test"
|
||||
), patch.object(system_endpoint.settings, "LLM_MODEL", ""):
|
||||
resp = asyncio.run(system_endpoint.llm_test(_="token"))
|
||||
|
||||
self.assertFalse(resp.success)
|
||||
self.assertEqual(resp.message, "请先配置 LLM 模型")
|
||||
|
||||
def test_llm_test_returns_successful_reply_preview(self):
|
||||
llm_test_mock = AsyncMock(
|
||||
return_value={
|
||||
"provider": "deepseek",
|
||||
"model": "deepseek-chat",
|
||||
"duration_ms": 321,
|
||||
"reply_preview": "OK",
|
||||
}
|
||||
)
|
||||
with patch.object(system_endpoint.settings, "AI_AGENT_ENABLE", True), patch.object(
|
||||
system_endpoint.settings, "LLM_PROVIDER", "deepseek"
|
||||
), patch.object(system_endpoint.settings, "LLM_MODEL", "deepseek-chat"), patch.object(
|
||||
system_endpoint.settings, "LLM_THINKING_LEVEL", "max"
|
||||
), patch.object(
|
||||
system_endpoint.settings, "LLM_API_KEY", "sk-test"
|
||||
), patch.object(
|
||||
system_endpoint.settings, "LLM_BASE_URL", "https://api.deepseek.com"
|
||||
), patch.object(
|
||||
system_endpoint.settings, "LLM_BASE_URL_PRESET", "deepseek-default"
|
||||
), patch.object(
|
||||
system_endpoint.settings, "LLM_USER_AGENT", "MoviePilot-Test/1.0"
|
||||
), patch.object(
|
||||
system_endpoint.settings, "LLM_USE_PROXY", True
|
||||
), patch.object(
|
||||
system_endpoint.settings, "LLM_API_PROTOCOL", "responses"
|
||||
), patch.object(
|
||||
system_endpoint.LLMHelper,
|
||||
"test_current_settings",
|
||||
llm_test_mock,
|
||||
create=True,
|
||||
):
|
||||
resp = asyncio.run(system_endpoint.llm_test(_="token"))
|
||||
|
||||
llm_test_mock.assert_awaited_once_with(
|
||||
provider="deepseek",
|
||||
model="deepseek-chat",
|
||||
thinking_level="max",
|
||||
api_key="sk-test",
|
||||
base_url="https://api.deepseek.com",
|
||||
base_url_preset="deepseek-default",
|
||||
user_agent="MoviePilot-Test/1.0",
|
||||
use_proxy=True,
|
||||
api_protocol="responses",
|
||||
web_search_mode="local",
|
||||
)
|
||||
self.assertTrue(resp.success)
|
||||
self.assertEqual(resp.data["provider"], "deepseek")
|
||||
self.assertEqual(resp.data["model"], "deepseek-chat")
|
||||
self.assertEqual(resp.data["duration_ms"], 321)
|
||||
self.assertEqual(resp.data["reply_preview"], "OK")
|
||||
|
||||
def test_llm_test_prefers_request_payload_over_saved_settings(self):
|
||||
llm_test_mock = AsyncMock(
|
||||
return_value={
|
||||
"provider": "openai",
|
||||
"model": "gpt-4.1-mini",
|
||||
"duration_ms": 123,
|
||||
"reply_preview": "OK",
|
||||
}
|
||||
)
|
||||
payload = system_endpoint.LlmTestRequest(
|
||||
enabled=True,
|
||||
provider="openai",
|
||||
model="gpt-4.1-mini",
|
||||
thinking_level="high",
|
||||
api_key="sk-live",
|
||||
base_url="https://example.com/v1",
|
||||
base_url_preset="openai-default",
|
||||
user_agent="MoviePilot-Custom/1.0",
|
||||
use_proxy=False,
|
||||
)
|
||||
|
||||
with patch.object(system_endpoint.settings, "AI_AGENT_ENABLE", False), patch.object(
|
||||
system_endpoint.settings, "LLM_PROVIDER", "deepseek"
|
||||
), patch.object(system_endpoint.settings, "LLM_MODEL", "deepseek-chat"), patch.object(
|
||||
system_endpoint.settings, "LLM_API_KEY", "sk-saved"
|
||||
), patch.object(
|
||||
system_endpoint.settings, "LLM_BASE_URL", "https://api.deepseek.com"
|
||||
), patch.object(
|
||||
system_endpoint.LLMHelper,
|
||||
"test_current_settings",
|
||||
llm_test_mock,
|
||||
create=True,
|
||||
):
|
||||
resp = asyncio.run(system_endpoint.llm_test(payload=payload, _="token"))
|
||||
|
||||
llm_test_mock.assert_awaited_once_with(
|
||||
provider="openai",
|
||||
model="gpt-4.1-mini",
|
||||
thinking_level="high",
|
||||
api_key="sk-live",
|
||||
base_url="https://example.com/v1",
|
||||
base_url_preset="openai-default",
|
||||
user_agent="MoviePilot-Custom/1.0",
|
||||
use_proxy=False,
|
||||
api_protocol=None,
|
||||
web_search_mode=None,
|
||||
)
|
||||
self.assertTrue(resp.success)
|
||||
self.assertEqual(resp.data["provider"], "openai")
|
||||
self.assertEqual(resp.data["model"], "gpt-4.1-mini")
|
||||
|
||||
def test_llm_test_supports_legacy_thinking_payload(self):
|
||||
llm_test_mock = AsyncMock(
|
||||
return_value={
|
||||
"provider": "deepseek",
|
||||
"model": "deepseek-v4-pro",
|
||||
"duration_ms": 123,
|
||||
"reply_preview": "OK",
|
||||
}
|
||||
)
|
||||
payload = system_endpoint.LlmTestRequest(
|
||||
enabled=True,
|
||||
provider="deepseek",
|
||||
model="deepseek-v4-pro",
|
||||
api_key="sk-live",
|
||||
base_url="https://api.deepseek.com",
|
||||
base_url_preset="deepseek-default",
|
||||
user_agent=None,
|
||||
use_proxy=None,
|
||||
)
|
||||
|
||||
with patch.object(system_endpoint.settings, "AI_AGENT_ENABLE", False), patch.object(
|
||||
system_endpoint.LLMHelper,
|
||||
"test_current_settings",
|
||||
llm_test_mock,
|
||||
create=True,
|
||||
):
|
||||
resp = asyncio.run(system_endpoint.llm_test(payload=payload, _="token"))
|
||||
|
||||
llm_test_mock.assert_awaited_once_with(
|
||||
provider="deepseek",
|
||||
model="deepseek-v4-pro",
|
||||
thinking_level=None,
|
||||
api_key="sk-live",
|
||||
base_url="https://api.deepseek.com",
|
||||
base_url_preset="deepseek-default",
|
||||
user_agent=None,
|
||||
use_proxy=None,
|
||||
api_protocol=None,
|
||||
web_search_mode=None,
|
||||
)
|
||||
self.assertTrue(resp.success)
|
||||
|
||||
def test_llm_test_rejects_empty_reply(self):
|
||||
with patch.object(system_endpoint.settings, "AI_AGENT_ENABLE", True), patch.object(
|
||||
system_endpoint.settings, "LLM_PROVIDER", "deepseek"
|
||||
), patch.object(system_endpoint.settings, "LLM_MODEL", "deepseek-chat"), patch.object(
|
||||
system_endpoint.settings, "LLM_API_KEY", "sk-test"
|
||||
), patch.object(
|
||||
system_endpoint.LLMHelper,
|
||||
"test_current_settings",
|
||||
AsyncMock(return_value={"provider": "deepseek", "model": "deepseek-chat", "duration_ms": 12}),
|
||||
create=True,
|
||||
):
|
||||
resp = asyncio.run(system_endpoint.llm_test(_="token"))
|
||||
|
||||
self.assertFalse(resp.success)
|
||||
self.assertEqual(resp.message, "模型响应为空")
|
||||
self.assertEqual(resp.data["duration_ms"], 12)
|
||||
|
||||
def test_llm_test_maps_timeout_error(self):
|
||||
with patch.object(system_endpoint.settings, "AI_AGENT_ENABLE", True), patch.object(
|
||||
system_endpoint.settings, "LLM_PROVIDER", "deepseek"
|
||||
), patch.object(system_endpoint.settings, "LLM_MODEL", "deepseek-chat"), patch.object(
|
||||
system_endpoint.settings, "LLM_API_KEY", "sk-test"
|
||||
), patch.object(
|
||||
system_endpoint.LLMHelper,
|
||||
"test_current_settings",
|
||||
AsyncMock(side_effect=TimeoutError("request timed out")),
|
||||
create=True,
|
||||
):
|
||||
resp = asyncio.run(system_endpoint.llm_test(_="token"))
|
||||
|
||||
self.assertFalse(resp.success)
|
||||
self.assertEqual(resp.message, "LLM 调用超时")
|
||||
|
||||
def test_llm_test_sanitizes_error_message(self):
|
||||
raw_error = (
|
||||
"request failed api_key=sk-secret "
|
||||
"Authorization: Bearer sk-secret "
|
||||
"base error sk-secret"
|
||||
)
|
||||
with patch.object(system_endpoint.settings, "AI_AGENT_ENABLE", True), patch.object(
|
||||
system_endpoint.settings, "LLM_API_KEY", "sk-secret"
|
||||
), patch.object(system_endpoint.settings, "LLM_PROVIDER", "deepseek"), patch.object(
|
||||
system_endpoint.settings, "LLM_MODEL", "deepseek-chat"
|
||||
), patch.object(
|
||||
system_endpoint.LLMHelper,
|
||||
"test_current_settings",
|
||||
AsyncMock(side_effect=RuntimeError(raw_error)),
|
||||
create=True,
|
||||
):
|
||||
resp = asyncio.run(system_endpoint.llm_test(_="token"))
|
||||
|
||||
self.assertFalse(resp.success)
|
||||
self.assertNotIn("sk-secret", resp.message)
|
||||
self.assertNotIn("Authorization: Bearer", resp.message)
|
||||
self.assertIn("***", resp.message)
|
||||
Reference in New Issue
Block a user