Как издеваться за подпроцессы, порожденные ProcessPoolexecutor в Python (здесь: Elasticsearch.elasticsearch)? ⇐ Python

[Anonymous]

Я пытаюсь имитировать класс elasticsearch.Elasticsearch, используя unittest.mock.patch в тесте, но макет не применяется к подпроцессам, созданным ProcessPoolExecutor. Вот код, с которым я работаю:
main.py

Код: Выделить всё

import concurrent.futures
import elasticsearch

class ThreadTest:
def __init__(self):
pass

def process(self):
futures = []
with concurrent.futures.ProcessPoolExecutor(max_workers=5) as executor:
future = executor.submit(
self.just_es,
)
futures.append(future)

concurrent.futures.wait(fs=futures)
for future in futures:
result = future.result()

def just_es(self):
print(elasticsearch.Elasticsearch)  # I want this to be mocked
return "hey"

test.py

Код: Выделить всё

import elasticsearch
import pytest
from main import ThreadTest

class MockElasticsearchClient:
def __init__(self):
pass

@property
def meta(self):
return

@pytest.fixture(autouse=True)
def mock_es(monkeypatch):
def fake_es_client_6_for_run_query(*args, **kwargs):
return MockElasticsearchClient()

monkeypatch.setattr(
elasticsearch,
"Elasticsearch",
fake_es_client_6_for_run_query,
)

def test_thread(monkeypatch, mock_es):
tt = ThreadTest()
tt.just_es()  # This works
tt.process()  # This doesn't work, Elasticsearch is not mocked in subprocesses

Выход:

Код: Выделить всё

 # output by just_es()

 # output by process which internally calls just_es()

Мок работает для основного процесса, но не для подпроцессов, порожденных ProcessPoolExecutor. Я считаю, что это связано с тем, что Monkeypatch применяется только в основном процессе, но мне нужно, чтобы макет распространялся на все подпроцессы.
Что еще я пробовал:
Я пробовал использовать инициализатор с ProcessPoolExecutor для применения макета, но макет по-прежнему не применяется в подпроцессах.

Подробнее здесь: https://stackoverflow.com/questions/793 ... n-here-ela