tests/test_ping.py

Rama main · réplica de lectura del repositorio original.

"""Tests du diagnostic /ping (chemin LLM maillon par maillon)."""

import httpx
import pytest

from app import ping


class FakeResponse:
    def __init__(self, status_code=200, payload=None):
        self.status_code = status_code
        self._payload = payload or {}

    def json(self):
        return self._payload


class FakeClient:
    """httpx.AsyncClient minimal : réponses par (méthode, url-part)."""

    def __init__(self, routes):
        self._routes = routes

    async def __aenter__(self):
        return self

    async def __aexit__(self, *exc):
        return False

    async def get(self, url):
        for pattern, response in self._routes.get("GET", []):
            if pattern in url:
                return response
        raise httpx.ConnectError("no route")

    async def post(self, url, json=None):
        for pattern, response in self._routes.get("POST", []):
            if pattern in url:
                return response
        raise httpx.ConnectError("no route")


@pytest.fixture
def healthy(monkeypatch):
    """Tous les maillons répondent (modèle = celui configuré du module)."""
    model = ping.OLLAMA_MODEL
    monkeypatch.setattr(
        ping.httpx,
        "AsyncClient",
        lambda **k: FakeClient({
            "GET": [
                ("/api/tags", FakeResponse(payload={"models": [{"name": model}]})),
                ("/api/ps", FakeResponse(payload={"models": [{"name": model}]})),
            ],
            "POST": [("/api/generate", FakeResponse(payload={"response": "OK"}))],
        }),
    )


async def test_ping_tous_maillons_verts(healthy):
    report = await ping.ping_report()
    assert "🟢 Bridge Ollama" in report
    assert "présent côté Ollama" in report
    assert "chargé en RAM" in report
    assert "✅" in report and "opérationnel" in report


async def test_ping_bridge_mort_nomme_le_maillon(monkeypatch):
    def boom(**k):
        raise httpx.ConnectError("refused")

    monkeypatch.setattr(ping.httpx, "AsyncClient", boom)
    report = await ping.ping_report()
    assert "🔴 Bridge Ollama" in report
    assert "vektor-ollama-bridge" in report  # la piste de réparation est citée
    assert "Premier maillon cassé" in report


async def test_ping_modele_absent_le_dit(monkeypatch):
    monkeypatch.setattr(
        ping.httpx,
        "AsyncClient",
        lambda **k: FakeClient({
            "GET": [
                ("/api/tags", FakeResponse(payload={"models": [{"name": "autre:7b"}]})),
                ("/api/ps", FakeResponse(payload={"models": []})),
            ],
            "POST": [("/api/generate", FakeResponse(payload={"response": "OK"}))],
        }),
    )
    report = await ping.ping_report()
    assert f"🔴 Modèle `{ping.OLLAMA_MODEL}` ABSENT" in report
    assert "Premier maillon cassé" in report


async def test_ping_modele_non_charge_est_un_avertissement_pas_une_panne(monkeypatch):
    model = ping.OLLAMA_MODEL
    monkeypatch.setattr(
        ping.httpx,
        "AsyncClient",
        lambda **k: FakeClient({
            "GET": [
                ("/api/tags", FakeResponse(payload={"models": [{"name": model}]})),
                ("/api/ps", FakeResponse(payload={"models": []})),  # pas en RAM
            ],
            "POST": [("/api/generate", FakeResponse(payload={"response": "OK"}))],
        }),
    )
    report = await ping.ping_report()
    assert "🟡" in report  # warning, pas 🔴
    assert "pas une panne" in report