"""Lua sliding-window counters (fakeredis + lupa): remaining, reset, window rollover, force, peek, fail-open."""
from __future__ import annotations

import pytest


@pytest.fixture
def rl(app):
    from ratelimit import redis_limiter
    redis_limiter.reset_for_tests()
    return redis_limiter


@pytest.fixture
def tiers(app):
    from ratelimit import tiers
    return tiers


T0 = 1_800_000_000_000  # ms


def test_requests_counter_and_reset(rl, tiers):
    keyless = tiers.TIERS["keyless"]
    d = rl.apply_tier("ip:test", keyless, req_cost=1, at_ms=T0)
    assert d.allowed and d.applied
    assert d.remaining_requests == 29 and d.limit_requests == 30
    assert d.remaining_rows == 100_000
    assert d.reset_requests == T0 // 1000 + 3600           # oldest bucket + window
    for _ in range(29):
        d = rl.apply_tier("ip:test", keyless, req_cost=1, at_ms=T0 + 5_000)
    assert d.allowed and d.remaining_requests == 0
    d = rl.apply_tier("ip:test", keyless, req_cost=1, at_ms=T0 + 6_000)
    assert not d.allowed and not d.allowed_requests and d.allowed_rows and not d.applied
    assert d.remaining_requests == 0
    assert d.reset_requests == T0 // 1000 + 3600


def test_window_rollover_frees_oldest_bucket(rl, tiers):
    free = tiers.TIERS["free"]   # 60 s window, 120 requests
    for i in range(120):
        d = rl.apply_tier("key:1", free, req_cost=1, at_ms=T0 + i * 100)   # all within 12 s
    assert d.remaining_requests == 0
    assert not rl.apply_tier("key:1", free, req_cost=1, at_ms=T0 + 30_000).allowed
    # 60 s after the first bucket the first second (10 requests) has rolled out of the window
    d = rl.apply_tier("key:1", free, req_cost=0, at_ms=T0 + 60_000)
    assert d.remaining_requests == 10
    d = rl.apply_tier("key:1", free, req_cost=1, at_ms=T0 + 60_000)
    assert d.allowed and d.remaining_requests == 9
    # far in the future: everything expired, hash pruned
    d = rl.apply_tier("key:1", free, req_cost=0, at_ms=T0 + 200_000)
    assert d.remaining_requests == 120
    assert rl.client().hlen("rl:key:1:req") == 0


def test_rows_counter_precheck_and_force(rl, tiers):
    keyless = tiers.TIERS["keyless"]
    # charge rows after the fact, even when it overshoots
    d = rl.apply_tier("ip:r", keyless, req_cost=0, rows_cost=99_990, force=True, at_ms=T0)
    assert d.applied and d.remaining_rows == 10
    d = rl.apply_tier("ip:r", keyless, req_cost=0, rows_cost=50, force=True, at_ms=T0 + 1000)
    assert d.applied and not d.allowed_rows and d.remaining_rows == 0
    # now the pre-check refuses on rows while requests are still available
    d = rl.apply_tier("ip:r", keyless, req_cost=1, at_ms=T0 + 2000)
    assert d.allowed_requests and not d.allowed_rows and not d.applied
    assert d.reset_rows == T0 // 1000 + 3600


def test_peek_does_not_write(rl, tiers):
    free = tiers.TIERS["free"]
    d = rl.peek_tier("key:9", free, at_ms=T0)
    assert d.remaining_requests == 120 and d.remaining_rows == 1_000_000
    assert rl.client().exists("rl:key:9:req") == 0


def test_two_principals_are_independent(rl, tiers):
    keyless = tiers.TIERS["keyless"]
    rl.apply_tier("ip:a", keyless, at_ms=T0)
    assert rl.peek_tier("ip:b", keyless, at_ms=T0).remaining_requests == 30
    assert rl.peek_tier("ip:a", keyless, at_ms=T0).remaining_requests == 29


def test_fail_open_when_redis_raises(rl, tiers, monkeypatch, caplog):
    import redis as redis_lib

    class Broken:
        def evalsha(self, *a, **k):
            raise redis_lib.exceptions.ConnectionError("down")

        def script_load(self, *a):
            raise redis_lib.exceptions.ConnectionError("down")

    monkeypatch.setattr(rl, "_client", Broken())
    monkeypatch.setattr(rl, "_sha", None)
    monkeypatch.setattr(rl, "_down_until", 0.0)
    monkeypatch.setattr(rl, "_last_warn", 0.0)
    with caplog.at_level("WARNING"):
        assert rl.apply_tier("ip:x", tiers.TIERS["keyless"], at_ms=T0) is None
    assert "fails open" in caplog.text
    assert not rl.available()          # circuit open for a few seconds


def test_tier_table_is_exact(tiers):
    t = tiers.TIERS
    assert (t["keyless"].window_s, t["keyless"].requests, t["keyless"].rows, t["keyless"].max_rows_per_request) == (3600, 30, 100_000, 5_000)
    assert (t["free"].window_s, t["free"].requests, t["free"].rows, t["free"].max_rows_per_request) == (60, 120, 1_000_000, 50_000)
    assert (t["high_usage"].window_s, t["high_usage"].requests, t["high_usage"].rows, t["high_usage"].max_rows_per_request) == (60, 600, 10_000_000, 200_000)
    assert t["keyless"].requests_type == "requests_per_hour" and t["free"].rows_type == "rows_per_minute"
    assert tiers.row_cost(101, status=200, media_type="application/vnd.apache.parquet", quota_exempt=False) == 51
    assert tiers.row_cost(500, status=304, media_type="application/json", quota_exempt=False) == 0
    assert tiers.row_cost(500, status=200, media_type="application/json", quota_exempt=True) == 0
    assert tiers.row_cost(500, status=404, media_type="application/json", quota_exempt=False) == 0