108 lines
3.2 KiB
Python
108 lines
3.2 KiB
Python
import threading
|
|
import unittest
|
|
|
|
from rate_limiter import RateLimiter
|
|
|
|
|
|
class TestRateLimiter(unittest.TestCase):
|
|
def test_basic_admit_and_reject(self):
|
|
t = [0.0]
|
|
rl = RateLimiter(limit=2, window=10.0, clock=lambda: t[0])
|
|
self.assertTrue(rl.allow())
|
|
self.assertTrue(rl.allow())
|
|
self.assertFalse(rl.allow())
|
|
|
|
def test_window_expiry(self):
|
|
t = [0.0]
|
|
rl = RateLimiter(limit=1, window=5.0, clock=lambda: t[0])
|
|
self.assertTrue(rl.allow())
|
|
t[0] = 4.999
|
|
self.assertFalse(rl.allow())
|
|
t[0] = 5.0
|
|
# exactly window old -> expired
|
|
self.assertTrue(rl.allow())
|
|
|
|
def test_boundary_half_open(self):
|
|
t = [0.0]
|
|
rl = RateLimiter(limit=2, window=1.0, clock=lambda: t[0])
|
|
self.assertTrue(rl.allow()) # ts 0
|
|
t[0] = 0.5
|
|
self.assertTrue(rl.allow()) # ts 0.5
|
|
t[0] = 1.0
|
|
# ts 0 expired (1.0-0>=1), ts 0.5 still live (1.0-0.5=0.5<1)
|
|
self.assertTrue(rl.allow()) # admits, now 2 live? ts0.5 + new
|
|
# window now has 0.5 and 1.0 -> full
|
|
self.assertFalse(rl.allow())
|
|
|
|
def test_limit_zero_rejects(self):
|
|
rl = RateLimiter(limit=0, window=10.0, clock=lambda: 0.0)
|
|
self.assertFalse(rl.allow())
|
|
|
|
def test_window_zero_rejects(self):
|
|
rl = RateLimiter(limit=5, window=0.0, clock=lambda: 0.0)
|
|
self.assertFalse(rl.allow())
|
|
|
|
def test_negative_args(self):
|
|
with self.assertRaises(ValueError):
|
|
RateLimiter(limit=-1, window=1.0)
|
|
with self.assertRaises(ValueError):
|
|
RateLimiter(limit=1, window=-1.0)
|
|
|
|
def test_reset(self):
|
|
t = [0.0]
|
|
rl = RateLimiter(limit=1, window=10.0, clock=lambda: t[0])
|
|
self.assertTrue(rl.allow())
|
|
self.assertFalse(rl.allow())
|
|
rl.reset()
|
|
self.assertTrue(rl.allow())
|
|
|
|
def test_count(self):
|
|
t = [0.0]
|
|
rl = RateLimiter(limit=5, window=10.0, clock=lambda: t[0])
|
|
rl.allow(); rl.allow()
|
|
self.assertEqual(rl.count(), 2)
|
|
t[0] = 20.0
|
|
self.assertEqual(rl.count(), 0)
|
|
|
|
def test_thread_safety(self):
|
|
t = [0.0]
|
|
rl = RateLimiter(limit=1000, window=10.0, clock=lambda: t[0])
|
|
admitted = [0]
|
|
lock = threading.Lock()
|
|
|
|
def worker():
|
|
for _ in range(500):
|
|
if rl.allow():
|
|
with lock:
|
|
admitted[0] += 1
|
|
|
|
threads = [threading.Thread(target=worker) for _ in range(4)]
|
|
for th in threads:
|
|
th.start()
|
|
for th in threads:
|
|
th.join()
|
|
self.assertEqual(admitted[0], 1000)
|
|
|
|
def test_concurrent_never_exceeds_limit(self):
|
|
t = [0.0]
|
|
rl = RateLimiter(limit=50, window=10.0, clock=lambda: t[0])
|
|
admitted = [0]
|
|
lock = threading.Lock()
|
|
|
|
def worker():
|
|
for _ in range(200):
|
|
if rl.allow():
|
|
with lock:
|
|
admitted[0] += 1
|
|
|
|
threads = [threading.Thread(target=worker) for _ in range(8)]
|
|
for th in threads:
|
|
th.start()
|
|
for th in threads:
|
|
th.join()
|
|
self.assertLessEqual(admitted[0], 50)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|