"""Tests for the generic guard mechanism (registry, adapters, transport normalization, retry)."""

import httpx
import pytest
import requests

from core.hardening.breaker import CircuitBreakerError
from core.hardening.guards import (
    DownstreamServerError,
    DownstreamTransportError,
    GuardRegistry,
    _http_status,
    httpx_timeout_adapter,
    is_downstream_failure,
    no_timeout_adapter,
    requests_timeout_adapter,
)


class Policy:
    """Minimal duck-typed policy for the generic registry/adapters."""

    def __init__(
        self,
        fail_max=3,
        reset_timeout=1.0,
        success_threshold=1,
        retries=0,
        connect_timeout=3.0,
        read_timeout=10.0,
    ):
        """Store the policy fields the mechanism reads."""
        self.fail_max = fail_max
        self.reset_timeout = reset_timeout
        self.success_threshold = success_threshold
        self.retries = retries
        self.connect_timeout = connect_timeout
        self.read_timeout = read_timeout


class Resp:
    """Minimal HTTP-response stand-in carrying a status_code."""

    def __init__(self, code):
        """Store the status code."""
        self.status_code = code


def raising(exc):
    """Return a call() that raises exc regardless of how it is invoked."""

    def _call(*_a, **_k):
        raise exc

    return _call


def reg(**kw):
    """Build a registry whose retry sleep is a no-op so tests don't actually wait."""
    kw.setdefault('sleep', lambda _d: None)
    return GuardRegistry(**kw)


# ----------------------------------------------------- adapters: transport normalization (#1)


def test_httpx_timeout_adapter_normalizes_httpx_read_timeout():
    """An httpx ReadTimeout becomes a DownstreamTransportError tagged with the resource."""
    with pytest.raises(DownstreamTransportError) as ei:
        httpx_timeout_adapter('acct', Policy(), raising(httpx.ReadTimeout('slow')))
    assert ei.value.resource == 'acct'


def test_httpx_timeout_adapter_normalizes_httpx_connect_error():
    """An httpx ConnectError becomes a DownstreamTransportError."""
    with pytest.raises(DownstreamTransportError):
        httpx_timeout_adapter('acct', Policy(), raising(httpx.ConnectError('refused')))


def test_requests_timeout_adapter_normalizes_requests_timeout():
    """A requests Timeout becomes a DownstreamTransportError."""
    with pytest.raises(DownstreamTransportError):
        requests_timeout_adapter(
            'mwaa', Policy(), raising(requests.exceptions.Timeout())
        )


def test_requests_timeout_adapter_normalizes_requests_connection_error():
    """A requests ConnectionError becomes a DownstreamTransportError."""
    with pytest.raises(DownstreamTransportError):
        requests_timeout_adapter(
            'mwaa', Policy(), raising(requests.exceptions.ConnectionError())
        )


# ----------------------------------------------------- adapters: status handling + timeout shape


def test_adapters_raise_on_5xx_and_pass_4xx_and_2xx():
    """5xx -> DownstreamServerError; 4xx/2xx returned untouched."""
    with pytest.raises(DownstreamServerError):
        httpx_timeout_adapter('acct', Policy(), lambda **t: Resp(503))
    assert (
        httpx_timeout_adapter('acct', Policy(), lambda **t: Resp(404)).status_code
        == 404
    )
    assert (
        requests_timeout_adapter('mwaa', Policy(), lambda **t: Resp(200)).status_code
        == 200
    )


def test_httpx_timeout_adapter_passes_httpx_timeout_shape():
    """The httpx adapter injects an httpx.Timeout built from the policy."""
    seen = {}
    httpx_timeout_adapter(
        'acct',
        Policy(connect_timeout=3.05, read_timeout=30),
        lambda **t: seen.update(t) or Resp(200),
    )
    timeout = seen['timeout']
    assert isinstance(timeout, httpx.Timeout)
    assert timeout.connect == 3.05
    assert timeout.read == timeout.write == timeout.pool == 30


