60 lines
2.0 KiB
Python
60 lines
2.0 KiB
Python
"""限流器测试."""
|
|
|
|
|
|
from src.ai.rate_limiter import RateLimiter, RateLimitResult
|
|
|
|
|
|
class TestRateLimiterDegraded:
|
|
"""Redis 不可用时的降级行为（全并行模式）."""
|
|
|
|
async def test_no_redis_allows_all(self) -> None:
|
|
"""Redis=None 时降级放行."""
|
|
limiter = RateLimiter(redis=None)
|
|
results = await limiter.check(user_id="u-1", ip="1.2.3.4", school_id="s-1")
|
|
assert len(results) == 3
|
|
assert all(r.allowed for r in results)
|
|
|
|
async def test_no_redis_no_identifiers(self) -> None:
|
|
"""无任何标识符时返回空列表."""
|
|
limiter = RateLimiter(redis=None)
|
|
results = await limiter.check()
|
|
assert results == []
|
|
|
|
async def test_no_redis_only_user(self) -> None:
|
|
"""仅 user_id."""
|
|
limiter = RateLimiter(redis=None)
|
|
results = await limiter.check(user_id="u-1")
|
|
assert len(results) == 1
|
|
assert results[0].dimension == "user"
|
|
assert results[0].allowed is True
|
|
assert results[0].remaining == 10 # default user_limit
|
|
|
|
async def test_no_redis_only_ip(self) -> None:
|
|
"""仅 ip."""
|
|
limiter = RateLimiter(redis=None)
|
|
results = await limiter.check(ip="1.2.3.4")
|
|
assert len(results) == 1
|
|
assert results[0].dimension == "ip"
|
|
assert results[0].limit == 30
|
|
|
|
async def test_no_redis_only_school(self) -> None:
|
|
"""仅 school_id."""
|
|
limiter = RateLimiter(redis=None)
|
|
results = await limiter.check(school_id="s-1")
|
|
assert len(results) == 1
|
|
assert results[0].dimension == "school"
|
|
assert results[0].limit == 100
|
|
|
|
|
|
class TestRateLimitResult:
|
|
"""RateLimitResult dataclass 测试."""
|
|
|
|
def test_default_values(self) -> None:
|
|
result = RateLimitResult(allowed=True, dimension="user", limit=10, remaining=5)
|
|
assert result.allowed is True
|
|
assert result.remaining == 5
|
|
|
|
def test_denied(self) -> None:
|
|
result = RateLimitResult(allowed=False, dimension="ip", limit=30, remaining=0)
|
|
assert result.allowed is False
|