unittest.mock 中的 patch() 在子进程中不起作用

Question

似乎

patch()

中的

unittest.mock

在使用

multiprocessing

模块生成的子进程中无法按预期工作。

这是我的代码：

a.py

import multiprocessing

from unittest.mock import patch

from b import run_new_process, call_mocked_function


def main():
    with patch("b.function_to_mock"):
        context = multiprocessing.get_context(multiprocessing.get_start_method())
        process = context.Process(target=run_new_process, args=())
        print("call from the main process")
        call_mocked_function()
        process.start()
        process.join()


if __name__ == '__main__':
    main()

b.py

def run_new_process():
    print("call from the subprocess")
    function_to_mock()


def call_mocked_function():
    function_to_mock()


def function_to_mock():
    print("inside real function body")

这段代码的输出是：

call from the main process
call from the subprocess
inside real function body

因此，当从同一进程调用该函数时，该函数会按预期进行模拟，但当我从子进程调用它时，会访问真实的函数体。为什么？当我运行 a.py 时，我希望不会看到 function_to_mock() 主体被调用。

预期输出：

call from the the main process
call from the subprocess

我正在使用Python 3.11.5。

我的上下文：这个示例实际上是我想要修改的 Airflow 测试代码的精简片段（https://github.com/apache/airflow/blob/b6318ffabce8cc3fdb02c30842726476b7e1fcca/tests/jobs/test_scheduler_job.py#L157） - 显然它适用于他们的 CI，但不适用于我的本地环境，所以我试图找出设置之间的差异。

Answer 1

我发现问题与 macOS（我的本地环境）和 Unix（另一个环境）处理子进程的方式差异有关。这里有部分解释：https://stackoverflow.com/a/61863247/2182542

patch()

设置的模拟在使用“spawn”命令（一个新的Python解释器进程）启动的子进程中不起作用，它必须是“fork”（父进程的副本）。

将创建上下文的行更改为：

后，程序会在 Unix 和 macOS 上按预期生成输出：

context = multiprocessing.get_context("fork")