Linux環境ではmock.patchがmultiprocessingを使って実行した関数内にも適用されるのに、MacOS環境でテストを実行した時に失敗する現象が起きたので、その回避策を共有します。

結論

multiprocessingがプロセスを開始する方法はいくつかあるが、デフォルトの開始方法はOSによって異なる
- デフォルトではLinux系のOSはforkが使用され、WindowsとMacOSはspawnが使用される
- forkは親プロセスの(mockを含む)全てのリソースが子プロセスに継承されるが、spawnではmockは継承されない
MacOSで子プロセスにmock.patchを適用するにはmultiprocessing.set_start_method("spawn", force=True)で開始方法としてforkを使うように強制する

自分の手元で動作確認したい人のために、以下にサンプルコードと動作確認手順を示します。

動作環境

Python3.10系
MacOS: Ventura 13.3.1
Linux(docker): python:3.10-alpine

動作確認で使ったサンプルコード

ディレクトリ構成

.
├── compose.yaml
├── src
│   └── demo.py
└── tests
    ├── __init__.py
    └── test_demo.py

compose.yaml

services:
    demo:
        container_name: demo
        image: python:3.10-alpine
        volumes:
            - ./:/opt
        tty: true
        working_dir: /opt

src/demo.py

"""demo"""
import time
import os

LOG_FILE_PATH = "/tmp/demo.log"


def log(msg: str):
    with open(LOG_FILE_PATH, "a", encoding="utf-8") as f:
        print(msg)
        print(msg, file=f)


class Server():

    def start(self):
        # リクエストを待ち受けてるつもり
        while True:
            log("waiting a request...")
            time.sleep(5)


def clean_up():
    # なんか終了処理
    log("cleaned up")


def main():
    try:
        print(f"{LOG_FILE_PATH=}")
        server = Server()
        server.start()
    except KeyboardInterrupt:
        clean_up()
        os._exit(1)


if __name__ == "__main__":
    main()

tests/test_demo.py

"""test of demo"""
import multiprocessing
import os
import signal
import tempfile
import time
import unittest
from unittest.mock import patch

from src.demo import main


class TestDemo(unittest.TestCase):

    def test_main_interrupt_ng_mac_os(self):
        # Setup & Execute
        _, tmp_file_path = tempfile.mkstemp()
        with patch("src.demo.LOG_FILE_PATH", tmp_file_path):
            p = multiprocessing.Process(target=main)
            p.start()
            time.sleep(0.5)
            os.kill(p.pid, signal.SIGINT)
            time.sleep(0.5)

        # Verify
        # 中断時に終了処理が行われること。
        with open(tmp_file_path, "r", encoding="utf-8") as f:
            self.assertIn("cleaned up", f.read().splitlines())

        # Cleanup
        os.remove(tmp_file_path)

    def test_main_interrupt_ok_mac_os(self):
        # Setup & Execute
        _, tmp_file_path = tempfile.mkstemp()
        with patch("src.demo.LOG_FILE_PATH", tmp_file_path):
            multiprocessing.set_start_method("fork", force=True)
            p = multiprocessing.Process(target=main)
            p.start()
            time.sleep(0.5)
            os.kill(p.pid, signal.SIGINT)
            time.sleep(0.5)

        # Verify
        # 中断時に終了処理が行われること。
        with open(tmp_file_path, "r", encoding="utf-8") as f:
            self.assertIn("cleaned up", f.read().splitlines())

        # Cleanup
        os.remove(tmp_file_path)


if __name__ == "__main__":
    unittest.main()

動作確認手順

まずLinux環境でちゃんと動くことを確認します。

Pythonコンテナ(alpine)を立ち上げます

ローカル環境

> docker compose up -d

コンテナが立ち上がったら、そのコンテナに入ります。

ローカル環境

> docker exec -it demo /bin/sh

コンテナ内でテストを実行します。

コンテナ内

> python -m unittest
LOG_FILE_PATH='/tmp/tmpyfj8omjg'
waiting a request...
cleaned up
.LOG_FILE_PATH='/tmp/tmpzqxd7ql4'
waiting a request...
cleaned up
.
----------------------------------------------------------------------
Ran 2 tests in 2.034s

OK

ちゃんと成功したのが確認できました。

LOG_FILE_PATHが一時ファイルになるようにpatchをしているので、LOG_FILE_PATH='/tmp/tmp...'と出力が出ていれば期待通りテストが実行されたことになります。

続いて、MacOS上でテストを実行してみます。

ローカル環境

> python -m unittest

LOG_FILE_PATH='/tmp/demo.log'
waiting a request...
cleaned up
FLOG_FILE_PATH='/var/folders/0s/y0mx0zc13wd1sf2lnq0xqc440000gn/T/tmpamnsa3jo'
waiting a request...
cleaned up
.
======================================================================
FAIL: test_main_interrupt_ng_mac_os (tests.test_demo.TestDemo)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/Users/miuramasaki/Devs/repos/demo/tests/test_demo.py", line 28, in test_main_interrupt_ng_mac_os
    self.assertIn("cleaned up", f.read().splitlines())
AssertionError: 'cleaned up' not found in []

----------------------------------------------------------------------
Ran 2 tests in 2.033s

FAILED (failures=1)

test_main_interrupt_ng_mac_osのテストが失敗しました。また、出力を見てみるとLOG_FILE_PATH='/tmp/demo.log'となっているためpatchが子プロセスまで適用されていないことが分かります。

参考