我正在尝试测试一个函数,该功能可选地使用像对象这样的多处理池。如果提供了一个,则将使用该池,如果不是一个池,它将创建一个用于使用的螺纹池。
我想测试这种行为。具体来说,该线池在适当时被调用,并且没有其他方式。
在下面的最小示例中,我正在尝试验证ThreadPool创建的调用状态。我使用一个oberthreadpool,因为我需要在测试环境中进行一些下游的东西,只能在串行操作中检查。
当前,这在tumptest.test_pool_created中失败。如何验证ThreadPool
被称为?
除了下面的示例外,我尝试在没有任何运气的情况下模拟__init__
的CC_2。
temp.py
from multiprocessing.pool import ThreadPool
def run(execution_pool=None):
values = [1, 2]
if execution_pool:
out = execution_pool.map(lambda x: x+1, values)
else:
with ThreadPool(2) as p:
out = p.map(lambda x: x+1, values)
return out
if __name__ == "__main__":
out = run()
print(out)
temp_test.py
import unittest
import unittest.mock as mock
from multiprocessing.pool import ThreadPool
from temp import run
# Mock ThreadPool for diverting parallel code to serial
class MockThreadPool:
def map(self, run_simulation, all_inputs, chunksize=1):
map(run_simulation, all_inputs)
class TempTest(unittest.TestCase):
def test_check_runs(self):
self.assertTrue(True)
# Want to test:
# - ThreadPool is created when no execution pool is passed to run()
# - ThreadPool is not created when an execution pool is passed to run()
@mock.patch('multiprocessing.pool.ThreadPool', return_value=MockThreadPool())
def test_pool_created(self, fcn_pool):
out = run(None)
self.assertTrue(fcn_pool.called)
@mock.patch('multiprocessing.pool.ThreadPool', return_value=MockThreadPool())
def test_pool_not_created(self, fcn_pool):
out = run(execution_pool=MockThreadPool())
self.assertFalse(fcn_pool.called)
我以前遇到过同样的问题。您正在修补multiprocessing.pool.ThreadPool
,但是temp
模块中的代码直接调用ThreadPool
。我敢肯定,如果您更改patch()
的电话:
@mock.patch('temp.ThreadPool', return_value=MockThreadPool())