【pytest】如何使用 pytest-rerunfailures 插件并自定义重跑操作

作用介绍

pytest-rerunfailures 插件可以使测试用例在执行测试失败的时候重新执行，并且会执行测试用例相对应的 setup 和 teardown 方法。通常有测试用例失败重跑需求的时候就可以使用这个插件。

但是 rerunfailures 插件无法控制用例在失败重跑时进行一些额外操作，这对于有一些自定义需求的业务场景来说很不方便，因此本文主要来介绍如何配合 rerunfailures 插件来进行一些自定义操作。

插件下载：

pip install pytest-rerunfailures

使用方法

• 命令行参数方法
• 装饰器注解方法

命令行参数方法

pytest --reruns 3

只需要在 pytest 的命令行后加上--reruns num即可，num 的值代表测试失败时重跑的次数

装饰器注解方法

@pytest.mark.flaky(reruns=5)
def test_complete_inspect_control():
    assert 1 == True

在测试用例上添加@pytest.mark.flaky(reruns=num)装饰器，num 的值代表测试失败时重跑的次数

自定义重跑配置

rerunfailures 装饰器在测试失败的时候会固定执行用例的 setup、call、teardown 方法，其中 call 方法代表用例本身的执行函数。

因此我们想要自定义用例重跑时的一些操作，可以从两个方面入手。

分别是：

• setup、teardown 方法
• rerunfailures 源代码

setup、teardown 方法

由于 setup 和 teardown 方法在每一个测试用例执行时都会执行一遍，而如果我们的自定义操作只针对于失败重跑的用例，则需要在 setup 和 teardown 方法里加一些判断条件。

我的方法是在测试用例的 class 对象中初始化一个 bool 变量，用来标志当前的测试用例是否需要进行重跑时的自定义操作。即：

class TestCase:
    restart_app = False

这里将 bool 变量命名为 restart_app，因为我的自定义操作是在失败时重启 app。

将这个变量初始化为 False，表示默认情况下当前执行的测试用例是不需要执行自定义操作的。

定义好了变量后，就该编写 setup 函数里的自定义操作的逻辑了。setup 具体逻辑如下：

class TestCaseBase:

    restart_app = False

    def setup_method(self):
        if  self.restart_app:
            # 这里想要自定义的操作
            self.restart_app = Flase # 失败重跑时执行完自定义操作后需要将标志复原

接下来还需要控制这个 restar_app 变量在什么时候变为 True。

pytest 框架有一个钩子函数，名为pytest_runtest_makereport。在这个函数里，每一个执行完毕的测试用例，不论是否成功，都可以在这个函数里获取他的测试结果，因此可以使用这个钩子函数来控制标志的值。

# conftest.py
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_makereport(item, call):
    result = yield
    rep = result.get_result()

    # 获取setup、call、teardown方法的执行失败的结果
    if rep.when in ("call", "setup", "teardown") and rep.failed:
        # 从报告中获取失败的详细信息
        failure_reason = rep.longrepr.reprcrash.message
        # 获取失败的测试用例名称
        test_name = item.name
        # 打印或记录失败信息
        logger.error(f"Test '{test_name}' failed with reason: {failure_reason}")
        if item.cls is not None:  # 确保测试用例属于某个类
            item.cls.restart_app = True # 设置重启标志为True

到这一步就大功告成了，每一次测试用例执行失败的时候，就会在pytest_runtest_makereport钩子函数中捕获到相应的信息，并将该用例的父类的标志对象设置为 True。接下来 rerunfailures 插件就会进行重新执行该用例的 setup、call、teardown 函数。运行 setup 函数时，判断到 restart_app 的值为 True，就会运行我们设置的自定义操作，运行完毕后将 restart_app 的值恢复为默认值。如此便实现了配合 rerunfailures 插件执行自定义重跑操作。

修改 rerunfailures 源代码

在一些需要高度自定义的需求下，单单只配置 setup、teardown 方法可能不够用，因此我们可以修改 rerunfailures 的源代码来实现我们想要的效果。

