Skip to content

multiprocessing spawn and monkeypatch #12045

@gpadres

Description

@gpadres

Monkeypatching in spawned up processes is not behaving as expected (e.g. as fork processes).
It seems that if we monkeypatch an attribute, and another object uses the attribute, the other object does not see the patch.

This is very relevant because for Python 3.14 forking up processes is going to be unavailable, see. Current behavior makes it impossible to test whatever goes on in the child process on a lower level.

I tested the following script on a docker container in a Mac, running
Ubuntu 20.04.6
Python 3.12.1
pytest==8.0.2.

In the script, I test 2 scenarios, the first one a spawn child process and the second one a fork child process.
For each scenario, I mock a function and execute it via a wrapper function (and directly). In the case of the spawn process via wrapper, the function is not mocked. Am I missing something?

Note: module name in monkeypatch needs to be specified for below script to run.

from multiprocessing import get_context
import os
import pytest
import time

def mock_function(wrapper_flag= None):
    print('Function was mocked! wrapped:', wrapper_flag, os.getpid())

def function_to_mock(wrapper_flag=None):
    print('Function is not mocked. wrapped:', wrapper_flag, os.getpid())

def wrapper(wrapper_flag):
    function_to_mock(wrapper_flag)

def create_process_and_execute_function_to_mock(ctx, wrapper_flag=False):
    to_target = function_to_mock
    if wrapper_flag:
        to_target = wrapper
    ctx_ = get_context(ctx)
    process = ctx_.Process(target=to_target, kwargs={'wrapper_flag': wrapper_flag})
    process.start()
    print(process)
    process.join()
    time.sleep(1)

def test_mocking_spawn(monkeypatch):
    monkeypatch.setattr("__name__", mock_function)
    print()
    print('In current process:')
    function_to_mock()
    print('In Spawn:')
    create_process_and_execute_function_to_mock('spawn')
    #The following line results in the unexpected behavior
    create_process_and_execute_function_to_mock('spawn', wrapper_flag=True)

def test_mocking_fork(monkeypatch):
    monkeypatch.setattr("__name__", mock_function)
    print()
    print('In current process:')
    function_to_mock()
    print('In Fork:')
    create_process_and_execute_function_to_mock('fork')
    create_process_and_execute_function_to_mock('fork', wrapper_flag=True)

if __name__ == "__main__":
    pytest.main(['-v', '-s',__file__])

Metadata

Metadata

Assignees

No one assigned

    Labels

    plugin: monkeypatchrelated to the monkeypatch builtin pluginstalestatus: needs informationreporter needs to provide more information; can be closed after 2 or more weeks of inactivity

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions