董宏凯andCursor
ba912acf72
新增 /api/admin/models 路由（读写 config.yaml models 段、连接测试、写前备份），以及工作区「模型配置」管理页与侧栏入口。 Co-authored-by: Cursor <cursoragent@cursor.com>
93 lines
2.9 KiB
Python
93 lines
2.9 KiB
Python
"""Focused regression tests for administrator model configuration helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from app.gateway.routers import model_management as router
|
|
|
|
|
|
def test_replace_models_block_keeps_other_sections_and_comments() -> None:
|
|
raw = """config_version: 1
|
|
models:
|
|
# Existing provider note
|
|
- name: old-model
|
|
use: langchain_openai:ChatOpenAI
|
|
model: old-model
|
|
# Disabled provider example remains useful to operators.
|
|
# - name: disabled-model
|
|
# model: disabled-model
|
|
agents:
|
|
default: lead_agent
|
|
"""
|
|
|
|
replaced = router._replace_models_block(
|
|
raw,
|
|
[
|
|
{
|
|
"name": "new-model",
|
|
"use": "langchain_openai:ChatOpenAI",
|
|
"model": "new-model",
|
|
}
|
|
],
|
|
)
|
|
|
|
parsed = yaml.safe_load(replaced)
|
|
assert parsed["models"] == [
|
|
{
|
|
"name": "new-model",
|
|
"use": "langchain_openai:ChatOpenAI",
|
|
"model": "new-model",
|
|
}
|
|
]
|
|
assert parsed["agents"] == {"default": "lead_agent"}
|
|
assert "# Existing provider note" in replaced
|
|
assert "# Disabled provider example remains useful to operators." in replaced
|
|
|
|
|
|
def test_connection_error_redaction_hides_literal_and_resolved_keys() -> None:
|
|
message = "provider rejected raw-secret and resolved-secret"
|
|
assert router._redact_text(message, ["raw-secret", "resolved-secret"]) == "provider rejected *** and ***"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connection_test_uses_one_minimal_prompt_without_writing_config(
|
|
tmp_path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
config_path = tmp_path / "config.yaml"
|
|
original = "models: []\nagent: lead_agent\n"
|
|
config_path.write_text(original, encoding="utf-8")
|
|
messages_seen = []
|
|
|
|
class FakeChatModel:
|
|
async def ainvoke(self, messages):
|
|
messages_seen.extend(messages)
|
|
return SimpleNamespace(content="OK")
|
|
|
|
monkeypatch.setattr(router, "_resolve_config_path", lambda: config_path)
|
|
monkeypatch.setattr(router, "create_chat_model", lambda *_args, **_kwargs: FakeChatModel())
|
|
request = SimpleNamespace(
|
|
state=SimpleNamespace(user=SimpleNamespace(system_role="admin")),
|
|
app=SimpleNamespace(state=SimpleNamespace(config=SimpleNamespace(model_copy=lambda **_kwargs: SimpleNamespace()))),
|
|
)
|
|
body = router.ModelTestRequest(
|
|
model=router.ModelUpsertRequest(
|
|
name="test-model",
|
|
use="langchain_openai:ChatOpenAI",
|
|
model="test-model",
|
|
api_key="raw-secret",
|
|
)
|
|
)
|
|
|
|
result = await router.test_model_connection(body, request)
|
|
|
|
assert result.success is True
|
|
assert result.preview == "OK"
|
|
assert len(messages_seen) == 1
|
|
assert messages_seen[0].content == "Reply with exactly: OK"
|
|
assert config_path.read_text(encoding="utf-8") == original
|