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.
221 lines
9.1 KiB
Python
221 lines
9.1 KiB
Python
"""
|
|
Tests for status_checker module
|
|
"""
|
|
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from services.health_checkers.base import HealthCheckResult
|
|
from services.status_checker import ServiceStatusChecker
|
|
|
|
|
|
class TestServiceStatusChecker:
|
|
"""Test ServiceStatusChecker class"""
|
|
|
|
@pytest.fixture
|
|
def status_checker(self):
|
|
"""Create ServiceStatusChecker instance"""
|
|
return ServiceStatusChecker(timeout=5.0)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_service_health_enabled(self, status_checker):
|
|
"""Test checking enabled service health"""
|
|
config = {"enabled": True, "url": "http://test.local:8080", "health_check_type": "api", "health_endpoint": "/health"}
|
|
|
|
# Mock the checker
|
|
mock_checker = AsyncMock()
|
|
mock_checker.check_health.return_value = HealthCheckResult("healthy", 1.5, uptime="2h 30m")
|
|
|
|
with patch.object(status_checker, "_get_checker_for_service", return_value=mock_checker):
|
|
result = await status_checker.check_service_health("test_service", config)
|
|
|
|
assert result["status"] == "healthy"
|
|
assert result["response_time"] == 1.5
|
|
assert result["uptime"] == "2h 30m"
|
|
assert result["error"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_service_health_disabled(self, status_checker):
|
|
"""Test checking disabled service health"""
|
|
config = {"enabled": False, "url": "http://test.local:8080"}
|
|
|
|
result = await status_checker.check_service_health("test_service", config)
|
|
|
|
assert result["status"] == "disabled"
|
|
assert result["response_time"] is None
|
|
assert result["error"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_service_health_exception(self, status_checker):
|
|
"""Test checking service health with exception"""
|
|
config = {"enabled": True, "url": "http://test.local:8080", "health_check_type": "api"}
|
|
|
|
with patch.object(status_checker, "_get_checker_for_service", side_effect=Exception("Test error")):
|
|
result = await status_checker.check_service_health("test_service", config)
|
|
|
|
assert result["status"] == "error"
|
|
assert "Test error" in result["error"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_checker_for_service_new(self, status_checker):
|
|
"""Test getting new checker for service"""
|
|
config = {"health_check_type": "api", "url": "http://test.local:8080"}
|
|
|
|
mock_checker = AsyncMock()
|
|
with patch("services.status_checker.factory.create_checker_for_service", return_value=mock_checker):
|
|
checker = await status_checker._get_checker_for_service("test_service", config)
|
|
|
|
assert checker == mock_checker
|
|
assert "test_service" in status_checker.checkers
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_checker_for_service_cached(self, status_checker):
|
|
"""Test getting cached checker for service"""
|
|
config = {"health_check_type": "api", "url": "http://test.local:8080"}
|
|
|
|
mock_checker = AsyncMock()
|
|
status_checker.checkers["test_service"] = mock_checker
|
|
|
|
checker = await status_checker._get_checker_for_service("test_service", config)
|
|
|
|
assert checker == mock_checker
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_all_services_success(self, status_checker):
|
|
"""Test checking all services successfully"""
|
|
# Mock the services configuration
|
|
with patch(
|
|
"services.status_checker.SERVICES",
|
|
{
|
|
"service1": {"enabled": True, "url": "http://test1.local:8080", "health_check_type": "api"},
|
|
"service2": {"enabled": True, "url": "http://test2.local:8080", "health_check_type": "api"},
|
|
"service3": {"enabled": False, "url": "http://test3.local:8080"},
|
|
},
|
|
):
|
|
# Mock individual health checks
|
|
with patch.object(status_checker, "check_service_health") as mock_check:
|
|
mock_check.side_effect = [
|
|
{"status": "healthy", "response_time": 1.0, "error": None, "uptime": "1h", "metadata": {}},
|
|
{"status": "unhealthy", "response_time": 2.0, "error": "Connection failed", "uptime": None, "metadata": {}},
|
|
{"status": "disabled", "response_time": None, "error": None, "uptime": None, "metadata": {}},
|
|
]
|
|
|
|
result = await status_checker.check_all_services()
|
|
|
|
assert "service1" in result
|
|
assert "service2" in result
|
|
assert "service3" in result
|
|
assert result["service1"]["status"] == "healthy"
|
|
assert result["service2"]["status"] == "unhealthy"
|
|
assert result["service3"]["status"] == "disabled"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_all_services_with_exceptions(self, status_checker):
|
|
"""Test checking all services with some exceptions"""
|
|
with patch(
|
|
"services.status_checker.SERVICES",
|
|
{
|
|
"service1": {"enabled": True, "url": "http://test1.local:8080", "health_check_type": "api"},
|
|
"service2": {"enabled": True, "url": "http://test2.local:8080", "health_check_type": "api"},
|
|
},
|
|
):
|
|
with patch.object(status_checker, "check_service_health") as mock_check:
|
|
mock_check.side_effect = [
|
|
{"status": "healthy", "response_time": 1.0, "error": None, "uptime": "1h", "metadata": {}},
|
|
Exception("Service error"),
|
|
]
|
|
|
|
result = await status_checker.check_all_services()
|
|
|
|
assert "service1" in result
|
|
assert "service2" in result
|
|
assert result["service1"]["status"] == "healthy"
|
|
assert result["service2"]["status"] == "error"
|
|
assert "Service error" in result["service2"]["error"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close(self, status_checker):
|
|
"""Test closing the status checker"""
|
|
# Add some mock checkers
|
|
mock_checker1 = AsyncMock()
|
|
mock_checker2 = AsyncMock()
|
|
status_checker.checkers = {"service1": mock_checker1, "service2": mock_checker2}
|
|
|
|
await status_checker.close()
|
|
|
|
# Verify close was called on all checkers
|
|
mock_checker1.close.assert_called_once()
|
|
mock_checker2.close.assert_called_once()
|
|
|
|
def test_status_checker_initialization(self, status_checker):
|
|
"""Test status checker initialization"""
|
|
assert status_checker.timeout == 5.0
|
|
assert status_checker.checkers == {}
|
|
|
|
|
|
class TestStatusCheckerIntegration:
|
|
"""Integration tests for status checker"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_full_health_check_flow(self):
|
|
"""Test the complete health check flow"""
|
|
status_checker = ServiceStatusChecker(timeout=5.0)
|
|
|
|
# Mock the factory and checkers
|
|
mock_checker = AsyncMock()
|
|
mock_checker.check_health.return_value = HealthCheckResult("healthy", 1.5, uptime="2h 30m")
|
|
|
|
with patch("services.status_checker.factory.create_checker_for_service", return_value=mock_checker):
|
|
with patch(
|
|
"services.status_checker.SERVICES",
|
|
{"test_service": {"enabled": True, "url": "http://test.local:8080", "health_check_type": "api", "health_endpoint": "/health"}},
|
|
):
|
|
result = await status_checker.check_all_services()
|
|
|
|
assert "test_service" in result
|
|
assert result["test_service"]["status"] == "healthy"
|
|
assert result["test_service"]["response_time"] == 1.5
|
|
assert result["test_service"]["uptime"] == "2h 30m"
|
|
|
|
# Clean up
|
|
await status_checker.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concurrent_health_checks(self):
|
|
"""Test that health checks run concurrently"""
|
|
status_checker = ServiceStatusChecker(timeout=5.0)
|
|
|
|
# Track the order of calls
|
|
call_order = []
|
|
|
|
async def mock_check_health(service_name, config):
|
|
call_order.append(service_name)
|
|
# Simulate some processing time
|
|
await asyncio.sleep(0.1)
|
|
return {"status": "healthy", "response_time": 0.1, "error": None, "uptime": "1h", "metadata": {}}
|
|
|
|
with patch.object(status_checker, "check_service_health", side_effect=mock_check_health):
|
|
with patch(
|
|
"services.status_checker.SERVICES",
|
|
{
|
|
"service1": {"enabled": True, "url": "http://test1.local:8080", "health_check_type": "api"},
|
|
"service2": {"enabled": True, "url": "http://test2.local:8080", "health_check_type": "api"},
|
|
"service3": {"enabled": True, "url": "http://test3.local:8080", "health_check_type": "api"},
|
|
},
|
|
):
|
|
start_time = asyncio.get_event_loop().time()
|
|
await status_checker.check_all_services()
|
|
end_time = asyncio.get_event_loop().time()
|
|
|
|
# Should complete in roughly 0.1s (concurrent) rather than 0.3s (sequential)
|
|
assert end_time - start_time < 0.2
|
|
assert len(call_order) == 3
|
|
assert "service1" in call_order
|
|
assert "service2" in call_order
|
|
assert "service3" in call_order
|
|
|
|
# Clean up
|
|
await status_checker.close()
|