我能够使用下面的代码在没有多处理的情况下测试实现。
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
1条答案
按热度按时间nafvub8i1#
这是一个未解决的问题。
“因为任何单独的Mock / MagicMock的类在mock模块的顶层都不可用,所以我认为这是无法修复的(基本的pickle限制)。”#139
以下是一些解决方法:如果您不需要神奇的方法(如
assert
和return_value
),则可以根据上面提到的问题#139创建一个自定义类:正如here所述,您可以使用多进程库的fork,我建议检查thread,看看您可能面临的问题。
我在用例中使用的解决方案是模拟
multiprocess.Pool()
,我将其用作上下文管理器,多亏了this thread,我才能够模拟和Assert用正确参数调用的方法。