3e9b2dc3b8
Implements a comprehensive operational monitoring UI at /monitoring covering all subsystems described in issue #862: - Agent Status: lists configured agents with name, model, status, last action - System Resources: RAM/disk/CPU usage with live progress bars via psutil - Economy: sats balance, earned/spent, transaction count (Lightning ledger) - Stream Health: viewer count, bitrate, uptime (graceful fallback when offline) - Content Pipeline: episode/highlight/clip counts from data/episodes/ - Alerts: auto-derived from resource thresholds, Ollama reachability, wallet balance Implementation details: - New route: GET /monitoring (HTML page), GET /monitoring/status (JSON), GET /monitoring/alerts (JSON) - /monitoring/status aggregates all subsystems concurrently with asyncio.gather - Frontend polls every 10 seconds with vanilla JS (no blocking) - All optional services degrade gracefully per project convention - CSS appended to mission-control.css (no inline styles) - "MONITORING" link added to desktop nav in base.html - 13 unit tests covering page render and all API endpoints Fixes #862 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
96 lines
3.6 KiB
Python
96 lines
3.6 KiB
Python
"""Tests for the real-time monitoring dashboard routes. Refs: #862"""
|
|
|
|
|
|
class TestMonitoringPage:
|
|
"""Tests for the monitoring dashboard HTML page."""
|
|
|
|
def test_monitoring_page_returns_200(self, client):
|
|
response = client.get("/monitoring")
|
|
assert response.status_code == 200
|
|
|
|
def test_monitoring_page_contains_key_headings(self, client):
|
|
response = client.get("/monitoring")
|
|
assert response.status_code == 200
|
|
body = response.text
|
|
assert "Real-Time Monitoring" in body
|
|
assert "Agent Status" in body
|
|
assert "System Resources" in body
|
|
assert "Economy" in body
|
|
assert "Stream Health" in body
|
|
assert "Content Pipeline" in body
|
|
|
|
|
|
class TestMonitoringStatusEndpoint:
|
|
"""Tests for /monitoring/status JSON endpoint."""
|
|
|
|
def test_status_returns_200(self, client):
|
|
response = client.get("/monitoring/status")
|
|
assert response.status_code == 200
|
|
|
|
def test_status_has_required_keys(self, client):
|
|
response = client.get("/monitoring/status")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
for key in ("timestamp", "uptime_seconds", "agents", "resources", "economy", "stream", "pipeline", "alerts"):
|
|
assert key in data, f"Missing key: {key}"
|
|
|
|
def test_agents_is_list(self, client):
|
|
response = client.get("/monitoring/status")
|
|
data = response.json()
|
|
assert isinstance(data["agents"], list)
|
|
|
|
def test_alerts_is_list(self, client):
|
|
response = client.get("/monitoring/status")
|
|
data = response.json()
|
|
assert isinstance(data["alerts"], list)
|
|
|
|
def test_resources_has_expected_fields(self, client):
|
|
response = client.get("/monitoring/status")
|
|
data = response.json()
|
|
resources = data["resources"]
|
|
for field in ("disk_percent", "disk_free_gb", "ollama_reachable", "loaded_models", "warnings"):
|
|
assert field in resources, f"Missing resource field: {field}"
|
|
|
|
def test_economy_has_expected_fields(self, client):
|
|
response = client.get("/monitoring/status")
|
|
data = response.json()
|
|
economy = data["economy"]
|
|
for field in ("balance_sats", "earned_sats", "spent_sats", "tx_count"):
|
|
assert field in economy, f"Missing economy field: {field}"
|
|
|
|
def test_stream_has_expected_fields(self, client):
|
|
response = client.get("/monitoring/status")
|
|
data = response.json()
|
|
stream = data["stream"]
|
|
for field in ("live", "viewer_count", "bitrate_kbps", "uptime_seconds"):
|
|
assert field in stream, f"Missing stream field: {field}"
|
|
|
|
def test_uptime_is_non_negative(self, client):
|
|
response = client.get("/monitoring/status")
|
|
data = response.json()
|
|
assert data["uptime_seconds"] >= 0
|
|
|
|
|
|
class TestMonitoringAlertsEndpoint:
|
|
"""Tests for /monitoring/alerts JSON endpoint."""
|
|
|
|
def test_alerts_returns_200(self, client):
|
|
response = client.get("/monitoring/alerts")
|
|
assert response.status_code == 200
|
|
|
|
def test_alerts_has_alerts_and_count(self, client):
|
|
response = client.get("/monitoring/alerts")
|
|
data = response.json()
|
|
assert "alerts" in data
|
|
assert "count" in data
|
|
assert isinstance(data["alerts"], list)
|
|
assert data["count"] == len(data["alerts"])
|
|
|
|
def test_alert_items_have_level_and_title(self, client):
|
|
response = client.get("/monitoring/alerts")
|
|
data = response.json()
|
|
for alert in data["alerts"]:
|
|
assert "level" in alert
|
|
assert "title" in alert
|
|
assert alert["level"] in ("info", "warning", "critical")
|