patch() from unittest.mock doesn't work in a subprocess

Question

It seems that patch() from unittest.mock does not work as expected in subprocesses spawned with multiprocessing module for me.

Here is my code:

a.py

import multiprocessing

from unittest.mock import patch

from b import run_new_process, call_mocked_function


def main():
    with patch("b.function_to_mock"):
        context = multiprocessing.get_context(multiprocessing.get_start_method())
        process = context.Process(target=run_new_process, args=())
        print("call from the main process")
        call_mocked_function()
        process.start()
        process.join()


if __name__ == '__main__':
    main()

b.py

def run_new_process():
    print("call from the subprocess")
    function_to_mock()


def call_mocked_function():
    function_to_mock()


def function_to_mock():
    print("inside real function body")

The output of this code is:

call from the main process
call from the subprocess
inside real function body

So the function is mocked as expected when invoked from the same process, but the real body is accessed when I call it from subprocess. Why? When I run a.py, I expect NOT to see function_to_mock() body ever invoked.

Expected output:

call from the the main process
call from the subprocess

I am using Python 3.11.5.

My context: this example is actually a trimmed down fragment of Airflow test code which I want to modify (https://github.com/apache/airflow/blob/b6318ffabce8cc3fdb02c30842726476b7e1fcca/tests/jobs/test_scheduler_job.py#L157) - apparently it works on their CI but not on my local environment so I'm trying to figure out the difference between setups.

Answer 1

I found that the problem was related to the difference between how macOS (my local env) and Unix (the other env) handle subprocesses. It is partially explained here: https://stackoverflow.com/a/61863247/2182542

Mocking set up by patch() doesn't work in subprocesses that were started with "spawn" command (a fresh Python interpreter process), it must be "fork" (a copy of the parent process).

The program produces output as expected on both Unix and macOS after changing the line creating context to:

context = multiprocessing.get_context("fork")