使用Pytest测试threaded_execute函数的TimeoutError与日志记录

问题背景 #

我在Python项目中使用pytest测试工具模块，需要验证threaded_execute函数在触发concurrent.futures.TimeoutError时的警告日志记录功能，要求不能安装额外第三方包。函数代码如下：

import logging
from concurrent import futures

logger = logging.getLogger(__name__)

THREAD_TIMEOUT: float = 60  # 为简化直接硬编码值

def threaded_execute(func, *args, timeout=None, **kwargs):
    timeout = timeout or THREAD_TIMEOUT
    while True:
        with futures.ThreadPoolExecutor(max_workers=1) as executor:
            future = executor.submit(func, *args, **kwargs)
            try:
                return future.result(timeout=timeout)
            except futures.TimeoutError as exc:
                logger.warning(exc)
                continue

解决方法 #

直接用pytest自带的caplog fixture就能搞定，步骤如下：

• 写一个可控的慢函数：给函数加调用计数器，第一次调用时sleep超过测试超时时间触发错误，第二次正常返回，避免原函数无限循环。
• 用caplog抓日志：设置日志级别后，就能捕获并验证警告日志。
• 用短超时参数：测试时传入极小的timeout，不用改原代码的常量，测试更快。

测试代码示例：

import time
from concurrent import futures
import pytest
from your_module import threaded_execute  # 替换成你的实际模块名

def test_threaded_execute_timeout_log(caplog):
    call_count = 0

    def slow_func():
        nonlocal call_count
        call_count += 1
        if call_count == 1:
            # 第一次调用sleep，超过0.1秒的超时设置
            time.sleep(0.2)
        return "success"

    # 设置caplog只捕获WARNING及以上级别的日志
    caplog.set_level(logging.WARNING)
    # 传入0.1秒超时，触发TimeoutError
    result = threaded_execute(slow_func, timeout=0.1)
    
    # 验证最终返回结果正确
    assert result == "success"
    # 验证函数被调用了两次（一次超时重试，一次成功）
    assert call_count == 2
    # 验证警告日志已生成
    assert len(caplog.records) == 1
    assert isinstance(caplog.records[0].msg, futures.TimeoutError)
    assert caplog.records[0].levelname == "WARNING"

说明 #

• caplog是pytest内置的fixture，完全不需要额外安装包，能轻松捕获日志内容并做断言。
• 用计数器控制慢函数的行为，既触发了超时重试，又不会让测试陷入无限循环。
• 手动传入短超时参数，避免等待原代码中60秒的默认超时，大幅提升测试速度。

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