"""Exhaustive tests for the thread-safe circuit breaker.

Organized by concern: CLOSED behavior, the OPEN reset-timeout boundary, HALF_OPEN probe
semantics, the single-probe gate, the failure classifier, control-flow (BaseException) handling,
the state-change callback, and concurrency stress.
"""

import threading
import time

import pytest

from core.hardening.breaker import CircuitBreaker, CircuitBreakerError


class ManualClock:
    """Deterministic monotonic clock so timeout-boundary tests don't depend on wall-clock sleeps."""

    def __init__(self):
        """Start at t=0."""
        self.t = 0.0

    def __call__(self):
        """Return the current time."""
        return self.t

    def advance(self, dt):
        """Advance the clock by dt seconds."""
        self.t += dt


def raises(exc):
    """Return a zero-arg callable that raises exc()."""

    def _fn():
        raise exc()

    return _fn


def ok(value='ok'):
    """Return a zero-arg callable that returns value."""

    def _fn():
        return value

    return _fn


def boom_callback(*_args):
    """Raise from an on_state_change callback (accepts the (name, old, new) args)."""
    raise RuntimeError('callback boom')


def make(**kw):
    """Build a CircuitBreaker with test-friendly defaults (override via kwargs)."""
    kw.setdefault('name', 't')
    kw.setdefault('fail_max', 3)
    kw.setdefault('reset_timeout', 0.2)
    kw.setdefault('success_threshold', 2)
    kw.setdefault('count_failure', lambda e: True)
    kw.setdefault('on_state_change', lambda *a: None)
    return CircuitBreaker(**kw)


def trip_open(cb, exc=ValueError):
    """Drive cb from CLOSED to OPEN by raising fail_max counted failures."""
    for _ in range(cb._fail_max):
        with pytest.raises(exc):
            cb.call(raises(exc))


# --------------------------------------------------------------------------- CLOSED


def test_closed_runs_fn_and_returns_value():
    """A closed breaker runs fn and returns its value."""
    cb = make()
    assert cb.call(ok('result')) == 'result'
    assert cb.state == 'closed'


def test_closed_stays_closed_across_many_successes():
    """Repeated successes keep the breaker closed."""
    cb = make(fail_max=2)
    for _ in range(50):
        assert cb.call(ok()) == 'ok'
    assert cb.state == 'closed'


def test_opens_only_after_exactly_fail_max_consecutive_failures():
    """fail_max-1 failures stay closed; the fail_max-th opens."""
    cb = make(fail_max=3)
    for _ in range(2):
        with pytest.raises(ValueError):
            cb.call(raises(ValueError))
    assert cb.state == 'closed'
    with pytest.raises(ValueError):
        cb.call(raises(ValueError))
    assert cb.state == 'open'


def test_success_resets_consecutive_failure_count():
    """A success in CLOSED clears the running failure count (failures must be consecutive)."""
    cb = make(fail_max=3)
    for _ in range(2):
        with pytest.raises(ValueError):
            cb.call(raises(ValueError))
    cb.call(ok())  # resets the counter
    for _ in range(2):
        with pytest.raises(ValueError):
            cb.call(raises(ValueError))
    assert cb.state == 'closed'  # only 2 since the reset, not 4
    with pytest.raises(ValueError):
        cb.call(raises(ValueError))
    assert cb.state == 'open'


# ----------------------------------------------------------------------------- OPEN


def test_open_fast_fails_without_running_fn():
    """An open breaker raises CircuitBreakerError and never invokes fn."""
    cb = make(fail_max=1)
    trip_open(cb)
    ran = {'n': 0}

    def fn():
        ran['n'] += 1
        return 'ok'

    with pytest.raises(CircuitBreakerError):
        cb.call(fn)
    assert ran['n'] == 0


def test_circuit_breaker_error_carries_name():
    """CircuitBreakerError exposes the breaker name."""
    cb = make(name='snowflake', fail_max=1)
    trip_open(cb)
    with pytest.raises(CircuitBreakerError) as ei:
        cb.call(ok())
    assert ei.value.name == 'snowflake'
    assert 'snowflake' in str(ei.value)


def test_open_rejects_just_before_reset_timeout():
    """Before reset_timeout elapses, the breaker keeps rejecting."""
    clock = ManualClock()
    cb = make(fail_max=1, reset_timeout=1.0, clock=clock)
    trip_open(cb)
    clock.advance(0.999)
    with pytest.raises(CircuitBreakerError):
        cb.call(ok())
    assert cb.state == 'open'


