如何检测Python子进程运行时长不超限？附Pytest/Mock/断言思路

一、基础实现：监控子进程运行时间 #

要检查子进程是否在设定时限内完成，最直接的方式是利用subprocess模块自带的超时机制，或者结合时间戳手动计算运行时长。

核心实现代码 #

import subprocess
import time

def run_with_timeout(cmd, timeout=5):
    """运行子进程并检查是否超时"""
    start_ts = time.time()
    try:
        # 使用subprocess.run的timeout参数直接管控
        proc_result = subprocess.run(
            cmd,
            timeout=timeout,
            capture_output=True,
            text=True,
            check=False
        )
        elapsed = time.time() - start_ts
        return {
            "completed": True,
            "elapsed_time": elapsed,
            "stdout": proc_result.stdout,
            "returncode": proc_result.returncode
        }
    except subprocess.TimeoutExpired:
        elapsed = time.time() - start_ts
        return {
            "completed": False,
            "elapsed_time": elapsed,
            "error": "Process exceeded timeout limit"
        }

关键说明 #

• subprocess.run的timeout参数是内核级别的超时管控，超时后会自动终止子进程，避免僵尸进程残留。
• 手动计算elapsed_time可以辅助验证实际运行时长是否符合预期。

二、Pytest测试实现思路 #

针对子进程超时场景，我们可以从真实进程测试和Mock隔离测试两个维度编写用例，结合assert完成断言验证。

1. 测试正常不超时场景 #

直接运行一个快速结束的命令，断言进程正常完成且运行时长小于设定超时值：

from my_module import run_with_timeout

def test_proc_within_timeout():
    result = run_with_timeout(["echo", "test"], timeout=5)
    assert result["completed"] is True
    assert result["elapsed_time"] < 5
    assert result["stdout"].strip() == "test"

2. 测试超时场景 #

运行一个会阻塞的命令（如sleep），设置较短超时值，断言进程超时终止：

import subprocess

def test_proc_timeout():
    # 直接用subprocess.run验证超时异常
    with pytest.raises(subprocess.TimeoutExpired):
        subprocess.run(["sleep", "10"], timeout=2)
    
    # 用自定义函数验证超时返回结果
    result = run_with_timeout(["sleep", "10"], timeout=2)
    assert result["completed"] is False
    assert result["error"] == "Process exceeded timeout limit"
    assert result["elapsed_time"] >= 2  # 实际时长会略大于超时值

3. 用Mock隔离外部依赖 #

通过unittest.mock.patch模拟subprocess.run的行为，避免真实执行外部命令，提升测试速度和稳定性：

from unittest.mock import patch, Mock

@patch("my_module.subprocess.run")
def test_mocked_proc_timeout(mock_run):
    # 模拟超时异常
    mock_run.side_effect = subprocess.TimeoutExpired(
        cmd=["sleep", "10"], timeout=2
    )
    result = run_with_timeout(["sleep", "10"], timeout=2)
    assert result["completed"] is False
    assert result["error"] == "Process exceeded timeout limit"
    
    # 模拟正常执行场景
    mock_proc_result = Mock()
    mock_proc_result.stdout = "mocked output"
    mock_proc_result.returncode = 0
    mock_run.side_effect = None
    mock_run.return_value = mock_proc_result
    
    result = run_with_timeout(["echo", "mock"], timeout=5)
    assert result["completed"] is True
    assert result["stdout"] == "mocked output"

三、核心注意事项 #

• 避免手动轮询子进程状态：subprocess.run的timeout是官方推荐的方式，比手动用Popen.poll()+计时更可靠。
• 断言时间时留有余量：超时后的实际运行时长可能略大于设定值（信号传递、进程终止需要时间），因此用>=而非==断言。
• Mock的路径要准确：patch时需指定subprocess.run在被测试模块中的导入路径（如my_module.subprocess.run），而非全局的subprocess.run。

内容的提问来源于stack exchange，提问作者dmp-98