Spaces:
Running
Running
| """ | |
| 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 | |
| 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") | |