ObsiGate/tests/test_ai_models.py

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


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


# ── 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")