自动化测试中的稳定性治理实战：定位并消除 Flaky Test 的方法与工具链设计

自动化测试最怕的，不是“测出 bug”，而是“今天红、明天绿，后天又红”。
这类测试我们通常叫 Flaky Test：在代码没有实质变化的情况下，测试结果不稳定、可重复性差。

如果团队里 Flaky Test 比例高，后果往往不是“偶尔烦一下”那么简单，而是：

• CI 红灯失去信任，开发看到失败先怀疑测试
• 合并效率下降，排队重跑成为日常
• 真问题被淹没，线上风险被放大
• 测试团队疲于救火，无法把精力投到更高价值的质量建设上

我自己在做测试平台和 CI 治理时，踩过一个很典型的坑：团队一开始把 Flaky Test 当成“测试代码写得不严谨”的局部问题处理，结果修了几个用例后，整体改善很有限。后来才发现，它其实更像一个 系统性问题：和测试设计、环境隔离、数据治理、调度机制、失败归因、重试策略都有关。

这篇文章不只讲“为什么会 flaky”，而是从架构治理的角度，带你搭一套可落地的思路：如何识别、定位、度量、止血，并逐步把 Flaky Test 纳入团队的工程闭环。

背景与问题

什么样的测试算 Flaky Test

一个直观定义是：

在相同代码版本、相近环境和相同输入下，多次执行结果不一致的测试，就是 Flaky Test。

它可能表现为：

• 同一个 PR 上第一次失败，重跑又通过
• 只在夜间批量任务失败，白天手工跑没问题
• 在某些机器上稳定，在另一些机器上随机挂掉
• 单独运行通过，放到整套回归里就失败

Flaky Test 的常见来源

我一般把来源拆成四类，便于治理时分层处理。

1. 时间与并发问题

   • 固定 sleep 等待异步完成
   • 定时任务、时区、夏令时、系统时间漂移
   • 竞态条件、资源争用、线程调度不确定

2. 环境与依赖问题

   • 测试环境不隔离，共享数据库/缓存/消息队列
   • 第三方服务不稳定
   • 网络抖动、容器冷启动、端口冲突

3. 数据与状态污染

   • 用例间共享数据，前一个测试影响后一个
   • 脏数据未清理
   • 随机数据不可复现，seed 未固定

4. 测试设计问题

   • 断言过于脆弱，比如依赖排序、文案、精确时间
   • UI 测试直接依赖瞬态 DOM 状态
   • 过度集成，一个测试验证了太多环节

为什么“重跑就好了”不是治理

很多团队会先上一个简单策略：失败自动重试 1~3 次。这个动作有价值，但它只能算止血，不是根治。

因为纯重试会带来三个副作用：

• 掩盖真实问题：偶发失败的根因被埋掉
• 拉长 CI 时间：重试越多，反馈越慢
• 扭曲质量指标：通过率看起来更高，实际可信度更差

所以更合理的做法是：

• 短期：重试兜底，避免阻塞主流程
• 中期：记录重试轨迹，识别可疑用例
• 长期：建立失败归因与治理机制，减少 flaky 存量和增量

核心原理

稳定性治理不是一个脚本，而是一条链路。核心思路可以概括成四步：

1. 识别：哪些测试疑似 flaky
2. 分类：失败发生在哪一层
3. 归因：为什么会不稳定
4. 治理：怎么修、怎么防止再长出来

治理闭环架构

flowchart LR
    A[测试执行] --> B[结果采集]
    B --> C[失败重试与轨迹记录]
    C --> D[Flaky 识别引擎]
    D --> E[归因分类]
    E --> F[治理工作台]
    F --> G[修复/隔离/降级]
    G --> H[规则沉淀]
    H --> A

上图的关键点在于：
不要把 flaky 处理停留在“看日志、手工重跑”阶段，而是要让测试执行平台天然产出治理数据。

识别：如何判断一个测试“疑似 flaky”

一个常用工程规则是：

• 首次失败
• 在相同代码版本上重跑后通过
• 且近期有多次类似波动记录

这时就可以打上“疑似 flaky”标签。

更进一步，可以维护一些指标：