def test_open_admits_probe_at_exactly_reset_timeout():
    """At exactly reset_timeout (boundary is '<', not '<='), the next call is admitted as a probe."""
    clock = ManualClock()
    cb = make(fail_max=1, reset_timeout=1.0, success_threshold=1, clock=clock)
    trip_open(cb)
    clock.advance(1.0)
    assert (
        cb.call(ok()) == 'ok'
    )  # admitted (would raise CircuitBreakerError if still open)
    assert cb.state == 'closed'


# ------------------------------------------------------------------------ HALF_OPEN


def test_half_open_single_success_closes():
    """With success_threshold=1 a single probe success closes the breaker."""
    cb = make(fail_max=1, reset_timeout=0.0, success_threshold=1)
    trip_open(cb)
    assert cb.call(ok()) == 'ok'
    assert cb.state == 'closed'


def test_half_open_requires_threshold_consecutive_successes():
    """Closing needs success_threshold consecutive probe successes; it stays half-open until then."""
    cb = make(fail_max=1, reset_timeout=0.0, success_threshold=3)
    trip_open(cb)
    cb.call(ok())
    assert cb.state == 'halfopen'
    cb.call(ok())
    assert cb.state == 'halfopen'
    cb.call(ok())
    assert cb.state == 'closed'


def test_half_open_failure_reopens_immediately():
    """A single failing probe sends the breaker straight back to OPEN."""
    cb = make(fail_max=1, reset_timeout=0.0, success_threshold=5)
    trip_open(cb)
    cb.call(ok())  # one good probe (succ=1, still half-open)
    assert cb.state == 'halfopen'
    with pytest.raises(ValueError):
        cb.call(raises(ValueError))
    assert cb.state == 'open'


def test_half_open_failure_resets_success_progress():
    """A failing probe discards accumulated success progress; closing restarts from zero."""
    clock = ManualClock()
    cb = make(fail_max=1, reset_timeout=1.0, success_threshold=2, clock=clock)
    trip_open(cb)
    clock.advance(1.0)
    cb.call(ok())  # succ=1
    assert cb.state == 'halfopen'
    with pytest.raises(ValueError):
        cb.call(raises(ValueError))  # reopen, succ reset
    assert cb.state == 'open'
    clock.advance(1.0)
    cb.call(ok())  # succ=1 again (NOT 2) -> still half-open
    assert cb.state == 'halfopen'
    cb.call(ok())  # succ=2 -> closed
    assert cb.state == 'closed'


def test_reopen_restarts_the_reset_timeout_window():
    """Reopening from half-open stamps a fresh opened_at, so the full timeout must elapse again."""
    clock = ManualClock()
    cb = make(fail_max=1, reset_timeout=1.0, success_threshold=1, clock=clock)
    trip_open(cb)  # opened_at = 0
    clock.advance(1.0)
    with pytest.raises(ValueError):
        cb.call(raises(ValueError))  # probe fails -> reopen, opened_at = 1.0
    assert cb.state == 'open'
    clock.advance(0.5)  # only 0.5 since reopen
    with pytest.raises(CircuitBreakerError):
        cb.call(ok())
    clock.advance(0.5)  # now a full 1.0 since reopen
    assert cb.call(ok()) == 'ok'
    assert cb.state == 'closed'


def test_half_open_probe_returns_value():
    """A probe call returns fn's value just like a closed call."""
    cb = make(fail_max=1, reset_timeout=0.0, success_threshold=1)
    trip_open(cb)
    assert cb.call(ok('payload')) == 'payload'


def test_non_counted_exception_in_half_open_stays_half_open():
    """A probe exception the classifier excludes neither reopens nor counts as success."""
    cb = make(
        fail_max=1,
        reset_timeout=0.0,
        success_threshold=2,
        count_failure=lambda e: not isinstance(e, KeyError),
    )
    with pytest.raises(ValueError):
        cb.call(raises(ValueError))  # counted -> open
    with pytest.raises(KeyError):
        cb.call(raises(KeyError))  # probe, NOT counted -> stays half-open
    assert cb.state == 'halfopen'
    cb.call(ok())  # succ=1
    cb.call(ok())  # succ=2 -> closed
    assert cb.state == 'closed'


# ------------------------------------------------------------------- single-probe gate


