wehub-resource-sync
e4dcfc49aa
Tests / Lint and Format (push) Waiting to run
Tests / Web Node Tests (push) Waiting to run
Tests / Import Check (Python 3.11) (push) Waiting to run
Tests / Import Check (Python 3.12) (push) Waiting to run
Tests / Import Check (Python 3.13) (push) Waiting to run
Tests / Import Check (Python 3.14) (push) Waiting to run
Tests / Python Tests (Python 3.11) (push) Blocked by required conditions
Tests / Python Tests (Python 3.12) (push) Blocked by required conditions
Tests / Python Tests (Python 3.13) (push) Blocked by required conditions
Tests / Python Tests (Python 3.14) (push) Blocked by required conditions
Tests / Test Summary (push) Blocked by required conditions
654 lines
24 KiB
Python
654 lines
24 KiB
Python
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from deeptutor.api.routers import settings as settings_router
|
|
from deeptutor.services.config.provider_runtime import (
|
|
ResolvedEmbeddingConfig,
|
|
ResolvedLLMConfig,
|
|
)
|
|
from deeptutor.services.config.runtime_settings import RuntimeSettingsService
|
|
from deeptutor.services.embedding import client as embedding_client_module
|
|
from deeptutor.services.embedding import config as embedding_config_module
|
|
from deeptutor.services.llm import client as llm_client_module
|
|
from deeptutor.services.llm import config as llm_config_module
|
|
|
|
|
|
class _FakeEmbeddingAdapter:
|
|
def __init__(self, config: dict[str, Any]):
|
|
self.config = config
|
|
|
|
async def embed(self, request):
|
|
return type("EmbeddingResponse", (), {"embeddings": [[] for _ in request.texts]})()
|
|
|
|
|
|
class _FakeCatalogService:
|
|
def __init__(self, catalog: dict[str, Any]):
|
|
self._catalog = deepcopy(catalog)
|
|
|
|
def save(self, catalog: dict[str, Any]) -> dict[str, Any]:
|
|
self._catalog = deepcopy(catalog)
|
|
return deepcopy(self._catalog)
|
|
|
|
def load(self) -> dict[str, Any]:
|
|
return deepcopy(self._catalog)
|
|
|
|
def apply(self, catalog: dict[str, Any]) -> dict[str, Any]:
|
|
current = self.save(catalog)
|
|
return {
|
|
"catalog_path": "memory://model_catalog.json",
|
|
"services": list(current["services"]),
|
|
}
|
|
|
|
|
|
def _build_catalog(
|
|
*,
|
|
llm_model: str,
|
|
llm_base_url: str,
|
|
llm_api_key: str,
|
|
embedding_model: str,
|
|
embedding_base_url: str,
|
|
embedding_api_key: str,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"version": 1,
|
|
"services": {
|
|
"llm": {
|
|
"active_profile_id": "llm-profile-default",
|
|
"active_model_id": "llm-model-default",
|
|
"profiles": [
|
|
{
|
|
"id": "llm-profile-default",
|
|
"name": "Default LLM Endpoint",
|
|
"binding": "openai",
|
|
"base_url": llm_base_url,
|
|
"api_key": llm_api_key,
|
|
"api_version": "",
|
|
"extra_headers": {},
|
|
"models": [
|
|
{
|
|
"id": "llm-model-default",
|
|
"name": llm_model,
|
|
"model": llm_model,
|
|
}
|
|
],
|
|
}
|
|
],
|
|
},
|
|
"embedding": {
|
|
"active_profile_id": "embedding-profile-default",
|
|
"active_model_id": "embedding-model-default",
|
|
"profiles": [
|
|
{
|
|
"id": "embedding-profile-default",
|
|
"name": "Default Embedding Endpoint",
|
|
"binding": "openai",
|
|
"base_url": embedding_base_url,
|
|
"api_key": embedding_api_key,
|
|
"api_version": "",
|
|
"extra_headers": {},
|
|
"models": [
|
|
{
|
|
"id": "embedding-model-default",
|
|
"name": embedding_model,
|
|
"model": embedding_model,
|
|
"dimension": "1536",
|
|
}
|
|
],
|
|
}
|
|
],
|
|
},
|
|
"search": {
|
|
"active_profile_id": None,
|
|
"profiles": [],
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
def _patch_runtime(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
service: _FakeCatalogService,
|
|
) -> None:
|
|
monkeypatch.setattr(settings_router, "get_model_catalog_service", lambda: service)
|
|
monkeypatch.setattr(
|
|
embedding_client_module,
|
|
"_resolve_adapter_class",
|
|
lambda _binding: _FakeEmbeddingAdapter,
|
|
)
|
|
|
|
def _resolve_llm_runtime_config() -> ResolvedLLMConfig:
|
|
catalog = service.load()
|
|
profile = catalog["services"]["llm"]["profiles"][0]
|
|
model = profile["models"][0]
|
|
return ResolvedLLMConfig(
|
|
model=model["model"],
|
|
provider_name=profile["binding"],
|
|
provider_mode="standard",
|
|
binding_hint=profile["binding"],
|
|
binding=profile["binding"],
|
|
api_key=profile["api_key"],
|
|
base_url=profile["base_url"],
|
|
effective_url=profile["base_url"],
|
|
api_version=None,
|
|
extra_headers={},
|
|
reasoning_effort=None,
|
|
)
|
|
|
|
def _resolve_embedding_runtime_config() -> ResolvedEmbeddingConfig:
|
|
catalog = service.load()
|
|
profile = catalog["services"]["embedding"]["profiles"][0]
|
|
model = profile["models"][0]
|
|
return ResolvedEmbeddingConfig(
|
|
model=model["model"],
|
|
provider_name=profile["binding"],
|
|
provider_mode="standard",
|
|
binding_hint=profile["binding"],
|
|
binding=profile["binding"],
|
|
api_key=profile["api_key"],
|
|
base_url=profile["base_url"],
|
|
effective_url=profile["base_url"],
|
|
api_version=None,
|
|
extra_headers={},
|
|
dimension=int(model["dimension"]),
|
|
request_timeout=60,
|
|
batch_size=10,
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
llm_config_module,
|
|
"resolve_llm_runtime_config",
|
|
_resolve_llm_runtime_config,
|
|
)
|
|
monkeypatch.setattr(
|
|
embedding_config_module,
|
|
"resolve_embedding_runtime_config",
|
|
_resolve_embedding_runtime_config,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_network_settings_roundtrip_normalizes_cors_origins(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path
|
|
) -> None:
|
|
service = RuntimeSettingsService(tmp_path / "settings", process_env={})
|
|
service.save_system({"backend_port": 8001, "frontend_port": 3782})
|
|
service.save_auth({"enabled": True, "cookie_secure": True})
|
|
monkeypatch.setattr(settings_router, "get_runtime_settings_service", lambda: service)
|
|
|
|
payload = settings_router.NetworkSettingsUpdate(
|
|
backend_port=8101,
|
|
frontend_port=3882,
|
|
public_api_base="https://api.example.com/deeptutor",
|
|
cors_origins=["app.example.com; https://learn.example.com/path"],
|
|
)
|
|
|
|
response = await settings_router.update_network_settings(payload)
|
|
|
|
assert response["settings"]["backend_port"] == 8101
|
|
assert response["settings"]["public_api_base"] == "https://api.example.com/deeptutor"
|
|
assert response["settings"]["cors_origins"] == [
|
|
"http://app.example.com",
|
|
"https://learn.example.com",
|
|
]
|
|
assert response["effective"]["cors_mode"] == "explicit"
|
|
assert response["auth"]["cross_site_cookie_ready"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mineru_settings_roundtrip_redacts_token(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path
|
|
) -> None:
|
|
service = RuntimeSettingsService(tmp_path / "settings", process_env={})
|
|
monkeypatch.setattr(settings_router, "get_runtime_settings_service", lambda: service)
|
|
|
|
payload = settings_router.MinerUSettingsUpdate(
|
|
mode="cloud",
|
|
api_base_url="https://mineru.net/",
|
|
api_token="secret-token",
|
|
model_version="vlm",
|
|
)
|
|
response = await settings_router.update_mineru_settings(payload)
|
|
|
|
# The raw token never leaves the backend; only a boolean flag does.
|
|
assert response["api_token_set"] is True
|
|
assert "api_token" not in response["settings"]
|
|
assert response["settings"]["mode"] == "cloud"
|
|
assert response["settings"]["api_base_url"] == "https://mineru.net"
|
|
assert response["settings"]["model_version"] == "vlm"
|
|
# Persisted on disk under the canonical key.
|
|
assert service.load_mineru()["api_token"] == "secret-token"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mineru_token_tristate_keep_then_clear(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path
|
|
) -> None:
|
|
service = RuntimeSettingsService(tmp_path / "settings", process_env={})
|
|
monkeypatch.setattr(settings_router, "get_runtime_settings_service", lambda: service)
|
|
service.save_mineru({"mode": "cloud", "api_token": "keep-me"})
|
|
|
|
# api_token=None → keep the stored token.
|
|
await settings_router.update_mineru_settings(
|
|
settings_router.MinerUSettingsUpdate(mode="cloud", api_token=None)
|
|
)
|
|
assert service.load_mineru()["api_token"] == "keep-me"
|
|
|
|
# api_token="" → explicitly clear it.
|
|
await settings_router.update_mineru_settings(
|
|
settings_router.MinerUSettingsUpdate(mode="cloud", api_token="")
|
|
)
|
|
assert service.load_mineru()["api_token"] == ""
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mineru_test_connection_reports_missing_token(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path
|
|
) -> None:
|
|
service = RuntimeSettingsService(tmp_path / "settings", process_env={})
|
|
monkeypatch.setattr(settings_router, "get_runtime_settings_service", lambda: service)
|
|
|
|
result = await settings_router.test_mineru_connection(
|
|
settings_router.MinerUSettingsUpdate(mode="cloud", api_token="")
|
|
)
|
|
assert result["ok"] is False
|
|
assert "token" in result["message"].lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mineru_payload_includes_local_cli_probe(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path
|
|
) -> None:
|
|
from deeptutor.services.parsing.engines.mineru import backend as mineru_backend
|
|
|
|
service = RuntimeSettingsService(tmp_path / "settings", process_env={})
|
|
monkeypatch.setattr(settings_router, "get_runtime_settings_service", lambda: service)
|
|
monkeypatch.setattr(
|
|
mineru_backend,
|
|
"local_cli_probe",
|
|
lambda *a: {"found": True, "command": "mineru", "path": "/env/bin/mineru"},
|
|
)
|
|
|
|
payload = await settings_router.get_mineru_settings()
|
|
assert payload["local_cli"] == {
|
|
"found": True,
|
|
"command": "mineru",
|
|
"path": "/env/bin/mineru",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mineru_test_connection_local_mode(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
|
|
from deeptutor.services.parsing.engines.mineru import backend as mineru_backend
|
|
|
|
service = RuntimeSettingsService(tmp_path / "settings", process_env={})
|
|
monkeypatch.setattr(settings_router, "get_runtime_settings_service", lambda: service)
|
|
|
|
# CLI present → ok with version detail.
|
|
monkeypatch.setattr(
|
|
mineru_backend,
|
|
"local_cli_probe",
|
|
lambda *a: {"found": True, "command": "mineru", "path": "/env/bin/mineru"},
|
|
)
|
|
monkeypatch.setattr(mineru_backend, "local_cli_version", lambda cmd: "mineru, version 2.5.0")
|
|
result = await settings_router.test_mineru_connection(
|
|
settings_router.MinerUSettingsUpdate(mode="local")
|
|
)
|
|
assert result["ok"] is True
|
|
assert "2.5.0" in result["message"]
|
|
|
|
# CLI absent → actionable failure message.
|
|
monkeypatch.setattr(
|
|
mineru_backend, "local_cli_probe", lambda *a: {"found": False, "command": "", "path": ""}
|
|
)
|
|
result = await settings_router.test_mineru_connection(
|
|
settings_router.MinerUSettingsUpdate(mode="local")
|
|
)
|
|
assert result["ok"] is False
|
|
assert "not found" in result["message"].lower()
|
|
|
|
# Bad configured path → message points at the path, not at PATH install.
|
|
monkeypatch.setattr(
|
|
mineru_backend,
|
|
"local_cli_probe",
|
|
lambda *a: {"found": False, "command": "", "path": "/bad/mineru", "source": "configured"},
|
|
)
|
|
result = await settings_router.test_mineru_connection(
|
|
settings_router.MinerUSettingsUpdate(mode="local", local_cli_path="/bad/mineru")
|
|
)
|
|
assert result["ok"] is False
|
|
assert "/bad/mineru" in result["message"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mineru_models_download_start_requires_downloader(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from deeptutor.services.parsing.engines.mineru import models as mineru_models
|
|
|
|
monkeypatch.setattr(
|
|
mineru_models, "resolve_models_downloader", lambda p: {"found": False, "path": ""}
|
|
)
|
|
result = await settings_router.start_mineru_models_download(
|
|
settings_router.MinerUModelDownloadPayload()
|
|
)
|
|
assert result["ok"] is False
|
|
assert "not found" in result["message"].lower()
|
|
|
|
# Configured CLI without a sibling downloader → message names the path.
|
|
monkeypatch.setattr(
|
|
mineru_models,
|
|
"resolve_models_downloader",
|
|
lambda p: {"found": False, "path": "/env/bin/mineru-models-download"},
|
|
)
|
|
result = await settings_router.start_mineru_models_download(
|
|
settings_router.MinerUModelDownloadPayload(local_cli_path="/env/bin/mineru")
|
|
)
|
|
assert result["ok"] is False
|
|
assert "/env/bin/mineru-models-download" in result["message"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mineru_models_download_start_and_status_passthrough(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from deeptutor.services.parsing.engines.mineru import models as mineru_models
|
|
|
|
calls: dict[str, object] = {}
|
|
|
|
class _FakeManager:
|
|
def start(self, **kwargs):
|
|
calls.update(kwargs)
|
|
return {"ok": True, "message": ""}
|
|
|
|
def status(self, cursor=0):
|
|
return {"state": "running", "lines": ["l1"], "next_cursor": 1, "message": ""}
|
|
|
|
def cancel(self):
|
|
return {"ok": True, "message": ""}
|
|
|
|
monkeypatch.setattr(
|
|
mineru_models,
|
|
"resolve_models_downloader",
|
|
lambda p: {"found": True, "path": "/env/bin/mineru-models-download"},
|
|
)
|
|
monkeypatch.setattr(mineru_models, "get_model_download_manager", lambda: _FakeManager())
|
|
|
|
result = await settings_router.start_mineru_models_download(
|
|
settings_router.MinerUModelDownloadPayload(
|
|
model_type="all", source="modelscope", endpoint="https://hf-mirror.com"
|
|
)
|
|
)
|
|
assert result["ok"] is True
|
|
assert calls["downloader"] == "/env/bin/mineru-models-download"
|
|
assert calls["model_type"] == "all"
|
|
assert calls["source"] == "modelscope"
|
|
|
|
status = await settings_router.mineru_models_download_status(cursor=0)
|
|
assert status["lines"] == ["l1"]
|
|
cancel = await settings_router.cancel_mineru_models_download()
|
|
assert cancel["ok"] is True
|
|
|
|
|
|
def test_embedding_provider_choices_use_full_endpoint_urls() -> None:
|
|
embedding = {item["value"]: item for item in settings_router._provider_choices()["embedding"]}
|
|
|
|
assert embedding["openrouter"]["base_url"] == "https://openrouter.ai/api/v1/embeddings"
|
|
assert embedding["ollama"]["base_url"] == "http://localhost:11434/api/embed"
|
|
assert embedding["openai"]["base_url"] == "https://api.openai.com/v1/embeddings"
|
|
assert "custom_openai_sdk" not in embedding
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_llm_options_returns_redacted_catalog(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
catalog = _build_catalog(
|
|
llm_model="gpt-4o-mini",
|
|
llm_base_url="https://llm.example/v1",
|
|
llm_api_key="secret-key",
|
|
embedding_model="text-embedding-3-small",
|
|
embedding_base_url="https://emb.example/v1/embeddings",
|
|
embedding_api_key="emb-key",
|
|
)
|
|
service = _FakeCatalogService(catalog)
|
|
monkeypatch.setattr(settings_router, "get_model_catalog_service", lambda: service)
|
|
|
|
response = await settings_router.get_llm_options()
|
|
|
|
assert response["active"] == {
|
|
"profile_id": "llm-profile-default",
|
|
"model_id": "llm-model-default",
|
|
}
|
|
assert response["options"][0]["model"] == "gpt-4o-mini"
|
|
assert "api_key" not in response["options"][0]
|
|
assert "base_url" not in response["options"][0]
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_runtime_state() -> None:
|
|
llm_config_module.clear_llm_config_cache()
|
|
llm_client_module.reset_llm_client()
|
|
embedding_client_module.reset_embedding_client()
|
|
yield
|
|
llm_config_module.clear_llm_config_cache()
|
|
llm_client_module.reset_llm_client()
|
|
embedding_client_module.reset_embedding_client()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_catalog_invalidates_runtime_caches(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
initial_catalog = _build_catalog(
|
|
llm_model="gpt-old",
|
|
llm_base_url="https://old-llm.example/v1",
|
|
llm_api_key="old-llm-key",
|
|
embedding_model="text-embedding-old",
|
|
embedding_base_url="https://old-embedding.example/v1/embeddings",
|
|
embedding_api_key="old-embedding-key",
|
|
)
|
|
updated_catalog = _build_catalog(
|
|
llm_model="gpt-new",
|
|
llm_base_url="https://new-llm.example/v1",
|
|
llm_api_key="new-llm-key",
|
|
embedding_model="text-embedding-new",
|
|
embedding_base_url="https://new-embedding.example/v1/embeddings",
|
|
embedding_api_key="new-embedding-key",
|
|
)
|
|
service = _FakeCatalogService(initial_catalog)
|
|
_patch_runtime(monkeypatch, service)
|
|
|
|
old_llm_config = llm_config_module.get_llm_config()
|
|
old_llm_client = llm_client_module.get_llm_client()
|
|
old_embedding_client = embedding_client_module.get_embedding_client()
|
|
|
|
response = await settings_router.update_catalog(
|
|
settings_router.CatalogPayload(catalog=updated_catalog)
|
|
)
|
|
|
|
new_llm_config = llm_config_module.get_llm_config()
|
|
new_llm_client = llm_client_module.get_llm_client()
|
|
new_embedding_client = embedding_client_module.get_embedding_client()
|
|
|
|
assert response == {"catalog": updated_catalog}
|
|
assert old_llm_config.model == "gpt-old"
|
|
assert new_llm_config.model == "gpt-new"
|
|
assert new_llm_config.base_url == "https://new-llm.example/v1"
|
|
assert new_llm_config is not old_llm_config
|
|
assert new_llm_client is not old_llm_client
|
|
assert new_llm_client.config.model == "gpt-new"
|
|
assert new_embedding_client is not old_embedding_client
|
|
assert new_embedding_client.config.model == "text-embedding-new"
|
|
assert new_embedding_client.config.base_url == "https://new-embedding.example/v1/embeddings"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_catalog_invalidates_runtime_caches(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
initial_catalog = _build_catalog(
|
|
llm_model="gpt-before-apply",
|
|
llm_base_url="https://before-apply-llm.example/v1",
|
|
llm_api_key="before-apply-llm-key",
|
|
embedding_model="text-embedding-before-apply",
|
|
embedding_base_url="https://before-apply-embedding.example/v1/embeddings",
|
|
embedding_api_key="before-apply-embedding-key",
|
|
)
|
|
applied_catalog = _build_catalog(
|
|
llm_model="gpt-after-apply",
|
|
llm_base_url="https://after-apply-llm.example/v1",
|
|
llm_api_key="after-apply-llm-key",
|
|
embedding_model="text-embedding-after-apply",
|
|
embedding_base_url="https://after-apply-embedding.example/v1/embeddings",
|
|
embedding_api_key="after-apply-embedding-key",
|
|
)
|
|
service = _FakeCatalogService(initial_catalog)
|
|
_patch_runtime(monkeypatch, service)
|
|
|
|
llm_config_module.get_llm_config()
|
|
old_llm_client = llm_client_module.get_llm_client()
|
|
old_embedding_client = embedding_client_module.get_embedding_client()
|
|
|
|
response = await settings_router.apply_catalog(
|
|
settings_router.CatalogPayload(catalog=applied_catalog)
|
|
)
|
|
|
|
new_llm_config = llm_config_module.get_llm_config()
|
|
new_llm_client = llm_client_module.get_llm_client()
|
|
new_embedding_client = embedding_client_module.get_embedding_client()
|
|
|
|
assert response["catalog"] == applied_catalog
|
|
assert response["runtime"]["catalog_path"]
|
|
assert new_llm_config.model == "gpt-after-apply"
|
|
assert new_llm_client is not old_llm_client
|
|
assert new_llm_client.config.base_url == "https://after-apply-llm.example/v1"
|
|
assert new_embedding_client is not old_embedding_client
|
|
assert new_embedding_client.config.model == "text-embedding-after-apply"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enabled_tools_roundtrip(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
|
|
settings_file = tmp_path / "interface.json"
|
|
monkeypatch.setattr(settings_router, "_settings_file", lambda: settings_file)
|
|
|
|
# Default state — no file yet, so the loader emits the full toggleable set.
|
|
assert set(settings_router.get_enabled_optional_tools()) == set(
|
|
settings_router.USER_TOGGLEABLE_TOOL_NAMES
|
|
)
|
|
|
|
# PUT a partial set; unknown tool names get filtered out.
|
|
update = settings_router.EnabledToolsUpdate(
|
|
enabled_tools=["web_search", "reason", "not_a_real_tool"]
|
|
)
|
|
response = await settings_router.update_enabled_tools(update)
|
|
assert response == {"enabled_optional_tools": ["web_search", "reason"]}
|
|
assert settings_router.get_enabled_optional_tools() == ["web_search", "reason"]
|
|
|
|
# Empty selection is a valid "all off" state.
|
|
response = await settings_router.update_enabled_tools(
|
|
settings_router.EnabledToolsUpdate(enabled_tools=[])
|
|
)
|
|
assert response == {"enabled_optional_tools": []}
|
|
assert settings_router.get_enabled_optional_tools() == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_complete_tour_invalidates_runtime_caches(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path
|
|
) -> None:
|
|
initial_catalog = _build_catalog(
|
|
llm_model="gpt-before-tour",
|
|
llm_base_url="https://before-tour-llm.example/v1",
|
|
llm_api_key="before-tour-llm-key",
|
|
embedding_model="text-embedding-before-tour",
|
|
embedding_base_url="https://before-tour-embedding.example/v1/embeddings",
|
|
embedding_api_key="before-tour-embedding-key",
|
|
)
|
|
completed_catalog = _build_catalog(
|
|
llm_model="gpt-after-tour",
|
|
llm_base_url="https://after-tour-llm.example/v1",
|
|
llm_api_key="after-tour-llm-key",
|
|
embedding_model="text-embedding-after-tour",
|
|
embedding_base_url="https://after-tour-embedding.example/v1/embeddings",
|
|
embedding_api_key="after-tour-embedding-key",
|
|
)
|
|
service = _FakeCatalogService(initial_catalog)
|
|
_patch_runtime(monkeypatch, service)
|
|
|
|
tour_cache = tmp_path / ".tour_cache.json"
|
|
tour_cache.write_text('{"status": "running"}', encoding="utf-8")
|
|
monkeypatch.setattr(settings_router, "TOUR_CACHE", tour_cache)
|
|
|
|
llm_config_module.get_llm_config()
|
|
old_llm_client = llm_client_module.get_llm_client()
|
|
old_embedding_client = embedding_client_module.get_embedding_client()
|
|
|
|
response = await settings_router.complete_tour(
|
|
settings_router.TourCompletePayload(catalog=completed_catalog)
|
|
)
|
|
|
|
new_llm_config = llm_config_module.get_llm_config()
|
|
new_llm_client = llm_client_module.get_llm_client()
|
|
new_embedding_client = embedding_client_module.get_embedding_client()
|
|
cache = tour_cache.read_text(encoding="utf-8")
|
|
|
|
assert response["runtime"]["catalog_path"]
|
|
assert response["status"] == "completed"
|
|
assert new_llm_config.model == "gpt-after-tour"
|
|
assert new_llm_client is not old_llm_client
|
|
assert new_embedding_client is not old_embedding_client
|
|
assert '"status": "completed"' in cache
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_models_returns_picker_options(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
import deeptutor.services.llm.factory as factory_module
|
|
|
|
async def _fake_fetch(binding: str, base_url: str, api_key: str | None = None):
|
|
assert binding == "openai" # "OpenAI" is normalized to lowercase
|
|
assert base_url == "https://api.example.com/v1"
|
|
assert api_key == "sk-x"
|
|
return ["gpt-4o", "gpt-4o-mini"]
|
|
|
|
monkeypatch.setattr(factory_module, "fetch_models", _fake_fetch)
|
|
|
|
response = await settings_router.fetch_models_from_provider(
|
|
settings_router.FetchModelsPayload(
|
|
binding="OpenAI", base_url="https://api.example.com/v1", api_key="sk-x"
|
|
)
|
|
)
|
|
|
|
assert response == {
|
|
"models": [
|
|
{"id": "gpt-4o", "name": "gpt-4o"},
|
|
{"id": "gpt-4o-mini", "name": "gpt-4o-mini"},
|
|
]
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_models_requires_base_url() -> None:
|
|
from fastapi import HTTPException
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await settings_router.fetch_models_from_provider(
|
|
settings_router.FetchModelsPayload(base_url=" ")
|
|
)
|
|
assert exc_info.value.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_models_maps_provider_error_to_502(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from fastapi import HTTPException
|
|
|
|
import deeptutor.services.llm.factory as factory_module
|
|
|
|
async def _boom(binding: str, base_url: str, api_key: str | None = None):
|
|
raise RuntimeError("connection refused")
|
|
|
|
monkeypatch.setattr(factory_module, "fetch_models", _boom)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await settings_router.fetch_models_from_provider(
|
|
settings_router.FetchModelsPayload(binding="custom", base_url="https://x/v1")
|
|
)
|
|
assert exc_info.value.status_code == 502
|