如何测试限速算法？答案

【问题标题】：How to test a rate limiting algorithm?如何测试限速算法？
【发布时间】：2022-10-20 09:37:51
【问题描述】：

这是我从this page 得到的滑动窗口速率限制算法：

from time import time, sleep

class SlidingWindow:

    def __init__(self, capacity, time_unit, forward_callback, drop_callback):
        self.capacity = capacity
        self.time_unit = time_unit
        self.forward_callback = forward_callback
        self.drop_callback = drop_callback

        self.cur_time = time()
        self.pre_count = capacity
        self.cur_count = 0

    def handle(self, packet):

        if (time() - self.cur_time) > self.time_unit:
            self.cur_time = time()
            self.pre_count = self.cur_count
            self.cur_count = 0

        ec = (self.pre_count * (self.time_unit - (time() - self.cur_time)) / self.time_unit) + self.cur_count

        if (ec > self.capacity):
            return self.drop_callback(packet)

        self.cur_count += 1
        return self.forward_callback(packet)


def forward(packet):
    print("Packet Forwarded: " + str(packet))


def drop(packet):
    print("Packet Dropped: " + str(packet))


throttle = SlidingWindow(5, 1, forward, drop)

packet = 0

while True:
    sleep(0.1)
    throttle.handle(packet)
    packet += 1

如何在 pytest 中测试这种时间敏感算法？我想测试当请求（通过调用handle()）通过时调用forward_callback，当请求失败时调用drop_callback。如何在 pytest 中模拟循环？

【问题讨论】：

标签： python pytest

【解决方案1】：

您应该直接测试SlidingWindow 类，最后不要测试循环。原因是SlidingWindow 是独立的，无论如何使用它都应该工作。

由于该类严重依赖当前时间，您可以使用pytest-freezegun 来模拟请求的时间。

测试示例类似于：

from datetime import datetime, timedelta
from mock import MagicMock, call

from sliding_window import SlidingWindow


def test_handle_overflow(freezer, mg):
    start = datetime(year=2012, month=1, day=14, hour=12, minute=0, second=1)
    freezer.move_to(start)
    forward, drop = MagicMock(), MagicMock()
    throttle = SlidingWindow(5, 1, forward, drop)  # 5 per second -> every 200ms

    # first two should be accepted, while the 3rd should be dropped
    throttle.handle(0)
    freezer.tick(timedelta(milliseconds=200))
    throttle.handle(1)
    freezer.tick(timedelta(milliseconds=50))
    throttle.handle(2)

    forward.assert_has_calls(calls=[call(0), call(1)])
    drop.assert_called_once_with(2)

【讨论】：