Monkeypatching a function that uses multiprocess (via concurrent.futures.ProcessPoolExecutor) does not work as expected. If the same function is written with a single process or multithread (via concurrent.futures.ThreadPoolExecutor), monkeypatch works as expected.
Why does multiprocess monkeypatch fail and how does one correctly monkeypatch a multiprocess function for testing?
Simplest code example to illustrate my question is below. In actual usage, I would try to monkeypatch a function imported from another module.
# file_a.py
import concurrent.futures as ccf
MY_CONSTANT = "hello"
def my_function():
return MY_CONSTANT
def singleprocess_f():
result = []
for _ in range(3):
result.append(my_function())
return result
def multithread_f():
result = []
with ccf.ThreadPoolExecutor() as executor:
futures = []
for _ in range(3):
future = executor.submit(my_function)
futures.append(future)
for future in ccf.as_completed(futures):
result.append(future.result())
return result
def multiprocess_f():
result = []
with ccf.ProcessPoolExecutor() as executor:
futures = []
for _ in range(3):
future = executor.submit(my_function)
futures.append(future)
for future in ccf.as_completed(futures):
result.append(future.result())
return result
I expected all tests to pass:
# test_file_a.py
from file_a import multiprocess_f, multithread_f, singleprocess_f
# PASSES:
def test_singleprocess_f(monkeypatch):
monkeypatch.setattr("file_a.MY_CONSTANT", "world")
result = singleprocess_f()
assert result == ["world"] * 3
# PASSES:
def test_multithread_f(monkeypatch):
monkeypatch.setattr("file_a.MY_CONSTANT", "world")
result = multithread_f()
assert result == ["world"] * 3
# FAILS:
def test_multiprocess_f(monkeypatch):
monkeypatch.setattr("file_a.MY_CONSTANT", "world")
result = multiprocess_f()
assert result == ["world"] * 3
Kacper is a new contributor to this site. Take care in asking for clarification, commenting, and answering.
Check out our Code of Conduct.