def test_requests_timeout_adapter_passes_connect_read_tuple():
    """The requests adapter injects a (connect, read) tuple from the policy."""
    seen = {}
    requests_timeout_adapter(
        'mwaa',
        Policy(connect_timeout=3.05, read_timeout=10),
        lambda **t: seen.update(t) or Resp(200),
    )
    assert seen['timeout'] == (3.05, 10)


def test_no_timeout_adapter_calls_without_timeout_and_raises_on_5xx():
    """The owsrequest adapter calls with no args (no per-call timeout) and still raises on 5xx."""
    seen = {'called': False}

    def call():
        seen['called'] = True
        return Resp(200)

    assert no_timeout_adapter('collab', Policy(), call).status_code == 200
    assert seen['called'] is True
    with pytest.raises(DownstreamServerError):
        no_timeout_adapter('collab', Policy(), lambda: Resp(500))


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


def test_is_downstream_failure_counts_server_and_transport_only():
    """The default classifier counts the two normalized errors and nothing else."""
    assert is_downstream_failure(DownstreamServerError('r', 500)) is True
    assert is_downstream_failure(DownstreamTransportError('r')) is True
    assert is_downstream_failure(ValueError()) is False
    assert is_downstream_failure(KeyError()) is False


# ------------------------------------------------------------------------------ registry


def test_register_and_call_returns_value():
    """A registered passthrough downstream runs and returns its adapter's result."""
    r = reg()
    r.register('x', Policy(), lambda res, pol, call: call(), is_downstream_failure)
    assert r.call('x', lambda: 'ok') == 'ok'


def test_call_unregistered_resource_raises_keyerror():
    """Calling an unregistered name fails loudly."""
    with pytest.raises(KeyError):
        reg().call('nope', lambda: None)


def test_guards_returns_every_registered_guard():
    """guards() exposes all registered guards (used for test-state reset)."""
    r = reg()
    r.register('a', Policy(), lambda *a: None, is_downstream_failure)
    r.register('b', Policy(), lambda *a: None, is_downstream_failure)
    assert {g.name for g in r.guards()} == {'a', 'b'}


def test_transport_failures_trip_the_breaker():
    """Regression for the classifier bug: a timeout/connect failure now opens the breaker."""
    r = reg()
    r.register('acct', Policy(fail_max=2), httpx_timeout_adapter, is_downstream_failure)
    for _ in range(2):
        with pytest.raises(DownstreamTransportError):
            r.call('acct', raising(httpx.ConnectError('down')))
    assert r.guard('acct').breaker.state == 'open'
    with pytest.raises(CircuitBreakerError):
        r.call('acct', raising(httpx.ConnectError('down')))


def test_registry_forwards_on_state_change_to_every_breaker():
    """A callback given to the registry fires on a real trip, tagged with the breaker name."""
    seen = []
    r = reg(on_state_change=lambda name, old, new: seen.append((name, old, new)))
    r.register('acct', Policy(fail_max=1), httpx_timeout_adapter, is_downstream_failure)
    with pytest.raises(DownstreamTransportError):
        r.call('acct', raising(httpx.ConnectError('down')))
    assert ('acct', 'closed', 'open') in seen


# -------------------------------------------------------------------- retry (classifier-gated)


def _counting_adapter(box, exc=None):
    """Build an adapter that counts calls and optionally raises exc."""

    def _adapter(resource, policy, call):
        box['n'] += 1
        if exc is not None:
            raise exc
        return 'ok'

    return _adapter


def test_idempotent_retries_retryable_errors_up_to_policy_retries():
    """idempotent=True retries a retryable (classified) error retries+1 times."""
    box = {'n': 0}
    r = reg()
    r.register(
        'x',
        Policy(retries=2, fail_max=10),
        _counting_adapter(box, DownstreamTransportError('x')),
        is_downstream_failure,
    )
    with pytest.raises(DownstreamTransportError):
        r.call('x', lambda: None, idempotent=True)
    assert box['n'] == 3