def test_probe_releases_gate_on_success():
    """A successful probe releases the gate so the next half-open admit succeeds."""
    cb = make(fail_max=1, reset_timeout=0.0, success_threshold=3)
    trip_open(cb)
    cb.call(ok())
    cb.call(ok())  # would CircuitBreakerError if the gate had leaked
    cb.call(ok())
    assert cb.state == 'closed'


def test_probe_releases_gate_on_exception():
    """A probe that raises still releases the gate (no permanent stuck-open)."""
    cb = make(fail_max=1, reset_timeout=0.0, success_threshold=1)
    trip_open(cb)
    with pytest.raises(ValueError):
        cb.call(raises(ValueError))  # probe raises; gate must release
    cb.call(ok())  # next probe admitted -> closes
    assert cb.state == 'closed'


def test_half_open_rejects_concurrent_callers_while_a_probe_is_in_flight():
    """While one probe is running, other callers are fast-failed (only one probe at a time)."""
    cb = make(fail_max=1, reset_timeout=0.0, success_threshold=5)
    trip_open(cb)
    in_probe = threading.Event()
    release = threading.Event()

    def slow_probe():
        in_probe.set()
        release.wait(2.0)
        return 'ok'

    t = threading.Thread(target=lambda: cb.call(slow_probe))
    t.start()
    assert in_probe.wait(2.0)  # the probe holds the gate
    with pytest.raises(CircuitBreakerError):
        cb.call(ok())  # second caller rejected while the probe is in flight
    release.set()
    t.join()


def test_gate_returns_to_single_permit_after_many_cycles():
    """After many open/probe/close cycles the gate still holds exactly one permit (no inflation/leak)."""
    clock = ManualClock()
    cb = make(fail_max=1, reset_timeout=1.0, success_threshold=1, clock=clock)
    for i in range(25):
        with pytest.raises(ValueError):
            cb.call(raises(ValueError))  # open
        clock.advance(1.0)
        cb.call(ok())  # probe -> closed
    assert cb.state == 'closed'
    # BoundedSemaphore(1): an idle gate must read exactly one permit.
    assert cb._half_open._value == 1


# ------------------------------------------------------------------------ classifier


def test_classifier_excluded_exception_never_trips():
    """Exceptions the classifier excludes are re-raised but never counted toward tripping."""
    cb = make(fail_max=2, count_failure=lambda e: not isinstance(e, KeyError))
    for _ in range(10):
        with pytest.raises(KeyError):
            cb.call(raises(KeyError))
    assert cb.state == 'closed'


def test_classifier_included_exception_trips():
    """Exceptions the classifier includes count and trip the breaker."""
    cb = make(fail_max=2, count_failure=lambda e: isinstance(e, ValueError))
    for _ in range(2):
        with pytest.raises(ValueError):
            cb.call(raises(ValueError))
    assert cb.state == 'open'


def test_classifier_can_select_by_exception_type():
    """A mixed stream only trips on the classified type."""
    cb = make(fail_max=2, count_failure=lambda e: isinstance(e, TimeoutError))
    with pytest.raises(KeyError):
        cb.call(raises(KeyError))  # ignored
    with pytest.raises(TimeoutError):
        cb.call(raises(TimeoutError))  # counted (1)
    assert cb.state == 'closed'
    with pytest.raises(TimeoutError):
        cb.call(raises(TimeoutError))  # counted (2) -> open
    assert cb.state == 'open'


def test_throwing_classifier_does_not_mask_downstream_error():
    """If count_failure raises, the original downstream error propagates (not the classifier's)."""

    def bad_classifier(e):
        raise RuntimeError('classifier bug')

    cb = make(fail_max=2, count_failure=bad_classifier)
    for _ in range(2):
        with pytest.raises(ValueError):  # ValueError, NOT RuntimeError
            cb.call(raises(ValueError))
    assert cb.state == 'open'  # the failure was still counted


# --------------------------------------------------------------- control-flow exceptions


def test_base_exception_is_not_counted_as_a_failure():
    """BaseException (e.g. KeyboardInterrupt) propagates and never trips the breaker."""
    cb = make(fail_max=1)
    with pytest.raises(KeyboardInterrupt):
        cb.call(raises(KeyboardInterrupt))
    assert cb.state == 'closed'


def test_base_exception_during_probe_releases_gate():
    """A BaseException raised by a probe still releases the gate."""
    cb = make(fail_max=1, reset_timeout=0.0, success_threshold=1)
    trip_open(cb)
    with pytest.raises(KeyboardInterrupt):
        cb.call(
            raises(KeyboardInterrupt)
        )  # probe; gate must release despite BaseException
    cb.call(ok())  # next probe admitted -> closes
    assert cb.state == 'closed'


