"""Regression: the MCP fixed-window rate limiters must not slide.

Both ``_check_pat_rate_limit`` and ``_check_auth_failure_rate_limit``
implement a *fixed* window — the TTL is anchored to the first request of
the window and the counter decays after ``window`` seconds.

A previous revision moved the limiter onto an atomic ``INCR`` + ``EXPIRE``
pipeline (to close a crash-between-INCR-and-EXPIRE race) but called
``EXPIRE`` *unconditionally on every request*. That turned the fixed
window into an ever-sliding one: under steady traffic the TTL was reset
to the full window on every call, so a PAT/IP that crossed the cap once
stayed blocked far past the intended window — effectively until traffic
stopped. These tests pin the fixed-window contract so the regression
cannot return.
"""
from __future__ import annotations

from app.core.redis_client import get_redis
from app.mcp import middleware


async def test_pat_rate_limit_window_is_fixed_not_sliding(monkeypatch) -> None:
    monkeypatch.setattr(middleware.settings, "MCP_RATE_LIMIT_PER_PAT_MAX", 100)
    monkeypatch.setattr(
        middleware.settings, "MCP_RATE_LIMIT_PER_PAT_WINDOW_SECONDS", 60
    )
    pat_id = "test-pat-window"
    key = f"mcp_ratelimit:pat:{pat_id}"
    redis = get_redis()
    await redis.delete(key)

    allowed, _ = await middleware._check_pat_rate_limit(pat_id)
    assert allowed is True

    # Simulate "near the end of the window": shrink the TTL to 500ms. A
    # correct fixed-window limiter leaves this alone on the next call
    # (SET ... NX is a no-op once the key exists); the sliding-window bug
    # would reset it back to the full 60s window.
    await redis.pexpire(key, 500)
    allowed, _ = await middleware._check_pat_rate_limit(pat_id)
    assert allowed is True

    ttl_ms = await redis.pttl(key)
    assert 0 < ttl_ms <= 1000, (
        f"window TTL was extended to {ttl_ms}ms — the per-PAT limiter is "
        f"sliding the window instead of holding it fixed"
    )
    await redis.delete(key)


async def test_pat_rate_limit_blocks_after_cap(monkeypatch) -> None:
    monkeypatch.setattr(middleware.settings, "MCP_RATE_LIMIT_PER_PAT_MAX", 3)
    monkeypatch.setattr(
        middleware.settings, "MCP_RATE_LIMIT_PER_PAT_WINDOW_SECONDS", 60
    )
    pat_id = "test-pat-cap"
    key = f"mcp_ratelimit:pat:{pat_id}"
    redis = get_redis()
    await redis.delete(key)

    for _ in range(3):
        allowed, retry_after = await middleware._check_pat_rate_limit(pat_id)
        assert allowed is True
        assert retry_after == 0

    allowed, retry_after = await middleware._check_pat_rate_limit(pat_id)
    assert allowed is False
    assert retry_after > 0
    await redis.delete(key)


async def test_auth_failure_window_is_fixed_not_sliding(monkeypatch) -> None:
    monkeypatch.setattr(middleware.settings, "MCP_AUTH_FAILURE_LIMIT_PER_IP_MAX", 100)
    monkeypatch.setattr(
        middleware.settings, "MCP_AUTH_FAILURE_LIMIT_PER_IP_WINDOW_SECONDS", 60
    )
    client_ip = "203.0.113.7"
    key = f"mcp_authfail:{client_ip}"
    redis = get_redis()
    await redis.delete(key)

    allowed, _ = await middleware._check_auth_failure_rate_limit(client_ip)
    assert allowed is True

    await redis.pexpire(key, 500)
    allowed, _ = await middleware._check_auth_failure_rate_limit(client_ip)
    assert allowed is True

    ttl_ms = await redis.pttl(key)
    assert 0 < ttl_ms <= 1000, (
        f"window TTL was extended to {ttl_ms}ms — the auth-failure limiter "
        f"is sliding the window instead of holding it fixed"
    )
    await redis.delete(key)


async def test_token_failure_peek_blocks_after_cap(monkeypatch) -> None:
    """Launch-audit regression (2026-06-10): the pre-DB peek is keyed by
    TOKEN hash, not IP — a stuck client retrying one dead PAT gets
    short-circuited before any DB work, while valid tokens (which never
    accumulate failures) are never blocked, so one colleague's retry
    loop cannot 429 the whole office NAT.
    """
    monkeypatch.setattr(
        middleware.settings, "MCP_AUTH_FAILURE_LIMIT_PER_IP_MAX", 3
    )
    monkeypatch.setattr(
        middleware.settings, "MCP_AUTH_FAILURE_LIMIT_PER_IP_WINDOW_SECONDS", 60
    )
    token = "helix_pat_unit_test_dead_token_aaaaaaaaaaaa"
    key = middleware._token_failure_key(token)
    redis = get_redis()
    await redis.delete(key)

    # Below the cap: peek allows.
    for _ in range(3):
        await middleware._check_token_failure_rate_limit(token)
    allowed, _ = await middleware._peek_token_failure_rate_limit(token)
    assert allowed is True

    # Crossing the cap: peek blocks with a sane retry_after.
    await middleware._check_token_failure_rate_limit(token)
    allowed, retry_after = await middleware._peek_token_failure_rate_limit(token)
    assert allowed is False
    assert 0 < retry_after <= 60

    # A DIFFERENT token from the "same client" is unaffected — the
    # gate is per-token, so valid users behind the same IP never 429.
    other = "helix_pat_unit_test_other_token_bbbbbbbbbbbb"
    await redis.delete(middleware._token_failure_key(other))
    allowed, _ = await middleware._peek_token_failure_rate_limit(other)
    assert allowed is True

    # The raw token must never appear in the Redis keyspace.
    assert token not in key
    await redis.delete(key)