Spaces:
Sleeping
Sleeping
Claude
Add a circuit breaker to the multi-backend search fallback (item from review)
ba2278a unverified | import circuit_breaker as cb_module | |
| from circuit_breaker import CircuitBreaker, CircuitOpenError | |
| import pytest | |
| class _FakeClock: | |
| """Deterministic stand-in for time.monotonic().""" | |
| def __init__(self, start: float = 0.0): | |
| self.now = start | |
| def __call__(self) -> float: | |
| return self.now | |
| def advance(self, seconds: float) -> None: | |
| self.now += seconds | |
| def clock(monkeypatch): | |
| fake = _FakeClock() | |
| monkeypatch.setattr(cb_module.time, "monotonic", fake) | |
| return fake | |
| def test_before_call_is_a_noop_below_the_failure_threshold(clock): | |
| breaker = CircuitBreaker(failure_threshold=3, cooldown_seconds=60) | |
| breaker.record_failure("bing") | |
| breaker.record_failure("bing") | |
| breaker.before_call("bing") # should not raise - only 2 failures so far | |
| def test_before_call_raises_once_threshold_is_reached(clock): | |
| breaker = CircuitBreaker(failure_threshold=3, cooldown_seconds=60) | |
| for _ in range(3): | |
| breaker.record_failure("bing") | |
| with pytest.raises(CircuitOpenError): | |
| breaker.before_call("bing") | |
| def test_record_success_closes_the_circuit(clock): | |
| breaker = CircuitBreaker(failure_threshold=3, cooldown_seconds=60) | |
| for _ in range(3): | |
| breaker.record_failure("bing") | |
| breaker.record_success("bing") | |
| breaker.before_call("bing") # back to closed - should not raise | |
| def test_before_call_allows_a_trial_after_the_cooldown_elapses(clock): | |
| breaker = CircuitBreaker(failure_threshold=3, cooldown_seconds=60) | |
| for _ in range(3): | |
| breaker.record_failure("bing") | |
| clock.advance(61) | |
| breaker.before_call("bing") # cooldown elapsed - half-open, should not raise | |
| def test_a_failed_trial_after_cooldown_reopens_for_another_full_cooldown(clock): | |
| breaker = CircuitBreaker(failure_threshold=3, cooldown_seconds=60) | |
| for _ in range(3): | |
| breaker.record_failure("bing") | |
| clock.advance(61) | |
| breaker.before_call("bing") # half-open trial begins | |
| breaker.record_failure("bing") # ...and it failed | |
| clock.advance(30) # only 30s into the new cooldown | |
| with pytest.raises(CircuitOpenError): | |
| breaker.before_call("bing") | |
| clock.advance(31) # now past the new 60s cooldown | |
| breaker.before_call("bing") # should not raise | |
| def test_keys_are_independent(clock): | |
| breaker = CircuitBreaker(failure_threshold=2, cooldown_seconds=60) | |
| for _ in range(2): | |
| breaker.record_failure("bing") | |
| breaker.before_call("brave") # untouched key - should not raise | |
| with pytest.raises(CircuitOpenError): | |
| breaker.before_call("bing") | |
| def test_circuit_open_error_carries_the_key_and_remaining_seconds(clock): | |
| breaker = CircuitBreaker(failure_threshold=1, cooldown_seconds=60) | |
| breaker.record_failure("bing") | |
| clock.advance(10) | |
| with pytest.raises(CircuitOpenError) as exc_info: | |
| breaker.before_call("bing") | |
| assert exc_info.value.key == "bing" | |
| assert exc_info.value.retry_after == pytest.approx(50, abs=0.01) | |