【Python】 MacOSで multiprocessing を使って非同期実行した関数に mock.patch が適用されない問題の対処法

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

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

```txt:ディレクトリ構成
.
├── compose.yaml
├── src
│   └── demo.py
└── tests
    ├── __init__.py
    └── test_demo.py
```

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

```python: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()
```

```python: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`)を立ち上げます

```sh:ローカル環境
> docker compose up -d
```

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

```sh:ローカル環境
> docker exec -it demo /bin/sh
```

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

```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上でテストを実行してみます。

```sh:ローカル環境
> 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が子プロセスまで適用されていないことが分かります。

## 参考

- <https://docs.python.org/ja/3/library/multiprocessing.html#module-multiprocessing>
- <https://stackoverflow.com/questions/70425526/mocking-methods-called-within-multiprocessing-doesnt-work-on-mac>