Jarian Cottingham
645d8820f1
- Wrap RateLimitMiddleware in Middleware() - bare class broke all requests (500) - Clamp/validate num param - ?num=abc no longer returns 500 - Query length cap (500), result counts clamped to 1..20 - Wire lifespan into Starlette so shared httpx client closes on shutdown - Purge stale per-IP rate limit entries to bound memory growth - URL-encode queries in the Playwright scraper path - Lazy lib imports so the package works without optional playwright - Add pyproject.toml (activates ruff/pytest/bandit in CI), README, LICENSE - Remove committed __pycache__, add .gitignore/.dockerignore - Fix SearXNG healthcheck path (/health -> /healthz) and compose docs
70 lines
2.0 KiB
Python
70 lines
2.0 KiB
Python
import asyncio
|
|
import time
|
|
|
|
from lib.rate_limiter import RateLimiter
|
|
|
|
|
|
class TestRateLimiter:
|
|
"""Test rate limiter enforces minimum delays between requests."""
|
|
|
|
def test_allows_first_request_immediately(self):
|
|
async def run():
|
|
limiter = RateLimiter(min_interval_seconds=0.5)
|
|
start = time.monotonic()
|
|
await limiter.acquire()
|
|
elapsed = time.monotonic() - start
|
|
return elapsed
|
|
|
|
assert asyncio.run(run()) < 0.1
|
|
|
|
def test_enforces_minimum_interval(self):
|
|
async def run():
|
|
limiter = RateLimiter(min_interval_seconds=0.3)
|
|
await limiter.acquire()
|
|
|
|
start = time.monotonic()
|
|
await limiter.acquire()
|
|
return time.monotonic() - start
|
|
|
|
assert asyncio.run(run()) >= 0.25
|
|
|
|
def test_consecutive_requests_space_correctly(self):
|
|
interval = 0.2
|
|
|
|
async def run():
|
|
limiter = RateLimiter(min_interval_seconds=interval)
|
|
|
|
times = []
|
|
for _ in range(5):
|
|
await limiter.acquire()
|
|
times.append(time.monotonic())
|
|
return times
|
|
|
|
times = asyncio.run(run())
|
|
for i in range(1, len(times)):
|
|
gap = times[i] - times[i - 1]
|
|
assert gap >= interval * 0.8
|
|
|
|
def test_custom_interval(self):
|
|
async def run():
|
|
limiter = RateLimiter(min_interval_seconds=0.1)
|
|
await limiter.acquire()
|
|
|
|
start = time.monotonic()
|
|
await limiter.acquire()
|
|
return time.monotonic() - start
|
|
|
|
assert asyncio.run(run()) >= 0.05
|
|
|
|
def test_no_delay_after_long_pause(self):
|
|
async def run():
|
|
limiter = RateLimiter(min_interval_seconds=0.3)
|
|
await limiter.acquire()
|
|
await asyncio.sleep(0.5)
|
|
|
|
start = time.monotonic()
|
|
await limiter.acquire()
|
|
return time.monotonic() - start
|
|
|
|
assert asyncio.run(run()) < 0.1
|