Spaces:
Running
Running
File size: 2,948 Bytes
889e86b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 | """
Unit tests for utils/rate_limiter.py's pure counting/enforcement logic.
Scope: enforce_rate_limit() itself (fixed-window counting, per-tier limits,
429 on breach). The with_rate_limit() FastAPI dependency composition and the
Firestore violation-logging side effect are thin wrappers around this and are
not separately unit-tested (per docs/qa/test-strategy.md, the FastAPI/
Firestore integration belongs at the integration layer, not here).
"""
import time
import pytest
from fastapi import HTTPException
import utils.rate_limiter as rl
@pytest.fixture(autouse=True)
def reset_limiter_state(monkeypatch):
# Each test gets a clean set of counters and a violation logger that
# never touches Firestore (no app is initialized in this test process).
monkeypatch.setattr(rl, "_counters", rl.defaultdict(rl._Window))
monkeypatch.setattr(rl, "_violation_counts", rl.defaultdict(int))
monkeypatch.setattr(rl, "_log_violation", lambda *a, **k: None)
def test_allows_requests_within_the_limit():
for _ in range(8):
rl.enforce_rate_limit("uid-1", "generation") # 8/hour is the limit — must not raise
def test_blocks_the_request_that_exceeds_the_limit():
for _ in range(8):
rl.enforce_rate_limit("uid-1", "generation")
with pytest.raises(HTTPException) as exc_info:
rl.enforce_rate_limit("uid-1", "generation")
assert exc_info.value.status_code == 429
def test_counters_are_isolated_per_uid():
for _ in range(8):
rl.enforce_rate_limit("uid-1", "generation")
# A different uid must not be affected by uid-1 exhausting its limit.
rl.enforce_rate_limit("uid-2", "generation")
def test_counters_are_isolated_per_tier():
for _ in range(8):
rl.enforce_rate_limit("uid-1", "generation")
# The same uid hitting a different tier must not be blocked by "generation".
rl.enforce_rate_limit("uid-1", "chat")
def test_a_tier_with_multiple_windows_enforces_the_tighter_one_first():
# "chat" is 10/minute and 200/day -- the per-minute window binds first.
for _ in range(10):
rl.enforce_rate_limit("uid-1", "chat")
with pytest.raises(HTTPException) as exc_info:
rl.enforce_rate_limit("uid-1", "chat")
assert exc_info.value.status_code == 429
assert "minute" in exc_info.value.detail
def test_window_resets_after_its_period_elapses(monkeypatch):
fake_now = [1_000_000.0]
monkeypatch.setattr(rl.time, "time", lambda: fake_now[0])
for _ in range(8):
rl.enforce_rate_limit("uid-1", "generation")
with pytest.raises(HTTPException):
rl.enforce_rate_limit("uid-1", "generation")
# Jump past the 1-hour window.
fake_now[0] += 3600 + 1
rl.enforce_rate_limit("uid-1", "generation") # must not raise — window reset
def test_unknown_tier_raises_a_programmer_error_not_a_silent_pass():
with pytest.raises(ValueError):
rl.enforce_rate_limit("uid-1", "not-a-real-tier")
|