diff --git a/tests/dashboard/test_health.py b/tests/dashboard/test_health.py new file mode 100644 index 00000000..7d6eded3 --- /dev/null +++ b/tests/dashboard/test_health.py @@ -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"