--Exécuter unittest avec le module multiprocessus dans setup.py entraîne une boucle infinie
Utilisez le module "multiprocessus" pour exécuter des fonctions dans plusieurs processus en python.
La formule est assez simple pour comprendre comment utiliser et un exemple de code. https://docs.python.org/ja/3/library/multiprocessing.html
Maintenant, regardez le code suivant
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()
J'ai omis "if \ _ \ _ name \ _ \ _ == '\ _ \ _ main \ _ \ _':" dans l'exemple de code officiel. Ce serait génial. (Entrez une boucle infinie et avez besoin d'un processus kill)
Cela semble être une spécification car il est officiellement mentionné https://docs.python.org/ja/3/library/multiprocessing.html#the-spawn-and-forkserver-start-methods
Pour le moment ** Ne pas omettre "if \ _ \ _ name \ _ \ _ == '\ _ \ _ main \ _ \ _':" dans le code utilisant le multiprocessus ** Cela signifie que.
Supposons maintenant que vous souhaitiez créer un module utilisant plusieurs processus et exécutant 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()
Testons-le.
$python test_mp_test.py
Ceci peut être fait.
Alors, que se passe-t-il si vous exécutez ce test via ** setup.py **?
Structure du répertoire.
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',
)
Courir
$python setup.py test
**C'est vrai. Même si je n'ai pas omis "if name == '__ main__':", j'obtiens une boucle infinie. ** **
Résultat d'exécution
======================================================================
ERROR: test_000_something (tests.test_mp_test.TestMp_test)
Test something.
----------------------------------------------------------------------
Traceback (most recent call last):
~ Omis ~
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
La boucle suivante
・
・
・
Je pense que le processus d'exécution des tests de setup.py ne répond pas aux spécifications du multiprocessus, La solution de contournement que j'ai trouvée était:
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
Réécrivez le "\ _ \ _ main \ _ \ _" en cours d'exécution. En faisant cela, vous pouvez également exécuter unittest à partir de setup.py.
De plus, si vous souhaitez exécuter un traitement multi-processus dans plusieurs cas de test, Vous pouvez en faire une classe de test dédiée et la décrire dans setUp et 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()
J'ai remarqué et étudié lors de la construction d'un environnement de test avec tox. Je ne le remarque pas avec VSC ou Unittest seul, donc je pense que ça peut devenir "Wow!" À la fin du développement. Cependant, il semble y avoir une solution de contournement plus intelligente, mais j'apprécierais que quelqu'un qui le sait puisse me le dire.
Recommended Posts