1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
|
from __future__ import annotations
import functools
import random
import pytest
import limits.aio.strategies
from limits import RateLimitItemPerMinute
from limits.storage import storage_from_string
from limits.strategies import (
FixedWindowRateLimiter,
MovingWindowRateLimiter,
SlidingWindowCounterRateLimiter,
)
from tests.utils import (
all_storage,
async_all_storage,
async_moving_window_storage,
async_sliding_window_counter_storage,
moving_window_storage,
sliding_window_counter_storage,
)
def hit_window(strategy, storage):
limit = RateLimitItemPerMinute(500)
uid = int(random.random() * 100)
strategy(storage).hit(limit, uid)
def hit_window_async(event_loop, strategy, storage):
limit = RateLimitItemPerMinute(500)
uid = int(random.random() * 100)
event_loop.run_until_complete(strategy(storage).hit(limit, uid))
@all_storage
@pytest.mark.benchmark(group="fixed-window")
def test_fixed_window(benchmark, uri, args, fixture):
benchmark(
functools.partial(
hit_window, FixedWindowRateLimiter, storage_from_string(uri, **args)
)
)
@sliding_window_counter_storage
@pytest.mark.benchmark(group="sliding-window-counter")
def test_sliding_window_counter(benchmark, uri, args, fixture):
benchmark(
functools.partial(
hit_window,
SlidingWindowCounterRateLimiter,
storage_from_string(uri, **args),
)
)
@moving_window_storage
@pytest.mark.benchmark(group="moving-window")
def test_moving_window(benchmark, uri, args, fixture):
benchmark(
functools.partial(
hit_window, MovingWindowRateLimiter, storage_from_string(uri, **args)
)
)
@async_all_storage
@pytest.mark.benchmark(group="async-fixed-window")
def test_fixed_window_async(event_loop, benchmark, uri, args, fixture):
benchmark(
functools.partial(
hit_window_async,
event_loop,
limits.aio.strategies.FixedWindowRateLimiter,
storage_from_string(uri, **args),
)
)
@async_moving_window_storage
@pytest.mark.benchmark(group="async-moving-window")
def test_moving_window_async(event_loop, benchmark, uri, args, fixture):
benchmark(
functools.partial(
hit_window_async,
event_loop,
limits.aio.strategies.MovingWindowRateLimiter,
storage_from_string(uri, **args),
)
)
@async_sliding_window_counter_storage
@pytest.mark.benchmark(group="async-sliding-window-counter")
def test_sliding_window_counter_async(event_loop, benchmark, uri, args, fixture):
benchmark(
functools.partial(
hit_window_async,
event_loop,
limits.aio.strategies.SlidingWindowCounterRateLimiter,
storage_from_string(uri, **args),
)
)
|
