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

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