test: add tests for /health/detailed endpoint and gateway health probe

- TestHealthDetailedEndpoint: 3 tests for the new API server endpoint
  (returns runtime data, handles missing status, no auth required)
- TestProbeGatewayHealth: 5 tests for _probe_gateway_health()
  (URL normalization, successful/failed probes, fallback chain)
- TestStatusRemoteGateway: 4 tests for /api/status remote fallback
  (remote probe triggers, skipped when local PID found, null PID handling)
This commit is contained in:
Teknium 2026-04-14 15:39:35 -07:00 committed by Teknium
parent 139a5e37a4
commit 353b5bacbd
2 changed files with 245 additions and 0 deletions

View file

@ -220,6 +220,7 @@ def _create_app(adapter: APIServerAdapter) -> web.Application:
app = web.Application(middlewares=mws)
app["api_server_adapter"] = adapter
app.router.add_get("/health", adapter._handle_health)
app.router.add_get("/health/detailed", adapter._handle_health_detailed)
app.router.add_get("/v1/health", adapter._handle_health)
app.router.add_get("/v1/models", adapter._handle_models)
app.router.add_post("/v1/chat/completions", adapter._handle_chat_completions)
@ -277,6 +278,58 @@ class TestHealthEndpoint:
assert data["platform"] == "hermes-agent"
# ---------------------------------------------------------------------------
# /health/detailed endpoint
# ---------------------------------------------------------------------------
class TestHealthDetailedEndpoint:
@pytest.mark.asyncio
async def test_health_detailed_returns_ok(self, adapter):
"""GET /health/detailed returns status, platform, and runtime fields."""
app = _create_app(adapter)
with patch("gateway.status.read_runtime_status", return_value={
"gateway_state": "running",
"platforms": {"telegram": {"state": "connected"}},
"active_agents": 2,
"exit_reason": None,
"updated_at": "2026-04-14T00:00:00Z",
}):
async with TestClient(TestServer(app)) as cli:
resp = await cli.get("/health/detailed")
assert resp.status == 200
data = await resp.json()
assert data["status"] == "ok"
assert data["platform"] == "hermes-agent"
assert data["gateway_state"] == "running"
assert data["platforms"] == {"telegram": {"state": "connected"}}
assert data["active_agents"] == 2
assert isinstance(data["pid"], int)
assert "updated_at" in data
@pytest.mark.asyncio
async def test_health_detailed_no_runtime_status(self, adapter):
"""When gateway_state.json is missing, fields are None."""
app = _create_app(adapter)
with patch("gateway.status.read_runtime_status", return_value=None):
async with TestClient(TestServer(app)) as cli:
resp = await cli.get("/health/detailed")
assert resp.status == 200
data = await resp.json()
assert data["status"] == "ok"
assert data["gateway_state"] is None
assert data["platforms"] == {}
@pytest.mark.asyncio
async def test_health_detailed_does_not_require_auth(self, auth_adapter):
"""Health detailed endpoint should be accessible without auth, like /health."""
app = _create_app(auth_adapter)
with patch("gateway.status.read_runtime_status", return_value=None):
async with TestClient(TestServer(app)) as cli:
resp = await cli.get("/health/detailed")
assert resp.status == 200
# ---------------------------------------------------------------------------
# /v1/models endpoint
# ---------------------------------------------------------------------------

View file

