如何重启twisted.trial._dist.disttrial.DistTrialRunner并重跑失败用例

问题原因 #

• twisted.trial._dist.disttrial.DistTrialRunner 属于Twisted内部非公开模块，本身没有设计为可多次实例化、多次调用run的场景
• 首次调用run()方法后，底层的Twisted反应器（reactor）会被终止，且全局状态没有重置，第二次实例化运行时会因为reactor状态异常抛出错误
• 原代码中每次循环都新建Runner实例也无法规避reactor全局状态污染的问题
• 原代码对run()方法的返回值解析不符合实际接口定义，DistTrialRunner的run()返回的是IReporter实例，不会直接返回状态码和失败用例列表

解决方法 #

方案1：重置reactor状态实现重试（适合代码调用场景） #

在两次运行之间手动重置reactor全局状态，同时正确提取失败用例构造新的测试集，最多重试3次的实现代码如下：

from twisted.trial._dist.disttrial import DistTrialRunner as Runner
from twisted.internet import reactor

max_retry = 3
current_run = 0
current_suite = suite  # 你的初始测试集

while current_run < max_retry:
    current_run += 1
    runner = Runner(workerNumber=3, workerArguments=[])
    run_result = runner.run(current_suite)
    if run_result.wasSuccessful():
        exit(0)
    # 提取失败、错误用例构造新的测试集
    current_suite = type(current_suite)()
    for test_case, _ in run_result.failures + run_result.errors:
        current_suite.addTest(test_case)
    # 重置reactor全局状态，消除上一次运行的残留影响
    reactor.__init__()

exit(1)

方案2：独立进程隔离运行（适合稳定性要求高的场景） #

如果重置reactor的方案仍有兼容问题，可以将每次测试运行放在独立子进程中执行，完全隔离运行状态，避免全局状态污染：

import subprocess
import sys

test_target = "your.test.module.path"  # 替换为实际测试目标路径
for _ in range(3):
    ret = subprocess.run(
        [sys.executable, "-m", "twisted.trial", "--jobs", "3", test_target],
        capture_output=False
    )
    if ret.returncode == 0:
        exit(0)
exit(1)

注意事项 #

• 非公开模块_dist.disttrial.DistTrialRunner的API可能在不同Twisted版本中发生变化，优先使用trial命令行工具内置的--retry参数实现失败用例重跑，稳定性更高
• 重置reactor的操作属于测试场景专用的非常规操作，禁止在生产服务代码中使用

内容的提问来源于stack exchange，提问作者hhh