How to unit-test code that uses python-multiprocessing

Question

I have some code that uses multiprocessing.Pool to fork workers and perform a task in parallel. I'm trying to find the right way to run unit tests of this code.

Note I am not trying to test serial code test cases in parallel which I know packages like nose support.

If I write a test function that tests said parallel code, and attempt to run the tests with nose via: nosetests tests/test_function.py the non-parallel test execute properly but the parallel tests fail when multiprocessing tries to fork because main is not importable:

      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__

Which just repeats until I terminate the task. I can run the tests successfully if I modify tests/test_function.py to include:

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

and then execute with python tests\test_function.py

So what is the "right" way to do this that will integrate with a unit test package (doesn't have to be nose)?

Environ: Python 2.7.10 amd64 on Windows 7 64-bit

Update 2020: With python 3 and pytest, this is not an issue, suggest upgrades!

Answer 1

It looks like (see 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

I see people advising:

• patch multiprocessing module itself (a blogpost from first link has a link to the patch. I currently cannot post more than two links here)
• or implement similar workaround in test module: an example of workaround in a test

A code (non-significantly modified) from "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

Disclaimer: I haven't tried either of those solutions by myself (at least yet). I came across them, and this question, while trying to solve different problem. If I try any of those solutions, I'll amend this post.