• Fail Rate：失败率
• Retry Pass Rate：重试通过率
• Flaky Score：基于最近 N 次执行结果计算的不稳定指数
• MTTR-Test：测试修复平均耗时
• Top Flaky Cases：波动最大的前若干用例

一个简单的打分思路：

Flaky Score = 重试转绿次数 / 总执行次数 * 权重
            + 环境相关失败比例 * 权重
            + 最近波动频次 * 权重

不是为了做学术模型，而是为了排优先级。

分类：按层次分桶，别把所有失败混在一起

我建议把失败先分层：

• 测试代码层：断言、等待方式、随机数据
• 应用层：接口超时、异步未完成、锁冲突
• 依赖层：数据库、缓存、MQ、外部 API
• 环境层：CPU 抢占、网络抖动、容器资源不足
• 平台层：调度器、并发执行器、日志采集缺失

这样做的价值很大：
开发、测试、SRE 三方看到同一条失败时，知道该归谁先看，而不是相互转单。

归因：稳定性问题的典型模式

下面这张图是我比较常用的归因模式图。

classDiagram
    class FlakyTest {
      +name
      +suite
      +history
      +retryTrace
      +envInfo
    }

    class RootCause {
      <<abstract>>
      +category
      +symptom
    }

    class TimingIssue {
      +fixedSleep
      +asyncRace
      +clockDrift
    }

    class DataPollution {
      +sharedData
      +dirtyState
      +orderDependence
    }

    class EnvInstability {
      +networkJitter
      +resourceStarvation
      +dependencyUnavailable
    }

    class TestDesignIssue {
      +fragileAssertion
      +randomInput
      +overIntegration
    }

    FlakyTest --> RootCause
    RootCause <|-- TimingIssue
    RootCause <|-- DataPollution
    RootCause <|-- EnvInstability
    RootCause <|-- TestDesignIssue

治理：分三层动作

1. 止血动作

适合 CI 已经被大量 flaky 拖垮的团队。

• 失败自动重试 1 次
• 高风险 flaky 用例隔离到非阻塞流水线
• 给可疑用例打标，不计入主质量门禁
• 收集失败现场：日志、截图、trace、环境信息

2. 修复动作

真正减少存量。

• 去掉固定 sleep，改显式等待
• 让测试数据独立、可回收、可追踪
• 隔离外部依赖，用 stub/mock 或契约测试替代
• 固定随机种子
• 清理跨用例共享状态

3. 预防动作

防止增量继续出现。

• 测试编码规范
• Flaky 检测任务进入 nightly
• PR 阶段新增用例必须通过稳定性检查
• 测试平台自动识别“首失败重试转绿”的案例并建单

方案对比与取舍分析

治理 Flaky Test，常见有三条路线。

路线一：纯人工排查

做法：失败后由测试/开发看日志、手工复现、改用例。
优点：启动成本低。
缺点：规模一大就崩，依赖经验，无法形成组织资产。

适用场景：

• 用例量不大
• 团队刚开始重视稳定性
• 还没有统一 CI 平台能力

路线二：平台重试 + 报表统计

做法：执行平台自动重试，并记录失败/重跑结果，出可疑用例榜单。
优点：能快速止血，也能沉淀数据。
缺点：如果没有归因机制，容易变成“统计很好看，问题还在”。

适用场景：

• 中等规模团队
• CI 已经统一
• 想先把问题可视化

路线三：稳定性治理平台化

做法：从执行、采集、归因、工单、治理规则到质量门禁形成闭环。
优点：长期收益最高，可持续。
缺点：建设成本高，需要测试平台、CI、日志系统协作。

适用场景：

• 多团队共享测试基础设施
• 回归套件大、执行成本高
• 需要管理层看到明确质量指标

一个务实建议

如果你现在团队问题很多，不要一口气搞“大而全平台”。
更现实的顺序通常是：

1. 先统一执行入口
2. 再加失败重试和轨迹记录
3. 再做可疑用例识别
4. 再做归因分桶和治理流程

实战代码（可运行）

下面我用 Python 做一个最小可运行的治理示例，演示三件事：

1. 模拟测试执行结果
2. 识别“首失败、重跑通过”的 flaky 用例
3. 输出一个简单报告

这不是完整平台，但它很好地体现了工具链设计的最小闭环。

