测试python中的多处理实现

svmlkihl 于 2022-11-19 发布在 Python

关注(0)|答案(1)|浏览(173)

我能够使用下面的代码在没有多处理的情况下测试实现。

import unittest
from unittest.mock import patch

def side_effect_cube(x):
    return x**3

@patch("test.data.geocode.test_geo_thread.cube", side_effect=side_effect_cube)
def test_sum(mock_cube):
    assert find_cube(7) == [1, 8, 27, 64, 125, 216]
    assert mock_cube.called

def find_cube(num):
    result = []
    for i in range(1,num):
        result.append(cube(i))
    return result

def cube(x):
    return x**3

然而，当我将多处理添加到实现中时，它开始失败。

import unittest
from unittest.mock import patch
import multiprocessing as mp

def side_effect_cube(x):
    return x**3

@patch("test.data.geocode.test_geo_thread.cube", side_effect=side_effect_cube)
def test_sum(mock_cube):
    assert find_cube(7) == [1, 8, 27, 64, 125, 216]
    assert mock_cube.called

def find_cube(num):
    pool = mp.Pool(processes=4)
    result = [pool.apply(cube, args=(x,)) for x in range(1, num)]
    return result

def cube(x):
    return x**3

下面是我看到的错误

cls = <class 'multiprocessing.reduction.ForkingPickler'>, obj = (0, 0, <MagicMock name='cube' id='61482936'>, (1,), {}), protocol = None

    @classmethod
    def dumps(cls, obj, protocol=None):
        buf = io.BytesIO()
>       cls(buf, protocol).dump(obj) E       _pickle.PicklingError: Can't pickle <class 'unittest.mock.MagicMock'>: it's not the same object as unittest.mock.MagicMock

    ..\..\appdata\local\programs\python\python37\lib\multiprocessing\reduction.py:51: PicklingError

python-3.x

来源：https://stackoverflow.com/questions/61622084/test-multiprocessing-implementation-in-python

1条答案

按热度按时间

nafvub8i1#

这是一个未解决的问题。
“因为任何单独的Mock / MagicMock的类在mock模块的顶层都不可用，所以我认为这是无法修复的（基本的pickle限制）。”#139
以下是一些解决方法：如果您不需要神奇的方法（如assert和return_value），则可以根据上面提到的问题#139创建一个自定义类：

class PickableMock(Mock):
def __reduce__(self):
    return Mock, ()

正如here所述，您可以使用多进程库的fork，我建议检查thread，看看您可能面临的问题。
我在用例中使用的解决方案是模拟multiprocess.Pool()，我将其用作上下文管理器，多亏了this thread，我才能够模拟和Assert用正确参数调用的方法。

import multiprocessing
import unittest
from unittest.mock import patch, MagicMock

def cube(x):
    return x**3

def find_cube(num):
    with multiprocessing.Pool() as executor:
        return [executor.apply(cube, args=(x,)) for x in range(1, num)]

class MyTestCase(unittest.TestCase):
    @patch("multiprocessing.Pool")
    def test_cube(self, pool):
        pool.return_value.__enter__.apply = MagicMock()
        num = 7
        find_cube(num)
        for i in range(1, num):
            self.assertEqual(i, 
pool.return_value.__enter__.mock_calls[i].kwargs['args'][0])

赞(0）回复(0）举报 2022-11-19

我来回答

测试python中的多处理实现

1条答案

相关问题

热门标签

最新问答