首先我们要找到 rerunfailures 的源代码文件。在 python 虚拟环境中的路径为：

.venv/Lib/site-packages/pytest_rerunfailures.py

如果使用的是 python 的全局环境，只需要在 python 的安装文件夹下的 lib 库里寻找即可。

打开源文件，找到pytest_runtest_protocol函数，这个函数也是 pytest 框架的钩子函数，不过我们没有必要自己使用这个钩子函数实现一些重跑逻辑，直接在现有的轮子上做增量即可。

这个函数的源代码如下：

def pytest_runtest_protocol(item, nextitem):
    """
    Run the test protocol.

    Note: when teardown fails, two reports are generated for the case, one for
    the test case and the other for the teardown error.
    """
    reruns = get_reruns_count(item)
    if reruns is None:
        # global setting is not specified, and this test is not marked with
        # flaky
        return

    # while this doesn't need to be run with every item, it will fail on the
    # first item if necessary
    check_options(item.session.config)
    delay = get_reruns_delay(item)
    parallel = not is_master(item.config)
    db = item.session.config.failures_db
    item.execution_count = db.get_test_failures(item.nodeid)
    db.set_test_reruns(item.nodeid, reruns)

    if item.execution_count > reruns:
        return True

    need_to_run = True
    while need_to_run:
        item.execution_count += 1
        item.ihook.pytest_runtest_logstart(nodeid=item.nodeid, location=item.location)
        reports = runtestprotocol(item, nextitem=nextitem, log=False)

        for report in reports:  # 3 reports: setup, call, teardown
            report.rerun = item.execution_count - 1
            if _should_not_rerun(item, report, reruns):
                # last run or no failure detected, log normally
                item.ihook.pytest_runtest_logreport(report=report)
            else:
                # failure detected and reruns not exhausted, since i < reruns
                report.outcome = "rerun"
                time.sleep(delay)
                if not parallel or works_with_current_xdist():
                    # will rerun test, log intermediate result
                    item.ihook.pytest_runtest_logreport(report=report)

                # cleanin item's cashed results from any level of setups
                _remove_cached_results_from_failed_fixtures(item)
                _remove_failed_setup_state_from_session(item)

                break  # trigger rerun
        else:
            need_to_run = False

        item.ihook.pytest_runtest_logfinish(nodeid=item.nodeid, location=item.location)

    return True

整段代码看起来很长，但是我们只需要重点关注两个地方：runtestprotocol和for report in reports。

runtestprotocol函数是控制整个测试用例执行 setup、call、teardown 方法的。
而for report in reports则是针对三个方法来分别识别其运行结果的，我们的一些自定义逻辑基本就写在这个循环下面。

通常来说，如果我们需要在测试用例执行前（setup、call、teardwon）执行一些操作的话，则可以在runtestprotocol函数前执行，如：

# do something special
reports = runtestprotocol(item, nextitem=nextitem, log=False)

如果需要在测试用例执行结束后，获取执行结果并进行操作的话，则需要在for report in reports循环下进行操作，如：

for report in reports:  # 3 reports: setup, call, teardown
    if report.outcome = "failed" :
        # do something special
        break
    report.rerun = item.execution_count - 1
    if _should_not_rerun(item, report, reruns):
        # last run or no failure detected, log normally
        item.ihook.pytest_runtest_logreport(report=report)
    else:
        # failure detected and reruns not exhausted, since i < reruns
        report.outcome = "rerun"
        time.sleep(delay)
        if not parallel or works_with_current_xdist():
            # will rerun test, log intermediate result
            item.ihook.pytest_runtest_logreport(report=report)

        # cleanin item's cashed results from any level of setups
        _remove_cached_results_from_failed_fixtures(item)
        _remove_failed_setup_state_from_session(item)

        break  # trigger rerun
else:
    need_to_run = False

这里只是加了一个无论设置了多少次重跑次数，只要执行失败（setup、call、teardown）都重新执行的逻辑。

通常大部分的需求会有一些更多的、额外的一些逻辑，需要大家自行处理。