60 lines
1.8 KiB
Python
60 lines
1.8 KiB
Python
|
"""Tests for circuit breaker and infrastructure."""
|
||
|
|
|
||
|
|
import time
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from world_intel_mcp.circuit_breaker import CircuitBreaker
|
||
|
|
|
||
|
|
|
||
|
|
def test_circuit_starts_closed() -> None:
|
||
|
|
cb = CircuitBreaker(failure_threshold=3, cooldown_seconds=1.0)
|
||
|
|
assert cb.is_available("test-source")
|
||
|
|
|
||
|
|
|
||
|
|
def test_circuit_trips_after_threshold() -> None:
|
||
|
|
cb = CircuitBreaker(failure_threshold=2, cooldown_seconds=10.0)
|
||
|
|
cb.record_failure("src")
|
||
|
|
assert cb.is_available("src") # 1 failure, threshold is 2
|
||
|
|
cb.record_failure("src")
|
||
|
|
assert not cb.is_available("src") # tripped
|
||
|
|
|
||
|
|
|
||
|
|
def test_circuit_recovers_after_cooldown() -> None:
|
||
|
|
cb = CircuitBreaker(failure_threshold=1, cooldown_seconds=0.5)
|
||
|
|
cb.record_failure("src")
|
||
|
|
assert not cb.is_available("src")
|
||
|
|
time.sleep(0.6)
|
||
|
|
assert cb.is_available("src") # half-open, allows probe
|
||
|
|
|
||
|
|
|
||
|
|
def test_success_resets_failures() -> None:
|
||
|
|
cb = CircuitBreaker(failure_threshold=3, cooldown_seconds=10.0)
|
||
|
|
cb.record_failure("src")
|
||
|
|
cb.record_failure("src")
|
||
|
|
cb.record_success("src")
|
||
|
|
assert cb.is_available("src")
|
||
|
|
# Even after 2 more failures, need 3 consecutive
|
||
|
|
cb.record_failure("src")
|
||
|
|
cb.record_failure("src")
|
||
|
|
assert cb.is_available("src") # only 2 since reset
|
||
|
|
|
||
|
|
|
||
|
|
def test_status_output() -> None:
|
||
|
|
cb = CircuitBreaker(failure_threshold=2, cooldown_seconds=60.0)
|
||
|
|
cb.record_success("healthy")
|
||
|
|
cb.record_failure("unhealthy")
|
||
|
|
cb.record_failure("unhealthy")
|
||
|
|
|
||
|
|
status = cb.status()
|
||
|
|
assert status["healthy"]["status"] == "closed"
|
||
|
|
assert status["unhealthy"]["status"] == "open"
|
||
|
|
assert status["unhealthy"]["total_trips"] == 1
|
||
|
|
|
||
|
|
|
||
|
|
def test_independent_sources() -> None:
|
||
|
|
cb = CircuitBreaker(failure_threshold=1, cooldown_seconds=60.0)
|
||
|
|
cb.record_failure("source_a")
|
||
|
|
assert not cb.is_available("source_a")
|
||
|
|
assert cb.is_available("source_b") # independent
|