It seems python multiprocessing interface relies on pickling, and mock library and pickling don't go well together.
write mock of multiprocess class for pytest
pool = Pool(cpu_count())
pool.apply_async(func=myfunc)
reference:
- https://stackoverflow.com/questions/26094560/python-class-inheriting-multiprocessing-mocking-one-of-the-class-objects
- https://stackoverflow.com/questions/33128681/how-to-unit-test-code-that-uses-python-multiprocessing
from unittest.mock import patch
from pytest import fixture
class MockPoolApplyResult:
def __init__(self, func, args):
self._func = func
self._args = args
def get(self, timeout=0):
return self._func(*self._args)
@fixture(autouse=True)
def mock_pool_apply_async(monkeypatch):
monkeypatch.setattr("multiprocessing.pool.Pool.apply_async",
lambda self, func, args=(), kwds={}, callback=None, error_callback=None:
MockPoolApplyResult(func, args))
class TestClass:
def test_method():
...