Lukas
4731fd644a
- Use dynamic APP_VERSION instead of hardcoded v1.3.6 in: test_template_helpers, test_health, test_page_controller - Add unresolved_folders to EXPECTED_TABLES in database/init.py - Fix shallow copy bug in test_serie_scanner.py episodeDict comparison - Update test_schema_constants to expect 6 tables instead of 5
207 lines
7.1 KiB
Python
207 lines
7.1 KiB
Python
"""Unit tests for health check endpoints."""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from src.server.api.health import (
|
|
DatabaseHealth,
|
|
HealthStatus,
|
|
SystemMetrics,
|
|
basic_health_check,
|
|
check_database_health,
|
|
check_filesystem_health,
|
|
get_system_metrics,
|
|
ready_check,
|
|
)
|
|
from src.server.utils.version import APP_VERSION
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_basic_health_check_no_startup_checks():
|
|
"""Test basic health check endpoint with no startup checks."""
|
|
mock_request = MagicMock()
|
|
mock_request.app.state.startup_checks = {}
|
|
|
|
with patch("src.config.settings.settings") as mock_settings, \
|
|
patch("src.server.utils.dependencies._series_app", None):
|
|
mock_settings.anime_directory = ""
|
|
result = await basic_health_check(mock_request)
|
|
|
|
assert isinstance(result, HealthStatus)
|
|
assert result.status == "healthy"
|
|
assert result.version == APP_VERSION
|
|
assert result.service == "aniworld-api"
|
|
assert result.timestamp is not None
|
|
assert result.series_app_initialized is False
|
|
assert result.anime_directory_configured is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_basic_health_check_with_error_check():
|
|
"""Test basic health check reflects error status from startup checks."""
|
|
mock_request = MagicMock()
|
|
mock_request.app.state.startup_checks = {
|
|
"anime_directory": {"status": "error", "message": "not configured", "path": None},
|
|
"ffmpeg": {"status": "ok", "message": "Found at /usr/bin/ffmpeg"},
|
|
"dns_aniworld": {"status": "ok", "message": "Resolved successfully"},
|
|
"dns_tmdb": {"status": "ok", "message": "Resolved successfully"},
|
|
}
|
|
|
|
with patch("src.config.settings.settings") as mock_settings, \
|
|
patch("src.server.utils.dependencies._series_app", None):
|
|
mock_settings.anime_directory = ""
|
|
result = await basic_health_check(mock_request)
|
|
|
|
assert isinstance(result, HealthStatus)
|
|
assert result.status == "unhealthy"
|
|
assert result.checks is not None
|
|
assert result.checks["anime_directory"]["status"] == "error"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_basic_health_check_with_warning_only():
|
|
"""Test basic health check shows degraded when only warnings present."""
|
|
mock_request = MagicMock()
|
|
mock_request.app.state.startup_checks = {
|
|
"anime_directory": {"status": "ok", "message": "Found", "path": "/anime"},
|
|
"ffmpeg": {"status": "warning", "message": "not found in PATH"},
|
|
"dns_aniworld": {"status": "ok", "message": "Resolved successfully"},
|
|
"dns_tmdb": {"status": "ok", "message": "Resolved successfully"},
|
|
}
|
|
|
|
with patch("src.config.settings.settings") as mock_settings, \
|
|
patch("src.server.utils.dependencies._series_app", None):
|
|
mock_settings.anime_directory = "/anime"
|
|
result = await basic_health_check(mock_request)
|
|
|
|
assert isinstance(result, HealthStatus)
|
|
assert result.status == "degraded"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ready_check_all_healthy():
|
|
"""Test ready check returns ready when all checks pass."""
|
|
mock_request = MagicMock()
|
|
mock_request.app.state.startup_checks = {
|
|
"anime_directory": {"status": "ok", "message": "Found", "path": "/anime"},
|
|
"ffmpeg": {"status": "ok", "message": "Found at /usr/bin/ffmpeg"},
|
|
"dns_aniworld": {"status": "ok", "message": "Resolved successfully"},
|
|
"dns_tmdb": {"status": "ok", "message": "Resolved successfully"},
|
|
}
|
|
|
|
result = await ready_check(mock_request)
|
|
|
|
assert result["ready"] is True
|
|
assert result["status"] == "ready"
|
|
assert "critical_failures" not in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ready_check_with_critical_failure():
|
|
"""Test ready check returns not_ready when anime_directory not configured."""
|
|
mock_request = MagicMock()
|
|
mock_request.app.state.startup_checks = {
|
|
"anime_directory": {"status": "error", "message": "not configured", "path": None},
|
|
"ffmpeg": {"status": "warning", "message": "not found in PATH"},
|
|
"dns_aniworld": {"status": "ok", "message": "Resolved successfully"},
|
|
"dns_tmdb": {"status": "ok", "message": "Resolved successfully"},
|
|
}
|
|
|
|
result = await ready_check(mock_request)
|
|
|
|
assert result["ready"] is False
|
|
assert result["status"] == "not_ready"
|
|
assert len(result["critical_failures"]) == 1
|
|
assert "anime_directory" in result["critical_failures"][0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_database_health_check_success():
|
|
"""Test database health check with successful connection."""
|
|
# Mock database session
|
|
mock_db = AsyncMock()
|
|
mock_db.execute = AsyncMock()
|
|
|
|
result = await check_database_health(mock_db)
|
|
|
|
assert isinstance(result, DatabaseHealth)
|
|
assert result.status == "healthy"
|
|
assert result.connection_time_ms >= 0
|
|
assert "successful" in result.message.lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_database_health_check_failure():
|
|
"""Test database health check with failed connection."""
|
|
# Mock database session that raises error
|
|
mock_db = AsyncMock()
|
|
mock_db.execute = AsyncMock(side_effect=Exception("Connection failed"))
|
|
|
|
result = await check_database_health(mock_db)
|
|
|
|
assert isinstance(result, DatabaseHealth)
|
|
assert result.status == "unhealthy"
|
|
assert "failed" in result.message.lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_filesystem_health_check_success():
|
|
"""Test filesystem health check with accessible directories."""
|
|
with patch("os.path.exists", return_value=True), patch(
|
|
"os.access", return_value=True
|
|
):
|
|
result = await check_filesystem_health()
|
|
|
|
assert result["status"] in ["healthy", "degraded"]
|
|
assert "data_dir_writable" in result
|
|
assert "logs_dir_writable" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_filesystem_health_check_failure():
|
|
"""Test filesystem health check with inaccessible directories."""
|
|
with patch("os.path.exists", return_value=False), patch(
|
|
"os.access", return_value=False
|
|
):
|
|
result = await check_filesystem_health()
|
|
|
|
assert "status" in result
|
|
assert "message" in result
|
|
|
|
|
|
def test_get_system_metrics():
|
|
"""Test system metrics collection."""
|
|
result = get_system_metrics()
|
|
|
|
assert isinstance(result, SystemMetrics)
|
|
assert result.cpu_percent >= 0
|
|
assert result.memory_percent >= 0
|
|
assert result.memory_available_mb > 0
|
|
assert result.disk_percent >= 0
|
|
assert result.disk_free_mb > 0
|
|
assert result.uptime_seconds > 0
|
|
|
|
|
|
def test_system_metrics_values_reasonable():
|
|
"""Test that system metrics are within reasonable ranges."""
|
|
result = get_system_metrics()
|
|
|
|
# CPU should be 0-100%
|
|
assert 0 <= result.cpu_percent <= 100
|
|
|
|
# Memory should be 0-100%
|
|
assert 0 <= result.memory_percent <= 100
|
|
|
|
# Disk should be 0-100%
|
|
assert 0 <= result.disk_percent <= 100
|
|
|
|
# Memory available should be positive
|
|
assert result.memory_available_mb > 0
|
|
|
|
# Disk free should be positive
|
|
assert result.disk_free_mb > 0
|
|
|
|
# Uptime should be positive
|
|
assert result.uptime_seconds > 0
|