在多处理中调用的模拟方法在 Mac 上不起作用

Question

我遇到了一个非常奇怪的错误。在我的项目中，我进行了单元测试，其中我模拟了在多处理操作中调用的一些方法（例如下载方法）。
这些单元测试在我的 CI 上运行良好，但是当我尝试在 Mac OSX 上本地运行它们时，没有考虑模拟。

我实现了以下最小可重现示例：

# test_mock_multiprocessing.py
import multiprocessing
from typing import List
from unittest import mock

import pytest


def _f(x: float) -> float:
    return x**2


def f(x: float) -> float:
    return _f(x)


def map_f(xs: List[float]) -> List[float]:
    return list(map(f, xs))


def multimap_f(xs: List[float]) -> List[float]:
    with multiprocessing.Pool(4) as pool:
        ys = list(pool.map(f, xs))
    return ys


@pytest.mark.parametrize("method", [map_f, multimap_f])
def test_original(method):
    xs = [-2, 3, 1]
    expected_ys = [4, 9, 1]
    ys = method(xs)
    assert ys == expected_ys


def mocked_f(x: float) -> float:
    return -x


@mock.patch("test_mock_multiprocessing._f", side_effect=mocked_f)
@pytest.mark.parametrize("method", [map_f, multimap_f])
def test_mocked(mocker, method):
    xs = [-2, 3, 1]
    expected_ys = [2, -3, -1]
    ys = method(xs)
    assert ys == expected_ys

我启动时使用的

pytest test_mock_multiprocessing.py

如果我在 Docker 映像 python3.8-buster（安装了 pytest）中启动这些测试，它们都会成功。 但是，如果我直接在我的主机（Mac OSX）上启动相同的测试，在一个只安装了 pytest 的 virtualenv 中，我有以下输出：

$ pytest test_mock_multiprocessing.py
=========================================================================================== test session starts ============================================================================================
platform darwin -- Python 3.8.12, pytest-6.2.5, py-1.10.0, pluggy-1.0.0
rootdir: ***
plugins: typeguard-2.13.0, cov-3.0.0
collected 4 items

test_mock_multiprocessing.py ...F                                                                                                                                                                    [100%]

================================================================================================= FAILURES =================================================================================================
_________________________________________________________________________________________ test_mocked[multimap_f] __________________________________________________________________________________________

mocker = <MagicMock name='_f' id='4377597216'>, method = <function multimap_f at 0x104e431f0>

    @mock.patch("test_mock_multiprocessing._f", side_effect=mocked_f)
    @pytest.mark.parametrize("method", [map_f, multimap_f])
    def test_mocked(mocker, method):
        xs = [-2, 3, 1]
        expected_ys = [2, -3, -1]
        ys = method(xs)
>       assert ys == expected_ys
E       assert [4, 9, 1] == [2, -3, -1]
E         At index 0 diff: 4 != 2
E         Use -v to get the full diff

test_mock_multiprocessing.py:44: AssertionError
========================================================================================= short test summary info ==========================================================================================
FAILED test_mock_multiprocessing.py::test_mocked[multimap_f] - assert [4, 9, 1] == [2, -3, -1]
======================================================================================= 1 failed, 3 passed in 1.06s ========================================================================================

Answer 1

Linux 和 MacOs 下的行为差异可能与多处理 start method 有关。在 Linux 上，默认的启动方式是fork，而在 MacOS 和 Windows 上是spawn。

如果使用fork，您的进程将在当前状态下分叉，包括模拟，而使用spawn，将启动一个新的 Python 解释器，其中模拟将不起作用。在 MacOs 下（但不是在 Windows 下），您可以在测试中将启动方法更改为 fork：

@mock.patch("test_mock_multiprocessing._f", side_effect=mocked_f)
@pytest.mark.parametrize("method", [map_f, multimap_f])
def test_mocked(mocker, method):
    start_method = multiprocessing.get_start_method()
    try:
        multiprocessing.set_start_method('fork', force=True)
        ... do the test
    finally: 
        multiprocessing.set_start_method(start_method, force=True)

为方便起见，您还可以将其包装到上下文管理器中：

@contextmanager
def set_start_method(method):
    start_method = multiprocessing.get_start_method()
    try:
        multiprocessing.set_start_method(method, force=True)
        yield
    finally:
        multiprocessing.set_start_method(start_method, force=True)

...
def test_something():
    with set_start_method('fork'):
        ... do the test