Linux環境ではmock.patchがmultiprocessingを使って実行した関数内にも適用されるのに、MacOS環境でテストを実行した時に失敗する現象が起きたので、その回避策を共有します。
結論
-
multiprocessingがプロセスを開始する方法はいくつかあるが、デフォルトの開始方法はOSによって異なる- デフォルトではLinux系のOSは
forkが使用され、WindowsとMacOSはspawnが使用される -
forkは親プロセスの(mockを含む)全てのリソースが子プロセスに継承されるが、spawnではmockは継承されない
- デフォルトではLinux系のOSは
- 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が子プロセスまで適用されていないことが分かります。