示例 1：Flaky 识别脚本

import random
import time
from collections import defaultdict
from dataclasses import dataclass, asdict

random.seed(42)

@dataclass
class TestResult:
    name: str
    build_id: str
    attempt: int
    status: str  # passed / failed
    duration_ms: int
    env: str
    error_type: str = ""

TESTS = [
    "test_login",
    "test_create_order",
    "test_refund",
    "test_search",
    "test_user_profile"
]

def run_test_case(name: str, env: str) -> TestResult:
    duration = random.randint(50, 500)

    # 模拟不同类型失败
    flaky_cases = {
        "test_search": 0.35,         # 典型 flaky
        "test_user_profile": 0.15    # 轻度不稳定
    }
    stable_fail_cases = {
        "test_refund": 0.80          # 大概率真实失败
    }

    if name in flaky_cases and random.random() < flaky_cases[name]:
        return TestResult(name, "", 0, "failed", duration, env, "TimeoutError")

    if name in stable_fail_cases and random.random() < stable_fail_cases[name]:
        return TestResult(name, "", 0, "failed", duration, env, "AssertionError")

    return TestResult(name, "", 0, "passed", duration, env)

def execute_build(build_id: str, env: str = "ci-worker-1", retry: int = 1):
    results = []
    for test_name in TESTS:
        first = run_test_case(test_name, env)
        first.build_id = build_id
        first.attempt = 1
        results.append(first)

        if first.status == "failed":
            for i in range(retry):
                rerun = run_test_case(test_name, env)
                rerun.build_id = build_id
                rerun.attempt = i + 2
                results.append(rerun)
                if rerun.status == "passed":
                    break
    return results

def detect_flaky(results):
    grouped = defaultdict(list)
    for r in results:
        grouped[(r.build_id, r.name)].append(r)

    flaky_report = []
    for (build_id, name), items in grouped.items():
        items = sorted(items, key=lambda x: x.attempt)
        first = items[0]
        later_pass = any(x.status == "passed" for x in items[1:])

        if first.status == "failed" and later_pass:
            flaky_report.append({
                "build_id": build_id,
                "test_name": name,
                "attempts": len(items),
                "first_error": first.error_type,
                "final_status": "passed_after_retry"
            })
    return flaky_report

def summarize(results):
    total = len([r for r in results if r.attempt == 1])
    first_pass = len([r for r in results if r.attempt == 1 and r.status == "passed"])
    first_fail = total - first_pass
    retry_count = len([r for r in results if r.attempt > 1])

    print("=== 执行摘要 ===")
    print(f"首轮总用例数: {total}")
    print(f"首轮通过数 : {first_pass}")
    print(f"首轮失败数 : {first_fail}")
    print(f"重试次数   : {retry_count}")
    print()

if __name__ == "__main__":
    all_results = []
    for i in range(1, 6):
        build_id = f"build-{i}"
        results = execute_build(build_id, retry=2)
        all_results.extend(results)

    summarize(all_results)
    flaky = detect_flaky(all_results)

    print("=== Flaky Report ===")
    for item in flaky:
        print(item)

    print(f"\n疑似 Flaky 数量: {len(flaky)}")

运行效果说明

这段代码会生成多次构建记录，并识别出：

• 首次失败
• 重试后通过

的测试用例。它的价值不在于“算法高级”，而在于告诉我们一个事实：

只要执行平台保留了构建号、测试名、尝试次数、环境、错误类型这些基础字段，后面的 flaky 治理能力就能逐步搭起来。

示例 2：一个更靠谱的等待方式

很多 flaky 都来自固定等待。比如下面这种写法：

import time

def test_async_job_bad():
    trigger_job()
    time.sleep(2)  # 坑点：异步任务不一定 2 秒完成
    assert query_job_status() == "done"

更稳妥的方式是轮询 + 超时控制：

import time

def wait_until(condition_fn, timeout=5, interval=0.2):
    start = time.time()
    while time.time() - start < timeout:
        if condition_fn():
            return True
        time.sleep(interval)
    return False

job_done = False

def trigger_job():
    global job_done
    job_done = False
    # 模拟异步完成
    def complete():
        global job_done
        time.sleep(1)
        job_done = True

    import threading
    threading.Thread(target=complete).start()

