--Running a unittest with a multiprocess module in setup.py results in an infinite loop --Apparently the specifications of the multiprocess module --Temporarily rewrite "\ _ \ _ main \ _ \ _" as a workaround
To run a function in multiple processes in python, use the "multiprocess" module.
The formula is easy enough to understand how to use and sample code. https://docs.python.org/ja/3/library/multiprocessing.html
Now look at the following code
from multiprocessing import Pool
def f(x):
return x*x
def main():
with Pool(5) as p:
print(p.map(f, [1, 2, 3]))
main()
I've omitted "if \ _ \ _ name \ _ \ _ =='\ _ \ _ main \ _ \ _':" from the official sample code. Doing this would be great. (You will need a process kill in an infinite loop)
This seems to be a specification as it is officially mentioned https://docs.python.org/ja/3/library/multiprocessing.html#the-spawn-and-forkserver-start-methods
For the time being ** Do not omit "if \ _ \ _ name \ _ \ _ =='\ _ \ _ main \ _ \ _':" in code using multiprocess ** It means that.
Now, suppose you want to create a module that uses multi-process and run unittest.
mp_test.py
from multiprocessing import Pool
def f(x):
return x*x
def main():
with Pool(5) as p:
print(p.map(f, [1, 2, 3]))
if __name__ == '__main__':
pass
test_mp_test.py
"""Tests for `multiprocess_test` package."""
import unittest
from mp_test import mp_test
class TestMultiprocess_test(unittest.TestCase):
"""Tests for `multiprocess_test` package."""
def setUp(self):
"""Set up test fixtures, if any."""
def tearDown(self):
"""Tear down test fixtures, if any."""
def test_000_something(self):
mp_test.main()
if __name__ == "__main__":
unittest.main()
Let's test it.
$python test_mp_test.py
This can be done.
So what happens if you run this test via ** setup.py **?
Directory structure.
mp_test
|-mp_test.py
tests
|-test_mp_test.py
setup.py
setup.py
#!/usr/bin/env python
"""The setup script."""
from setuptools import setup, find_packages
setup(
author="rr28",
author_email='[email protected]',
python_requires='>=3.5',
description="multiprocess test.",
entry_points={
'console_scripts': [
'mp_test=mp_test.mp_test:main',
],
},
name='mp_test',
packages=find_packages(include=['mp_test', 'mp_test.*']),
test_suite='tests',
version='0.1.0',
)
Run
$python setup.py test
**That's right. Even though I didn't omit "if name =='main':", I get an infinite loop. ** **
Execution result
======================================================================
ERROR: test_000_something (tests.test_mp_test.TestMp_test)
Test something.
----------------------------------------------------------------------
Traceback (most recent call last):
~ Omitted ~
RuntimeError:
An attempt has been made to start a new process before the
current process has finished its bootstrapping phase.
This probably means that you are not using fork to start your
child processes and you have forgotten to use the proper idiom
in the main module:
if __name__ == '__main__':
freeze_support()
...
The "freeze_support()" line can be omitted if the program
is not going to be frozen to produce an executable.
----------------------------------------------------------------------
Ran 1 test in 0.007s
FAILED (errors=1)
Test failed: <unittest.runner.TextTestResult run=1 errors=1 failures=0>
test_000_something (tests.test_mp_test.TestMp_test)
Test something. ... ERROR
The following loop
・
・
・
I think that the test execution process of setup.py does not meet the multiprocess specification, The workaround I came up with was:
test_mp_test.py
"""Tests for `multiprocess_test` package."""
import unittest
import sys
from mp_test import mp_test
class TestMp_test(unittest.TestCase):
"""Tests for `mp_test` package."""
def setUp(self):
"""Set up test fixtures, if any."""
def tearDown(self):
"""Tear down test fixtures, if any."""
def test_000_something(self):
old_main = sys.modules["__main__"]
old_main_file = sys.modules["__main__"].__file__
sys.modules["__main__"] = sys.modules["mp_test.mp_test"]
sys.modules["__main__"].__file__ = sys.modules["mp_test.mp_test"].__file__
mp_test.main()
sys.modules["__main__"] = old_main
sys.modules["__main__"].__file__ = old_main_file
if __name__ == "__main__":
unittest.main()
https://stackoverflow.com/questions/33128681/how-to-unit-test-code-that-uses-python-multiprocessing
Rewrite the running "\ _ \ _ main \ _ \ _". By doing this, you can also run unittests from setup.py.
Also, if you want to execute multi-process processing in multiple test cases, You may make it a dedicated test class and describe it in setUp and tearDown.
test_mp_test.py
import unittest
import sys
from mp_test import mp_test
class TestMp_test(unittest.TestCase):
"""Tests for `mp_test` package."""
def setUp(self):
"""Set up test fixtures, if any."""
self._old_main = sys.modules["__main__"]
self._old_main_file = sys.modules["__main__"].__file__
sys.modules["__main__"] = sys.modules["mp_test.mp_test"]
sys.modules["__main__"].__file__ = sys.modules["mp_test.mp_test"].__file__
def tearDown(self):
"""Tear down test fixtures, if any."""
sys.modules["__main__"] = self._old_main
sys.modules["__main__"].__file__ = self._old_main_file
def test_000_something(self):
"""Test something."""
mp_test.main()
if __name__=="__main__":
unittest.main()
I noticed and investigated while building a test environment with tox. I don't notice it with VSC or Unittest alone, so I think it may become "Wow!" At the end of development. However, there seems to be a smarter workaround, but I would appreciate it if anyone who knows it could tell me.
Recommended Posts