b3f2a8b091
- Unit tests for RateLimiter: token refill, per-user isolation, reset - HTTP tests: 429 response, X-RateLimit headers, per-user enforcement - Uses rate_limited_client fixture with limit=3 for easy testing
80 lines
2.4 KiB
Python
80 lines
2.4 KiB
Python
"""Tests for RateLimiter — per-user token-bucket rate limiting."""
|
|
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from nexus.multi_user_bridge import RateLimiter
|
|
|
|
|
|
class TestRateLimiter:
|
|
|
|
def test_allows_within_limit(self):
|
|
rl = RateLimiter(max_tokens=5, window_seconds=1.0)
|
|
for i in range(5):
|
|
assert rl.check("user1") is True
|
|
|
|
def test_blocks_after_limit(self):
|
|
rl = RateLimiter(max_tokens=3, window_seconds=1.0)
|
|
rl.check("user1")
|
|
rl.check("user1")
|
|
rl.check("user1")
|
|
assert rl.check("user1") is False
|
|
|
|
def test_per_user_isolation(self):
|
|
rl = RateLimiter(max_tokens=2, window_seconds=1.0)
|
|
rl.check("alice")
|
|
rl.check("alice")
|
|
assert rl.check("alice") is False # exhausted
|
|
assert rl.check("bob") is True # independent bucket
|
|
|
|
def test_remaining_count(self):
|
|
rl = RateLimiter(max_tokens=10, window_seconds=60.0)
|
|
assert rl.remaining("user1") == 10
|
|
rl.check("user1")
|
|
assert rl.remaining("user1") == 9
|
|
rl.check("user1")
|
|
rl.check("user1")
|
|
assert rl.remaining("user1") == 7
|
|
|
|
def test_token_refill_over_time(self):
|
|
rl = RateLimiter(max_tokens=10, window_seconds=1.0)
|
|
# Exhaust all tokens
|
|
for _ in range(10):
|
|
rl.check("user1")
|
|
assert rl.check("user1") is False
|
|
|
|
# Wait for tokens to refill (1 window = 10 tokens in 1 second)
|
|
time.sleep(1.1)
|
|
|
|
# Should have tokens again
|
|
assert rl.check("user1") is True
|
|
|
|
def test_reset_clears_bucket(self):
|
|
rl = RateLimiter(max_tokens=5, window_seconds=60.0)
|
|
for _ in range(5):
|
|
rl.check("user1")
|
|
assert rl.check("user1") is False
|
|
|
|
rl.reset("user1")
|
|
assert rl.check("user1") is True
|
|
assert rl.remaining("user1") == 4
|
|
|
|
def test_separate_limits_per_user(self):
|
|
rl = RateLimiter(max_tokens=1, window_seconds=60.0)
|
|
assert rl.check("a") is True
|
|
assert rl.check("a") is False
|
|
assert rl.check("b") is True
|
|
assert rl.check("c") is True
|
|
assert rl.check("b") is False
|
|
assert rl.check("c") is False
|
|
|
|
def test_default_config(self):
|
|
rl = RateLimiter()
|
|
assert rl._max_tokens == 60
|
|
assert rl._window == 60.0
|
|
|
|
def test_unknown_user_gets_full_bucket(self):
|
|
rl = RateLimiter(max_tokens=5, window_seconds=60.0)
|
|
assert rl.remaining("new_user") == 5
|