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

I prefer to mock multiprocessing in unit tests using python mock. Because unit tests should be independent and repeatable. That's why usually I'm creating mock version of multiprocessing classes (Process and Pool). Just to be sure that my tests are executed in right manner.