mock.patch和多处理

Question

我很难在多处理环境中使用mock.patch，而在没有多处理的情况下，mock.patch可以正常工作。文件名：test_mp.py

import multiprocessing
import mock
def inner():
return sub()
def sub():
return "abc"
def test_local():
assert inner()=="abc"
def test_mp():    
with multiprocessing.Pool() as pool:
assert pool.apply(inner,args=[])=='abc'
def test_mock():    
with mock.patch('test_mp.sub', return_value='xxx') as xx:
assert inner()=="xxx"
xx.assert_called_once()
def test_mp_mock():    
with multiprocessing.Pool() as pool:
with mock.patch('test_mp.sub', return_value='xyz') as xx:
assert pool.apply(inner,args=[])=='xyz'
xx.assert_called_once()

• 测试test_local、test_mock和test_mock成功完成
• 但是test_mp_mock失败

=================================== FAILURES ===================================
_________________________________ test_mp_mock _________________________________
def test_mp_mock():
with multiprocessing.Pool() as pool:
with mock.patch('test_mp.sub', return_value='xyz') as xx:
>               assert pool.apply(inner,args=[])=='xyz'
E               AssertionError: assert 'abc' == 'xyz'
E                 - abc
E                 + xyz
projects/mfhealth/test_mp.py:25: AssertionError

更新：

基于https://medium.com/uckey/how-mock-patch-decorator-works-in-python-37acd8b78ae并借助CCD_ 8https://github.com/elritsch/python-sharedmock我能够走得更远，但仍然没有完成。

我为扩展了test_mp.py

from sharedmock.mock import SharedMock
def inner2(sm):
with mock.patch('test_mp.sub', sm) as xx:
return inner()

def test_mp_smock():
with multiprocessing.Pool() as pool:
sm=SharedMock()
sm.return_value="xyz"
with mock.patch('test_mp.sub', sm) as xx:
assert pool.apply(inner2,args=[sm])=='xyz'
assert xx.call_count == 1
def test_mp_mock2():
with multiprocessing.Pool() as pool:
sm=mock.Mock()
sm.return_value="xyz"
print(f"before patch {sub}, {locals()}")
with mock.patch('test_mp.sub', sm) as xx:
print(f"after patch {sub}")
assert pool.apply(inner2,args=[sm])=='xyz'
assert xx.call_count == 1

结果如下：

• test_mp_smock成功完成
• test_mp_mock2与_pickle.PicklingError: Can't pickle <class 'mock.mock.Mock'>: it's not the same object as mock.mock.Mock一起失败

test_mp_smock的主要缺点是必须引入一种新的方法inner2来通过mock.patch激活修补。关于如何在不引入包装器方法inner2的情况下将补丁从test_mp_smock传播到测试中的代码的任何想法，因为我无法覆盖。

pool.apply(inner,args=[])

Answer 1

最后，我让它工作起来了。

SharedMock不像Mock那样灵活，例如缺少assert_called_once_with。。。，但是您可以设置返回值，或者检查调用次数或参数。

import multiprocessing
import mock
from sharedmock.mock import SharedMock

def inner():
return sub(x="xxx")
def sub(x=""):
return f"abc"
def fun_under_test():
with multiprocessing.Pool() as pool:
assert pool.apply(inner,args=[])=='xyz'
def test_final():
sm=SharedMock()
sm.return_value="xyz"
with mock.patch('test_mp.sub', sm) as xx:
fun_under_test()
assert xx.call_count == 1 #number of calls of sub function
assert xx.mock_calls[0][2]['x']=="xxx" # value of parameters ie sub(x="xxx")