# -------------------------------------------------------------------- state callback


def test_on_state_change_fires_for_every_transition():
    """The callback observes closed->open, open->halfopen, halfopen->open and halfopen->closed."""
    seen = []
    clock = ManualClock()
    cb = make(
        fail_max=1,
        reset_timeout=1.0,
        success_threshold=1,
        clock=clock,
        on_state_change=lambda name, old, new: seen.append((old, new)),
    )
    with pytest.raises(ValueError):
        cb.call(raises(ValueError))  # closed -> open
    clock.advance(1.0)
    with pytest.raises(ValueError):
        cb.call(raises(ValueError))  # open -> halfopen -> open
    clock.advance(1.0)
    cb.call(ok())  # open -> halfopen -> closed
    assert ('closed', 'open') in seen
    assert ('open', 'halfopen') in seen
    assert ('halfopen', 'open') in seen
    assert ('halfopen', 'closed') in seen


def test_on_state_change_receives_breaker_name():
    """The callback's first argument is the breaker name."""
    names = []
    cb = make(
        name='abacus', fail_max=1, on_state_change=lambda name, o, n: names.append(name)
    )
    trip_open(cb)
    assert names == ['abacus']


def test_state_change_callback_runs_outside_the_lock():
    """A blocking callback does not serialize other threads' admissions."""
    in_callback = threading.Event()
    releasing = threading.Event()

    def blocking_cb(name, old, new):
        if new == 'open':
            in_callback.set()
            releasing.wait(2.0)

    cb = make(fail_max=1, on_state_change=blocking_cb)

    def trip():
        try:
            cb.call(raises(ValueError))
        except ValueError:
            pass

    a = threading.Thread(target=trip)
    a.start()
    assert in_callback.wait(2.0)  # A is blocked inside the OPEN callback
    t0 = time.time()
    with pytest.raises(
        CircuitBreakerError
    ):  # breaker already OPEN; B must fast-fail...
        cb.call(ok())
    assert time.time() - t0 < 0.5  # ...not serialize behind A's blocked callback
    releasing.set()
    a.join()


def test_throwing_callback_does_not_corrupt_state():
    """A throwing callback is logged, not propagated, and never strands the state machine."""
    cb = make(fail_max=1, on_state_change=boom_callback)
    with pytest.raises(ValueError):
        cb.call(raises(ValueError))
    assert cb.state == 'open'  # transition still happened


def test_throwing_callback_does_not_strand_the_probe_gate():
    """Even if the open-callback throws, the breaker can still admit a probe and recover."""
    cb = make(
        fail_max=1,
        reset_timeout=0.0,
        success_threshold=1,
        on_state_change=boom_callback,
    )
    with pytest.raises(ValueError):
        cb.call(raises(ValueError))  # open (callback throws, swallowed+logged)
    assert cb.call(ok()) == 'ok'  # probe admitted despite the throwing transitions
    assert cb.state == 'closed'


# --------------------------------------------------------------------------- concurrency


def test_no_io_serialization():
    """Concurrent calls run in parallel; the lock never wraps the work."""
    cb = make(fail_max=100)
    conc = {'n': 0, 'max': 0}
    lk = threading.Lock()

    def work():
        with lk:
            conc['n'] += 1
            conc['max'] = max(conc['max'], conc['n'])
        time.sleep(0.1)
        with lk:
            conc['n'] -= 1

    threads = [threading.Thread(target=lambda: cb.call(work)) for _ in range(15)]
    t = time.time()
    [x.start() for x in threads]
    [x.join() for x in threads]
    assert conc['max'] >= 10 and (time.time() - t) < 1.0  # concurrent, not serialized


def test_concurrent_successes_all_run_and_keep_closed():
    """Many concurrent successful calls all execute, the breaker stays closed, the gate stays intact."""
    cb = make(fail_max=5)
    done = {'n': 0}
    lk = threading.Lock()

    def call_ok():
        cb.call(ok())
        with lk:
            done['n'] += 1

    threads = [threading.Thread(target=call_ok) for _ in range(60)]
    [x.start() for x in threads]
    [x.join() for x in threads]
    assert done['n'] == 60
    assert cb.state == 'closed'
    assert cb._half_open._value == 1


