41 lines
1.1 KiB
Python
41 lines
1.1 KiB
Python
import sys
|
|
sys.path.insert(0, "/root/.hermes/wizard-outputs/beam")
|
|
from sliding_window import SlidingWindow
|
|
|
|
def test_basic():
|
|
w = SlidingWindow(2, 10.0)
|
|
assert w.allow("k", 0) is True
|
|
assert w.allow("k", 1) is True
|
|
assert w.allow("k", 2) is False
|
|
assert w.allow("k", 11) is True # first expired
|
|
|
|
def test_window_boundary():
|
|
w = SlidingWindow(1, 5.0)
|
|
assert w.allow("k", 0) is True
|
|
assert w.allow("k", 5) is True # exactly window: old event cutoff 0 <= 0 expired
|
|
|
|
def test_per_key():
|
|
w = SlidingWindow(1, 10.0)
|
|
assert w.allow("a", 0) is True
|
|
assert w.allow("b", 0) is True
|
|
assert w.allow("a", 1) is False
|
|
|
|
def test_invalid():
|
|
for kw in ({"limit":0,"window":1}, {"limit":1,"window":0}, {"limit":1,"window":-1}):
|
|
try:
|
|
SlidingWindow(**kw); raise SystemExit("expected ValueError")
|
|
except ValueError:
|
|
pass
|
|
|
|
def test_reset():
|
|
w = SlidingWindow(1, 10.0)
|
|
assert w.allow("k",0) is True
|
|
assert w.allow("k",1) is False
|
|
w.reset("k")
|
|
assert w.allow("k",1) is True
|
|
w.allow("a",0); w.allow("b",0)
|
|
w.reset()
|
|
assert w.allow("a",0) is True
|
|
|
|
print("ALL TESTS PASSED")
|