mirror of
https://github.com/msoedov/agentic_security.git
synced 2026-09-29 11:22:25 +02:00
fei
fcd1fd20ae
record_failure ignored the half_open state: the minimum-sample gate (total >= 10) and the success counter meant a failure during probation neither re-opened the circuit nor prevented it from closing. One failure followed by three successes closed the circuit with the failure forgotten. Trip back to open on the first half-open failure, with fresh counters.
255 lines
7.9 KiB
Python
255 lines
7.9 KiB
Python
"""Tests for CircuitBreaker."""
|
|
|
|
import time
|
|
from agentic_security.executor.circuit_breaker import CircuitBreaker
|
|
|
|
|
|
class TestCircuitBreaker:
|
|
"""Test CircuitBreaker functionality."""
|
|
|
|
def test_initialization(self):
|
|
"""Test circuit breaker initialization."""
|
|
breaker = CircuitBreaker(failure_threshold=0.5, recovery_timeout=30)
|
|
|
|
assert breaker.failure_threshold == 0.5
|
|
assert breaker.recovery_timeout == 30
|
|
assert breaker.state == "closed"
|
|
assert breaker.failures == 0
|
|
assert breaker.successes == 0
|
|
|
|
def test_record_success(self):
|
|
"""Test recording successful requests."""
|
|
breaker = CircuitBreaker()
|
|
|
|
breaker.record_success()
|
|
assert breaker.successes == 1
|
|
assert breaker.failures == 0
|
|
assert breaker.state == "closed"
|
|
|
|
def test_record_failure(self):
|
|
"""Test recording failed requests."""
|
|
breaker = CircuitBreaker()
|
|
|
|
breaker.record_failure()
|
|
assert breaker.failures == 1
|
|
assert breaker.successes == 0
|
|
assert breaker.last_failure_time is not None
|
|
|
|
def test_circuit_opens_on_failure_threshold(self):
|
|
"""Test that circuit opens when failure threshold is exceeded."""
|
|
breaker = CircuitBreaker(failure_threshold=0.5, recovery_timeout=30)
|
|
|
|
# Record 10 requests: 6 failures, 4 successes (60% failure rate)
|
|
for _ in range(4):
|
|
breaker.record_success()
|
|
for _ in range(6):
|
|
breaker.record_failure()
|
|
|
|
# Circuit should be open (60% > 50% threshold)
|
|
assert breaker.state == "open"
|
|
assert breaker.is_open() is True
|
|
|
|
def test_circuit_stays_closed_below_threshold(self):
|
|
"""Test that circuit stays closed when below threshold."""
|
|
breaker = CircuitBreaker(failure_threshold=0.5, recovery_timeout=30)
|
|
|
|
# Record 10 requests: 4 failures, 6 successes (40% failure rate)
|
|
for _ in range(6):
|
|
breaker.record_success()
|
|
for _ in range(4):
|
|
breaker.record_failure()
|
|
|
|
# Circuit should stay closed (40% < 50% threshold)
|
|
assert breaker.state == "closed"
|
|
assert breaker.is_open() is False
|
|
|
|
def test_minimum_sample_size_required(self):
|
|
"""Test that minimum sample size is required before opening."""
|
|
breaker = CircuitBreaker(failure_threshold=0.5)
|
|
|
|
# Only 5 failures (below minimum of 10 total requests)
|
|
for _ in range(5):
|
|
breaker.record_failure()
|
|
|
|
# Circuit should stay closed (not enough samples)
|
|
assert breaker.state == "closed"
|
|
assert breaker.is_open() is False
|
|
|
|
def test_circuit_recovery_after_timeout(self):
|
|
"""Test that circuit enters half-open state after recovery timeout."""
|
|
breaker = CircuitBreaker(failure_threshold=0.5, recovery_timeout=1)
|
|
|
|
# Open the circuit
|
|
for _ in range(4):
|
|
breaker.record_success()
|
|
for _ in range(6):
|
|
breaker.record_failure()
|
|
|
|
assert breaker.state == "open"
|
|
|
|
# Wait for recovery timeout
|
|
time.sleep(1.1)
|
|
|
|
# Check if circuit moves to half-open
|
|
is_open = breaker.is_open()
|
|
assert is_open is False
|
|
assert breaker.state == "half_open"
|
|
|
|
def test_half_open_to_closed_on_successes(self):
|
|
"""Test that circuit closes from half-open after enough successes."""
|
|
breaker = CircuitBreaker(failure_threshold=0.5, recovery_timeout=1)
|
|
|
|
# Open the circuit
|
|
for _ in range(4):
|
|
breaker.record_success()
|
|
for _ in range(6):
|
|
breaker.record_failure()
|
|
|
|
# Wait for recovery
|
|
time.sleep(1.1)
|
|
breaker.is_open() # Triggers transition to half-open
|
|
|
|
assert breaker.state == "half_open"
|
|
|
|
# Record 3 successes
|
|
breaker.record_success()
|
|
breaker.record_success()
|
|
breaker.record_success()
|
|
|
|
# Should transition to closed
|
|
assert breaker.state == "closed"
|
|
|
|
def test_half_open_failure_reopens_circuit(self):
|
|
"""Test that a failure during half-open probation reopens the circuit."""
|
|
breaker = CircuitBreaker(failure_threshold=0.5, recovery_timeout=1)
|
|
|
|
# Open the circuit
|
|
for _ in range(4):
|
|
breaker.record_success()
|
|
for _ in range(6):
|
|
breaker.record_failure()
|
|
assert breaker.state == "open"
|
|
|
|
# Enter half-open after recovery timeout
|
|
time.sleep(1.1)
|
|
assert breaker.is_open() is False
|
|
assert breaker.state == "half_open"
|
|
|
|
# A single failure during probation trips the circuit back open
|
|
breaker.record_failure()
|
|
assert breaker.state == "open"
|
|
assert breaker.is_open() is True
|
|
|
|
def test_half_open_failure_not_forgotten_by_successes(self):
|
|
"""Test that a half-open failure is not forgotten by later successes."""
|
|
breaker = CircuitBreaker(failure_threshold=0.5, recovery_timeout=1)
|
|
|
|
# Open the circuit
|
|
for _ in range(4):
|
|
breaker.record_success()
|
|
for _ in range(6):
|
|
breaker.record_failure()
|
|
assert breaker.state == "open"
|
|
|
|
# Enter half-open after recovery timeout
|
|
time.sleep(1.1)
|
|
breaker.is_open()
|
|
assert breaker.state == "half_open"
|
|
|
|
# One failure followed by enough successes to close a healthy circuit
|
|
breaker.record_failure()
|
|
for _ in range(3):
|
|
breaker.record_success()
|
|
|
|
# The failure must keep the circuit open
|
|
assert breaker.state == "open"
|
|
|
|
def test_get_state(self):
|
|
"""Test get_state method."""
|
|
breaker = CircuitBreaker()
|
|
|
|
assert breaker.get_state() == "closed"
|
|
|
|
# Open the circuit
|
|
for _ in range(10):
|
|
breaker.record_failure()
|
|
|
|
assert breaker.get_state() == "open"
|
|
|
|
def test_get_failure_rate(self):
|
|
"""Test get_failure_rate method."""
|
|
breaker = CircuitBreaker()
|
|
|
|
# No requests
|
|
assert breaker.get_failure_rate() == 0.0
|
|
|
|
# 3 failures, 7 successes (30% failure rate)
|
|
for _ in range(7):
|
|
breaker.record_success()
|
|
for _ in range(3):
|
|
breaker.record_failure()
|
|
|
|
assert breaker.get_failure_rate() == 0.3
|
|
|
|
def test_reset(self):
|
|
"""Test reset method."""
|
|
breaker = CircuitBreaker()
|
|
|
|
# Record some activity
|
|
breaker.record_success()
|
|
breaker.record_failure()
|
|
for _ in range(10):
|
|
breaker.record_failure()
|
|
|
|
# Reset
|
|
breaker.reset()
|
|
|
|
# Should be back to initial state
|
|
assert breaker.state == "closed"
|
|
assert breaker.failures == 0
|
|
assert breaker.successes == 0
|
|
assert breaker.last_failure_time is None
|
|
|
|
def test_exact_failure_threshold(self):
|
|
"""Test behavior at exact failure threshold."""
|
|
breaker = CircuitBreaker(failure_threshold=0.5)
|
|
|
|
# Exactly 50% failure rate (5 failures, 5 successes)
|
|
for _ in range(5):
|
|
breaker.record_success()
|
|
for _ in range(5):
|
|
breaker.record_failure()
|
|
|
|
# Should be open (>= threshold)
|
|
assert breaker.state == "open"
|
|
|
|
def test_high_failure_threshold(self):
|
|
"""Test with high failure threshold."""
|
|
breaker = CircuitBreaker(failure_threshold=0.9)
|
|
|
|
# 80% failure rate (8 failures, 2 successes)
|
|
for _ in range(2):
|
|
breaker.record_success()
|
|
for _ in range(8):
|
|
breaker.record_failure()
|
|
|
|
# Should stay closed (80% < 90%)
|
|
assert breaker.state == "closed"
|
|
|
|
def test_zero_recovery_timeout(self):
|
|
"""Test with zero recovery timeout."""
|
|
breaker = CircuitBreaker(failure_threshold=0.5, recovery_timeout=0)
|
|
|
|
# Open the circuit
|
|
for _ in range(10):
|
|
breaker.record_failure()
|
|
|
|
assert breaker.state == "open"
|
|
|
|
# Should immediately allow recovery attempt
|
|
time.sleep(0.01)
|
|
is_open = breaker.is_open()
|
|
|
|
assert is_open is False
|
|
assert breaker.state == "half_open"
|