test: add unit tests for health.py route module
Adds 41 unit tests covering: - Pydantic models (DependencyStatus, SovereigntyReport, HealthStatus) - Helper functions (_calculate_overall_score, _generate_recommendations) - Ollama sync probe (_check_ollama_sync) with mock HTTP responses - Async Ollama check with TTL cache behavior (hit, miss, expiry) - Legacy bool wrapper (check_ollama) - SQLite and Lightning dependency checks - All five endpoints (/health, /health/status, /health/sovereignty, /health/components, /health/snapshot) - Graceful degradation when Ollama is down or snapshot module fails Fixes #945 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
496
tests/dashboard/test_health.py
Normal file
496
tests/dashboard/test_health.py
Normal file
@@ -0,0 +1,496 @@
|
||||
"""Unit tests for dashboard/routes/health.py.
|
||||
|
||||
Covers helper functions, caching, endpoint responses, and graceful
|
||||
degradation when subsystems (Ollama, SQLite) are unavailable.
|
||||
|
||||
Fixes #945
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from dashboard.routes.health import (
|
||||
DependencyStatus,
|
||||
HealthStatus,
|
||||
SovereigntyReport,
|
||||
_calculate_overall_score,
|
||||
_check_lightning,
|
||||
_check_ollama_sync,
|
||||
_check_sqlite,
|
||||
_generate_recommendations,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pydantic models
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestDependencyStatusModel:
|
||||
"""Validate DependencyStatus model."""
|
||||
|
||||
def test_fields(self):
|
||||
dep = DependencyStatus(
|
||||
name="Test", status="healthy", sovereignty_score=8, details={"key": "val"}
|
||||
)
|
||||
assert dep.name == "Test"
|
||||
assert dep.status == "healthy"
|
||||
assert dep.sovereignty_score == 8
|
||||
assert dep.details == {"key": "val"}
|
||||
|
||||
def test_empty_details(self):
|
||||
dep = DependencyStatus(name="X", status="unavailable", sovereignty_score=0, details={})
|
||||
assert dep.details == {}
|
||||
|
||||
|
||||
class TestSovereigntyReportModel:
|
||||
"""Validate SovereigntyReport model."""
|
||||
|
||||
def test_fields(self):
|
||||
report = SovereigntyReport(
|
||||
overall_score=9.3,
|
||||
dependencies=[],
|
||||
timestamp="2026-01-01T00:00:00+00:00",
|
||||
recommendations=["All good"],
|
||||
)
|
||||
assert report.overall_score == 9.3
|
||||
assert report.dependencies == []
|
||||
assert report.recommendations == ["All good"]
|
||||
|
||||
|
||||
class TestHealthStatusModel:
|
||||
"""Validate HealthStatus model."""
|
||||
|
||||
def test_fields(self):
|
||||
hs = HealthStatus(
|
||||
status="ok",
|
||||
timestamp="2026-01-01T00:00:00+00:00",
|
||||
version="2.0.0",
|
||||
uptime_seconds=42.5,
|
||||
)
|
||||
assert hs.status == "ok"
|
||||
assert hs.uptime_seconds == 42.5
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helper functions
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCalculateOverallScore:
|
||||
"""Test _calculate_overall_score."""
|
||||
|
||||
def test_empty_deps(self):
|
||||
assert _calculate_overall_score([]) == 0.0
|
||||
|
||||
def test_single_dep(self):
|
||||
deps = [DependencyStatus(name="A", status="healthy", sovereignty_score=7, details={})]
|
||||
assert _calculate_overall_score(deps) == 7.0
|
||||
|
||||
def test_averages_multiple(self):
|
||||
deps = [
|
||||
DependencyStatus(name="A", status="healthy", sovereignty_score=10, details={}),
|
||||
DependencyStatus(name="B", status="healthy", sovereignty_score=8, details={}),
|
||||
DependencyStatus(name="C", status="unavailable", sovereignty_score=6, details={}),
|
||||
]
|
||||
assert _calculate_overall_score(deps) == 8.0
|
||||
|
||||
def test_rounding(self):
|
||||
deps = [
|
||||
DependencyStatus(name="A", status="healthy", sovereignty_score=10, details={}),
|
||||
DependencyStatus(name="B", status="healthy", sovereignty_score=9, details={}),
|
||||
DependencyStatus(name="C", status="healthy", sovereignty_score=10, details={}),
|
||||
]
|
||||
assert _calculate_overall_score(deps) == 9.7
|
||||
|
||||
|
||||
class TestGenerateRecommendations:
|
||||
"""Test _generate_recommendations."""
|
||||
|
||||
def test_all_healthy(self):
|
||||
deps = [DependencyStatus(name="X", status="healthy", sovereignty_score=10, details={})]
|
||||
recs = _generate_recommendations(deps)
|
||||
assert recs == ["System operating optimally - all dependencies healthy"]
|
||||
|
||||
def test_unavailable_service(self):
|
||||
deps = [
|
||||
DependencyStatus(name="Ollama AI", status="unavailable", sovereignty_score=10, details={})
|
||||
]
|
||||
recs = _generate_recommendations(deps)
|
||||
assert any("Ollama AI is unavailable" in r for r in recs)
|
||||
|
||||
def test_degraded_lightning_mock(self):
|
||||
deps = [
|
||||
DependencyStatus(
|
||||
name="Lightning Payments",
|
||||
status="degraded",
|
||||
sovereignty_score=8,
|
||||
details={"backend": "mock"},
|
||||
)
|
||||
]
|
||||
recs = _generate_recommendations(deps)
|
||||
assert any("Switch to real Lightning" in r for r in recs)
|
||||
|
||||
def test_degraded_non_lightning(self):
|
||||
"""Degraded non-Lightning dep produces no specific recommendation."""
|
||||
deps = [
|
||||
DependencyStatus(name="Redis", status="degraded", sovereignty_score=5, details={})
|
||||
]
|
||||
recs = _generate_recommendations(deps)
|
||||
assert recs == ["System operating optimally - all dependencies healthy"]
|
||||
|
||||
def test_multiple_unavailable(self):
|
||||
deps = [
|
||||
DependencyStatus(name="A", status="unavailable", sovereignty_score=5, details={}),
|
||||
DependencyStatus(name="B", status="unavailable", sovereignty_score=5, details={}),
|
||||
]
|
||||
recs = _generate_recommendations(deps)
|
||||
assert len(recs) == 2
|
||||
assert "A is unavailable" in recs[0]
|
||||
assert "B is unavailable" in recs[1]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _check_lightning (static)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCheckLightning:
|
||||
"""Test _check_lightning — always returns unavailable for now."""
|
||||
|
||||
def test_returns_unavailable(self):
|
||||
dep = _check_lightning()
|
||||
assert dep.name == "Lightning Payments"
|
||||
assert dep.status == "unavailable"
|
||||
assert dep.sovereignty_score == 8
|
||||
assert "removed" in dep.details.get("note", "").lower()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _check_ollama_sync
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCheckOllamaSync:
|
||||
"""Test synchronous Ollama health probe."""
|
||||
|
||||
def test_healthy_when_reachable(self):
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.status = 200
|
||||
mock_resp.__enter__ = MagicMock(return_value=mock_resp)
|
||||
mock_resp.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
with patch("urllib.request.urlopen", return_value=mock_resp):
|
||||
dep = _check_ollama_sync()
|
||||
|
||||
assert dep.status == "healthy"
|
||||
assert dep.name == "Ollama AI"
|
||||
assert dep.sovereignty_score == 10
|
||||
|
||||
def test_unavailable_on_connection_error(self):
|
||||
with patch(
|
||||
"urllib.request.urlopen",
|
||||
side_effect=ConnectionError("refused"),
|
||||
):
|
||||
dep = _check_ollama_sync()
|
||||
|
||||
assert dep.status == "unavailable"
|
||||
assert "Cannot connect" in dep.details.get("error", "")
|
||||
|
||||
def test_unavailable_on_timeout(self):
|
||||
from urllib.error import URLError
|
||||
|
||||
with patch(
|
||||
"urllib.request.urlopen",
|
||||
side_effect=URLError("timeout"),
|
||||
):
|
||||
dep = _check_ollama_sync()
|
||||
|
||||
assert dep.status == "unavailable"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _check_sqlite
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCheckSQLite:
|
||||
"""Test SQLite health probe."""
|
||||
|
||||
def test_healthy_when_db_reachable(self, tmp_path):
|
||||
import sqlite3
|
||||
|
||||
db_path = tmp_path / "data" / "timmy.db"
|
||||
db_path.parent.mkdir(parents=True)
|
||||
sqlite3.connect(str(db_path)).close()
|
||||
|
||||
with patch("dashboard.routes.health.settings") as mock_settings:
|
||||
mock_settings.repo_root = str(tmp_path)
|
||||
dep = _check_sqlite()
|
||||
|
||||
assert dep.status == "healthy"
|
||||
assert dep.name == "SQLite Database"
|
||||
|
||||
def test_unavailable_on_missing_db(self, tmp_path):
|
||||
with patch("dashboard.routes.health.settings") as mock_settings:
|
||||
mock_settings.repo_root = str(tmp_path / "nonexistent")
|
||||
dep = _check_sqlite()
|
||||
|
||||
assert dep.status == "unavailable"
|
||||
assert "error" in dep.details
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _check_ollama (async, with caching)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCheckOllamaAsync:
|
||||
"""Test async Ollama check with TTL cache."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_cache(self):
|
||||
"""Clear the module-level Ollama cache before each test."""
|
||||
import dashboard.routes.health as mod
|
||||
|
||||
mod._ollama_cache = None
|
||||
mod._ollama_cache_ts = 0.0
|
||||
yield
|
||||
mod._ollama_cache = None
|
||||
mod._ollama_cache_ts = 0.0
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_returns_dependency_status(self):
|
||||
healthy = DependencyStatus(
|
||||
name="Ollama AI", status="healthy", sovereignty_score=10, details={}
|
||||
)
|
||||
with patch(
|
||||
"dashboard.routes.health._check_ollama_sync",
|
||||
return_value=healthy,
|
||||
):
|
||||
from dashboard.routes.health import _check_ollama
|
||||
|
||||
result = await _check_ollama()
|
||||
|
||||
assert result.status == "healthy"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_caches_result(self):
|
||||
healthy = DependencyStatus(
|
||||
name="Ollama AI", status="healthy", sovereignty_score=10, details={}
|
||||
)
|
||||
with patch(
|
||||
"dashboard.routes.health._check_ollama_sync",
|
||||
return_value=healthy,
|
||||
) as mock_sync:
|
||||
from dashboard.routes.health import _check_ollama
|
||||
|
||||
await _check_ollama()
|
||||
await _check_ollama()
|
||||
|
||||
# Should only call the sync function once due to cache
|
||||
assert mock_sync.call_count == 1
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cache_expires(self):
|
||||
healthy = DependencyStatus(
|
||||
name="Ollama AI", status="healthy", sovereignty_score=10, details={}
|
||||
)
|
||||
import dashboard.routes.health as mod
|
||||
|
||||
with patch(
|
||||
"dashboard.routes.health._check_ollama_sync",
|
||||
return_value=healthy,
|
||||
) as mock_sync:
|
||||
from dashboard.routes.health import _check_ollama
|
||||
|
||||
await _check_ollama()
|
||||
# Expire the cache
|
||||
mod._ollama_cache_ts = time.monotonic() - 60
|
||||
await _check_ollama()
|
||||
|
||||
assert mock_sync.call_count == 2
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_fallback_on_thread_exception(self):
|
||||
"""If to_thread raises, return unavailable status."""
|
||||
import asyncio
|
||||
|
||||
with patch.object(
|
||||
asyncio,
|
||||
"to_thread",
|
||||
side_effect=RuntimeError("thread pool exhausted"),
|
||||
):
|
||||
from dashboard.routes.health import _check_ollama
|
||||
|
||||
result = await _check_ollama()
|
||||
|
||||
assert result.status == "unavailable"
|
||||
|
||||
|
||||
class TestCheckOllamaBool:
|
||||
"""Test the legacy bool wrapper."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_cache(self):
|
||||
import dashboard.routes.health as mod
|
||||
|
||||
mod._ollama_cache = None
|
||||
mod._ollama_cache_ts = 0.0
|
||||
yield
|
||||
mod._ollama_cache = None
|
||||
mod._ollama_cache_ts = 0.0
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_true_when_healthy(self):
|
||||
healthy = DependencyStatus(
|
||||
name="Ollama AI", status="healthy", sovereignty_score=10, details={}
|
||||
)
|
||||
with patch("dashboard.routes.health._check_ollama_sync", return_value=healthy):
|
||||
from dashboard.routes.health import check_ollama
|
||||
|
||||
assert await check_ollama() is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_false_when_unavailable(self):
|
||||
down = DependencyStatus(
|
||||
name="Ollama AI", status="unavailable", sovereignty_score=10, details={}
|
||||
)
|
||||
with patch("dashboard.routes.health._check_ollama_sync", return_value=down):
|
||||
from dashboard.routes.health import check_ollama
|
||||
|
||||
assert await check_ollama() is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Endpoint tests via FastAPI TestClient
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestHealthEndpoint:
|
||||
"""Tests for GET /health."""
|
||||
|
||||
def test_returns_200(self, client):
|
||||
response = client.get("/health")
|
||||
assert response.status_code == 200
|
||||
|
||||
def test_ok_when_ollama_up(self, client):
|
||||
with patch("dashboard.routes.health.check_ollama", new_callable=AsyncMock, return_value=True):
|
||||
data = client.get("/health").json()
|
||||
|
||||
assert data["status"] == "ok"
|
||||
assert data["services"]["ollama"] == "up"
|
||||
assert data["agents"]["agent"]["status"] == "idle"
|
||||
|
||||
def test_degraded_when_ollama_down(self, client):
|
||||
with patch(
|
||||
"dashboard.routes.health.check_ollama", new_callable=AsyncMock, return_value=False
|
||||
):
|
||||
data = client.get("/health").json()
|
||||
|
||||
assert data["status"] == "degraded"
|
||||
assert data["services"]["ollama"] == "down"
|
||||
assert data["agents"]["agent"]["status"] == "offline"
|
||||
|
||||
def test_extended_fields(self, client):
|
||||
data = client.get("/health").json()
|
||||
assert "timestamp" in data
|
||||
assert "version" in data
|
||||
assert "uptime_seconds" in data
|
||||
assert isinstance(data["uptime_seconds"], (int, float))
|
||||
assert "llm_backend" in data
|
||||
assert "llm_model" in data
|
||||
|
||||
|
||||
class TestHealthStatusPanel:
|
||||
"""Tests for GET /health/status (HTML response)."""
|
||||
|
||||
def test_returns_html(self, client):
|
||||
response = client.get("/health/status")
|
||||
assert response.status_code == 200
|
||||
assert "text/html" in response.headers["content-type"]
|
||||
|
||||
def test_shows_up_when_ollama_healthy(self, client):
|
||||
with patch("dashboard.routes.health.check_ollama", new_callable=AsyncMock, return_value=True):
|
||||
text = client.get("/health/status").text
|
||||
|
||||
assert "UP" in text
|
||||
|
||||
def test_shows_down_when_ollama_unhealthy(self, client):
|
||||
with patch(
|
||||
"dashboard.routes.health.check_ollama", new_callable=AsyncMock, return_value=False
|
||||
):
|
||||
text = client.get("/health/status").text
|
||||
|
||||
assert "DOWN" in text
|
||||
|
||||
def test_includes_model_name(self, client):
|
||||
text = client.get("/health/status").text
|
||||
assert "Model:" in text
|
||||
|
||||
|
||||
class TestSovereigntyEndpoint:
|
||||
"""Tests for GET /health/sovereignty."""
|
||||
|
||||
def test_aggregates_three_subsystems(self, client):
|
||||
data = client.get("/health/sovereignty").json()
|
||||
names = [d["name"] for d in data["dependencies"]]
|
||||
assert "Ollama AI" in names
|
||||
assert "Lightning Payments" in names
|
||||
assert "SQLite Database" in names
|
||||
|
||||
def test_score_range(self, client):
|
||||
data = client.get("/health/sovereignty").json()
|
||||
assert 0 <= data["overall_score"] <= 10
|
||||
|
||||
|
||||
class TestComponentsEndpoint:
|
||||
"""Tests for GET /health/components."""
|
||||
|
||||
def test_returns_timestamp(self, client):
|
||||
data = client.get("/health/components").json()
|
||||
assert "timestamp" in data
|
||||
|
||||
def test_config_keys(self, client):
|
||||
data = client.get("/health/components").json()
|
||||
cfg = data["config"]
|
||||
assert "debug" in cfg
|
||||
assert "model_backend" in cfg
|
||||
assert "ollama_model" in cfg
|
||||
|
||||
|
||||
class TestSnapshotEndpoint:
|
||||
"""Tests for GET /health/snapshot."""
|
||||
|
||||
def test_returns_200(self, client):
|
||||
response = client.get("/health/snapshot")
|
||||
assert response.status_code == 200
|
||||
|
||||
def test_overall_status_valid(self, client):
|
||||
data = client.get("/health/snapshot").json()
|
||||
assert data["overall_status"] in ["green", "yellow", "red", "unknown"]
|
||||
|
||||
def test_graceful_fallback_on_import_error(self, client):
|
||||
"""Snapshot degrades gracefully when automation module fails."""
|
||||
with patch(
|
||||
"dashboard.routes.health.asyncio.to_thread",
|
||||
side_effect=ImportError("no module"),
|
||||
):
|
||||
data = client.get("/health/snapshot").json()
|
||||
|
||||
assert data["overall_status"] == "unknown"
|
||||
assert "error" in data
|
||||
assert data["ci"]["status"] == "unknown"
|
||||
|
||||
def test_graceful_fallback_on_runtime_error(self, client):
|
||||
with patch(
|
||||
"dashboard.routes.health.asyncio.to_thread",
|
||||
side_effect=RuntimeError("boom"),
|
||||
):
|
||||
data = client.get("/health/snapshot").json()
|
||||
|
||||
assert data["overall_status"] == "unknown"
|
||||
Reference in New Issue
Block a user