要約

setup.pyでmultiprocessモジュールを含むunittestを実行すると無限ループになる
どうやらmultiprocessモジュールの仕様らしい
回避策として「__main__」を一時的に書き換える

前提

pythonでマルチプロセスで関数を動作させるのは「multiprocess」モジュールを使用します。

使い方やサンプルコードは公式が十分に分かりやすいです。
https://docs.python.org/ja/3/library/multiprocessing.html

さて、次のコードを見てください


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()

公式のサンプルコードから「if __name__ == '__main__':」を省きました。
これを実行するとえらいことになります。
（無限ループに入ってプロセスキルが必要になります）

公式にも記載があるので、これは仕様のようです
https://docs.python.org/ja/3/library/multiprocessing.html#the-spawn-and-forkserver-start-methods

ひとまず
multiprocessを使用したコードでは「if __name__ == '__main__':」を省いてはいけない
ということになります。

本題

さて、マルチプロセスを使ったモジュールを作成して、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()

ではテストしてみましょう。

$python test_mp_test.py

これは実行できます。

では、このテスト、setup.pyを経由して実行するとどうなるでしょうか。

ディレクトリ構成.

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='rr28_yosizumi@hotmail.com',
    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',
)

実行

$python setup.py test

そうです。「if name == 'main':」を省いていないにも関わらず、無限ループになります。

実行結果

======================================================================
ERROR: test_000_something (tests.test_mp_test.TestMp_test)
Test something.
----------------------------------------------------------------------
Traceback (most recent call last):
  ～省略～
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

以下ループ
・
・
・

回避策

setup.pyのテスト実行プロセスがmultiprocessの仕様にあっていないのではとは思うのですが、
行きついた回避策は次のようなものでした

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()

実行中の"__main__"を書き換えます。
こうすることで、setup.pyからのunittestも実行可能です。

また、複数のテストケースでマルチプロセスによる処理を実行するのであれば、
専用のテストクラスにしてsetUpと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()

おわりに

toxでテスト環境を構築しているときに気が付いて調査しました。
VSCや単体でのUnittestでは気が付かないので、開発終盤で「わぁぁぁぁぁっぁぁ」となることもあるのではないでしょうか。
ただ、もっとスマートな回避策がありそうなのですが、知っている人がいれば教えてもらえるとありがたいです。

multiprocessを使ったコードをpython setup.py testする

要約

前提

本題

回避策

おわりに