@ -984,3 +984,195 @@ class TestModelInfoEndpoint:
assert resp.status_code == 200
data = resp.json()
assert data["auto_context_length"] == 0
# ---------------------------------------------------------------------------
# Gateway health probe tests
# ---------------------------------------------------------------------------
class TestProbeGatewayHealth:
"""Tests for _probe_gateway_health() — cross-container gateway detection."""
def test_returns_false_when_no_url_configured(self, monkeypatch):
"""When GATEWAY_HEALTH_URL is unset, the probe returns (False, None)."""
import hermes_cli.web_server as ws
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", None)
alive, body = ws._probe_gateway_health()
assert alive is False
assert body is None
def test_normalizes_url_with_health_suffix(self, monkeypatch):
"""If the user sets the URL to include /health, it's stripped to base."""
import hermes_cli.web_server as ws
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642/health")
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_TIMEOUT", 1)
# Both paths should fail (no server), but we verify they were constructed
# correctly by checking the URLs attempted.
calls = []
original_urlopen = ws.urllib.request.urlopen
def mock_urlopen(req, **kwargs):
calls.append(req.full_url)
raise ConnectionError("mock")
monkeypatch.setattr(ws.urllib.request, "urlopen", mock_urlopen)
alive, body = ws._probe_gateway_health()
assert alive is False
assert "http://gw:8642/health/detailed" in calls
assert "http://gw:8642/health" in calls
def test_normalizes_url_with_health_detailed_suffix(self, monkeypatch):
"""If the user sets the URL to include /health/detailed, it's stripped to base."""
import hermes_cli.web_server as ws
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642/health/detailed")
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_TIMEOUT", 1)
calls = []
def mock_urlopen(req, **kwargs):
calls.append(req.full_url)
raise ConnectionError("mock")
monkeypatch.setattr(ws.urllib.request, "urlopen", mock_urlopen)
ws._probe_gateway_health()
assert "http://gw:8642/health/detailed" in calls
assert "http://gw:8642/health" in calls
def test_successful_detailed_probe(self, monkeypatch):
"""Successful /health/detailed probe returns (True, body_dict)."""
import hermes_cli.web_server as ws
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642")
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_TIMEOUT", 1)
response_body = json.dumps({
"status": "ok",
"gateway_state": "running",
"pid": 42,
})
mock_resp = MagicMock()
mock_resp.status = 200
mock_resp.read.return_value = response_body.encode()
mock_resp.__enter__ = MagicMock(return_value=mock_resp)
mock_resp.__exit__ = MagicMock(return_value=False)
monkeypatch.setattr(ws.urllib.request, "urlopen", lambda req, **kw: mock_resp)
alive, body = ws._probe_gateway_health()
assert alive is True
assert body["status"] == "ok"
assert body["pid"] == 42
def test_detailed_fails_falls_back_to_simple_health(self, monkeypatch):
"""If /health/detailed fails, falls back to /health."""
import hermes_cli.web_server as ws
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642")
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_TIMEOUT", 1)
call_count = [0]
def mock_urlopen(req, **kwargs):
call_count[0] += 1
if call_count[0] == 1:
raise ConnectionError("detailed failed")
mock_resp = MagicMock()
mock_resp.status = 200
mock_resp.read.return_value = json.dumps({"status": "ok"}).encode()
mock_resp.__enter__ = MagicMock(return_value=mock_resp)
mock_resp.__exit__ = MagicMock(return_value=False)
return mock_resp
monkeypatch.setattr(ws.urllib.request, "urlopen", mock_urlopen)
alive, body = ws._probe_gateway_health()
assert alive is True
assert body["status"] == "ok"
assert call_count[0] == 2
class TestStatusRemoteGateway:
"""Tests for /api/status with remote gateway health fallback."""
@pytest.fixture(autouse=True)
def _setup_test_client(self):
try:
from starlette.testclient import TestClient
except ImportError:
pytest.skip("fastapi/starlette not installed")
from hermes_cli.web_server import app, _SESSION_TOKEN
self.client = TestClient(app)
self.client.headers["Authorization"] = f"Bearer {_SESSION_TOKEN}"
def test_status_falls_back_to_remote_probe(self, monkeypatch):
"""When local PID check fails and remote probe succeeds, gateway shows running."""
import hermes_cli.web_server as ws
monkeypatch.setattr(ws, "get_running_pid", lambda: None)
monkeypatch.setattr(ws, "read_runtime_status", lambda: None)
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642")
monkeypatch.setattr(ws, "_probe_gateway_health", lambda: (True, {
"status": "ok",
"gateway_state": "running",
"platforms": {"telegram": {"state": "connected"}},
"pid": 999,
}))
resp = self.client.get("/api/status")
assert resp.status_code == 200
data = resp.json()
assert data["gateway_running"] is True
assert data["gateway_pid"] == 999
assert data["gateway_state"] == "running"
def test_status_remote_probe_not_attempted_when_local_pid_found(self, monkeypatch):
"""When local PID check succeeds, the remote probe is never called."""
import hermes_cli.web_server as ws
monkeypatch.setattr(ws, "get_running_pid", lambda: 1234)
monkeypatch.setattr(ws, "read_runtime_status", lambda: {
"gateway_state": "running",
"platforms": {},
})
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642")
probe_called = [False]
original = ws._probe_gateway_health
def track_probe():
probe_called[0] = True
return original()
monkeypatch.setattr(ws, "_probe_gateway_health", track_probe)
resp = self.client.get("/api/status")
assert resp.status_code == 200
assert not probe_called[0]
def test_status_remote_probe_not_attempted_when_no_url(self, monkeypatch):
"""When GATEWAY_HEALTH_URL is unset, no probe is attempted."""
import hermes_cli.web_server as ws
monkeypatch.setattr(ws, "get_running_pid", lambda: None)
monkeypatch.setattr(ws, "read_runtime_status", lambda: None)
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", None)
resp = self.client.get("/api/status")
assert resp.status_code == 200
data = resp.json()
assert data["gateway_running"] is False
def test_status_remote_running_null_pid(self, monkeypatch):
"""Remote gateway running but PID not in response — pid should be None."""
import hermes_cli.web_server as ws
monkeypatch.setattr(ws, "get_running_pid", lambda: None)
monkeypatch.setattr(ws, "read_runtime_status", lambda: None)
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642")
monkeypatch.setattr(ws, "_probe_gateway_health", lambda: (True, {
"status": "ok",
}))
resp = self.client.get("/api/status")
assert resp.status_code == 200
data = resp.json()
assert data["gateway_running"] is True
assert data["gateway_pid"] is None
assert data["gateway_state"] == "running"