deerflow-code/offline-backend-20260512/backend/tests/test_model_management.py

"""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


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_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