Skip to content
Advertisement

patch object spawned in sub-processs

I am using multiprocessing package for crating sub-processes. I need to handle exceptions from sub-process. catch, report, terminate and re-spawn sub-process. I struggle to create test for it.

I would like to patch object which represent my sub-process and raise exception to see if handling is correct. But it looks like that object is patched only in main process and in spawned process is unchanged version. Any ideas how to accomplish requested functionality?

Example:

import multiprocessing
import time

class SubprocessClass(multiprocessing.Process):
    def __init__(self) -> None:
        super().__init__()

    def simple_method(self):
        return 42

    def run(self):
        try:
            self.simple_method()
        except Exception:
            # ok, exception handled
            pass
        else:
            # I wanted exception ! <- code goes here
            assert False

@mock.patch.object(SubprocessClass, "simple_method")
def test_patch_subprocess(mock_simple_method):
    mock_simple_method.side_effect = Exception("exception from mock")
    subprocess = SubprocessClass()
    subprocess.run()
    subprocess.start()
    time.sleep(0.1)
    subprocess.join()

Advertisement

Answer

you can monkey-patch the object before it is started (it is a bit iffy but you will get actual process running that code)

def _this_always_raises(*args, **kwargs):
    raise RuntimeError("I am overridden")


def test_patch_subprocess():
    subprocess = SubprocessClass()

    subprocess.simple_method = _this_always_raises

    subprocess.start()
    time.sleep(0.1)
    subprocess.join()
    assert subprocess.exitcode == 0

you could also mock multiprocessing to behave like threading but that is a bit unpredictable

if you want to do it genericly for all objects you can mock the class for another one derived from the original one with only one method overriden

class SubprocessClassThatRaisesInSimpleMethod(SubprocessClass):
    def simple_method(self):
        raise RuntimeError("I am overridden")

# then mock with unittest mock the process spawner to use this class instead of SubprocessClass
User contributions licensed under: CC BY-SA
6 People found this is helpful
Advertisement