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

Question

我有一些代码使用 multiprocessing.Pool 分叉工作人员并并行执行任务。我正在尝试找到 正确的方法 到运行这段代码的单元测试。

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

如果我编写一个测试所述并行代码的测试函数，并尝试运行通过 nose 进行测试： nosetests tests/test_function.py 非并行测试正确执行但并行测试在多处理时失败尝试分叉，因为 main 不可导入：

      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

执行

那么 "right" 集成单元测试包的方法是什么（不一定是 nose）？

环境：Python 2.7.10 amd64 on Windows 7 64-bit

2020 年更新：使用 python3 和 pytest，这不是问题，建议升级！

Answer 1

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

Answer 2

看起来像（见http://7fttallrussian.blogspot.com/2014/04/fix-for-bug-with-unittest-and.html）

There is a bug in all Pythons up to 2.7.6 (i.e. all 2.x so far, April 17 2014) that breaks using unittest and multiprocessing module on Windows. ... It's fixed in new Pythons 3.x but haven't been backported to 2.x yet

我看到有人建议：

补丁多处理模块本身（第一个link的博客post有一个补丁link。我目前不能post超过两个link在这里）
或在测试模块中实施类似的解决方法：an example of workaround in a test

来自 "an example of workaround in a test" 的代码（未进行重大修改）：

import unittest
import sys
class TestSample(unittest.TestCase):
    def test(self):

        # To fix the Windows forking system it's necessary to point __main__ to
        # the module we want to execute in the forked process
        old_main =                          sys.modules["__main__"]
        old_main_file =                     sys.modules["__main__"].__file__
        sys.modules["__main__"] =           sys.modules["app"]
        sys.modules["__main__"].__file__ =  sys.modules["app"].__file__

        # do your testing here

        sys.modules["__main__"] =           old_main
        sys.modules["__main__"].__file__ =  old_main_file

免责声明：我自己还没有尝试过这两种解决方案（至少目前还没有）。我在尝试解决不同的问题时遇到了他们和这个问题。如果我尝试任何这些解决方案，我将修改此 post.

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

How to unit-test code that uses python-multiprocessing

python

unit-testing

nose

python-multiprocessing