Files
ObsiGate/tests/test_ai_models.py
T
bruno b69cb9b0f8
CI / lint (push) Successful in 1m20s
CI / security (push) Successful in 53s
CI / test (push) Successful in 2m27s
CI / build (push) Successful in 50s
CI / e2e (push) Successful in 10m48s
fix(ai): BUG-044 capacités des modèles lues chez le fournisseur (Mistral vision)
Le panneau de modèle par défaut et la bulle ⓘ n'affichaient aucun modèle Mistral
« Vision capable » alors que GET api.mistral.ai/v1/models en déclare 28 : la table
de capacités était entièrement statique et aucun de ses motifs ne correspondait aux
familles Mistral actuelles (seul `pixtral`, retiré de l'API, les matchait).

- backend/provider_capabilities.py (nouveau) : capacités déclarées par le
  fournisseur (Mistral `capabilities`, OpenRouter `architecture`), détectées par
  la forme du payload, snapshot en cache process-wide (TTL 30 min, surchargeable
  par AI_CAPABILITIES_TTL_SECONDS) rempli par GET /api/config/ai-models.
- backend/model_capabilities.py : une déclaration prime sur la table statique
  pour chaque drapeau mentionné ; la table ne comble que le reste (Mistral ne
  déclare jamais `embedding`). Table corrigée pour le repli hors ligne : familles
  vision Mistral (ministral, magistral, mistral-small, mistral-medium,
  mistral-vibe-cli, labs-leanstral), mistral-ocr = vision sans chat, et défaut du
  fournisseur Mistral sans `embeddings` (mistral-large / codestral n'étaient plus
  des « embedders »).
- backend/ai_routes.py : GET /api/ai/model-capabilities reste sans appel réseau
  (cache froid → table statique).
- Tests : tests/test_provider_capabilities.py (nouveau), TestMistralFamilies et
  TestDeclaredCapabilities (bout en bout via l'API).
- Docs : CHANGELOG [Unreleased], registre + journal ISSUES_TODOLIST,
  fiche docs/features/ai-provider-picker.md (§L).

Vérifié : 28/28 modèles vision déclarés par Mistral détectés (0 avant), 0 écart
dans les deux sens ; pytest 1007 passed / 6 skipped ; ruff 0 (backend) ; mypy 0 ;
tests frontend unit 9/9 + IA 66/66 + validate-imports 37 modules ; instance de test
reconstruite et vérifiée sur http://localhost:2020.
2026-09-15 09:26:31 -04:00

471 lines
19 KiB
Python

"""Tests for the AI models listing endpoint + curated fallback lists (ROADMAP #74/#71).
Covers:
- GET /api/config/ai-models always returns a non-empty list for known providers,
even when the live API call fails (network, auth, etc.).
- The fallback list is curated with at least one model per provider.
- Gemini parsing strips the "models/" prefix.
- The default model is prepended if not already in the list.
"""
from __future__ import annotations
import pytest
from fastapi.testclient import TestClient
from backend.main import _FALLBACK_MODELS, app
#: Trimmed-down copy of what api.mistral.ai/v1/models really returns (BUG-044).
MISTRAL_LIVE_PAYLOAD = {
"object": "list",
"data": [
{"id": "mistral-small-latest", "capabilities": {"completion_chat": True, "vision": True}},
{"id": "mistral-medium-latest", "capabilities": {"completion_chat": True, "vision": True}},
{"id": "mistral-embed", "capabilities": {"completion_chat": False, "vision": False}},
],
}
@pytest.fixture
def admin_client(tmp_path):
"""Minimal admin client for the /api/config/ai-models endpoint."""
from backend.auth.password import hash_password
import json
import os
from pathlib import Path
data_dir = tmp_path / "data"
data_dir.mkdir()
users = {
"version": 1,
"users": {
"admin": {
"id": "admin-1",
"username": "admin",
"display_name": "admin",
"password_hash": hash_password("chab30"),
"role": "admin",
"vaults": ["*"],
"active": True,
"created_at": "2026-01-01T00:00:00",
},
},
}
(data_dir / "users.json").write_text(json.dumps(users), encoding="utf-8")
src_secret = Path("data/secret.key")
if src_secret.exists():
import shutil
shutil.copy2(str(src_secret), str(data_dir / "secret.key"))
orig_cwd = os.getcwd()
os.chdir(str(tmp_path))
os.environ["VAULT_1_NAME"] = "TestVault"
os.environ["VAULT_1_PATH"] = str(Path("test-vault").resolve())
os.environ["OBSIGATE_AUTH_ENABLED"] = "true"
os.environ["OBSIGATE_ADMIN_USER"] = "admin"
os.environ["OBSIGATE_ADMIN_PASSWORD"] = "chab30"
os.environ["OBSIGATE_WATCHER_ENABLED"] = "false"
import backend.main
backend.main._load_config = lambda: {"watcher_enabled": False}
from backend.indexer import build_index, index
import asyncio
for key in list(index.keys()):
del index[key]
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(build_index())
client = TestClient(app)
yield client
client.close()
os.chdir(orig_cwd)
for k in ["VAULT_1_NAME", "VAULT_1_PATH", "OBSIGATE_AUTH_ENABLED",
"OBSIGATE_ADMIN_USER", "OBSIGATE_ADMIN_PASSWORD", "OBSIGATE_WATCHER_ENABLED"]:
os.environ.pop(k, None)
def _login_admin(client):
resp = client.post("/api/auth/login",
json={"username": "admin", "password": "chab30"})
assert resp.status_code == 200, resp.text
return resp.json()["access_token"]
@pytest.fixture(autouse=True)
def _no_declared_caps():
"""Provider declarations are cached process-wide: isolate every test."""
from backend.provider_capabilities import clear_declared_capabilities
clear_declared_capabilities()
yield
clear_declared_capabilities()
# ── Unit tests for the fallback table itself ─────────────────────────────
class TestFallbackTable:
def test_every_known_provider_has_fallback_list(self):
"""Every provider in the dropdown must have a non-empty fallback list."""
expected_providers = {
"deepseek", "openrouter", "gemini", "nvidia",
"qwencloud", "xiaomi", "mistral",
}
assert set(_FALLBACK_MODELS.keys()) >= expected_providers
for p in expected_providers:
assert _FALLBACK_MODELS[p], f"fallback list for {p!r} is empty"
assert all(isinstance(m, str) and m.strip() for m in _FALLBACK_MODELS[p]), \
f"fallback list for {p!r} contains invalid entries: {_FALLBACK_MODELS[p]}"
def test_fallback_lists_are_short_and_focused(self):
"""Fallbacks should be short (≤10 models) and well-known."""
for p, models in _FALLBACK_MODELS.items():
assert len(models) <= 10, f"too many fallbacks for {p}: {len(models)}"
def test_xiaomi_fallback_contains_mimo_models(self):
"""Xiaomi's MiMo models must be in the fallback list."""
mimos = [m for m in _FALLBACK_MODELS["xiaomi"] if "mimo" in m.lower()]
assert mimos, "no MiMo model in xiaomi fallback"
# ── Endpoint integration tests ────────────────────────────────────────────
class TestListModelsEndpoint:
"""Verify /api/config/ai-models?provider=X always returns a usable list."""
def test_unknown_provider_returns_empty(self, admin_client):
token = _login_admin(admin_client)
resp = admin_client.get(
"/api/config/ai-models",
params={"provider": "nonexistent-provider"},
headers={"Authorization": f"Bearer {token}"},
)
assert resp.status_code == 200
data = resp.json()
assert data["models"] == []
assert "error" in data or "source" in data
def test_provider_without_key_returns_fallback(self, admin_client, monkeypatch):
"""When the provider has no API key configured, return the curated fallback list."""
# Force every provider key to be empty so the endpoint hits its
# 'no key configured' branch.
from backend import ai
monkeypatch.setattr(ai, "get_ai_key", lambda name: None)
token = _login_admin(admin_client)
for provider in ("xiaomi", "nvidia", "deepseek", "mistral"):
resp = admin_client.get(
"/api/config/ai-models",
params={"provider": provider},
headers={"Authorization": f"Bearer {token}"},
)
assert resp.status_code == 200, f"{provider}: {resp.status_code}"
data = resp.json()
assert data["models"], f"{provider}: fallback list is empty"
assert data.get("source") == "fallback", (
f"{provider}: expected source=fallback, got {data.get('source')!r}"
)
def test_provider_with_unreachable_api_returns_fallback(
self, admin_client, monkeypatch,
):
"""When the live API call fails (network/DNS), the fallback list is used.
This test patches both the key lookup AND the urlopen call so we
deterministically hit the network-failure branch.
"""
from backend import ai as aimod
# Fake key so we don't take the 'no key' short-circuit.
monkeypatch.setattr(aimod, "get_ai_key", lambda name: "fake-key-for-test")
# Patch urllib.request.urlopen to always raise — simulating network down.
# NOTE: main.py imports urllib.request at module load, so we patch the
# symbol it actually uses (urllib.request.urlopen).
import urllib.request
def _boom(*args, **kwargs):
raise OSError("simulated network down")
monkeypatch.setattr(urllib.request, "urlopen", _boom)
token = _login_admin(admin_client)
# Pick a non-Gemini provider so we hit the network code path.
resp = admin_client.get(
"/api/config/ai-models",
params={"provider": "nvidia"},
headers={"Authorization": f"Bearer {token}"},
)
assert resp.status_code == 200
data = resp.json()
# The response should be a usable list — either via fallback after
# network failure, or the 'no key' shortcut. Both are acceptable as
# long as models is non-empty.
assert data["models"], "model list should be non-empty"
assert data.get("source") in ("fallback",), (
f"unexpected source: {data.get('source')!r}"
)
def test_response_includes_source_field(self, admin_client, monkeypatch):
"""All successful responses must include a 'source' field for UI hinting."""
from backend import ai as aimod
monkeypatch.setattr(aimod, "get_ai_key", lambda name: "fake-key")
token = _login_admin(admin_client)
resp = admin_client.get(
"/api/config/ai-models",
params={"provider": "gemini"},
headers={"Authorization": f"Bearer {token}"},
)
assert resp.status_code == 200
data = resp.json()
assert "source" in data
assert data["source"] in ("live", "fallback")
def test_xiaomi_uses_api_key_header_not_bearer(self, admin_client, monkeypatch):
"""Regression test: Xiaomi MiMo must use `api-key` header, NOT Authorization.
Without this, the live call always fails with 401 even with a valid key.
"""
# Fake key — patch BOTH the source module AND the imported reference
# in backend.main (which does `from backend.ai import ... get_ai_key`).
import backend.ai as aimod
import backend.main as bmain
monkeypatch.setattr(aimod, "get_ai_key", lambda name: "fake-xiaomi-key")
if hasattr(bmain, "get_ai_key"):
monkeypatch.setattr(bmain, "get_ai_key", lambda name: "fake-xiaomi-key")
captured = {}
def fake_Request(url, *args, **kwargs):
captured["url"] = url
captured["headers"] = dict(kwargs.get("headers") or {})
class FakeResp:
def __enter__(self): return self
def __exit__(self, *a): pass
def read(self):
return b'{"object":"list","data":[{"id":"mimo-v2.5-pro","object":"model","owned_by":"xiaomi"}]}'
captured["response"] = FakeResp()
return captured["response"]
def fake_urlopen(req, timeout=None):
return req
# Patch urllib.request at the point where backend.main imported it.
import urllib.request as global_urllib_mod
monkeypatch.setattr(global_urllib_mod, "Request", fake_Request, raising=True)
monkeypatch.setattr(global_urllib_mod, "urlopen", fake_urlopen, raising=True)
token = _login_admin(admin_client)
resp = admin_client.get(
"/api/config/ai-models",
params={"provider": "xiaomi"},
headers={"Authorization": f"Bearer {token}"},
)
assert resp.status_code == 200, f"resp: {resp.text[:300]}"
# Verify the URL + headers used.
assert captured.get("url"), f"Request was not called (captured={captured!r})"
assert captured["url"].startswith("https://api.xiaomimimo.com/v1/models"), (
f"unexpected URL: {captured.get('url')!r}"
)
hdrs = {k.lower(): v for k, v in captured.get("headers", {}).items()}
assert "api-key" in hdrs, f"missing api-key header in {hdrs!r}"
assert hdrs["api-key"] == "fake-xiaomi-key"
assert "authorization" not in hdrs, f"Authorization leaked: {hdrs!r}"
def test_models_include_capabilities(self, admin_client, monkeypatch):
"""The model list must expose per-model capability flags (#81)."""
from backend import ai as aimod
monkeypatch.setattr(aimod, "get_ai_key", lambda name: None)
token = _login_admin(admin_client)
resp = admin_client.get(
"/api/config/ai-models",
params={"provider": "qwencloud"},
headers={"Authorization": f"Bearer {token}"},
)
assert resp.status_code == 200
data = resp.json()
caps = data.get("capabilities") or {}
assert "qwen-vl-max" in caps
assert caps["qwen-vl-max"]["vision"] is True
def test_model_capabilities_endpoint(self, admin_client):
"""GET /api/ai/model-capabilities returns the curated flags."""
token = _login_admin(admin_client)
resp = admin_client.get(
"/api/ai/model-capabilities",
params={"provider": "deepseek", "model": "deepseek-chat"},
headers={"Authorization": f"Bearer {token}"},
)
assert resp.status_code == 200
data = resp.json()
assert data["provider"] == "deepseek"
assert data["capabilities"]["chat"] is True
assert data["capabilities"]["vision"] is False
def test_xiaomi_test_endpoint_uses_real_url(self, admin_client, monkeypatch):
"""The /api/config/ai-keys/test endpoint must also use api.xiaomimimo.com.
Regression: it previously used api.xiaomi.com which doesn't exist
(DNS error Name or service not known).
"""
# Patch BOTH the source module AND the imported reference in backend.main
import backend.ai as aimod
import backend.main as bmain
monkeypatch.setattr(aimod, "get_ai_key", lambda name: "fake-key")
if hasattr(bmain, "get_ai_key"):
monkeypatch.setattr(bmain, "get_ai_key", lambda name: "fake-key")
import urllib.request as global_urllib_mod
captured_urls = []
captured_headers = []
def fake_urlopen(req, timeout=None):
captured_urls.append(req.full_url)
captured_headers.append(dict(req.headers))
raise OSError("simulated network error")
monkeypatch.setattr(global_urllib_mod, "urlopen", fake_urlopen, raising=True)
token = _login_admin(admin_client)
admin_client.post(
"/api/config/ai-keys/test",
headers={"Authorization": f"Bearer {token}"},
)
# Find the xiaomi URL among the captured calls.
xiaomi_idx = next(
(i for i, u in enumerate(captured_urls) if "xiaomi" in u.lower()),
None,
)
assert xiaomi_idx is not None, (
f"xiaomi URL not found in captured URLs: {captured_urls!r}"
)
xiaomi_url = captured_urls[xiaomi_idx]
# Must use the real MiMo endpoint, NOT the dead api.xiaomi.com.
assert "api.xiaomimimo.com" in xiaomi_url, (
f"xiaomi test URL is wrong: {xiaomi_url!r}"
)
# Must use api-key header, not Authorization. urllib.request
# normalizes header names to title-case ("Api-key"), but HTTP
# headers are case-insensitive on the wire — both forms are valid.
xiaomi_hdrs = {k.lower(): v for k, v in captured_headers[xiaomi_idx].items()}
assert "api-key" in xiaomi_hdrs, (
f"xiaomi api-key header missing: {xiaomi_hdrs!r}"
)
assert xiaomi_hdrs["api-key"] == "fake-key"
# Bearer prefix must NOT be in the api-key value
assert "Bearer" not in xiaomi_hdrs["api-key"]
class TestDeclaredCapabilities:
"""BUG-044 — the provider's own ``capabilities[]`` must reach the UI.
``GET /api/config/ai-models`` is what fills the capability cache; the picker
bubble (``GET /api/ai/model-capabilities``) and the vision gate then read it.
Before the fix no Mistral model was ever reported as vision-capable.
"""
def _serve_payload(self, monkeypatch, payload):
"""Make the provider models call return ``payload`` verbatim."""
import json
import urllib.request as global_urllib_mod
body = json.dumps(payload).encode()
def fake_Request(url, *args, **kwargs):
class FakeResp:
def __enter__(self): return self
def __exit__(self, *a): pass
def read(self): return body
return FakeResp()
monkeypatch.setattr(global_urllib_mod, "Request", fake_Request, raising=True)
monkeypatch.setattr(
global_urllib_mod, "urlopen", lambda req, timeout=None: req, raising=True
)
def _fake_key(self, monkeypatch):
"""main.py binds get_ai_key at import: patch that binding too."""
import backend.ai as aimod
import backend.main as bmain
monkeypatch.setattr(aimod, "get_ai_key", lambda name: "fake-mistral-key")
if hasattr(bmain, "get_ai_key"):
monkeypatch.setattr(bmain, "get_ai_key", lambda name: "fake-mistral-key")
def test_declared_vision_reaches_both_endpoints(self, admin_client, monkeypatch):
self._fake_key(monkeypatch)
self._serve_payload(monkeypatch, MISTRAL_LIVE_PAYLOAD)
token = _login_admin(admin_client)
headers = {"Authorization": f"Bearer {token}"}
listing = admin_client.get(
"/api/config/ai-models",
params={"provider": "mistral"},
headers=headers,
)
assert listing.status_code == 200, listing.text
data = listing.json()
assert data["source"] == "live"
assert data["capabilities"]["mistral-small-latest"]["vision"] is True
assert data["capabilities"]["mistral-medium-latest"]["vision"] is True
# The declaration also revokes: mistral-embed is not a chat model.
assert data["capabilities"]["mistral-embed"]["chat"] is False
assert data["capabilities"]["mistral-embed"]["embeddings"] is True
for model in ("mistral-small-latest", "mistral-medium-latest"):
detail = admin_client.get(
"/api/ai/model-capabilities",
params={"provider": "mistral", "model": model},
headers=headers,
)
assert detail.status_code == 200, detail.text
assert detail.json()["capabilities"]["vision"] is True, model
def test_text_only_mistral_models_stay_text_only(self, admin_client, monkeypatch):
"""mistral-large-latest declares no vision in the real API — keep it so."""
self._fake_key(monkeypatch)
self._serve_payload(monkeypatch, MISTRAL_LIVE_PAYLOAD)
token = _login_admin(admin_client)
resp = admin_client.get(
"/api/config/ai-models",
params={"provider": "mistral"},
headers={"Authorization": f"Bearer {token}"},
)
assert resp.status_code == 200, resp.text
caps = resp.json()["capabilities"]["mistral-large-latest"]
assert caps["vision"] is False
assert caps["embeddings"] is False
def test_curated_fallback_covers_mistral_without_any_api_call(
self, admin_client, monkeypatch,
):
"""No API key: the curated list must still answer correctly (offline)."""
import backend.ai as aimod
import backend.main as bmain
monkeypatch.setattr(aimod, "get_ai_key", lambda name: None)
if hasattr(bmain, "get_ai_key"):
monkeypatch.setattr(bmain, "get_ai_key", lambda name: None)
token = _login_admin(admin_client)
resp = admin_client.get(
"/api/config/ai-models",
params={"provider": "mistral"},
headers={"Authorization": f"Bearer {token}"},
)
assert resp.status_code == 200, resp.text
data = resp.json()
assert data["source"] == "fallback"
caps = data["capabilities"]
assert caps["mistral-small-latest"]["vision"] is True
assert caps["mistral-large-latest"]["vision"] is False
assert caps["mistral-large-latest"]["embeddings"] is False