1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
import time
from limits import storage, RateLimitItemPerSecond
from limits import strategies
memcached_storage = storage.MemcachedStorage("memcached://localhost:22122")
# Flush memcached in case we are testing multiple Python versions in a row
memcached_storage.storage.flush_all()
fixed_window = strategies.FixedWindowRateLimiter(memcached_storage)
one_per_second = RateLimitItemPerSecond(1, 1)
assert True == fixed_window.hit(one_per_second, "test_namespace", "foo")
assert False == fixed_window.hit(one_per_second, "test_namespace", "foo")
time.sleep(1)
assert True == fixed_window.hit(one_per_second, "test_namespace", "foo")
|
