如何在Pytest中验证循环操作的休眠时长？

我们需要在Pytest中验证Loop类的LoopJob方法里的等待逻辑：循环最多执行3次，每次请求失败时休眠30秒，期望总休眠时长达到90秒。目前已经能通过requests_mock的请求历史验证循环次数，但不清楚如何断言休眠时长。

业务代码（Loop.py） #

import requests
import json
import time

class CommunicationError(Exception):
    pass

class Loop:
    def __init__(self):
        self.hostname = ""
        self.timeout = 10
    
    def LoopJob(self, job_id, date, subject, fromDate, toDate):
        # 示例payload，实际业务中根据参数生成
        payload = {
            "job_id": job_id,
            "date": date,
            "subject": subject,
            "fromDate": fromDate,
            "toDate": toDate
        }
        
        response = None
        for error_count in range(3):
            try:
                request_url = 'http://' + self.hostname + '/samplejob'
                headers = {'content-type': 'application/json'}
                response = requests.post(
                    request_url,
                    data=json.dumps(payload),
                    timeout=self.timeout,
                    headers=headers
                )
            except Exception:
                response = None

            if response is None:
                time.sleep(30)
            else:
                break

        if response is None or response.status_code != 202:
            error_message = "error message on parent method"
            raise CommunicationError(error_message)

        return response

当前测试代码（Pytest.py） #

def test_looptimes(monkeypatch, requests_mock):
    monkeypatch.setenv('ENV_ID', 'Local')
    headers = {'content-type': 'application/json'}
    
    # 模拟请求失败，让response为None
    requests_mock.post(
        "http://localhost:8080/samplejob",
        headers=headers,
        side_effect=Exception("Mocked request failure")
    )

    # 初始化Loop实例
    loop = Loop()
    loop.hostname = "localhost:8080"
    loop.timeout = 10

    # 执行方法，预期抛出异常
    try:
        loop.LoopJob("test_id", "2024-01-01", "test_subject", "2024-01-01", "2024-01-01")
    except CommunicationError:
        pass

    # 验证请求次数
    history = requests_mock.request_history
    assert len(history) == 3
    assert all(h.url == 'http://localhost:8080/samplejob' for h in history)

核心思路是mock替换time.sleep方法，记录它的调用次数和每次传入的时长参数，不需要真实等待，既高效又能准确验证逻辑。

方案一：使用monkeypatch统计休眠调用 #

修改测试代码，新增对time.sleep的mock和断言：

import time

def test_loop_sleep_duration(monkeypatch, requests_mock):
    monkeypatch.setenv('ENV_ID', 'Local')
    headers = {'content-type': 'application/json'}

    # 模拟3次请求全部失败
    requests_mock.post(
        "http://localhost:8080/samplejob",
        headers=headers,
        side_effect=Exception("Mocked request failure")
    )

    # 记录sleep的调用参数
    sleep_records = []
    def mock_sleep(seconds):
        sleep_records.append(seconds)
    
    # 替换系统的time.sleep为我们的mock函数
    monkeypatch.setattr(time, 'sleep', mock_sleep)

    # 初始化并执行测试
    loop = Loop()
    loop.hostname = "localhost:8080"
    loop.timeout = 10

    try:
        loop.LoopJob("test_id", "2024-01-01", "test_subject", "2024-01-01", "2024-01-01")
    except CommunicationError:
        pass

    # 断言休眠逻辑符合预期
    assert len(sleep_records) == 3  # 循环3次，每次都触发休眠
    assert all(sec == 30 for sec in sleep_records)  # 每次休眠30秒
    assert sum(sleep_records) == 90  # 总休眠时长90秒

    # 同时验证请求次数
    history = requests_mock.request_history
    assert len(history) == 3

方案二：使用unittest.mock.patch验证休眠调用 #

如果习惯用patch装饰器，也可以这样实现：

from unittest.mock import patch

def test_loop_sleep_with_patch(monkeypatch, requests_mock):
    monkeypatch.setenv('ENV_ID', 'Local')
    headers = {'content-type': 'application/json'}

    # 模拟请求失败
    requests_mock.post(
        "http://localhost:8080/samplejob",
        headers=headers,
        side_effect=Exception("Mocked request failure")
    )

    # 用patch临时替换time.sleep
    with patch('time.sleep') as mock_sleep:
        loop = Loop()
        loop.hostname = "localhost:8080"
        loop.timeout = 10

        try:
            loop.LoopJob("test_id", "2024-01-01", "test_subject", "2024-01-01", "2024-01-01")
        except CommunicationError:
            pass

        # 验证sleep的调用情况
        assert mock_sleep.call_count == 3
        mock_sleep.assert_has_calls([
            patch.call(30),
            patch.call(30),
            patch.call(30)
        ])

关键注意事项 #

1. 补全异常捕获：原业务代码中如果请求抛出异常，response不会被赋值为None，必须补充except Exception: response = None才能触发休眠逻辑，否则测试无法覆盖休眠分支。
2. 避免真实等待：通过mocktime.sleep，测试无需等待90秒，执行速度大幅提升。
3. 边界场景测试：还可以测试“第2次请求成功”的场景，此时应该只触发1次休眠，断言sleep_records长度为1，总时长30秒。

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