Pytest多进程函数Monkeypatch失效原因及正确测试方法

问题现象 #

在Pytest中对使用ProcessPoolExecutor的多进程函数做Monkeypatch时，修改不生效；但单进程或用ThreadPoolExecutor的多线程函数可以正常Monkeypatch。以下是可复现的代码：

被测试代码（file_a.py） #

import concurrent.futures as ccf

MY_CONSTANT = "hello"

def my_function():
    return MY_CONSTANT


def singleprocess_f():
    result = []
    for _ in range(3):
        result.append(my_function())
    return result


def multithread_f():
    result = []
    with ccf.ThreadPoolExecutor() as executor:
        futures = []
        for _ in range(3):
            future = executor.submit(my_function)
            futures.append(future)
        for future in ccf.as_completed(futures):
            result.append(future.result())
    return result


def multiprocess_f():
    result = []
    with ccf.ProcessPoolExecutor() as executor:
        futures = []
        for _ in range(3):
            future = executor.submit(my_function)
            futures.append(future)
        for future in ccf.as_completed(futures):
            result.append(future.result())
    return result

测试代码（test_file_a.py） #

from file_a import multiprocess_f, multithread_f, singleprocess_f

# 测试通过
def test_singleprocess_f(monkeypatch):
    monkeypatch.setattr("file_a.MY_CONSTANT", "world")
    result = singleprocess_f()
    assert result == ["world"] * 3

# 测试通过
def test_multithread_f(monkeypatch):
    monkeypatch.setattr("file_a.MY_CONSTANT", "world")
    result = multithread_f()
    assert result == ["world"] * 3

# 测试失败：返回的是["hello"]*3而非预期的["world"]*3
def test_multiprocess_f(monkeypatch):
    monkeypatch.setattr("file_a.MY_CONSTANT", "world")
    result = multiprocess_f()
    assert result == ["world"] * 3

失效原因 #

• 多线程场景：ThreadPoolExecutor的所有线程共享同一进程的内存空间，父进程中用Monkeypatch修改的模块属性，所有线程都能直接读取到，所以修改生效。
• 多进程场景：ProcessPoolExecutor通过创建独立进程执行任务：
  • Windows系统使用spawn方式启动子进程，会重新启动Python解释器并重新导入所有模块，完全忽略父进程的内存修改。
  • Unix系统使用fork方式，虽然会复制父进程内存，但子进程执行submit的函数时，会从模块文件重新加载目标函数（而非复用父进程中已修改的函数实例），导致Monkeypatch的修改无法被子进程识别。

解决方法 #

方法1：依赖注入（推荐） #

修改被测试函数，让它支持传入依赖的参数，避免直接依赖模块级常量，从根源上提升可测试性：

修改file_a.py中的multiprocess_f：

def multiprocess_f(target_constant=None):
    # 如果没有传入参数，使用原始常量
    use_constant = target_constant if target_constant is not None else MY_CONSTANT
    
    def wrapped_func():
        return use_constant
    
    result = []
    with ccf.ProcessPoolExecutor() as executor:
        futures = [executor.submit(wrapped_func) for _ in range(3)]
        for future in ccf.as_completed(futures):
            result.append(future.result())
    return result

对应的测试代码无需Monkeypatch，直接传参：

def test_multiprocess_f():
    result = multiprocess_f(target_constant="world")
    assert result == ["world"] * 3

方法2：子进程初始化时执行Monkeypatch #

利用ProcessPoolExecutor的initializer和initargs参数，在每个子进程启动时统一执行修改：

修改测试代码：

def test_multiprocess_f():
    # 子进程初始化函数：修改模块常量
    def init_child(new_constant):
        import file_a
        file_a.MY_CONSTANT = new_constant

    result = []
    # 创建Executor时指定初始化逻辑
    from file_a import ccf, my_function
    with ccf.ProcessPoolExecutor(initializer=init_child, initargs=("world",)) as executor:
        futures = [executor.submit(my_function) for _ in range(3)]
        for future in ccf.as_completed(futures):
            result.append(future.result())
    assert result == ["world"] * 3

如果要测试原有的multiprocess_f函数，可以通过Monkeypatch替换ProcessPoolExecutor的初始化参数：

def test_multiprocess_f(monkeypatch):
    def init_child(new_constant):
        import file_a
        file_a.MY_CONSTANT = new_constant

    # 替换ProcessPoolExecutor的构造参数
    from file_a import ccf
    original_executor = ccf.ProcessPoolExecutor
    def patched_executor(*args, **kwargs):
        kwargs["initializer"] = init_child
        kwargs["initargs"] = ("world",)
        return original_executor(*args, **kwargs)
    
    monkeypatch.setattr("file_a.ccf.ProcessPoolExecutor", patched_executor)
    
    result = multiprocess_f()
    assert result == ["world"] * 3

验证 #

执行pytest test_file_a.py，所有测试均可通过。

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