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 @pytest.fixture 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)