2026-05-09 12:15:44 +00:00
|
|
|
from time import sleep
|
2026-05-09 20:17:23 +00:00
|
|
|
|
|
|
|
|
import pytest
|
|
|
|
|
|
|
|
|
|
from token_bucket import RateLimitExceeded, TokenBucketLimiter
|
|
|
|
|
|
2026-05-09 12:15:44 +00:00
|
|
|
|
|
|
|
|
def test_limit_respected() -> None:
|
|
|
|
|
limiter = TokenBucketLimiter(10, 10)
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
for i in range(30):
|
|
|
|
|
with limiter:
|
|
|
|
|
sleep(0.1)
|
|
|
|
|
except RateLimitExceeded:
|
|
|
|
|
assert False, "Rate limit exceeded!"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_limit_exceeded() -> None:
|
|
|
|
|
limiter = TokenBucketLimiter(10, 10)
|
|
|
|
|
|
|
|
|
|
with pytest.raises(RateLimitExceeded) as e:
|
|
|
|
|
for i in range(20):
|
|
|
|
|
with limiter:
|
|
|
|
|
continue
|
|
|
|
|
|
2026-05-09 20:17:23 +00:00
|
|
|
assert "Rate exceeded" in str(e.value)
|