"""Tests for health_tools module.
Tests health check endpoints for Docker and Kubernetes orchestration.
Phase: Week 1 Day 1 - Quick Win Coverage (0% → 80%+)
"""
from __future__ import annotations
import time
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
class TestGetHealthStatus:
"""Test get_health_status function for health checks."""
@pytest.mark.asyncio
async def test_get_health_status_returns_dict(self) -> None:
"""Should return dictionary with health status."""
from session_buddy.tools.health_tools import get_health_status
# Mock health checks to return healthy components
mock_components = [
{
"name": "database",
"status": "healthy",
"message": "Database operational",
"latency_ms": 12.5,
},
]
with patch("session_buddy.health_checks.get_all_health_checks") as mock_checks:
mock_checks.return_value = mock_components
result = await get_health_status()
assert isinstance(result, dict)
assert "status" in result
assert "timestamp" in result
assert "version" in result
assert "uptime_seconds" in result
@pytest.mark.asyncio
async def test_get_health_status_liveness_check(self) -> None:
"""Should perform liveness check when ready=False (default)."""
from session_buddy.tools.health_tools import get_health_status
mock_components = [
{"name": "test", "status": "healthy", "message": "OK"},
]
with patch("session_buddy.health_checks.get_all_health_checks") as mock_checks:
mock_checks.return_value = mock_components
result = await get_health_status(ready=False)
assert isinstance(result, dict)
assert "alive" in result
assert result["alive"] is True
# Liveness check should have looser criteria
assert "metadata" in result
if "metadata" in result:
assert result["metadata"].get("check_type") == "liveness"
@pytest.mark.asyncio
async def test_get_health_status_readiness_check(self) -> None:
"""Should perform readiness check when ready=True."""
from session_buddy.tools.health_tools import get_health_status
mock_components = [
{"name": "test", "status": "healthy", "message": "OK"},
]
with patch("session_buddy.health_checks.get_all_health_checks") as mock_checks:
mock_checks.return_value = mock_components
result = await get_health_status(ready=True)
assert isinstance(result, dict)
assert "ready" in result
assert result["ready"] is True
# Readiness check should be stricter
assert "metadata" in result
if "metadata" in result:
assert result["metadata"].get("check_type") == "readiness"
@pytest.mark.asyncio
async def test_get_health_status_includes_version(self) -> None:
"""Should include version information."""
from session_buddy.tools.health_tools import get_health_status
mock_components = []
with patch("session_buddy.health_checks.get_all_health_checks") as mock_checks:
mock_checks.return_value = mock_components
result = await get_health_status()
assert "version" in result
assert isinstance(result["version"], str)
# Version should either be actual version or "unknown"
assert len(result["version"]) > 0
@pytest.mark.asyncio
async def test_get_health_status_includes_uptime(self) -> None:
"""Should calculate and include uptime seconds."""
from session_buddy.tools.health_tools import get_health_status
mock_components = []
with patch("session_buddy.health_checks.get_all_health_checks") as mock_checks:
mock_checks.return_value = mock_components
result = await get_health_status()
assert "uptime_seconds" in result
assert isinstance(result["uptime_seconds"], (int, float))
assert result["uptime_seconds"] >= 0
@pytest.mark.asyncio
async def test_get_health_status_includes_components(self) -> None:
"""Should include component health checks in response."""
from session_buddy.tools.health_tools import get_health_status
mock_components = [
{"name": "database", "status": "healthy", "message": "DB OK"},
{"name": "cache", "status": "healthy", "message": "Cache OK"},
]
with patch("session_buddy.health_checks.get_all_health_checks") as mock_checks:
mock_checks.return_value = mock_components
result = await get_health_status()
assert "components" in result
assert isinstance(result["components"], list)
assert len(result["components"]) == 2
@pytest.mark.asyncio
async def test_get_health_status_handles_unhealthy_components_liveness(
self,
) -> None:
"""Liveness check should be loose - only UNHEALTHY fails."""
from session_buddy.tools.health_tools import get_health_status
# Mock components with degraded status (should still pass liveness)
mock_components = [
{"name": "database", "status": "degraded", "message": "DB slow"},
]
with patch("session_buddy.health_checks.get_all_health_checks") as mock_checks:
mock_checks.return_value = mock_components
result = await get_health_status(ready=False)
# Liveness should still pass for degraded components
assert "alive" in result
# Exact behavior depends on HealthCheckResponse implementation
@pytest.mark.asyncio
async def test_get_health_status_handles_unhealthy_components_readiness(
self,
) -> None:
"""Readiness check should be strict - only HEALTHY passes."""
from session_buddy.tools.health_tools import get_health_status
# Mock components with degraded status (should fail readiness)
mock_components = [
{"name": "database", "status": "degraded", "message": "DB slow"},
]
with patch("session_buddy.health_checks.get_all_health_checks") as mock_checks:
mock_checks.return_value = mock_components
result = await get_health_status(ready=True)
# Readiness check should fail for degraded components
assert "ready" in result
# Exact behavior depends on HealthCheckResponse implementation
@pytest.mark.asyncio
async def test_get_health_status_handles_version_import_error(self) -> None:
"""Should handle missing version gracefully."""
from session_buddy.tools.health_tools import get_health_status
mock_components = []
# Mock version import failure
with (
patch("session_buddy.health_checks.get_all_health_checks") as mock_checks,
patch.dict("sys.modules", {"session_buddy": MagicMock()}),
):
mock_checks.return_value = mock_components
result = await get_health_status()
assert "version" in result
# Should fall back to "unknown" or similar
assert isinstance(result["version"], str)
@pytest.mark.asyncio
async def test_get_health_status_includes_timestamp(self) -> None:
"""Should include ISO timestamp in response."""
from session_buddy.tools.health_tools import get_health_status
mock_components = []
with patch("session_buddy.health_checks.get_all_health_checks") as mock_checks:
mock_checks.return_value = mock_components
time.time()
result = await get_health_status()
time.time()
assert "timestamp" in result
assert isinstance(result["timestamp"], str)
# Timestamp should be within test execution window
# (Actual validation would require parsing ISO format)
@pytest.mark.asyncio
async def test_get_health_status_with_empty_components(self) -> None:
"""Should handle empty component list gracefully."""
from session_buddy.tools.health_tools import get_health_status
mock_components = []
with patch("session_buddy.health_checks.get_all_health_checks") as mock_checks:
mock_checks.return_value = mock_components
result = await get_health_status()
assert isinstance(result, dict)
assert "components" in result
assert isinstance(result["components"], list)
@pytest.mark.asyncio
async def test_get_health_status_with_many_components(self) -> None:
"""Should handle multiple health check components."""
from session_buddy.tools.health_tools import get_health_status
# Create many mock components
mock_components = [
{"name": f"service_{i}", "status": "healthy", "message": f"Service {i} OK"}
for i in range(10)
]
with patch("session_buddy.health_checks.get_all_health_checks") as mock_checks:
mock_checks.return_value = mock_components
result = await get_health_status()
assert "components" in result
assert len(result["components"]) == 10
@pytest.mark.asyncio
async def test_get_health_status_preserves_component_metadata(self) -> None:
"""Should preserve component latency and other metadata."""
from session_buddy.tools.health_tools import get_health_status
mock_components = [
{
"name": "database",
"status": "healthy",
"message": "Database operational",
"latency_ms": 12.5,
"connection_pool": "10/50",
},
]
with patch("session_buddy.health_checks.get_all_health_checks") as mock_checks:
mock_checks.return_value = mock_components
result = await get_health_status()
assert "components" in result
# Metadata should be preserved in components
if result["components"]:
component = result["components"][0]
assert "name" in component
@pytest.mark.asyncio
async def test_get_health_status_server_start_time_constant(self) -> None:
"""Should use constant server start time for uptime calculation."""
from session_buddy.tools.health_tools import (
_SERVER_START_TIME,
get_health_status,
)
mock_components = []
with patch("session_buddy.health_checks.get_all_health_checks") as mock_checks:
mock_checks.return_value = mock_components
# Server start time should be set when module is imported
assert isinstance(_SERVER_START_TIME, float)
assert _SERVER_START_TIME > 0
result1 = await get_health_status()
# Small delay to ensure uptime increases
time.sleep(0.01)
result2 = await get_health_status()
# Uptime should increase between calls
assert result2["uptime_seconds"] >= result1["uptime_seconds"]
class TestHealthToolsModule:
"""Test module-level functionality and exports."""
def test_module_exports_get_health_status(self) -> None:
"""Should export get_health_status in __all__."""
from session_buddy.tools import health_tools
assert hasattr(health_tools, "__all__")
assert "get_health_status" in health_tools.__all__
def test_module_has_server_start_time(self) -> None:
"""Should have _SERVER_START_TIME module variable."""
from session_buddy.tools import health_tools
assert hasattr(health_tools, "_SERVER_START_TIME")
assert isinstance(health_tools._SERVER_START_TIME, float)
assert health_tools._SERVER_START_TIME > 0
