pybreaker circuit breaker closed, open and half-open transitions: working versions (tested 2026-10-16)

pybreaker==1.4.1
pytest==9.1.1

from __future__ import annotations

from pybreaker import CircuitBreaker, CircuitBreakerListener


class GatewayDown(Exception):
    """Backend failure: counts against the breaker."""


class InvalidCharge(Exception):
    """Business error: must never trip the breaker."""


class FakeGateway:
    """In-memory stand-in for a flaky payment gateway (no network)."""

    def __init__(self) -> None:
        self.up = True
        self.calls = 0

    def charge(self, amount: int) -> str:
        self.calls += 1
        if amount <= 0:
            raise InvalidCharge("amount must be positive")
        if not self.up:
            raise GatewayDown(f"gateway unavailable for {amount}")
        return f"charged:{amount}"


class StateLog(CircuitBreakerListener):
    """Records every transition as "old->new" so tests can assert on them."""

    def __init__(self) -> None:
        self.transitions: list[str] = []

    def state_change(self, cb, old_state, new_state) -> None:
        old = old_state.name if old_state is not None else "none"
        self.transitions.append(f"{old}->{new_state.name}")


class ChargeService:
    """Charges through a breaker so a sick gateway cannot keep taking traffic."""

    def __init__(self, gateway: FakeGateway, breaker: CircuitBreaker) -> None:
        self.gateway = gateway
        self.breaker = breaker

    def charge(self, amount: int) -> str:
        return self.breaker.call(self.gateway.charge, amount)

import pytest
from pybreaker import CircuitBreaker, CircuitBreakerError

from breaker import ChargeService, FakeGateway, GatewayDown, InvalidCharge, StateLog


def make_service(gateway, **kwargs):
    log = StateLog()
    breaker = CircuitBreaker(listeners=[log], name="payments", **kwargs)
    return ChargeService(gateway, breaker), log


@pytest.fixture
def gateway():
    return FakeGateway()


def test_trip_raises_breaker_error_not_original(gateway):
    svc, log = make_service(gateway, fail_max=2, reset_timeout=60)
    gateway.up = False
    with pytest.raises(GatewayDown):
        svc.charge(10)
    with pytest.raises(CircuitBreakerError) as exc:
        svc.charge(10)
    assert "opened" in str(exc.value)
    assert svc.breaker.current_state == "open"
    assert log.transitions == ["closed->open"]


def test_open_short_circuits_without_calling_backend(gateway):
    svc, _ = make_service(gateway, fail_max=1, reset_timeout=60)
    gateway.up = False
    with pytest.raises(CircuitBreakerError):
        svc.charge(10)
    assert gateway.calls == 1
    with pytest.raises(CircuitBreakerError):
        svc.charge(10)
    assert gateway.calls == 1
    assert svc.breaker.current_state == "open"


def test_elapsed_timeout_half_opens_then_closes(gateway):
    svc, log = make_service(gateway, fail_max=1, reset_timeout=0)
    gateway.up = False
    with pytest.raises(CircuitBreakerError):
        svc.charge(10)
    gateway.up = True
    assert svc.charge(10) == "charged:10"
    assert svc.breaker.current_state == "closed"
    assert log.transitions == ["closed->open", "open->half-open", "half-open->closed"]


def test_failed_trial_reopens(gateway):
    svc, log = make_service(gateway, fail_max=1, reset_timeout=0)
    gateway.up = False
    with pytest.raises(CircuitBreakerError):
        svc.charge(10)
    with pytest.raises(CircuitBreakerError):
        svc.charge(10)
    assert svc.breaker.current_state == "open"
    assert log.transitions == ["closed->open", "open->half-open", "half-open->open"]


def test_half_open_needs_success_threshold(gateway):
    svc, _ = make_service(gateway, fail_max=1, reset_timeout=0, success_threshold=2)
    gateway.up = False
    with pytest.raises(CircuitBreakerError):
        svc.charge(10)
    gateway.up = True
    svc.charge(10)
    assert svc.breaker.current_state == "half-open"
    svc.charge(10)
    assert svc.breaker.current_state == "closed"


def test_excluded_exception_never_trips(gateway):
    svc, _ = make_service(gateway, fail_max=1, reset_timeout=60, exclude=[InvalidCharge])
    for _ in range(3):
        with pytest.raises(InvalidCharge):
            svc.charge(0)
    assert svc.breaker.current_state == "closed"
    assert svc.breaker.fail_counter == 0

[pytest]
pythonpath = .