Is it possible to mock.patch
a function that will be run in a subprocess, which is created using the multiprocessing
module with the start method
set to spawn
?
If there is no solution to patching a child process which is not forked
, what would be the right solution to bypass this problem?
It's important to say that switching to use fork is not a solution to my problem.
As of python3.8-macOS, the default behavior for the multiprocessing start method is spawn
.
From the multiprocessing
documentation:
Changed in version 3.8: On macOS, the spawn start method is now the default. The fork start method should be considered unsafe as it can lead to crashes of the subprocess. See bpo-33725.
.
Example code (Run on macos with python >= 3.8):
import multiprocessing
import unittest
from unittest import mock
def baz(i):
print(i)
def bar(i): # Middle function to avoid a pickeling problem with mocks
baz(i)
def foo():
with multiprocessing.Pool(2) as pool:
pool.map(bar, [i for i in range(10)])
class TestClass(unittest.TestCase):
@mock.patch(f'{__name__}.baz', return_value=None)
def test_case(self, mock):
# multiprocessing.set_start_method('fork', force=True)
foo()
The baz
function is not patched in the spawned processes (and hence, it prints) as can be seen in the following output.
Changing the default start method (commented in code) solves the problem
Output:
============================================================================== test session starts ===============================================================================
platform darwin -- Python 3.8.0, pytest-5.4.3, py-1.8.2, pluggy-0.13.1
rootdir: /Users/alonmenczer/dev/path_test/proj
collected 1 item
mp_spawn.py 0
1
4
5
6
7
8
9
2
3
.
=============================================================================== 1 passed in 0.27s ================================================================================
multiprocessing. dummy replicates the API of multiprocessing but is no more than a wrapper around the threading module. That means you're restricted by the Global Interpreter Lock (GIL), and only one thread can actually execute CPU-bound operations at a time. That's going to keep you from fully utilizing your CPUs.
When you patch a function using mock, you have the option to specify autospec as True: If you set autospec=True then the mock with be created with a spec from the object being replaced. All attributes of the mock will also have the spec of the corresponding attribute of the object being replaced.
unittest. mock provides a powerful mechanism for mocking objects, called patch() , which looks up an object in a given module and replaces that object with a Mock . Usually, you use patch() as a decorator or a context manager to provide a scope in which you will mock the target object.
Understand multiprocessing in no more than 6 minutes Multiprocessing is quintessential when a long-running process has to be speeded up or multiple processes have to execute parallelly. Executing a process on a single core confines its capability, which could otherwise spread its tentacles across multiple cores.
I assume you would rather prefer to use mock
. But if the library doesn't work, mocking can be achieved directly. E.g.:
def my_mock_baz(func):
def _mock_baz(i):
print(42)
def inner(*args, **kwargs):
global baz
backup = baz
baz = _mock_baz
res = func(*args, **kwargs)
baz = backup
return res
return inner
And then in the test:
class TestClass(unittest.TestCase):
@my_mock_baz
def test_case(self):
foo()
Of course, I other cases, mocking would be more difficult, but with the flexibility of Python it should be not too hard.
Side note: your problem doesn't reproduce in my Ubuntu 20.04, Python 3.8.2.
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With