[claude] Add unit tests for health.py (#945) (#1002)

Co-authored-by: Claude (Opus 4.6) <claude@hermes.local>
Co-committed-by: Claude (Opus 4.6) <claude@hermes.local>

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"