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