103 lines
2.9 KiB
Python
103 lines
2.9 KiB
Python
"""Tests for Festinger health reporting used by gnommoweb state-machine checks."""
|
|
import asyncio
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from festinger import main
|
|
|
|
|
|
class DummyResponse:
|
|
def __init__(self, status_code=200, payload=None, text=""):
|
|
self.status_code = status_code
|
|
self._payload = payload or {}
|
|
self.text = text
|
|
|
|
@property
|
|
def is_success(self):
|
|
return 200 <= self.status_code < 300
|
|
|
|
def json(self):
|
|
return self._payload
|
|
|
|
|
|
class FakeAsyncClient:
|
|
requested_urls = []
|
|
|
|
def __init__(self, *, timeout):
|
|
self.timeout = timeout
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
async def get(self, url):
|
|
self.requested_urls.append(url)
|
|
return DummyResponse(
|
|
200,
|
|
{
|
|
"object": "list",
|
|
"data": [
|
|
{"id": "local-model", "object": "model"},
|
|
],
|
|
},
|
|
)
|
|
|
|
|
|
def test_probe_openai_compatible_upstream_reports_models_without_double_v1(monkeypatch):
|
|
FakeAsyncClient.requested_urls = []
|
|
monkeypatch.setattr(main.httpx, "AsyncClient", FakeAsyncClient)
|
|
|
|
result = asyncio.run(
|
|
main._probe_upstream_connection(
|
|
name="lm_studio",
|
|
base_url="http://lmstudio.local:1234/v1",
|
|
probe_path="/v1/models",
|
|
)
|
|
)
|
|
|
|
assert result["ok"] is True
|
|
assert result["status"] == "ok"
|
|
assert result["url"] == "http://lmstudio.local:1234/v1/models"
|
|
assert FakeAsyncClient.requested_urls == ["http://lmstudio.local:1234/v1/models"]
|
|
assert result["models"] == ["local-model"]
|
|
assert result["status_message"] == "lm_studio reachable: 1 model(s) available"
|
|
|
|
|
|
def test_health_endpoint_includes_connection_status_messages(monkeypatch):
|
|
async def fake_probe(name, base_url, probe_path):
|
|
return {
|
|
"name": name,
|
|
"ok": True,
|
|
"status": "ok",
|
|
"url": f"{base_url.rstrip('/')}{probe_path}",
|
|
"status_code": 200,
|
|
"latency_ms": 12,
|
|
"models": ["local-model"] if name == "lm_studio" else [],
|
|
"status_message": f"{name} reachable",
|
|
}
|
|
|
|
monkeypatch.setattr(main, "_probe_upstream_connection", fake_probe)
|
|
request = SimpleNamespace(
|
|
app=SimpleNamespace(
|
|
state=SimpleNamespace(
|
|
yaml_config={
|
|
"upstream_ollama": "http://ollama.local:11434",
|
|
"upstream_openai": "http://lmstudio.local:1234",
|
|
}
|
|
)
|
|
)
|
|
)
|
|
|
|
result = asyncio.run(main.health(request))
|
|
|
|
assert result["status"] == "ok"
|
|
assert result["connections"]["lm_studio"]["ok"] is True
|
|
assert result["connections"]["ollama"]["ok"] is True
|
|
assert result["status_messages"] == [
|
|
"lm_studio reachable",
|
|
"ollama reachable",
|
|
]
|