def query_job_status():
    return "done" if job_done else "running"

def test_async_job_good():
    trigger_job()
    assert wait_until(lambda: query_job_status() == "done", timeout=3), "job not done in time"

if __name__ == "__main__":
    test_async_job_good()
    print("test passed")

这个例子很基础，但非常有代表性：
固定 sleep 往往是 flaky 的温床。

工具链设计：从执行到治理的落地形态

如果把它做成一条实际可用的工具链，我建议至少包括下面几个组件。

1. 测试执行器

负责：

• 执行测试
• 控制重试次数
• 记录 attempt 级别结果
• 采集日志、截图、trace、容器信息

2. 结果采集与存储

建议至少存这些字段：

这里的 env_hash 很有用。它可以把镜像版本、依赖版本、系统信息聚合成一个环境签名，帮助你看出问题是不是集中出现在某类机器或镜像上。

3. Flaky 识别引擎

最小版本可以按规则判断：

• fail -> pass_after_retry
• 单测独立跑通过，套跑失败
• 特定 worker 高失败率
• 近期失败集中在某时间窗口

4. 归因工作台

这里不用一开始追求 AI 化，规则化先做起来就很值钱。

例如：

• TimeoutError + 耗时接近超时阈值 -> 等待/性能问题
• 仅特定 worker 失败 -> 环境问题
• 顺序变了就失败 -> 共享状态污染
• 只在并发执行时失败 -> 竞态/资源争用

5. 治理流程与质量门禁

建议把治理动作明确分级：

• P0：阻塞主干、重试也不稳，立即修
• P1：重试可过但高频波动，本周内治理
• P2：低频不稳定，纳入专项清理
• P3：暂时隔离，不阻塞发布但持续观察

容量估算与工程边界

做平台时，经常有人忽略“数据量”。Flaky 治理的数据不是特别复杂，但量可能不小。

假设：

• 每天 300 次构建
• 每次 2000 个测试
• 平均失败重试 5%
• 每个结果记录 1 KB

粗略估算：

• 首轮记录：300 × 2000 = 60 万条/天
• 重试记录：60 万 × 5% ≈ 3 万条/天
• 总量约 63 万条/天
• 按 1 KB 计，约 600+ MB/天，仅结果元数据

如果再加上：

• 日志
• 截图
• 浏览器 trace
• 视频

存储压力会上一个量级。

因此工程上应考虑：

• 元数据进关系型数据库或检索引擎
• 大对象进对象存储
• 原始日志按时间淘汰
• 聚合指标长期保留，明细短期保留

边界条件也要明确：

• 不是所有 flaky 都值得修，低价值、即将下线的用例可以隔离
• 不是所有失败都该重试，真实失败高概率用例应少重试
• UI 端到端测试天然比单元测试更易波动，指标不能一刀切

常见坑与排查

这一节我尽量讲得接地气一点，都是现场经常遇到的。

坑 1：把所有失败都自动重试三次

看上去“通过率提高了”，实际上：

• CI 变慢
• 真实问题延迟暴露
• 团队对失败失去敏感度

建议：

• 只对可疑 flaky 类别启用有限重试
• 保留首次失败结果，不要用最终绿灯覆盖原始状态
• 报表里区分“首次通过”和“重试后通过”

坑 2：日志太少，根本无法归因

很多平台只存最终状态：pass / fail。
这种数据对治理几乎没帮助。

建议至少采集：

• 错误类型
• 栈信息摘要
• 执行节点
• 环境版本
• 开始/结束时间
• 首次失败截图或 trace

坑 3：共享测试环境，彼此污染

典型表现：

• 单跑过，套跑挂
• 某个用户数据反复被抢占
• 用例顺序改变后结果不同

排查路径：

1. 对失败套件做随机顺序执行
2. 对失败用例做独立进程/独立库重跑
3. 检查是否存在固定账号、固定订单号、固定主键

坑 4：用固定 sleep 掩盖异步问题

这是最常见的“看起来能跑，实际上不稳”。

排查方式：

• 查测试代码里是否大量出现 sleep(1)、sleep(3)
• 对失败样本看耗时分布，是否接近阈值
• 看机器负载升高时失败率是否显著上升

