64 lines
1.5 KiB
Python
64 lines
1.5 KiB
Python
from app.ratelimit import MAX_FAILURES, WINDOW_SECONDS, FailureLimiter
|
|
|
|
|
|
class Clock:
|
|
def __init__(self):
|
|
self.now = 1000.0
|
|
|
|
def __call__(self):
|
|
return self.now
|
|
|
|
|
|
def make():
|
|
clock = Clock()
|
|
return FailureLimiter(clock=clock), clock
|
|
|
|
|
|
def test_blocks_after_max_failures():
|
|
limiter, _ = make()
|
|
for _ in range(MAX_FAILURES - 1):
|
|
limiter.record("alice")
|
|
assert not limiter.blocked("alice")
|
|
limiter.record("alice")
|
|
assert limiter.blocked("alice")
|
|
|
|
|
|
def test_keys_are_independent():
|
|
limiter, _ = make()
|
|
for _ in range(MAX_FAILURES):
|
|
limiter.record("alice")
|
|
assert not limiter.blocked("bob")
|
|
|
|
|
|
def test_block_expires_with_clock():
|
|
limiter, clock = make()
|
|
for _ in range(MAX_FAILURES):
|
|
limiter.record("alice")
|
|
clock.now += WINDOW_SECONDS - 1
|
|
assert limiter.blocked("alice")
|
|
clock.now += 2
|
|
assert not limiter.blocked("alice")
|
|
|
|
|
|
def test_old_failures_do_not_accumulate():
|
|
limiter, clock = make()
|
|
for _ in range(MAX_FAILURES - 1):
|
|
limiter.record("alice")
|
|
clock.now += WINDOW_SECONDS + 1
|
|
limiter.record("alice")
|
|
assert not limiter.blocked("alice")
|
|
|
|
|
|
def test_reset_clears_block():
|
|
limiter, _ = make()
|
|
for _ in range(MAX_FAILURES):
|
|
limiter.record("alice")
|
|
limiter.reset("alice")
|
|
assert not limiter.blocked("alice")
|
|
|
|
|
|
def test_reset_unknown_key_is_noop():
|
|
limiter, _ = make()
|
|
limiter.reset("nobody")
|
|
assert not limiter.blocked("nobody")
|