问题描述
我正在努力在多处理环境中使用 mock.patch
而没有多处理 mock.patch
工作正常.文件名:test_mp.py
I'm struggling to use mock.patch
in a multiprocessing environment while without multiprocessing mock.patch
works fine.Filename: 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
失败了 - Test
test_local
,test_mock
andtest_mock
finishes successfully - However
test_mp_mock
fails with
=================================== 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 并借助 sharedmock
https://github.com/elritsch/python-sharedmock 我能够走得更远,但仍然没有完成.
Based on https://medium.com/uckey/how-mock-patch-decorator-works-in-python-37acd8b78ae and with help of sharedmock
https://github.com/elritsch/python-sharedmock I was able to get further, but still not complete.
我为
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
finishes successfully.test_mp_mock2
fails with_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
激活补丁.关于如何将补丁从 test_mp_smock
传播到被测代码而不引入包装方法 inner2
的任何想法,因为我无法覆盖.
The main disadvantage of test_mp_smock
is that a new method inner2
must be introduced to activate patching via mock.patch
. Any idea of how to propagate patching from test_mp_smock
to code under test without introducing a wrapper method inner2
because I cannot override.
pool.apply(inner,args=[])
推荐答案
我终于搞定了.
SharedMock
不像 Mock
那样灵活,例如.缺少assert_call_once_with
,...,但您可以设置返回值或检查调用次数或其参数.
SharedMock
is not that flexible like Mock
, eg. missing assert_called_once_with
, ..., but you can set returned value or check the number of calls or its arguments.
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")
这篇关于mock.patch 和多处理的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持!