567697a115
Some checks failed
Integration Tests / integration-tests (push) Failing after 19s
Integration Tests / performance-tests (push) Has been skipped
Service Adapters (Python FastAPI) / test (3.11) (push) Successful in 1m13s
Service Adapters (Python FastAPI) / test (3.12) (push) Successful in 1m19s
Service Adapters (Python FastAPI) / test (3.13) (push) Successful in 1m17s
Service Adapters (Python FastAPI) / build (push) Successful in 16s
### Summary of Changes - Refactored the `test_get_services` method to enhance the organization of mock responses and improve test clarity. - Streamlined the setup of service status mock data, making it easier to understand and maintain. ### Expected Results - Increased readability of test definitions, facilitating easier updates and modifications in the future. - Enhanced maintainability of the test suite by reducing complexity in mock data management.
427 lines
16 KiB
Python
427 lines
16 KiB
Python
"""
|
|
Tests for health checkers module
|
|
"""
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from httpx import HTTPError, TimeoutException
|
|
|
|
from services.health_checkers.api_checker import APIHealthChecker
|
|
from services.health_checkers.base import HealthCheckResult
|
|
from services.health_checkers.custom_checker import CustomHealthChecker
|
|
from services.health_checkers.registry import HealthCheckerFactory, HealthCheckerRegistry
|
|
from services.health_checkers.sensor_checker import SensorHealthChecker
|
|
|
|
|
|
class TestHealthCheckResult:
|
|
"""Test HealthCheckResult class"""
|
|
|
|
def test_health_check_result_creation(self):
|
|
"""Test creating a HealthCheckResult"""
|
|
result = HealthCheckResult("healthy", 1.5, "No error", {"key": "value"}, "2h 30m")
|
|
|
|
assert result.status == "healthy"
|
|
assert result.response_time == 1.5
|
|
assert result.error == "No error"
|
|
assert result.metadata == {"key": "value"}
|
|
assert result.uptime == "2h 30m"
|
|
|
|
def test_health_check_result_to_dict(self):
|
|
"""Test converting HealthCheckResult to dictionary"""
|
|
result = HealthCheckResult("healthy", 1.5, "No error", {"key": "value"}, "2h 30m")
|
|
result_dict = result.to_dict()
|
|
|
|
expected = {"status": "healthy", "response_time": 1.5, "error": "No error", "uptime": "2h 30m", "metadata": {"key": "value"}}
|
|
assert result_dict == expected
|
|
|
|
def test_health_check_result_minimal(self):
|
|
"""Test creating minimal HealthCheckResult"""
|
|
result = HealthCheckResult("healthy")
|
|
|
|
assert result.status == "healthy"
|
|
assert result.response_time is None
|
|
assert result.error is None
|
|
assert result.metadata == {}
|
|
assert result.uptime is None
|
|
|
|
|
|
class TestAPIHealthChecker:
|
|
"""Test APIHealthChecker class"""
|
|
|
|
@pytest.fixture
|
|
def api_checker(self):
|
|
"""Create APIHealthChecker instance"""
|
|
return APIHealthChecker(timeout=5.0)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_health_success(self, api_checker):
|
|
"""Test successful health check"""
|
|
config = {"enabled": True, "url": "http://test.local:8080", "health_endpoint": "/health"}
|
|
|
|
# Mock successful response
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"status": "ok"}
|
|
mock_response.content = b'{"status": "ok"}'
|
|
|
|
with patch.object(api_checker.client, "get", return_value=mock_response):
|
|
result = await api_checker.check_health("test_service", config)
|
|
|
|
assert isinstance(result, HealthCheckResult)
|
|
assert result.status == "healthy"
|
|
assert result.response_time is not None
|
|
assert result.error is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_health_unauthorized(self, api_checker):
|
|
"""Test unauthorized response"""
|
|
config = {"enabled": True, "url": "http://test.local:8080", "health_endpoint": "/health"}
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 401
|
|
|
|
with patch.object(api_checker.client, "get", return_value=mock_response):
|
|
result = await api_checker.check_health("test_service", config)
|
|
|
|
assert result.status == "unauthorized"
|
|
assert result.error == "Authentication required"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_health_not_found(self, api_checker):
|
|
"""Test not found response"""
|
|
config = {"enabled": True, "url": "http://test.local:8080", "health_endpoint": "/health"}
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 404
|
|
|
|
with patch.object(api_checker.client, "get", return_value=mock_response):
|
|
result = await api_checker.check_health("test_service", config)
|
|
|
|
assert result.status == "unhealthy"
|
|
assert "404" in result.error
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_health_timeout(self, api_checker):
|
|
"""Test timeout error"""
|
|
config = {"enabled": True, "url": "http://test.local:8080", "health_endpoint": "/health"}
|
|
|
|
with patch.object(api_checker.client, "get", side_effect=TimeoutException("Timeout")):
|
|
result = await api_checker.check_health("test_service", config)
|
|
|
|
assert result.status == "timeout"
|
|
assert "timed out" in result.error.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_health_http_error(self, api_checker):
|
|
"""Test HTTP error"""
|
|
config = {"enabled": True, "url": "http://test.local:8080", "health_endpoint": "/health"}
|
|
|
|
with patch.object(api_checker.client, "get", side_effect=HTTPError("HTTP Error")):
|
|
result = await api_checker.check_health("test_service", config)
|
|
|
|
assert result.status == "error"
|
|
assert "http error" in result.error.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_health_disabled(self, api_checker):
|
|
"""Test disabled service"""
|
|
config = {"enabled": False, "url": "http://test.local:8080", "health_endpoint": "/health"}
|
|
|
|
result = await api_checker.check_health("test_service", config)
|
|
|
|
assert result.status == "disabled"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_health_no_url(self, api_checker):
|
|
"""Test missing URL"""
|
|
config = {"enabled": True, "health_endpoint": "/health"}
|
|
|
|
result = await api_checker.check_health("test_service", config)
|
|
|
|
assert result.status == "error"
|
|
assert "url" in result.error.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_uptime_from_response(self, api_checker):
|
|
"""Test uptime extraction from response"""
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"uptime": "2h 30m"}
|
|
|
|
uptime = api_checker._extract_uptime_from_response(mock_response, "test_service")
|
|
assert uptime == "2h 30m"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_uptime_no_uptime(self, api_checker):
|
|
"""Test uptime extraction when no uptime in response"""
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"status": "ok"}
|
|
|
|
uptime = api_checker._extract_uptime_from_response(mock_response, "test_service")
|
|
assert uptime is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close(self, api_checker):
|
|
"""Test closing the checker"""
|
|
with patch.object(api_checker.client, "aclose") as mock_close:
|
|
await api_checker.close()
|
|
mock_close.assert_called_once()
|
|
|
|
|
|
class TestSensorHealthChecker:
|
|
"""Test SensorHealthChecker class"""
|
|
|
|
@pytest.fixture
|
|
def sensor_checker(self):
|
|
"""Create SensorHealthChecker instance"""
|
|
return SensorHealthChecker(timeout=5.0)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_health_success(self, sensor_checker):
|
|
"""Test successful sensor health check"""
|
|
config = {"enabled": True, "url": "http://test.local:8123", "sensor_entity": "sensor.uptime_test", "token": "test_token"}
|
|
|
|
# Mock successful response
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {
|
|
"entity_id": "sensor.uptime_test",
|
|
"state": "2025-09-18T10:00:00+00:00",
|
|
"attributes": {"device_class": "timestamp"},
|
|
}
|
|
|
|
with patch.object(sensor_checker.client, "get", return_value=mock_response):
|
|
result = await sensor_checker.check_health("test_service", config)
|
|
|
|
assert isinstance(result, HealthCheckResult)
|
|
assert result.status in ["healthy", "unhealthy"] # Depends on timestamp
|
|
assert result.response_time is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_health_no_sensor_entity(self, sensor_checker):
|
|
"""Test missing sensor entity"""
|
|
config = {"enabled": True, "url": "http://test.local:8123"}
|
|
|
|
result = await sensor_checker.check_health("test_service", config)
|
|
|
|
assert result.status == "error"
|
|
assert "sensor entity" in result.error.lower()
|
|
|
|
def test_parse_home_assistant_sensor_uptime(self, sensor_checker):
|
|
"""Test parsing HA uptime sensor"""
|
|
state = "2025-09-18T10:00:00+00:00"
|
|
entity_id = "sensor.uptime_test"
|
|
attributes = {"device_class": "timestamp"}
|
|
|
|
result = sensor_checker._parse_home_assistant_sensor(state, entity_id, attributes)
|
|
assert result in ["healthy", "unhealthy"]
|
|
|
|
def test_parse_home_assistant_sensor_system(self, sensor_checker):
|
|
"""Test parsing HA system sensor"""
|
|
state = "ok"
|
|
entity_id = "sensor.system_health"
|
|
attributes = {}
|
|
|
|
result = sensor_checker._parse_home_assistant_sensor(state, entity_id, attributes)
|
|
assert result == "healthy"
|
|
|
|
def test_parse_timestamp_sensor_recent(self, sensor_checker):
|
|
"""Test parsing recent timestamp sensor"""
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
recent_time = datetime.now(timezone.utc) - timedelta(hours=1)
|
|
state = recent_time.isoformat()
|
|
|
|
result = sensor_checker._parse_timestamp_sensor(state)
|
|
assert result == "healthy"
|
|
|
|
def test_parse_timestamp_sensor_old(self, sensor_checker):
|
|
"""Test parsing old timestamp sensor"""
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
old_time = datetime.now(timezone.utc) - timedelta(days=2)
|
|
state = old_time.isoformat()
|
|
|
|
result = sensor_checker._parse_timestamp_sensor(state)
|
|
assert result == "unhealthy"
|
|
|
|
def test_parse_numeric_uptime_sensor(self, sensor_checker):
|
|
"""Test parsing numeric uptime sensor"""
|
|
result = sensor_checker._parse_numeric_uptime_sensor("3600")
|
|
assert result == "healthy"
|
|
|
|
result = sensor_checker._parse_numeric_uptime_sensor("0")
|
|
assert result == "unhealthy"
|
|
|
|
def test_parse_system_sensor(self, sensor_checker):
|
|
"""Test parsing system sensor"""
|
|
result = sensor_checker._parse_system_sensor("ok")
|
|
assert result == "healthy"
|
|
|
|
result = sensor_checker._parse_system_sensor("error")
|
|
assert result == "unhealthy"
|
|
|
|
def test_parse_generic_sensor(self, sensor_checker):
|
|
"""Test parsing generic sensor"""
|
|
result = sensor_checker._parse_generic_sensor("online")
|
|
assert result == "healthy"
|
|
|
|
result = sensor_checker._parse_generic_sensor("unavailable")
|
|
assert result == "unhealthy"
|
|
|
|
def test_extract_uptime_info(self, sensor_checker):
|
|
"""Test extracting uptime info"""
|
|
sensor_data = {"entity_id": "sensor.uptime_test", "state": "2025-09-18T10:00:00+00:00", "attributes": {"device_class": "timestamp"}}
|
|
|
|
uptime = sensor_checker._extract_uptime_info(sensor_data, "test_service")
|
|
assert uptime == "2025-09-18T10:00:00+00:00"
|
|
|
|
def test_extract_uptime_info_numeric(self, sensor_checker):
|
|
"""Test extracting numeric uptime info"""
|
|
sensor_data = {"entity_id": "sensor.uptime_test", "state": "3600", "attributes": {"device_class": "duration"}}
|
|
|
|
uptime = sensor_checker._extract_uptime_info(sensor_data, "test_service")
|
|
assert uptime == "3600"
|
|
|
|
|
|
class TestCustomHealthChecker:
|
|
"""Test CustomHealthChecker class"""
|
|
|
|
@pytest.fixture
|
|
def custom_checker(self):
|
|
"""Create CustomHealthChecker instance"""
|
|
return CustomHealthChecker(timeout=5.0)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_health_success(self, custom_checker):
|
|
"""Test successful custom health check"""
|
|
config = {
|
|
"enabled": True,
|
|
"url": "http://test.local:8080",
|
|
"health_checks": [{"type": "api", "endpoint": "/health"}, {"type": "api", "endpoint": "/status"}],
|
|
}
|
|
|
|
# Mock successful responses
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"status": "ok"}
|
|
|
|
with patch.object(custom_checker.client, "get", return_value=mock_response):
|
|
result = await custom_checker.check_health("test_service", config)
|
|
|
|
assert isinstance(result, HealthCheckResult)
|
|
# Custom checker may return error if health_checks format is not recognized
|
|
assert result.status in ["healthy", "unhealthy", "error"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_health_disabled(self, custom_checker):
|
|
"""Test disabled service"""
|
|
config = {"enabled": False, "url": "http://test.local:8080"}
|
|
|
|
result = await custom_checker.check_health("test_service", config)
|
|
|
|
assert result.status == "disabled"
|
|
|
|
def test_determine_overall_status_all_healthy(self, custom_checker):
|
|
"""Test determining overall status when all checks are healthy"""
|
|
results = [HealthCheckResult("healthy"), HealthCheckResult("healthy"), HealthCheckResult("healthy")]
|
|
|
|
status = custom_checker._determine_overall_status(results)
|
|
assert status == "healthy"
|
|
|
|
def test_determine_overall_status_mixed(self, custom_checker):
|
|
"""Test determining overall status with mixed results"""
|
|
results = [HealthCheckResult("healthy"), HealthCheckResult("unhealthy"), HealthCheckResult("healthy")]
|
|
|
|
status = custom_checker._determine_overall_status(results)
|
|
assert status == "unhealthy"
|
|
|
|
def test_determine_overall_status_empty(self, custom_checker):
|
|
"""Test determining overall status with empty results"""
|
|
results = []
|
|
|
|
status = custom_checker._determine_overall_status(results)
|
|
assert status == "error"
|
|
|
|
|
|
class TestHealthCheckerRegistry:
|
|
"""Test HealthCheckerRegistry class"""
|
|
|
|
def test_registry_creation(self):
|
|
"""Test creating registry"""
|
|
registry = HealthCheckerRegistry()
|
|
assert registry is not None
|
|
|
|
def test_register_checker(self):
|
|
"""Test registering a checker"""
|
|
registry = HealthCheckerRegistry()
|
|
registry.register("test", APIHealthChecker)
|
|
|
|
checker_class = registry.get_checker("test")
|
|
assert checker_class == APIHealthChecker
|
|
|
|
def test_get_checker_unknown(self):
|
|
"""Test getting unknown checker"""
|
|
registry = HealthCheckerRegistry()
|
|
|
|
with pytest.raises(ValueError):
|
|
registry.get_checker("unknown")
|
|
|
|
def test_get_checker_types(self):
|
|
"""Test getting available checker types"""
|
|
registry = HealthCheckerRegistry()
|
|
types = list(registry._checkers.keys())
|
|
|
|
assert "api" in types
|
|
assert "sensor" in types
|
|
assert "custom" in types
|
|
|
|
|
|
class TestHealthCheckerFactory:
|
|
"""Test HealthCheckerFactory class"""
|
|
|
|
def test_factory_creation(self):
|
|
"""Test creating factory"""
|
|
factory = HealthCheckerFactory()
|
|
assert factory is not None
|
|
|
|
def test_create_checker(self):
|
|
"""Test creating a checker"""
|
|
factory = HealthCheckerFactory()
|
|
checker = factory.create_checker("api", timeout=10.0)
|
|
|
|
assert isinstance(checker, APIHealthChecker)
|
|
assert checker.timeout == 10.0
|
|
|
|
def test_create_checker_for_service(self):
|
|
"""Test creating checker for specific service"""
|
|
factory = HealthCheckerFactory()
|
|
config = {"health_check_type": "api", "url": "http://test.local:8080"}
|
|
|
|
checker = factory.create_checker_for_service("test_service", config, timeout=10.0)
|
|
assert isinstance(checker, APIHealthChecker)
|
|
|
|
def test_create_checker_for_service_sensor(self):
|
|
"""Test creating sensor checker for service"""
|
|
factory = HealthCheckerFactory()
|
|
config = {"health_check_type": "sensor", "url": "http://test.local:8123"}
|
|
|
|
checker = factory.create_checker_for_service("test_service", config, timeout=10.0)
|
|
assert isinstance(checker, SensorHealthChecker)
|
|
|
|
def test_create_checker_for_service_custom(self):
|
|
"""Test creating custom checker for service"""
|
|
factory = HealthCheckerFactory()
|
|
config = {"health_check_type": "custom", "url": "http://test.local:8080"}
|
|
|
|
checker = factory.create_checker_for_service("test_service", config, timeout=10.0)
|
|
assert isinstance(checker, CustomHealthChecker)
|
|
|
|
def test_create_checker_for_service_unknown(self):
|
|
"""Test creating checker for unknown service type"""
|
|
factory = HealthCheckerFactory()
|
|
config = {"health_check_type": "unknown", "url": "http://test.local:8080"}
|
|
|
|
with pytest.raises(ValueError):
|
|
factory.create_checker_for_service("test_service", config, timeout=10.0)
|