坑 5：误把环境抖动当测试问题

有时测试本身没问题，真正有问题的是：

• 某批 worker CPU 被抢占
• 网络出口抖动
• 容器镜像版本漂移

建议：

• 报表按 worker / image / region 分组
• 引入环境健康度指标，与测试失败率联动分析

安全/性能最佳实践

稳定性治理不只是质量问题，也和安全、性能治理有交集。

安全最佳实践

1. 测试数据脱敏

不要为了复现 flaky，直接把生产敏感数据复制到测试环境。

建议：

• 构造最小化测试数据
• 用户信息、手机号、身份证等字段脱敏
• 日志里避免输出 access token、cookie、密钥

2. 外部依赖凭证最小权限

测试平台常常会访问数据库、对象存储、第三方服务。

建议：

• 使用只读或最小权限账号
• 凭证通过密钥管理系统注入
• 避免把凭证打印进失败日志

3. 隔离失败现场采集范围

截图、trace、HAR 文件可能包含敏感页面信息。

建议：

• 对采集内容设置保留期
• 按角色控制访问
• 只保留定位所需的最小数据

性能最佳实践

1. 控制重试预算

重试不是免费的。
可以按套件和优先级设置预算：

• 核心链路：最多重试 1 次
• 非核心回归：允许 2 次
• 明确高风险真实失败类：不重试

2. 将 flaky 治理与并发调度联动

如果某类测试对资源很敏感，不要盲目拉高并发。

可采用：

• 按资源类型分池
• 高干扰用例串行
• 重型 UI 测试单独 worker 池

3. 做趋势，不只看单次构建

单次失败往往说明不了太多。更有价值的是：

• 最近 7 天 flaky 排行
• 按环境维度趋势对比
• 修复前后波动下降情况

一个推荐的落地路径

如果你正准备在团队里推动这件事，我建议按下面顺序做，成功率更高。

sequenceDiagram
    participant Dev as 开发
    participant CI as CI平台
    participant Collector as 结果采集器
    participant Analyzer as Flaky分析器
    participant Board as 治理看板

    Dev->>CI: 提交代码触发流水线
    CI->>CI: 执行测试并按策略重试
    CI->>Collector: 上报 attempt 级结果与日志
    Collector->>Analyzer: 聚合历史执行记录
    Analyzer->>Board: 标记疑似 flaky 与归因类别
    Board->>Dev: 指派修复/隔离/观察

第一步：统一“测试结果数据模型”

先别急着搞识别算法，先保证所有测试结果能结构化上报。

第二步：上线最小重试与可疑识别

规则很简单也没关系，关键是把“失败后又通过”的案例找出来。

第三步：建立 Top N 清单

每周看一次：

• 波动最大用例
• 最常见错误类型
• 最不稳定 worker
• 修复 SLA

第四步：把修复经验沉淀成规则

例如：

• 禁止固定 sleep
• 测试数据必须带唯一前缀
• 用例不得依赖执行顺序
• 外部依赖优先 stub 化

第五步：纳入质量门禁

对新增测试做稳定性约束，而不是只清理老问题。

总结

Flaky Test 难治，不是因为“测试写得差”，而是它本质上是一个跨测试、应用、环境、平台的系统性问题。

真正有效的治理，不是单纯多重跑几次，而是建立一条完整闭环：

• 执行时保留 attempt 级数据
• 通过重试轨迹识别疑似 flaky
• 按层次做失败分类与归因
• 用止血、修复、预防三层动作持续收敛
• 把经验沉淀成平台规则和测试规范

如果只给一个最可执行的建议，那就是：

从今天开始，不要再只记录“这个测试失败了”，而要记录“它第几次失败、在哪里失败、为何失败、重跑后怎样”。

这一步一旦做起来，Flaky Test 就不再是“玄学问题”，而会变成一个可以度量、可以分派、可以持续优化的工程问题。

最后补一句边界条件：
别追求把所有 flaky 一次性清零。 更现实的目标是先降低最影响交付效率和团队信任的那一批，再逐步建立预防机制。稳定性治理，本来就是一场长期工程，而不是一次专项清理。