"""Tests for the resilient rate-limit store: Redis primary with in-memory fallback.

Covers the plain memory:// passthrough, the NEVER-500 fallback path on primary errors, the
per-request degradation latch, breaker tripping, boot-time construction failures, and
concurrency safety.
"""

import threading

from limits import parse
from limits.storage import MemoryStorage
from limits.strategies import STRATEGIES

from core.hardening import rate_limit_storage as rls
from core.hardening.rate_limit_storage import ResilientRateLimiter, build_rate_limiter
from core.hardening.rate_policy import RATE_STRATEGY

ITEM = parse('5/second')


class _FakeStorage(MemoryStorage):
    """A real limits MemoryStorage.

    isinstance(_, Storage) holds -- the strategy asserts it -- but its sliding-window ops raise
    on demand, and it counts primary calls.
    """

    def __init__(self):
        """Start un-failing, with the primary-call counter at zero."""
        super().__init__()
        self.fail_with = None
        self.calls = 0

    def _guard(self):
        """Count this call and raise fail_with if one has been armed."""
        self.calls += 1
        if self.fail_with is not None:
            raise self.fail_with

    def acquire_sliding_window_entry(self, *a, **k):
        """Guard, then delegate to the real MemoryStorage implementation."""
        self._guard()
        return super().acquire_sliding_window_entry(*a, **k)

    def get_sliding_window(self, *a, **k):
        """Guard, then delegate to the real MemoryStorage implementation."""
        self._guard()
        return super().get_sliding_window(*a, **k)


def _resilient(fake, on_change=lambda *a: None):
    """Build a ResilientRateLimiter over fake (primary) and a fresh MemoryStorage (fallback).

    Returns the limiter and the breaker's manual clock dict (key 't') for time control.
    """
    from core.hardening.breaker import CircuitBreaker

    primary = STRATEGIES[RATE_STRATEGY](fake)
    fallback = STRATEGIES[RATE_STRATEGY](MemoryStorage())
    clock = {'t': 0.0}
    breaker = CircuitBreaker(
        name='ratelimit-redis',
        fail_max=2,
        reset_timeout=10.0,
        success_threshold=1,
        count_failure=rls._is_redis_failure,
        on_state_change=on_change,
        clock=lambda: clock['t'],
    )
    return ResilientRateLimiter(primary, fallback, breaker), clock


def test_memory_uri_returns_plain_factory_unchanged():
    """memory:// builds a plain factory (not ResilientRateLimiter) whose limiter still hits."""
    factory = build_rate_limiter('memory://')
    assert not isinstance(factory, ResilientRateLimiter)
    limiter = factory.for_request()
    assert limiter.hit(ITEM, 'k') is True  # behaves like the plain strategy


def test_redis_uri_returns_resilient():
    """A redis*/valkey* URI builds a ResilientRateLimiter."""
    assert isinstance(build_rate_limiter('rediss://127.0.0.1:1'), ResilientRateLimiter)


def test_no_flask_context_needed_and_never_raises():
    """hit() works with no app/request context pushed, served by the fallback without raising."""
    fake = _FakeStorage()
    fake.fail_with = OSError('down')
    factory, _ = _resilient(fake)
    # No app/request context pushed at all:
    assert factory.for_request().hit(ITEM, 'k') is True  # served by fallback, no raise


def test_redis_error_falls_back_and_counts(monkeypatch):
    """A Redis/OSError on the primary falls back and counts a 'redis_error' fallback."""
    calls = []
    monkeypatch.setattr(
        rls.observability, 'ratelimit_fallback', lambda r: calls.append(r)
    )
    fake = _FakeStorage()
    fake.fail_with = OSError('down')
    factory, _ = _resilient(fake)
    assert factory.for_request().hit(ITEM, 'k') is True
    assert calls == ['redis_error']


def test_non_redis_error_falls_back_counts_and_logs(monkeypatch):
    """A non-Redis primary error still falls back (never-500) and counts 'other_error'."""
    calls = []
    monkeypatch.setattr(
        rls.observability, 'ratelimit_fallback', lambda r: calls.append(r)
    )
    fake = _FakeStorage()
    fake.fail_with = ValueError('bug')
    factory, _ = _resilient(fake)
    assert factory.for_request().hit(ITEM, 'k') is True  # never 500
    assert calls == ['other_error']


