Pytest Monkeypatch — многопроцессная функция для тестирования

Я предполагаю, что вы работаете на какой-то платформе, которая по умолчанию использует метод spawn для создания новых процессов (например, Windows). В этом случае не имеет значения, какое значение вы присвоите MY_CONSTANT в основном процессе, используя Monkeypatch или иным образом, поскольку при создании процессов дочернего пула они инициализируют свою память, повторно выполняя оператор MY_CONSTANT = "hello". То есть дочерние процессы, созданные с помощью spawn, не наследуют никаких значений от родительского процесса, а вместо этого инициализируют память, выполняя все операторы в глобальной области видимости (например, оператор импорта, объявления функций и переменных и т. д.).

Если это так, то вам необходимо предоставить функцию инициализатора пула, используя аргумент инициализатора при создании экземпляра ProcessPoolExecutor. Функция multiprocessing_f должна участвовать в тестировании, принимая необязательный аргумент исполнителя (многопроцессорный пул), который будет использоваться для тестирования:

Файл 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(executor=None):
    """To allow unit testing, we may be passed an executor
    to use."""

    if executor is None:
        executor = ccf.ProcessPoolExecutor()

    result = []
    with 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

Файл test_file_a.py

import concurrent.futures as ccf

import file_a

def init_pool_processes():
    file_a.MY_CONSTANT = "world"


# PASSES:
def test_singleprocess_f(monkeypatch):
    with monkeypatch.context() as m:
        monkeypatch.setattr(file_a, "MY_CONSTANT", "world")
        result = file_a.singleprocess_f()
        assert result == ["world"] * 3

# PASSES:
def test_multithread_f(monkeypatch):
    monkeypatch.setattr(file_a, "MY_CONSTANT", "world")
    result = file_a.multithread_f()
    assert result == ["world"] * 3

# PASSES:
def test_multiprocess_f():
    # Special executor for testing:
    executor = ccf.ProcessPoolExecutor(initializer=init_pool_processes)
    result = file_a.multiprocess_f(executor)
    assert result == ["world"] * 3