84 lines
2.3 KiB
Python
84 lines
2.3 KiB
Python
import asyncio
|
|
import time
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
|
|
from lib.rate_limiter import RateLimiter
|
|
|
|
|
|
@pytest.fixture
|
|
def event_loop():
|
|
loop = asyncio.new_event_loop()
|
|
yield loop
|
|
loop.close()
|
|
|
|
|
|
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
|
|
assert elapsed < 0.1
|
|
|
|
asyncio.get_event_loop().run_until_complete(run())
|
|
|
|
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()
|
|
elapsed = time.monotonic() - start
|
|
|
|
assert elapsed >= 0.25
|
|
|
|
asyncio.get_event_loop().run_until_complete(run())
|
|
|
|
def test_consecutive_requests_space_correctly(self):
|
|
async def run():
|
|
interval = 0.2
|
|
limiter = RateLimiter(min_interval_seconds=interval)
|
|
|
|
times = []
|
|
for _ in range(5):
|
|
await limiter.acquire()
|
|
times.append(time.monotonic())
|
|
|
|
for i in range(1, len(times)):
|
|
gap = times[i] - times[i - 1]
|
|
assert gap >= interval * 0.8
|
|
|
|
asyncio.get_event_loop().run_until_complete(run())
|
|
|
|
def test_custom_interval(self):
|
|
async def run():
|
|
limiter = RateLimiter(min_interval_seconds=0.1)
|
|
await limiter.acquire()
|
|
|
|
start = time.monotonic()
|
|
await limiter.acquire()
|
|
elapsed = time.monotonic() - start
|
|
|
|
assert elapsed >= 0.05
|
|
|
|
asyncio.get_event_loop().run_until_complete(run())
|
|
|
|
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()
|
|
elapsed = time.monotonic() - start
|
|
|
|
assert elapsed < 0.1
|
|
|
|
asyncio.get_event_loop().run_until_complete(run()) |