def test_non_retryable_error_is_not_retried():
    """A classifier-negative error propagates immediately, even when idempotent."""
    box = {'n': 0}
    r = reg()
    r.register(
        'x',
        Policy(retries=5, fail_max=10),
        _counting_adapter(box, ValueError('bug')),
        is_downstream_failure,
    )
    with pytest.raises(ValueError):
        r.call('x', lambda: None, idempotent=True)
    assert box['n'] == 1  # not retried


def test_writes_are_not_retried_even_on_a_retryable_error():
    """idempotent=False runs exactly once even when the error would otherwise be retried."""
    box = {'n': 0}
    r = reg()
    r.register(
        'x',
        Policy(retries=5, fail_max=10),
        _counting_adapter(box, DownstreamTransportError('x')),
        is_downstream_failure,
    )
    with pytest.raises(DownstreamTransportError):
        r.call('x', lambda: None, idempotent=False)
    assert box['n'] == 1


def test_retry_backoff_is_exponential_with_full_jitter():
    """Backoff is rng() * base_delay * 2**i; with rng=1.0 it's the exponential ceiling."""
    delays = []
    box = {'n': 0}
    r = reg(sleep=delays.append, base_delay=0.1, rng=lambda: 1.0)
    r.register(
        'x',
        Policy(retries=2, fail_max=10),
        _counting_adapter(box, DownstreamTransportError('x')),
        is_downstream_failure,
    )
    with pytest.raises(DownstreamTransportError):
        r.call('x', lambda: None, idempotent=True)
    assert delays == [0.1, 0.2]  # 2 gaps between 3 attempts; ceiling at rng=1.0


def test_retry_backoff_scales_by_the_jitter_rng():
    """The jitter rng scales each delay (rng=0.5 halves the exponential ceiling)."""
    delays = []
    box = {'n': 0}
    r = reg(sleep=delays.append, base_delay=0.1, rng=lambda: 0.5)
    r.register(
        'x',
        Policy(retries=2, fail_max=10),
        _counting_adapter(box, DownstreamTransportError('x')),
        is_downstream_failure,
    )
    with pytest.raises(DownstreamTransportError):
        r.call('x', lambda: None, idempotent=True)
    assert delays == [0.05, 0.1]


def test_retry_classifier_that_raises_propagates_the_original_error_without_retrying():
    """A throwing retry classifier doesn't mask the downstream error and doesn't trigger a retry."""
    box = {'n': 0}

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

    r = reg()
    r.register(
        'x',
        Policy(retries=5, fail_max=10),
        _counting_adapter(box, ValueError('real')),
        boom_classifier,
    )
    with pytest.raises(
        ValueError, match='real'
    ):  # the ORIGINAL error, not RuntimeError
        r.call('x', lambda: None, idempotent=True)
    assert box['n'] == 1  # treated as non-retryable


def test_retry_returns_on_recovery():
    """A retryable error that later succeeds returns the success without raising."""
    box = {'n': 0}

    def flaky(resource, policy, call):
        box['n'] += 1
        if box['n'] < 3:
            raise DownstreamTransportError('x')
        return 'recovered'

    r = reg()
    r.register('x', Policy(retries=5, fail_max=10), flaky, is_downstream_failure)
    assert r.call('x', lambda: None, idempotent=True) == 'recovered'
    assert box['n'] == 3


# ------------------------------------------------------------------------------- helpers


def test_http_status_prefers_status_code_then_status_then_none():
    """_http_status reads status_code, falls back to status, else None."""

    class A:
        status_code = 201

    class B:
        status = 502

    class C:
        status_code = None
        status = 503

    assert _http_status(A()) == 201
    assert _http_status(B()) == 502
    assert _http_status(C()) == 503  # status_code=None falls through to status
    assert _http_status(object()) is None