146 lines
4.7 KiB
Python
146 lines
4.7 KiB
Python
"""Unit tests for health monitor."""
|
|
|
|
import pytest
|
|
from datetime import datetime, timedelta
|
|
|
|
from src.data.health_monitor import HealthMonitor, HealthMetrics, HealthStatus
|
|
|
|
|
|
class TestHealthMetrics:
|
|
"""Tests for HealthMetrics."""
|
|
|
|
def test_record_success(self):
|
|
"""Test recording successful operation."""
|
|
metrics = HealthMetrics()
|
|
metrics.record_success(0.5)
|
|
|
|
assert metrics.status == HealthStatus.HEALTHY
|
|
assert metrics.success_count == 1
|
|
assert metrics.consecutive_failures == 0
|
|
assert len(metrics.response_times) == 1
|
|
|
|
def test_record_failure(self):
|
|
"""Test recording failed operation."""
|
|
metrics = HealthMetrics()
|
|
metrics.record_failure()
|
|
|
|
assert metrics.failure_count == 1
|
|
assert metrics.consecutive_failures == 1
|
|
|
|
def test_circuit_breaker(self):
|
|
"""Test circuit breaker opening."""
|
|
metrics = HealthMetrics()
|
|
|
|
# Record 5 failures
|
|
for _ in range(5):
|
|
metrics.record_failure()
|
|
|
|
assert metrics.circuit_breaker_open is True
|
|
assert metrics.consecutive_failures == 5
|
|
|
|
def test_should_attempt(self):
|
|
"""Test should_attempt logic."""
|
|
metrics = HealthMetrics()
|
|
|
|
# Should attempt if circuit breaker not open
|
|
assert metrics.should_attempt() is True
|
|
|
|
# Open circuit breaker
|
|
for _ in range(5):
|
|
metrics.record_failure()
|
|
|
|
# Should not attempt immediately
|
|
assert metrics.should_attempt(circuit_breaker_timeout=60) is False
|
|
|
|
def test_get_avg_response_time(self):
|
|
"""Test average response time calculation."""
|
|
metrics = HealthMetrics()
|
|
metrics.response_times.extend([0.1, 0.2, 0.3])
|
|
|
|
avg = metrics.get_avg_response_time()
|
|
assert avg == 0.2
|
|
|
|
|
|
class TestHealthMonitor:
|
|
"""Tests for HealthMonitor."""
|
|
|
|
@pytest.fixture
|
|
def monitor(self):
|
|
"""Create a health monitor instance."""
|
|
return HealthMonitor()
|
|
|
|
def test_record_success(self, monitor):
|
|
"""Test recording success."""
|
|
monitor.record_success("provider1", 0.5)
|
|
|
|
metrics = monitor.get_metrics("provider1")
|
|
assert metrics is not None
|
|
assert metrics.status == HealthStatus.HEALTHY
|
|
assert metrics.success_count == 1
|
|
|
|
def test_record_failure(self, monitor):
|
|
"""Test recording failure."""
|
|
monitor.record_failure("provider1")
|
|
|
|
metrics = monitor.get_metrics("provider1")
|
|
assert metrics is not None
|
|
assert metrics.failure_count == 1
|
|
assert metrics.consecutive_failures == 1
|
|
|
|
def test_is_healthy(self, monitor):
|
|
"""Test health checking."""
|
|
# No metrics yet - assume healthy
|
|
assert monitor.is_healthy("provider1") is True
|
|
|
|
# Record success
|
|
monitor.record_success("provider1", 0.5)
|
|
assert monitor.is_healthy("provider1") is True
|
|
|
|
# Record many failures
|
|
for _ in range(10):
|
|
monitor.record_failure("provider1")
|
|
|
|
assert monitor.is_healthy("provider1") is False
|
|
|
|
def test_get_health_status(self, monitor):
|
|
"""Test getting health status."""
|
|
monitor.record_success("provider1", 0.5)
|
|
status = monitor.get_health_status("provider1")
|
|
assert status == HealthStatus.HEALTHY
|
|
|
|
def test_select_healthiest(self, monitor):
|
|
"""Test selecting healthiest provider."""
|
|
# Make provider1 healthy
|
|
monitor.record_success("provider1", 0.1)
|
|
monitor.record_success("provider1", 0.2)
|
|
|
|
# Make provider2 unhealthy
|
|
monitor.record_failure("provider2")
|
|
monitor.record_failure("provider2")
|
|
monitor.record_failure("provider2")
|
|
|
|
healthiest = monitor.select_healthiest(["provider1", "provider2"])
|
|
assert healthiest == "provider1"
|
|
|
|
def test_reset_circuit_breaker(self, monitor):
|
|
"""Test resetting circuit breaker."""
|
|
# Open circuit breaker
|
|
for _ in range(5):
|
|
monitor.record_failure("provider1")
|
|
|
|
assert monitor.get_metrics("provider1").circuit_breaker_open is True
|
|
|
|
monitor.reset_circuit_breaker("provider1")
|
|
|
|
metrics = monitor.get_metrics("provider1")
|
|
assert metrics.circuit_breaker_open is False
|
|
assert metrics.consecutive_failures == 0
|
|
|
|
def test_reset_metrics(self, monitor):
|
|
"""Test resetting metrics."""
|
|
monitor.record_success("provider1", 0.5)
|
|
assert monitor.get_metrics("provider1") is not None
|
|
|
|
monitor.reset_metrics("provider1")
|
|
assert monitor.get_metrics("provider1") is None
|