python - 如何对使用 python-multiprocessing 的代码进行单元测试

Question

我有一些代码使用 multiprocessing.Pool 来 fork worker 并并行执行任务。我正试图找到正确的方法来运行这段代码的单元测试。

请注意，我不是尝试并行测试串行代码测试用例，我知道像 nose support 这样的包。

如果我编写一个测试所述并行代码的测试函数，并尝试通过 nose 运行测试:nosetests tests/test_function.py 非并行测试正确执行但并行测试失败当 multiprocessing 由于 main 不可导入而尝试 fork 时:

      File "C:\python-2.7.10.amd64\lib\multiprocessing\forking.py", line 488, in prepare
assert main_name not in sys.modules, main_name
AssertionError: __main__
    assert main_name not in sys.modules, main_name
AssertionError:     _assert main_name not in sys.modules, main_name
_main__AssertionError
: __main__

这只是重复，直到我终止任务。如果我修改 tests/test_function.py 以包含:

，我可以成功运行测试

if __name__ == '__main__':
    import nose
    nose.main()

然后用python tests\test_function.py执行

那么与单元测试包集成的“正确”方法是什么(不一定是 Nose )？

环境:Windows 7 64 位上的 Python 2.7.10 amd64

2020 年更新:使用 python 3 和 pytest，这不是问题，建议升级!

Answer 1

我更喜欢使用 python mock 在单元测试中模拟多处理.因为单元测试应该是独立和可重复的。这就是为什么我通常创建多处理类(Process 和 Pool)的模拟版本。只是为了确保我的测试以正确的方式执行。