test_ratelimits.py 文件源码

def test_bucket_acquire_timing():
    from time import monotonic

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

    frequency = SECONDS / PERMITS

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

    # Shouldn't wait for BURST calls
    for i in range(BURST - 1):
        assert times[i + 1] - times[i] < frequency - EPSILON

    # After burst, the rest should fall at about the expected frequency
    times = times[BURST:]
    epsilon = 0.04
    for i, time in enumerate(times):
        assert expected_times[i] - epsilon <= time <= expected_times[i] + epsilon