def test_per_request_latch_skips_primary_after_first_failure():
    """Once a request degrades it stays on the fallback even if the primary recovers.

    A fresh request is not latched and retries the (now-healthy) primary.
    """
    fake = _FakeStorage()
    fake.fail_with = OSError('down')
    factory, _ = _resilient(fake)
    req = factory.for_request()
    assert req.hit(ITEM, 'k') is True  # first call fails -> latch, served by fallback
    assert req._degraded is True
    fake.fail_with = None  # redis "recovers"
    fake.calls = 0
    assert (
        req.hit(ITEM, 'k') is True
    )  # SAME request must stay on fallback, NOT touch primary
    assert fake.calls == 0  # proves the latch: primary was not called again
    # a fresh request is not latched and tries the (now-healthy) primary again:
    assert factory.for_request().hit(ITEM, 'k') is True
    assert fake.calls == 1


def test_breaker_opens_after_fail_max_requests():
    """The breaker opens exactly once after fail_max failing requests, firing one callback."""
    changes = []
    fake = _FakeStorage()
    fake.fail_with = OSError('down')
    factory, _ = _resilient(fake, on_change=lambda n, o, w: changes.append(w))
    for _ in range(3):  # fail_max=2 -> opens (each request = 1 breaker failure)
        factory.for_request().hit(ITEM, 'k')
    assert factory._breaker.state == 'open'
    assert changes == ['open']  # opened exactly once


def test_boot_with_unreachable_redis_does_not_raise():
    """Construction against an unreachable Redis does not raise, and the first hit() degrades."""
    # 127.0.0.1:1 -> no DNS, connection refused fast (bounded by socket_connect_timeout).
    factory = build_rate_limiter('rediss://127.0.0.1:1')  # construction must not raise
    assert factory.for_request().hit(ITEM, 'k') is True  # first call degrades cleanly


def test_both_backends_failing_fails_open_never_raises(monkeypatch):
    """When both primary and fallback raise, hit() still fails open and never raises."""
    # Primary AND fallback both raise -> absolute never-500: hit() returns True, no exception.
    calls = []
    monkeypatch.setattr(
        rls.observability, 'ratelimit_fallback', lambda r: calls.append(r)
    )
    fake = _FakeStorage()
    fake.fail_with = OSError('down')
    factory, _ = _resilient(fake)
    # Make the fallback raise too:
    monkeypatch.setattr(
        factory._fallback,
        'hit',
        lambda *a, **k: (_ for _ in ()).throw(RuntimeError('fallback bug')),
    )
    assert factory.for_request().hit(ITEM, 'k') is True  # failed open
    assert 'fallback_error' in calls


def test_boot_construction_error_falls_back_to_memory(monkeypatch):
    """A construction-time error degrades to a plain memory factory and counts 'boot_error'."""
    calls = []
    monkeypatch.setattr(
        rls.observability, 'ratelimit_fallback', lambda r: calls.append(r)
    )
    # Force construction to raise a non-Redis error:
    monkeypatch.setattr(
        rls,
        'storage_from_string',
        lambda *a, **k: (_ for _ in ()).throw(ValueError('bad')),
    )
    factory = build_rate_limiter('rediss://x:6379')
    assert not isinstance(factory, ResilientRateLimiter)  # fell back to memory factory
    assert calls == ['boot_error']


def test_concurrent_hammering_never_raises_and_stays_consistent():
    """Many threads hammering a failing primary never raise, and the breaker ends up open."""
    fake = _FakeStorage()
    fake.fail_with = OSError('down')
    factory, _ = _resilient(fake)
    errors = []

    def worker():
        try:
            for _ in range(50):
                factory.for_request().hit(ITEM, 'k')
        except Exception as e:  # noqa: BLE001 -- test asserts nothing escapes
            errors.append(e)

    threads = [threading.Thread(target=worker) for _ in range(15)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    assert errors == []
    assert factory._breaker.state == 'open'