def test_concurrent_failures_trip_the_breaker_then_fast_fail():
    """Concurrent failures trip the breaker exactly once; afterwards calls fast-fail."""
    cb = make(fail_max=5)
    runs = {'n': 0}
    lk = threading.Lock()

    def attempt():
        def fn():
            with lk:
                runs['n'] += 1
            raise ValueError()

        try:
            cb.call(fn)
        except (ValueError, CircuitBreakerError):
            pass

    threads = [threading.Thread(target=attempt) for _ in range(40)]
    [x.start() for x in threads]
    [x.join() for x in threads]
    assert cb.state == 'open'  # it tripped
    assert runs['n'] >= 5  # at least fail_max real failures occurred
    with pytest.raises(CircuitBreakerError):  # and now it fast-fails
        cb.call(ok())


def test_exactly_one_probe_admitted_under_contention():
    """Under contention past reset_timeout, exactly one probe is admitted; the rest fast-fail."""
    clock = ManualClock()
    cb = make(fail_max=1, reset_timeout=0.05, success_threshold=10, clock=clock)
    trip_open(cb)
    clock.advance(0.06)  # deterministically past reset_timeout
    ran = {'n': 0}
    lk = threading.Lock()
    release = threading.Event()
    start = threading.Barrier(12)

    def probe():
        with lk:
            ran['n'] += 1
        release.wait(2.0)  # hold the admitted probe so the others contend for the gate
        return 'ok'

    def attempt():
        start.wait()
        try:
            cb.call(probe)
        except CircuitBreakerError:
            pass

    threads = [threading.Thread(target=attempt) for _ in range(12)]
    [t.start() for t in threads]
    deadline = time.time() + 2.0
    while (
        ran['n'] < 1 or sum(t.is_alive() for t in threads) > 1
    ) and time.time() < deadline:
        time.sleep(0.005)
    assert ran['n'] == 1  # exactly one probe admitted; the other 11 fast-failed
    release.set()
    [t.join() for t in threads]


# --------------------------------------------------------------------- decorator form


def test_decorator_runs_function_through_breaker_and_returns_value():
    """Used as @breaker, the wrapped function passes args/kwargs through and returns its value."""
    cb = make(fail_max=2)

    @cb
    def add(a, b, *, c=0):
        return a + b + c

    assert add(1, 2, c=3) == 6
    assert cb.state == 'closed'


def test_decorator_trips_and_then_fast_fails():
    """A decorated function's failures trip the breaker; afterwards it fast-fails without running."""
    cb = make(fail_max=2)
    runs = {'n': 0}

    @cb
    def flaky():
        runs['n'] += 1
        raise ValueError()

    for _ in range(2):
        with pytest.raises(ValueError):
            flaky()
    assert cb.state == 'open'
    with pytest.raises(CircuitBreakerError):
        flaky()
    assert runs['n'] == 2  # the open breaker never invoked the body again


def test_decorator_preserves_function_metadata():
    """functools.wraps keeps the wrapped function's name and docstring."""
    cb = make()

    @cb
    def fetch_account():
        """Fetch an account."""
        return 'ok'

    assert fetch_account.__name__ == 'fetch_account'
    assert fetch_account.__doc__ == 'Fetch an account.'


def test_failures_while_already_open_do_not_re_trip_or_slide_opened_at():
    """A straggler failing while the breaker is already OPEN must not re-trip it.

    It must not re-transition OPEN->OPEN, slide the reset clock, or re-fire on_state_change.
    fail_max=1 so a SINGLE straggler reproduces the pre-fix bug (with fail_max>1, the post-trip
    _fails reset means one straggler can't reach the threshold, so the test would pass on the
    buggy code and give no signal).
    """
    clock = {'t': 0.0}
    changes = []
    cb = CircuitBreaker(
        name='t',
        fail_max=1,
        reset_timeout=10.0,
        success_threshold=1,
        count_failure=lambda e: True,
        on_state_change=lambda n, o, w: changes.append((o, w)),
        clock=lambda: clock['t'],
    )

    def boom():
        raise RuntimeError('x')

    with pytest.raises(RuntimeError):
        cb.call(boom)  # 1 failure -> OPEN at t=0
    assert cb.state == 'open'
    assert changes == [('closed', 'open')]
    assert cb._opened_at == 0.0

    # A straggler in-flight call fails while OPEN (simulate by driving the internal path directly).
    clock['t'] = 5.0
    cb._fire(cb._on_failure())

    assert cb.state == 'open'
    assert changes == [('closed', 'open')]  # NOT a second ('open','open')
    assert cb._opened_at == 0.0  # not slid forward to 5.0