rq SimpleWorker测试中如何mock sleep验证requests_ratelimiter限流

你patch失败核心是两个问题：

• 不了解unittest.mock的patch规则：patch必须打在目标对象被调用方导入的命名空间路径上，而非对象定义的原始路径，且必须在目标代码执行前完成patch
• 部分尝试的路径本身错误，或测试逻辑存在疏漏（比如路由定义是POST，测试里误用GET发起请求）

方案1：直接patch第三方库内部sleep（无需修改业务代码） #

你之前尝试的pyrate_limiter.limit_context_decorator.sleep路径本身是正确的，不生效是因为patch时机晚于代码执行，且测试请求方法写错了。
保持原有业务代码不变，测试代码按如下编写：

import requests_mock
from unittest.mock import patch

def test_rate_limit(client):
    # patch必须放在最外层，在任务入队、worker执行前就生效
    with patch("pyrate_limiter.limit_context_decorator.sleep") as mock_sleep:
        # 注意你的路由定义是POST方法，不要误用get
        client.post("/do_work/")
        
        with requests_mock.Mocker() as m:
            response_success = {"result": "All good"}
            m.get(WORK_URL, json=response_success)
            
            worker = SimpleWorker([my_queue], connection=redis_conn)
            worker.work(burst=True)
        
        # 按你的限流规则断言调用次数，例如per_second=1时，连续2次请求会触发1次sleep
        # assert mock_sleep.call_count == 1

方案2：显式传入sleep函数（更稳定，不依赖第三方库内部结构） #

如果担心后续requests_ratelimiter/pyrate_limiter版本升级调整内部sleep的导入路径，可在初始化LimiterSession时显式传入sleep函数，直接patch自己业务模块的命名空间即可，无需感知第三方库内部逻辑。

1. 修改worker.py代码：

from time import sleep
from requests_ratelimiter import LimiterSession

@job('my_queue', connection=redis_conn, timeout=3600, result_ttl=24 * 60 * 60)
def do_work():
    # 显式传入sleep函数，指定限流时用我们自己导入的sleep
    session = LimiterSession(per_second=1, sleep_function=sleep)
    r = session.get(WORK_URL)

2. 测试代码直接patch业务模块下的sleep：

import requests_mock
from unittest.mock import patch

def test_rate_limit(client):
    with patch("my_app.worker.sleep") as mock_sleep:
        client.post("/do_work/")
        
        with requests_mock.Mocker() as m:
            response_success = {"result": "All good"}
            m.get(WORK_URL, json=response_success)
            
            worker = SimpleWorker([my_queue], connection=redis_conn)
            worker.work(burst=True)
        
        # 按实际限流规则断言即可

• 方案1：路径正确，但patch时机不对，且测试请求方法写错导致任务可能根本没入队执行
• 方案2：from time import time导入的是time函数本身，函数对象没有sleep属性；且传入time_function只是替换获取当前时间的逻辑，和限流触发的sleep完全无关，统计方向错误
• 方案3：路径错误，requests_ratelimiter是独立安装的第三方库，不是my_app.worker下的子模块，自然找不到对应路径

内容的提问来源于stack exchange，提问作者Nick K9