118 lines
3.8 KiB
Python
118 lines
3.8 KiB
Python
"""Tests for the central AI monitoring routes (Val 1).
|
|
|
|
External sources (service health, RabbitMQ, Prometheus) are mocked with respx /
|
|
monkeypatch — no real infra needed.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
|
|
from dashboard.api.routes import monitoring
|
|
from dashboard.config import DashboardSettings, SettingsCache
|
|
|
|
|
|
def _settings(**overrides) -> DashboardSettings:
|
|
base = {
|
|
"database_url": "postgresql+asyncpg://u:p@localhost/db",
|
|
"external_url": "http://localhost:51300",
|
|
}
|
|
base.update(overrides)
|
|
s = DashboardSettings(**base)
|
|
SettingsCache.set(s)
|
|
return s
|
|
|
|
|
|
# --- services aggregation -------------------------------------------------
|
|
|
|
|
|
async def test_services_summary(monkeypatch):
|
|
_settings()
|
|
|
|
async def fake_probe(client, mid, url):
|
|
status = "down" if mid == "video" else "healthy"
|
|
return {"module": mid, "url": url, "status": status, "latency_ms": 1.0}
|
|
|
|
monkeypatch.setattr(monitoring, "_probe", fake_probe)
|
|
out = await monitoring.monitoring_services()
|
|
|
|
assert out["summary"]["total"] == len(monitoring._service_endpoints())
|
|
assert out["summary"]["down"] == 1
|
|
assert out["summary"]["healthy"] == out["summary"]["total"] - 1
|
|
assert any(s["module"] == "video" and s["status"] == "down" for s in out["services"])
|
|
|
|
|
|
# --- RabbitMQ queues ------------------------------------------------------
|
|
|
|
|
|
async def test_queues_disabled_when_unset():
|
|
_settings(rabbitmq_mgmt_url=None)
|
|
out = await monitoring.monitoring_queues()
|
|
assert out["enabled"] is False
|
|
assert out["queues"] == []
|
|
|
|
|
|
@respx.mock
|
|
async def test_queues_parsed_and_sorted():
|
|
_settings(rabbitmq_mgmt_url="http://rmq:15672")
|
|
respx.get("http://rmq:15672/api/queues").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
json=[
|
|
{"name": "low", "vhost": "/", "messages": 2, "messages_ready": 2,
|
|
"messages_unacknowledged": 0, "consumers": 1, "state": "running"},
|
|
{"name": "busy", "vhost": "/", "messages": 50, "messages_ready": 48,
|
|
"messages_unacknowledged": 2, "consumers": 3, "state": "running"},
|
|
],
|
|
)
|
|
)
|
|
out = await monitoring.monitoring_queues()
|
|
assert out["enabled"] is True
|
|
# sorted by messages desc
|
|
assert [q["name"] for q in out["queues"]] == ["busy", "low"]
|
|
assert out["queues"][0]["unacked"] == 2
|
|
|
|
|
|
@respx.mock
|
|
async def test_queues_failopen_on_error():
|
|
_settings(rabbitmq_mgmt_url="http://rmq:15672")
|
|
respx.get("http://rmq:15672/api/queues").mock(side_effect=httpx.ConnectError("x"))
|
|
out = await monitoring.monitoring_queues()
|
|
assert out["enabled"] is True
|
|
assert out["queues"] == []
|
|
assert "error" in out
|
|
|
|
|
|
# --- Prometheus latency ---------------------------------------------------
|
|
|
|
|
|
async def test_latency_disabled_when_unset():
|
|
_settings(prometheus_url=None)
|
|
out = await monitoring.monitoring_latency()
|
|
assert out["enabled"] is False
|
|
|
|
|
|
@respx.mock
|
|
async def test_latency_parsed_per_job():
|
|
_settings(prometheus_url="http://prom:9090")
|
|
# Same response for all three quantile queries → equal p50/p90/p99.
|
|
respx.get("http://prom:9090/api/v1/query").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
json={
|
|
"data": {
|
|
"result": [
|
|
{"metric": {"job": "llm"}, "value": [0, "0.250"]},
|
|
{"metric": {"job": "web"}, "value": [0, "1.500"]},
|
|
]
|
|
}
|
|
},
|
|
)
|
|
)
|
|
out = await monitoring.monitoring_latency()
|
|
assert out["enabled"] is True
|
|
by_job = {s["job"]: s for s in out["services"]}
|
|
assert by_job["llm"]["p50_ms"] == 250.0 # 0.250s → 250ms
|
|
assert by_job["web"]["p99_ms"] == 1500.0
|