如何在 Pytest 中正确 monkeypatch 多进程函数进行单元测试

pytest 的 monkeypatch 对多进程（processpoolexecutor）无效，是因为子进程通过 spawn 方式启动时会重新执行模块全局代码，无法继承父进程的 patch；解决方案是通过 initializer 参数为进程池显式初始化共享状态。

pytest 的 monkeypatch 对多进程（processpoolexecutor）无效，是因为子进程通过 spawn 方式启动时会重新执行模块全局代码，无法继承父进程的 patch；解决方案是通过 initializer 参数为进程池显式初始化共享状态。

在 Python 单元测试中，pytest.monkeypatch 是修改模块级变量或函数行为的常用手段，但它对多进程场景存在根本性限制——子进程不共享父进程的内存空间。当使用 concurrent.futures.ProcessPoolExecutor 时，底层依赖 multiprocessing 模块创建新进程。在 Windows 和 macOS（默认 spawn 启动方式）下，每个子进程会重新导入模块并执行全局语句（如 MY_CONSTANT = “hello”），因此 monkeypatch.setattr() 在主进程中所做的修改对子进程完全不可见。

相比之下，ThreadPoolExecutor 共享主线程内存，所以 monkeypatch 生效；单进程逻辑自然也生效。问题本质不是 Pytest 的缺陷，而是操作系统进程模型的必然结果。

✅ 正确的测试方案：使用 initializer 初始化子进程

核心思路是：让每个子进程在启动时主动应用所需配置。ProcessPoolExecutor 支持 initializer 参数，它会在每个工作进程启动时调用一次指定函数，用于设置该进程的初始状态。

以下是改造后的完整实践：

1. 修改被测函数，支持传入自定义 executor（便于注入测试专用实例）

# file_a.py
import concurrent.futures as ccf

MY_CONSTANT = "hello"

def my_function():
    return MY_CONSTANT

def multiprocess_f(executor=None):
    """支持传入 executor，便于测试时注入带 initializer 的实例"""
    if executor is None:
        executor = ccf.ProcessPoolExecutor()

    result = []
    with executor:  # 注意：with 块确保 executor 正确 shutdown
        futures = [executor.submit(my_function) for _ in range(3)]
        for future in ccf.as_completed(futures):
            result.append(future.result())
    return result

✅ 小技巧：使用列表推导式简化 futures 提交；as_completed 返回顺序与提交无关，但本例中结果一致性不依赖顺序。

2. 在测试中定义 initializer，并构造专用 executor

# test_file_a.py
import concurrent.futures as ccf
import file_a

def init_pool_processes():
    """在每个子进程中执行：覆盖 MY_CONSTANT"""
    file_a.MY_CONSTANT = "world"

def test_multiprocess_f():
    # 创建带 initializer 的进程池，确保每个子进程都运行 init_pool_processes()
    executor = ccf.ProcessPoolExecutor(initializer=init_pool_processes)
    result = file_a.multiprocess_f(executor)
    assert result == ["world"] * 3

⚠️ 注意事项

• 不要在 initializer 中引用闭包变量或 monkeypatch 对象：initializer 函数必须可被序列化（pickle），且仅能访问模块级作用域或其参数（但 initializer 不接受额外参数）。因此直接赋值 file_a.MY_CONSTANT = … 是安全且推荐的方式。
• 避免全局 monkeypatch 风险：原测试中 monkeypatch.setattr(“file_a.MY_CONSTANT”, …) 使用字符串路径易出错；建议改用 monkeypatch.setattr(file_a, “MY_CONSTANT”, …) 提高类型安全性和 IDE 可导航性。
• 资源清理：ProcessPoolExecutor 在 with 块退出时自动调用 shutdown(wait=True)，无需手动管理，但务必保证 executor 被正确上下文管理。
• 跨平台兼容性：Linux 默认使用 fork，理论上可继承父进程 patch，但为统一行为、避免隐式差异，始终使用 initializer 是最佳实践。

总结

通过将测试控制权从“打补丁”转向“主动初始化”，我们既尊重了多进程的隔离性设计，又实现了可靠、可重复、跨平台的单元测试。这是 Python 并行编程测试中的关键范式转换。