首页 > 解决方案 > 用于复杂多处理和多线程模糊测试的 Python 测试工具

问题描述

的背景:

我的团队需要测试一个分布式协议实现(例如:RAFT算法),该实现是基于一个rpc层(例如:grpc),我们基于multiprocessing库提供的utils实现了一个mocked rpc层,我们可以进行操作如reorder、drop、delay来模拟真实的网络环境。该mock库用于注入模拟函数。我们在分布式协议中生成子进程作为代理。

当前测试工具:

我们使用 pytest 作为我们的主要框架,pytest 非常适合单线程测试,但不适合这种情况,因为异常无法在测试函数产生的线程/进程中捕获。我们尝试创建可以捕获异常并将异常发送回并重新引发的运行器:

result = [None, None, None]
    finished = [False, False, False]
    while not all(finished):
        for p, pi, i in zip(procs, proc_pipes, [0, 1, 2]):
            if pi.poll(timeout=1e-3):
                status, r = pi.recv()
                if status:
                    finished[i] = True
                    result[i] = r
                else:
                    raise r
    return result

但是这种方法仅能捕获测试过程中创建的代理进程抛出的异常(例如:created with pytest.fixture),并不能监控算法产生的后台线程中抛出的异常。

一些要求:

如果有专门用于测试分布式程序的pytest以外的工具,那将是最好的。

如果不存在这样的工具,那么我们希望找到一个 pytest 插件,它可以监视在测试执行期间产生的线程和进程,最好带有过滤功能以忽略某些异常。

如果这不可能,那么如果在任何线程/进程中抛出任何异常,它应该立即终止测试,我们将使用其他日志记录工具来捕获执行过程。

感谢您的慷慨帮助!

标签: pythonmultithreadingtestingmultiprocessingpytest

解决方案


也许pytest-reraise可以在这里帮助你?它提供了一个上下文管理器夹具来捕获来自其他线程的异常,并在捕获到异常时使测试用例失败。这是一个最小的示例测试用例,它会失败并提供有关断言错误的详细信息:

from threading import Thread

def test_assert(reraise):

    def run():
        with reraise:
            assert False

    Thread(target=run).start()

推荐阅读