子进程中生成的补丁对象

Question

我正在使用 multiprocessing 包来创建子流程。我需要处理子流程的异常。捕获、报告、终止和重新生成子进程。我很难为它创建测试。

我想修补代表我的子流程的对象并引发异常以查看处理是否正确。但看起来该对象仅在主进程中进行了修补，而在派生进程中是未更改的版本。对如何完成请求的功能有任何想法吗？

示例：

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()

Answer 1

您可以在对象启动之前对其进行猴子修补（这有点不确定，但你会得到实际过程运行该代码）

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

您也可以模拟多处理使其表现得像线程，但这有点不可预测

如果你想对所有对象进行通用操作，你可以模拟 class 另一个从原始对象派生的对象，只覆盖一个方法

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

子进程中生成的补丁对象

patch object spawned in sub-processs

python

multiprocessing

pytest