Lukas
3551838887
- Add _run_startup_health_checks() function in fastapi_app.py - Check ffmpeg availability (warning) - Check DNS resolution for aniworld.to and api.themoviedb.org (warning) - Check anime_directory configuration and writability (error) - Store startup checks in app.state for health endpoint access - Add /health/ready endpoint for container orchestrators - Returns not_ready with 503 when critical failures present - Includes critical_failures list for debugging - Update /health endpoint to include startup check results - Status reflects worst check (error > warning > ok) - Document health check endpoints in DEVELOPMENT.md - Add unit tests for startup health checks - Add unit tests for /health/ready endpoint Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
135 lines
5.8 KiB
Python
135 lines
5.8 KiB
Python
"""Unit tests for startup health checks in fastapi_app.py."""
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
class TestStartupHealthChecks:
|
|
"""Test startup health check function."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ffmpeg_missing_sets_warning(self):
|
|
"""Test ffmpeg missing results in warning status."""
|
|
mock_logger = MagicMock()
|
|
|
|
with patch("shutil.which", return_value=None):
|
|
from src.server.fastapi_app import _run_startup_health_checks
|
|
result = await _run_startup_health_checks(mock_logger)
|
|
|
|
assert result["ffmpeg"]["status"] == "warning"
|
|
assert "not found in PATH" in result["ffmpeg"]["message"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ffmpeg_present_sets_ok(self):
|
|
"""Test ffmpeg present results in ok status."""
|
|
mock_logger = MagicMock()
|
|
|
|
with patch("shutil.which", return_value="/usr/bin/ffmpeg"):
|
|
from src.server.fastapi_app import _run_startup_health_checks
|
|
result = await _run_startup_health_checks(mock_logger)
|
|
|
|
assert result["ffmpeg"]["status"] == "ok"
|
|
assert "Found at" in result["ffmpeg"]["message"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_anime_directory_not_configured_sets_error(self):
|
|
"""Test anime_directory not configured results in error status."""
|
|
mock_logger = MagicMock()
|
|
|
|
with patch("src.config.settings.settings") as mock_settings:
|
|
mock_settings.anime_directory = ""
|
|
|
|
from src.server.fastapi_app import _run_startup_health_checks
|
|
result = await _run_startup_health_checks(mock_logger)
|
|
|
|
assert result["anime_directory"]["status"] == "error"
|
|
assert result["anime_directory"]["path"] is None
|
|
assert "not configured" in result["anime_directory"]["message"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_anime_directory_not_exists_sets_error(self):
|
|
"""Test anime_directory path not existing results in error status."""
|
|
mock_logger = MagicMock()
|
|
|
|
with patch("src.config.settings.settings") as mock_settings:
|
|
mock_settings.anime_directory = "/nonexistent/path"
|
|
|
|
with patch("os.path.isdir", return_value=False):
|
|
from src.server.fastapi_app import _run_startup_health_checks
|
|
result = await _run_startup_health_checks(mock_logger)
|
|
|
|
assert result["anime_directory"]["status"] == "error"
|
|
assert "does not exist" in result["anime_directory"]["message"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_anime_directory_not_writable_sets_error(self):
|
|
"""Test anime_directory not writable results in error status."""
|
|
mock_logger = MagicMock()
|
|
|
|
with patch("src.config.settings.settings") as mock_settings:
|
|
mock_settings.anime_directory = "/some/path"
|
|
|
|
with patch("os.path.isdir", return_value=True):
|
|
with patch("os.access", return_value=False):
|
|
from src.server.fastapi_app import _run_startup_health_checks
|
|
result = await _run_startup_health_checks(mock_logger)
|
|
|
|
assert result["anime_directory"]["status"] == "error"
|
|
assert "not writable" in result["anime_directory"]["message"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_anime_directory_ok_when_writable(self):
|
|
"""Test anime_directory exists and writable results in ok status."""
|
|
mock_logger = MagicMock()
|
|
|
|
with patch("src.config.settings.settings") as mock_settings:
|
|
mock_settings.anime_directory = "/valid/path"
|
|
|
|
with patch("os.path.isdir", return_value=True):
|
|
with patch("os.access", return_value=True):
|
|
from src.server.fastapi_app import _run_startup_health_checks
|
|
result = await _run_startup_health_checks(mock_logger)
|
|
|
|
assert result["anime_directory"]["status"] == "ok"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dns_aniworld_failure_sets_warning(self):
|
|
"""Test DNS failure for aniworld.to sets warning status."""
|
|
mock_logger = MagicMock()
|
|
|
|
import socket
|
|
with patch("socket.gethostbyname", side_effect=socket.gaierror("DNS failed")):
|
|
from src.server.fastapi_app import _run_startup_health_checks
|
|
result = await _run_startup_health_checks(mock_logger)
|
|
|
|
assert result["dns_aniworld"]["status"] == "warning"
|
|
assert "DNS resolution failed" in result["dns_aniworld"]["message"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dns_tmdb_failure_sets_warning(self):
|
|
"""Test DNS failure for api.themoviedb.org sets warning status."""
|
|
mock_logger = MagicMock()
|
|
|
|
import socket
|
|
with patch("socket.gethostbyname", side_effect=socket.gaierror("DNS failed")):
|
|
from src.server.fastapi_app import _run_startup_health_checks
|
|
result = await _run_startup_health_checks(mock_logger)
|
|
|
|
assert result["dns_tmdb"]["status"] == "warning"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_all_checks_returned(self):
|
|
"""Test all health checks are present in result."""
|
|
mock_logger = MagicMock()
|
|
|
|
with patch("src.config.settings.settings") as mock_settings:
|
|
mock_settings.anime_directory = ""
|
|
|
|
from src.server.fastapi_app import _run_startup_health_checks
|
|
result = await _run_startup_health_checks(mock_logger)
|
|
|
|
assert "ffmpeg" in result
|
|
assert "dns_aniworld" in result
|
|
assert "dns_tmdb" in result
|
|
assert "anime_directory" in result |