786 lines
26 KiB
Python
786 lines
26 KiB
Python
"""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,
|
|
DetailedHealthStatus,
|
|
HealthStatus,
|
|
LivenessStatus,
|
|
ReadinessStatus,
|
|
SovereigntyReport,
|
|
_calculate_overall_score,
|
|
_check_lightning,
|
|
_check_ollama_sync,
|
|
_check_sqlite,
|
|
_generate_recommendations,
|
|
get_shutdown_info,
|
|
is_shutting_down,
|
|
request_shutdown,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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"
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Shutdown State Tests
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
class TestShutdownState:
|
|
"""Tests for shutdown state tracking."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_shutdown_state(self):
|
|
"""Reset shutdown state before each test."""
|
|
import dashboard.routes.health as mod
|
|
|
|
mod._shutdown_requested = False
|
|
mod._shutdown_reason = None
|
|
mod._shutdown_start_time = None
|
|
yield
|
|
mod._shutdown_requested = False
|
|
mod._shutdown_reason = None
|
|
mod._shutdown_start_time = None
|
|
|
|
def test_is_shutting_down_initial(self):
|
|
assert is_shutting_down() is False
|
|
|
|
def test_request_shutdown_sets_state(self):
|
|
request_shutdown(reason="test")
|
|
assert is_shutting_down() is True
|
|
|
|
def test_get_shutdown_info_when_not_shutting_down(self):
|
|
info = get_shutdown_info()
|
|
assert info is None
|
|
|
|
def test_get_shutdown_info_when_shutting_down(self):
|
|
request_shutdown(reason="test_reason")
|
|
info = get_shutdown_info()
|
|
assert info is not None
|
|
assert info["requested"] is True
|
|
assert info["reason"] == "test_reason"
|
|
assert "elapsed_seconds" in info
|
|
assert "timeout_seconds" in info
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Detailed Health Endpoint Tests
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
class TestDetailedHealthEndpoint:
|
|
"""Tests for GET /health/detailed."""
|
|
|
|
def test_returns_200_when_healthy(self, client):
|
|
with patch(
|
|
"dashboard.routes.health._check_ollama_sync",
|
|
return_value=DependencyStatus(
|
|
name="Ollama AI", status="healthy", sovereignty_score=10, details={}
|
|
),
|
|
):
|
|
response = client.get("/health/detailed")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] in ["healthy", "degraded", "unhealthy"]
|
|
assert "timestamp" in data
|
|
assert "version" in data
|
|
assert "uptime_seconds" in data
|
|
assert "services" in data
|
|
assert "system" in data
|
|
|
|
def test_returns_503_when_service_unhealthy(self, client):
|
|
with patch(
|
|
"dashboard.routes.health._check_ollama_sync",
|
|
return_value=DependencyStatus(
|
|
name="Ollama AI",
|
|
status="unavailable",
|
|
sovereignty_score=10,
|
|
details={"error": "down"},
|
|
),
|
|
):
|
|
response = client.get("/health/detailed")
|
|
|
|
assert response.status_code == 503
|
|
data = response.json()
|
|
assert data["status"] == "unhealthy"
|
|
|
|
def test_includes_shutdown_info_when_shutting_down(self, client):
|
|
with patch(
|
|
"dashboard.routes.health._check_ollama_sync",
|
|
return_value=DependencyStatus(
|
|
name="Ollama AI", status="healthy", sovereignty_score=10, details={}
|
|
),
|
|
):
|
|
with patch("dashboard.routes.health.is_shutting_down", return_value=True):
|
|
with patch(
|
|
"dashboard.routes.health.get_shutdown_info",
|
|
return_value={
|
|
"requested": True,
|
|
"reason": "test",
|
|
"elapsed_seconds": 1.5,
|
|
"timeout_seconds": 30.0,
|
|
},
|
|
):
|
|
response = client.get("/health/detailed")
|
|
|
|
assert response.status_code == 503
|
|
data = response.json()
|
|
assert "shutdown" in data
|
|
assert data["shutdown"]["requested"] is True
|
|
|
|
def test_services_structure(self, client):
|
|
with patch(
|
|
"dashboard.routes.health._check_ollama_sync",
|
|
return_value=DependencyStatus(
|
|
name="Ollama AI", status="healthy", sovereignty_score=10, details={"model": "test"}
|
|
),
|
|
):
|
|
response = client.get("/health/detailed")
|
|
|
|
data = response.json()
|
|
assert "services" in data
|
|
assert "ollama" in data["services"]
|
|
assert "sqlite" in data["services"]
|
|
# Each service should have status, healthy flag, and details
|
|
for _svc_name, svc_data in data["services"].items():
|
|
assert "status" in svc_data
|
|
assert "healthy" in svc_data
|
|
assert isinstance(svc_data["healthy"], bool)
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Readiness Probe Tests
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
class TestReadinessProbe:
|
|
"""Tests for GET /ready."""
|
|
|
|
def test_returns_200_when_ready(self, client):
|
|
# Wait for startup to complete
|
|
response = client.get("/ready")
|
|
data = response.json()
|
|
|
|
# Should return either 200 (ready) or 503 (not ready)
|
|
assert response.status_code in [200, 503]
|
|
assert "ready" in data
|
|
assert isinstance(data["ready"], bool)
|
|
assert "timestamp" in data
|
|
assert "checks" in data
|
|
|
|
def test_checks_structure(self, client):
|
|
response = client.get("/ready")
|
|
data = response.json()
|
|
|
|
assert "checks" in data
|
|
checks = data["checks"]
|
|
# Core checks that should be present
|
|
assert "startup_complete" in checks
|
|
assert "database" in checks
|
|
assert "not_shutting_down" in checks
|
|
|
|
def test_not_ready_during_shutdown(self, client):
|
|
with patch("dashboard.routes.health.is_shutting_down", return_value=True):
|
|
with patch(
|
|
"dashboard.routes.health._shutdown_reason",
|
|
"test shutdown",
|
|
):
|
|
response = client.get("/ready")
|
|
|
|
assert response.status_code == 503
|
|
data = response.json()
|
|
assert data["ready"] is False
|
|
assert data["checks"]["not_shutting_down"] is False
|
|
assert "reason" in data
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Liveness Probe Tests
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
class TestLivenessProbe:
|
|
"""Tests for GET /live."""
|
|
|
|
def test_returns_200_when_alive(self, client):
|
|
response = client.get("/live")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["alive"] is True
|
|
assert "timestamp" in data
|
|
assert "uptime_seconds" in data
|
|
assert "shutdown_requested" in data
|
|
|
|
def test_shutdown_requested_field(self, client):
|
|
with patch("dashboard.routes.health.is_shutting_down", return_value=False):
|
|
response = client.get("/live")
|
|
|
|
data = response.json()
|
|
assert data["shutdown_requested"] is False
|
|
|
|
def test_alive_false_after_shutdown_timeout(self, client):
|
|
import dashboard.routes.health as mod
|
|
|
|
with patch.object(mod, "_shutdown_requested", True):
|
|
with patch.object(mod, "_shutdown_start_time", time.monotonic() - 999):
|
|
with patch.object(mod, "GRACEFUL_SHUTDOWN_TIMEOUT", 30.0):
|
|
response = client.get("/live")
|
|
|
|
assert response.status_code == 503
|
|
data = response.json()
|
|
assert data["alive"] is False
|
|
assert data["shutdown_requested"] is True
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# New Pydantic Model Tests
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
class TestDetailedHealthStatusModel:
|
|
"""Validate DetailedHealthStatus model."""
|
|
|
|
def test_fields(self):
|
|
hs = DetailedHealthStatus(
|
|
status="healthy",
|
|
timestamp="2026-01-01T00:00:00+00:00",
|
|
version="2.0.0",
|
|
uptime_seconds=42.5,
|
|
services={"db": {"status": "up", "healthy": True, "details": {}}},
|
|
system={"memory_mb": 100.5},
|
|
)
|
|
assert hs.status == "healthy"
|
|
assert hs.services["db"]["healthy"] is True
|
|
|
|
|
|
class TestReadinessStatusModel:
|
|
"""Validate ReadinessStatus model."""
|
|
|
|
def test_fields(self):
|
|
rs = ReadinessStatus(
|
|
ready=True,
|
|
timestamp="2026-01-01T00:00:00+00:00",
|
|
checks={"db": True, "cache": True},
|
|
)
|
|
assert rs.ready is True
|
|
assert rs.checks["db"] is True
|
|
|
|
def test_with_reason(self):
|
|
rs = ReadinessStatus(
|
|
ready=False,
|
|
timestamp="2026-01-01T00:00:00+00:00",
|
|
checks={"db": False},
|
|
reason="Database unavailable",
|
|
)
|
|
assert rs.ready is False
|
|
assert rs.reason == "Database unavailable"
|
|
|
|
|
|
class TestLivenessStatusModel:
|
|
"""Validate LivenessStatus model."""
|
|
|
|
def test_fields(self):
|
|
ls = LivenessStatus(
|
|
alive=True,
|
|
timestamp="2026-01-01T00:00:00+00:00",
|
|
uptime_seconds=3600.0,
|
|
shutdown_requested=False,
|
|
)
|
|
assert ls.alive is True
|
|
assert ls.uptime_seconds == 3600.0
|
|
assert ls.shutdown_requested is False
|
|
|
|
def test_defaults(self):
|
|
ls = LivenessStatus(
|
|
alive=True,
|
|
timestamp="2026-01-01T00:00:00+00:00",
|
|
uptime_seconds=0.0,
|
|
)
|
|
assert ls.shutdown_requested is False
|