test_ratelimits.py 文件源码

def test_windowed_bucket_acquire_timing():
    from time import monotonic

    limiter = WindowedTokenBucketRateLimiter(SECONDS, PERMITS, BURST, TOKENS)
    times = []
    for _ in range(VALUE_COUNT):
        with limiter:
            times.append(monotonic())

    frequency = SECONDS / PERMITS

    first_time = times[0]
    expected_times = [first_time]
    for i in range(1, len(times)):
        expected_times.append(first_time + i * frequency)

    epsilon = 0.04
    for i, time in enumerate(times):
        assert expected_times[i] - epsilon <= time <= expected_times[i] + epsilon