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