brain-of-reese/tests/unit/test_rate_limit.py

"""Unit: the sliding-window failed sign-in limiter (phase 81, task 01).

Covers ``app.core.rate_limit`` — the per-client-IP sliding window that
both login routes share (the audit SEC-03 fix). Pins, in the task's
order:

* 9 failures → still allowed (``remaining_wait == 0``);
* the 10th failure → blocked (``remaining_wait > 0``, ``<= WINDOW_SECONDS``);
* independent IPs never share counters;
* window slide/expiry — a failure older than ``WINDOW_SECONDS`` stops
  counting (both pre-seeded module state and a fake monotonic clock);
* ``remaining_wait`` is the WHOLE-window wait until the OLDEST in-window
  failure expires (the ``Retry-After`` contract), not a per-failure age;
* ``reset`` (a success on either route) unblocks immediately;
* an unknown IP never blocks;
* FAIL-OPEN — a corrupted, non-deque state entry and a broken clock
  both degrade to "allow": ``remaining_wait`` returns 0 and
  ``record_failure`` drops the count without raising. A limiter bug
  must never lock the owner out.
"""
from __future__ import annotations

import time
from collections import deque
from collections.abc import Iterator

import pytest

from app.core import rate_limit as rl


@pytest.fixture(autouse=True)
def _clean_state() -> Iterator[None]:
    """The limiter is module-global: no counter may leak between tests."""
    rl._failures.clear()
    yield
    rl._failures.clear()


def test_nine_failures_still_allowed() -> None:
    """Below the threshold: 9 in-window failures → no wait (the 10th
    failure is the one that crosses the bar)."""
    for _ in range(rl.MAX_FAILURES - 1):
        rl.record_failure("ip")
    assert rl.remaining_wait("ip") == 0


def test_tenth_failure_blocks_until_the_window_clears() -> None:
    """The 10th recorded failure crosses the threshold: the routes'
    pre-check (the 11th attempt) sees a positive wait — bounded by the
    whole window and integral for the ``Retry-After`` header."""
    for _ in range(rl.MAX_FAILURES):
        rl.record_failure("ip")
    wait = rl.remaining_wait("ip")
    assert isinstance(wait, int)
    assert wait > 0
    assert wait <= rl.WINDOW_SECONDS


def test_ips_are_independent() -> None:
    """One IP's failures never count toward another IP's window."""
    for _ in range(rl.MAX_FAILURES):
        rl.record_failure("ip")
    assert rl.remaining_wait("ip") > 0
    assert rl.remaining_wait("ip2") == 0


def test_failures_older_than_the_window_stop_counting_preseeded() -> None:
    """Reach into the module state: timestamps ``WINDOW_SECONDS + 1`` in
    the past have slid out — 10 such 'failures' leave the IP unblocked."""
    now = time.monotonic()
    rl._failures["ip"] = deque([now - rl.WINDOW_SECONDS - 1.0] * rl.MAX_FAILURES)
    assert rl.remaining_wait("ip") == 0


def test_window_slides_and_wait_is_measured_from_the_oldest(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Fake monotonic clock, 10 failures one second apart (t = 1000..1009):

    * the wait is the WHOLE-window seconds until the OLDEST in-window
      failure expires — ``ceil(900 - 9) == 891`` at t = 1009, not a
      per-failure age;
    * the wait counts down with the oldest (300 at t = 1600, clamped to
      1 at t = 1899);
    * the block lifts the moment the oldest failure slides out (t =
      1900): 9 failures are then still in the window, yet the count is
      back below the threshold.
    """
    clock = {"t": 1000.0}
    monkeypatch.setattr(time, "monotonic", lambda: clock["t"])
    for i in range(rl.MAX_FAILURES):
        clock["t"] = 1000.0 + i  # failures at 1000, 1001, ..., 1009
        rl.record_failure("ip")

    assert rl.remaining_wait("ip") == 891  # ceil(900 - 9)
    clock["t"] = 1600.0
    assert rl.remaining_wait("ip") == 300  # oldest is 600 s old
    clock["t"] = 1899.0
    assert rl.remaining_wait("ip") == 1  # ceil(900 - 899), clamped >= 1
    clock["t"] = 1900.0
    assert rl.remaining_wait("ip") == 0  # oldest just expired → 9 left


def test_reset_unblocks_immediately() -> None:
    """A success on either route clears the IP's counter: the owner who
    fat-fingered is not poisoned by their own earlier misses."""
    for _ in range(rl.MAX_FAILURES):
        rl.record_failure("ip")
    assert rl.remaining_wait("ip") > 0
    rl.reset("ip")
    assert rl.remaining_wait("ip") == 0


def test_unknown_ip_never_blocks() -> None:
    """No failures recorded → no wait (the routes' pre-check fast path)."""
    assert rl.remaining_wait("never-seen") == 0


def test_fail_open_on_corrupt_state_entry() -> None:
    """A non-deque entry in the module state (corruption):
    ``remaining_wait`` must return 0 (never block) and ``record_failure``
    must drop the count WITHOUT raising (fail-open — a limiter bug never
    locks the owner out). The IP stays unblocked afterwards."""
    rl._failures["bad"] = object()  # type: ignore[assignment]
    assert rl.remaining_wait("bad") == 0
    rl.record_failure("bad")  # must not raise
    assert rl.remaining_wait("bad") == 0


def test_fail_open_on_broken_clock(monkeypatch: pytest.MonkeyPatch) -> None:
    """A clock anomaly (monotonic raising) is equally fail-open: 0 wait
    and a silent drop — the login route itself must never 500 because
    of the limiter."""

    def broken_clock() -> float:
        raise ValueError("clock skew")

    monkeypatch.setattr(time, "monotonic", broken_clock)
    assert rl.remaining_wait("any") == 0
    rl.record_failure("any")  # must not raise
    assert rl.remaining_wait("any") == 0


def test_reset_unknown_ip_is_a_noop() -> None:
    """No entry to drop: ``reset`` must not raise (the routes call it on
    every success unconditionally)."""
    rl.reset("ghost")
    assert rl.remaining_wait("ghost") == 0