How to fix the DeprecationWarning in the following pytest example? Note this is minimalistic reproducer.
Use case is rather simple:
- I have an object (represented as
Obj
) that is tested; Obj
interacts with an external server;- For testing purposes, external server is mocked using
multiprocessing.Process()
; Obj
usesProcessPoolExecutor
to do some CPU-bound operations;
import asyncio
import concurrent.futures
import http.server
import multiprocessing
import pytest
import pytest_asyncio
# This does not work, as mock_server is not pickable...
# multiprocessing.set_start_method("spawn")
def hello_world():
print("Hello, World!")
class Obj():
def __init__(self):
self.pool = concurrent.futures.ProcessPoolExecutor(max_workers=1)
asyncio.get_running_loop().run_in_executor(self.pool, hello_world)
@pytest_asyncio.fixture
async def obj_to_test():
return Obj()
@pytest.fixture
def mock_server(tmpdir):
def server(path, host, port):
os.chdir(path)
httpd = http.server.HTTPServer(
(host, port),
http.server.SimpleHTTPRequestHandler
)
httpd.serve_forever()
process = multiprocessing.Process(target=server, args=(".", "127.0.0.1", 9999))
process.start()
yield process
process.terminate()
process.join()
def test_something(obj_to_test, mock_server):
# normally obj_to_test will interact here with mock_server
pass
My environment is:
platform linux — Python 3.12.3, pytest-8.2.0, pluggy-1.5.0
plugins: asyncio-0.23.6
I am using Python installation from this Docker image: python:3.12-slim@sha256:2be8daddbb82756f7d1f2c7ece706aadcb284bf6ab6d769ea695cc3ed6016743
pip install pytest==8.2.0 pytest-asyncio==0.23.6
After above, I am getting the following warning, which I want to fix:
/usr/local/lib/python3.12/multiprocessing/popen_fork.py:66: DeprecationWarning: This process (pid=523) is multi-threaded, use of fork() may lead to deadlocks in the child.
Using multiprocessing.set_start_method("spawn")
does not help.
I was able to reproduce DeprecationWarning outside of pytest
environment as well:
import concurrent.futures
import multiprocessing
import time
pool = concurrent.futures.ProcessPoolExecutor(max_workers=1)
job = pool.submit(lambda: time.sleep(5))
process = multiprocessing.Process(target=lambda: None_)
process.start()
process.terminate()
and its run:
PYTHONWARNINGS=default python3 t5.py
/usr/local/lib/python3.12/multiprocessing/popen_fork.py:66: DeprecationWarning: This process (pid=1024) is multi-threaded, use of fork() may lead to deadlocks in the child.
self.pid = os.fork()