Skip to content

Instantly share code, notes, and snippets.

@tuantranf
Last active January 2, 2023 22:41
Show Gist options
  • Save tuantranf/c4c22eec108285a7b3d1dde5528edbe6 to your computer and use it in GitHub Desktop.
Save tuantranf/c4c22eec108285a7b3d1dde5528edbe6 to your computer and use it in GitHub Desktop.
mocking multiprocess pool for unittest

Mocking multiprocess pool apply_async method

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

module code

pool = Pool(cpu_count())
pool.apply_async(func=myfunc)

Pytest

reference:

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():
    ...
      
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment