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

Question

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

注意我并不是要尝试并行测试串行代码测试用例,我知道像nose支持的软件包.

如果我编写一个测试所述并行代码的测试函数,并尝试使用nose运行测试:nosetests tests/test_function.py非并行测试正确执行,但多处理尝试fork时并行测试失败,因为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

那么,与单元测试包集成的"正确"方法是什么(不一定是鼻子)？

Environ:Windows 7 64位上的Python 2.7.10 amd64

Answer 1

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

Answer 2

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

2.7.6 之前的所有 Python（即迄今为止的所有 2.x，2014 年 4 月 17 日）中都存在一个错误，该错误会破坏在 Windows 上使用单元测试和多处理模块。...它已在新的 Python 3.x 中修复，但尚未向后移植到 2.x

我看到人们建议：

• 补丁多处理模块本身（第一个链接的博客文章有一个补丁的链接。我目前不能在这里发布两个以上的链接）
• 或在测试模块中实现类似的解决方法：测试中的解决方法示例

来自“测试中解决方法的示例”的代码（未进行重大修改）：

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

免责声明：我自己还没有尝试过这些解决方案（至少还没有）。我在尝试解决不同的问题时遇到了它们和这个问题。如果我尝试其中任何解决方案，我将修改这篇文章。