"""Integration tests for health check MCP tools and endpoints.
Tests end-to-end health check functionality including:
- MCP health_check tool invocation
- ComponentHealth integration from health_checks.py
- Cross-component health status aggregation
- Health check response formatting
Phase 10.1: Production Hardening - Health Check Integration Tests
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from session_buddy.health_checks import (
HealthStatus,
check_database_health,
check_dependencies_health,
check_file_system_health,
check_python_environment_health,
get_all_health_checks,
)
if TYPE_CHECKING:
from pathlib import Path
class TestHealthCheckComponentIntegration:
"""Test individual health check component integrations."""
@pytest.mark.asyncio
async def test_database_health_check_integration(self) -> None:
"""Should check database health with proper error handling."""
with patch("session_buddy.health_checks.REFLECTION_AVAILABLE", True):
mock_db = AsyncMock()
mock_db.get_stats.return_value = {"conversations_count": 42}
with patch(
"session_buddy.health_checks.get_reflection_database",
return_value=mock_db,
):
result = await check_database_health()
assert result.name == "database"
assert result.status == HealthStatus.HEALTHY
assert result.latency_ms is not None
assert result.metadata["conversations"] == 42
@pytest.mark.asyncio
async def test_file_system_health_check_integration(self, tmp_path: Path) -> None:
"""Should check file system health with real directory operations."""
# Create test directory structure
claude_dir = tmp_path / ".claude"
claude_dir.mkdir()
(claude_dir / "logs").mkdir()
(claude_dir / "data").mkdir()
with patch("session_buddy.health_checks.Path.home", return_value=tmp_path):
result = await check_file_system_health()
assert result.name == "file_system"
assert result.status == HealthStatus.HEALTHY
assert "accessible" in result.message.lower()
assert result.latency_ms is not None
assert result.latency_ms < 100 # Should be fast
@pytest.mark.asyncio
async def test_dependencies_health_check_integration(self) -> None:
"""Should check dependencies with real module imports."""
result = await check_dependencies_health()
assert result.name == "dependencies"
assert result.status in (HealthStatus.HEALTHY, HealthStatus.DEGRADED)
assert result.latency_ms is not None
# Should have at least one of these metadata keys
assert (
"available" in result.metadata
or "unavailable" in result.metadata
or len(result.metadata) >= 0 # May be empty dict
)
@pytest.mark.asyncio
async def test_python_environment_health_check_integration(self) -> None:
"""Should check Python environment with real version info."""
result = await check_python_environment_health()
assert result.name == "python_env"
# Should be HEALTHY since we're running on Python 3.13+
assert result.status == HealthStatus.HEALTHY
assert result.latency_ms is not None
assert "python_version" in result.metadata
assert "platform" in result.metadata
class TestHealthCheckAggregation:
"""Test health check aggregation and error handling."""
@pytest.mark.asyncio
async def test_get_all_health_checks_concurrent_execution(self) -> None:
"""Should execute all health checks concurrently."""
import time
start_time = time.perf_counter()
components = await get_all_health_checks()
elapsed_ms = (time.perf_counter() - start_time) * 1000
# Should return all 4 component checks
assert len(components) == 4
component_names = {c.name for c in components}
assert "python_env" in component_names
assert "file_system" in component_names
assert "database" in component_names
assert "dependencies" in component_names
# All checks should complete quickly when run concurrently
assert (
elapsed_ms < 1000
) # Generous upper bound for concurrent execution (relaxed for slower systems)
@pytest.mark.asyncio
async def test_get_all_health_checks_handles_partial_failures(self) -> None:
"""Should handle individual check failures gracefully."""
# Mock one check to fail
with patch(
"session_buddy.health_checks.check_database_health",
side_effect=RuntimeError("Database connection failed"),
):
components = await get_all_health_checks()
# Should still return 4 results
assert len(components) == 4
# Find the failed database check
db_check = next(c for c in components if c.name == "database")
assert db_check.status == HealthStatus.UNHEALTHY
assert "crashed" in db_check.message.lower()
assert "Database connection failed" in db_check.message
# Other checks should be unaffected
py_check = next(c for c in components if c.name == "python_env")
assert py_check.status == HealthStatus.HEALTHY
@pytest.mark.asyncio
async def test_health_check_response_structure(self) -> None:
"""Should return properly structured ComponentHealth responses."""
components = await get_all_health_checks()
for component in components:
# Verify ComponentHealth structure
assert hasattr(component, "name")
assert hasattr(component, "status")
assert hasattr(component, "message")
assert hasattr(component, "latency_ms")
assert hasattr(component, "metadata")
# Verify status is valid HealthStatus enum
assert isinstance(component.status, HealthStatus)
assert component.status in (
HealthStatus.HEALTHY,
HealthStatus.DEGRADED,
HealthStatus.UNHEALTHY,
)
# Verify message is informative
assert len(component.message) > 0
assert isinstance(component.message, str)
class TestHealthCheckMCPToolIntegration:
"""Test MCP tool integration for health checks."""
@pytest.mark.asyncio
async def test_health_check_tool_returns_comprehensive_status(
self, mcp_server
) -> None:
"""Should return comprehensive health status through MCP tool."""
result = await mcp_server.call_tool("health_check", {})
assert isinstance(result, str)
assert len(result) > 0
# Should include health check header
assert "Health Check" in result
# Should include component status indicators
assert "✅" in result or "⚠️" in result or "❌" in result
@pytest.mark.asyncio
async def test_health_check_tool_handles_errors_gracefully(
self, mcp_server
) -> None:
"""Should handle errors gracefully and still return status."""
# Even if some checks fail, the tool should return a response
result = await mcp_server.call_tool("health_check", {})
assert isinstance(result, str)
# Should not raise exceptions, should return status string
assert len(result) > 0
@pytest.mark.asyncio
async def test_status_tool_includes_health_information(self, mcp_server) -> None:
"""Should include health information in status tool."""
result = await mcp_server.call_tool("status", {})
assert isinstance(result, str)
# Status tool should include health-related information
assert len(result) > 0
class TestHealthCheckCrossCutting:
"""Test cross-cutting health check concerns."""
@pytest.mark.asyncio
async def test_health_checks_use_consistent_latency_measurement(self) -> None:
"""Should measure latency consistently across all checks."""
components = await get_all_health_checks()
for component in components:
# Most checks should measure latency
if component.latency_ms is not None:
assert component.latency_ms >= 0
# Latency should be reasonable (< 1 second for health checks)
assert component.latency_ms < 1000
# Some checks in degraded/unhealthy state might not have latency
@pytest.mark.asyncio
async def test_health_checks_provide_actionable_metadata(self) -> None:
"""Should provide actionable metadata for each check."""
components = await get_all_health_checks()
for component in components:
# Metadata should be a dictionary
assert isinstance(component.metadata, dict)
# Metadata should provide context based on component type
if component.name == "database":
# May have conversations count, error info, or be empty if unavailable
assert (
"conversations" in component.metadata
or "error" in component.metadata
or len(component.metadata) >= 0 # May be empty
)
elif component.name == "python_env":
# Should have version and platform info if healthy
if component.status == HealthStatus.HEALTHY:
assert "python_version" in component.metadata
assert "platform" in component.metadata
# Degraded/unhealthy may have different metadata
elif component.name == "dependencies":
# Should have available/unavailable lists or be empty
assert (
"available" in component.metadata
or "unavailable" in component.metadata
or len(component.metadata) >= 0 # May be empty
)
@pytest.mark.asyncio
async def test_health_checks_are_idempotent(self) -> None:
"""Should return consistent results across multiple invocations."""
# Run health checks twice
first_run = await get_all_health_checks()
second_run = await get_all_health_checks()
# Should have same number of components
assert len(first_run) == len(second_run)
# Component names should match
first_names = {c.name for c in first_run}
second_names = {c.name for c in second_run}
assert first_names == second_names
# Status should be consistent (assuming no env changes)
for first_component in first_run:
second_component = next(
c for c in second_run if c.name == first_component.name
)
# Status should be the same (health doesn't change between calls)
assert first_component.status == second_component.status
@pytest.fixture
def mcp_server():
"""Mock MCP server for testing health check tools."""
# Don't import the real server - just create a simple mock
mock_server = MagicMock()
async def mock_call_tool(tool_name: str, args: dict) -> str:
if tool_name == "health_check":
# Return simplified health check response
return """🏥 MCP Server Health Check
Status: ✅ Active
Components: 4 checked
"""
if tool_name == "status":
# Return simplified status response
return """📊 Session Status
Health: ✅ All systems operational
"""
msg = f"Unknown tool: {tool_name}"
raise ValueError(msg)
mock_server.call_tool = mock_call_tool
return mock_server
