验证模拟对象是在Python单元测试中创建的



我正在尝试测试一个函数,该功能可选地使用像对象这样的多处理池。如果提供了一个,则将使用该池,如果不是一个池,它将创建一个用于使用的螺纹池。

我想测试这种行为。具体来说,该线池在适当时被调用,并且没有其他方式。

在下面的最小示例中,我正在尝试验证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())

最新更新