242 lines
7.6 KiB
Python
242 lines
7.6 KiB
Python
"""Focused regression tests for administrator model configuration helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
import yaml
|
|
from starlette.routing import Match
|
|
|
|
from app.gateway.routers import model_management as router
|
|
from deerflow.config.app_config import AppConfig
|
|
from deerflow.config.model_config import ModelConfig
|
|
from deerflow.models.factory import create_chat_model
|
|
|
|
|
|
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 ***"
|
|
|
|
|
|
def test_model_mutation_routes_accept_names_with_slashes() -> None:
|
|
for method in ("PUT", "DELETE"):
|
|
route = next(
|
|
item
|
|
for item in router.router.routes
|
|
if getattr(item, "methods", set()) == {method} and getattr(item, "path", "") == "/api/admin/models/{model_name:path}"
|
|
)
|
|
match, child_scope = route.matches(
|
|
{
|
|
"type": "http",
|
|
"method": method,
|
|
"path": "/api/admin/models/zai-org/GLM-5-FP8",
|
|
}
|
|
)
|
|
assert match is Match.FULL
|
|
assert child_scope["path_params"] == {"model_name": "zai-org/GLM-5-FP8"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_model_configuration_removes_matching_model_and_activates_config(
|
|
tmp_path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
config_path = tmp_path / "config.yaml"
|
|
config_path.write_text(
|
|
"""models:
|
|
- name: zai-org/GLM-5-FP8
|
|
use: langchain_openai:ChatOpenAI
|
|
model: GLM-5-FP8
|
|
- name: keep-me
|
|
use: langchain_openai:ChatOpenAI
|
|
model: keep-me
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
saved: dict[str, object] = {}
|
|
|
|
async def save_models(_request, raw_before: str, models: list[dict[str, object]]) -> list[dict[str, object]]:
|
|
saved["raw_before"] = raw_before
|
|
saved["models"] = models
|
|
return models
|
|
|
|
monkeypatch.setattr(router, "_resolve_config_path", lambda: config_path)
|
|
monkeypatch.setattr(router, "_save_models", save_models)
|
|
request = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(system_role="admin")))
|
|
|
|
await router.delete_model_configuration("zai-org/GLM-5-FP8", request)
|
|
|
|
assert "zai-org/GLM-5-FP8" in str(saved["raw_before"])
|
|
assert saved["models"] == [
|
|
{
|
|
"name": "keep-me",
|
|
"use": "langchain_openai:ChatOpenAI",
|
|
"model": "keep-me",
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reorder_model_configurations_persists_a_complete_permutation(
|
|
tmp_path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
config_path = tmp_path / "config.yaml"
|
|
config_path.write_text(
|
|
"""models:
|
|
- name: first
|
|
use: langchain_openai:ChatOpenAI
|
|
model: first
|
|
- name: second
|
|
use: langchain_openai:ChatOpenAI
|
|
model: second
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
saved: dict[str, object] = {}
|
|
|
|
async def save_models(_request, raw_before: str, models: list[dict[str, object]]) -> list[dict[str, object]]:
|
|
saved["raw_before"] = raw_before
|
|
saved["models"] = models
|
|
return models
|
|
|
|
monkeypatch.setattr(router, "_resolve_config_path", lambda: config_path)
|
|
monkeypatch.setattr(router, "_save_models", save_models)
|
|
request = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(system_role="admin")))
|
|
|
|
result = await router.reorder_model_configurations(
|
|
router.ModelOrderRequest(names=["second", "first"]),
|
|
request,
|
|
)
|
|
|
|
assert [item.name for item in result] == ["second", "first"]
|
|
assert [item["name"] for item in saved["models"]] == ["second", "first"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reorder_model_configurations_rejects_stale_or_partial_orders(
|
|
tmp_path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
config_path = tmp_path / "config.yaml"
|
|
config_path.write_text(
|
|
"""models:
|
|
- name: first
|
|
use: langchain_openai:ChatOpenAI
|
|
model: first
|
|
- name: second
|
|
use: langchain_openai:ChatOpenAI
|
|
model: second
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setattr(router, "_resolve_config_path", lambda: config_path)
|
|
request = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(system_role="admin")))
|
|
|
|
with pytest.raises(Exception) as exc_info:
|
|
await router.reorder_model_configurations(router.ModelOrderRequest(names=["first"]), request)
|
|
|
|
assert getattr(exc_info.value, "status_code", None) == 409
|
|
|
|
|
|
@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
|
|
|
|
|
|
def test_force_disable_thinking_skips_generic_openai_models_without_thinking_support() -> None:
|
|
"""LiteLLM proxies reject vendor-specific thinking payloads for ordinary chat models."""
|
|
config = AppConfig.model_construct(
|
|
models=[
|
|
ModelConfig(
|
|
name="litellm-qwen",
|
|
use="langchain_openai:ChatOpenAI",
|
|
model="qwen3.6",
|
|
api_key="test-key",
|
|
base_url="http://litellm.example/v1",
|
|
supports_thinking=False,
|
|
)
|
|
]
|
|
)
|
|
|
|
chat_model = create_chat_model(
|
|
"litellm-qwen",
|
|
app_config=config,
|
|
force_disable_thinking=True,
|
|
)
|
|
|
|
assert chat_model.extra_body is None
|