自动化测试中的稳定性治理实战：从脆弱用例识别到持续反馈闭环搭建

自动化测试做久了，很多团队都会遇到一个很现实的问题：不是没有测试，而是测试“不可信”。

我见过不少项目，CI 流水线每天都在跑，失败率也不低，但大家点开报告后的第一反应不是“是不是代码坏了”，而是“这次又是哪条老毛病用例炸了”。一旦团队形成这种心理预期，自动化测试的价值就开始快速缩水：失败告警没人第一时间处理，发布前还要手工回归兜底，最后自动化沦为“看起来很忙”。

这篇文章不讲太多空泛原则，而是从一个更实战的角度，带你搭一个稳定性治理闭环：先识别脆弱用例，再给出治理优先级，最后把结果持续反馈到研发流程里。

背景与问题

所谓“脆弱用例”（Flaky Test），通常指的是：

同样的代码、同样的环境，测试结果却偶现失败
失败原因不稳定，可能与时间、环境、数据、依赖服务状态有关
重试后经常恢复通过

它的危害不只是“烦”，而是会带来一连串连锁反应：

误报变多：研发难以区分真实缺陷和噪音失败
流水线变慢：重试、复跑、人工确认都会拉长反馈时间
信任下降：团队对自动化测试失去信心
隐性成本上升：排查时间、环境维护成本、发布风险同步增加

很多团队一开始会选择“先加重试”，这在短期内确实能止血，但如果没有治理闭环，重试只是在延迟问题暴露。

前置知识与环境准备

本文示例采用 Python，目的是把治理思路讲清楚，不依赖某个特定测试平台。你可以很容易映射到 Jenkins、GitLab CI、GitHub Actions 或内部平台。

你需要具备的基础

知道自动化测试和 CI 的基本概念
能读懂 Python 基础语法
理解测试报告中“通过/失败/耗时/重试次数”等常见字段

本文示例环境

Python 3.10+
无第三方依赖也可运行
示例数据使用本地 JSON 文件模拟

核心原理

稳定性治理不要一上来就“全量修测试”，否则很容易陷入两种局面：

范围太大，没人知道先改哪条
修完没有持续跟踪，过几周又回到原样

更有效的方法是把治理拆成四步：

采集运行数据
识别脆弱用例
按影响度排序治理
构建持续反馈闭环

可以把它理解成一个质量运营系统，而不只是测试脚本优化。

1. 采集什么数据

至少要有这些字段：

用例名称
所属模块
构建编号 / 提交版本
执行结果：pass / fail
执行耗时
是否重试
失败原因摘要
执行环境：分支、浏览器、机器、区域等

如果这些数据都没有，后面“脆弱识别”基本无从谈起。

2. 怎么判断一条用例脆弱

常见判断信号有：

失败后重跑通过
最近 N 次执行通过率显著波动
失败原因分布发散
强依赖外部时序 / 网络 / 异步回调
同模块其他用例稳定，唯独它经常抖动

注意一个边界：
不是所有失败率高的用例都叫脆弱。
如果某条用例稳定失败，那更可能是产品缺陷、脚本缺陷或断言过期，不属于 flaky，而是“确定性失败”。

3. 为什么要做优先级排序

治理不是平均用力。优先修复的应当是：

高频失败
阻断主干合并 / 发布
影响关键业务链路
排查成本高
重试掩盖真实问题严重

一个简单的优先级公式可以是：

治理优先级 = 失败频次 × 影响范围 × 恢复成本

4. 闭环的关键不是“发现问题”，而是“问题能流回流程”

一条脆弱用例被识别出来后，理想动作应包括：

自动打标：flaky / infra / product bug / script bug
自动通知责任人或模块群
自动生成趋势图
达到阈值后进入治理池
修复后继续观察，避免反复回潮

稳定性治理总体流程

flowchart TD
    A[CI 执行测试] --> B[采集测试结果与日志]
    B --> C[计算通过率/波动率/重试恢复率]
    C --> D{是否疑似脆弱用例}
    D -- 否 --> E[按正常失败处理]
    D -- 是 --> F[进入脆弱用例池]
    F --> G[按影响度排序]
    G --> H[分派治理责任人]
    H --> I[修复脚本/环境/数据/依赖]
    I --> J[持续观察回归效果]
    J --> C

这张图里最容易被忽略的是 J --> C。
很多团队会做到“识别”和“分派”，但没有“修复后观察”，结果同一问题被反复提单、反复打回，大家最后都疲了。

核心原理拆解：脆弱用例识别模型

为了让过程更可执行，我们用一个轻量模型来识别脆弱用例：

通过率（pass_rate）：最近 N 次通过比例
恢复率（retry_recovery_rate）：失败后经重试转通过的比例
波动指数（volatility）：结果在 pass/fail 之间来回切换的频度
错误分散度（error_diversity）：失败原因是否多样化

一个实用经验是：

通过率低但始终失败：更像真实问题
通过率不算太低，但切换频繁、重试恢复率高：更像脆弱问题

stateDiagram-v2
    [*] --> Stable
    Stable --> SuspectedFlaky: 结果波动升高
    SuspectedFlaky --> ConfirmedFlaky: 多次重试恢复/跨构建反复出现
    SuspectedFlaky --> RealDefect: 持续稳定失败
    ConfirmedFlaky --> Fixing: 进入治理池
    Fixing --> Observing: 修复后观察
    Observing --> Stable: 连续稳定通过
    Observing --> ConfirmedFlaky: 再次波动

实战代码（可运行）

下面我们从零开始做一个简单版本：

准备测试执行历史数据
计算脆弱指标
输出治理候选列表
模拟闭环告警

第一步：准备示例数据

保存为 test_history.json：

[
  {"test_name": "test_login", "module": "auth", "build_id": 101, "status": "pass", "retried": false, "duration": 1.2, "error": ""},
  {"test_name": "test_login", "module": "auth", "build_id": 102, "status": "fail", "retried": true, "duration": 1.3, "error": "Timeout waiting for element"},
  {"test_name": "test_login", "module": "auth", "build_id": 103, "status": "pass", "retried": false, "duration": 1.1, "error": ""},
  {"test_name": "test_login", "module": "auth", "build_id": 104, "status": "fail", "retried": true, "duration": 1.4, "error": "Element not clickable"},
  {"test_name": "test_login", "module": "auth", "build_id": 105, "status": "pass", "retried": false, "duration": 1.0, "error": ""},

  {"test_name": "test_create_order", "module": "order", "build_id": 101, "status": "fail", "retried": false, "duration": 2.8, "error": "AssertionError: total mismatch"},
  {"test_name": "test_create_order", "module": "order", "build_id": 102, "status": "fail", "retried": false, "duration": 2.6, "error": "AssertionError: total mismatch"},
  {"test_name": "test_create_order", "module": "order", "build_id": 103, "status": "fail", "retried": false, "duration": 2.7, "error": "AssertionError: total mismatch"},
  {"test_name": "test_create_order", "module": "order", "build_id": 104, "status": "fail", "retried": false, "duration": 2.9, "error": "AssertionError: total mismatch"},
  {"test_name": "test_create_order", "module": "order", "build_id": 105, "status": "fail", "retried": false, "duration": 2.5, "error": "AssertionError: total mismatch"},

  {"test_name": "test_search", "module": "search", "build_id": 101, "status": "pass", "retried": false, "duration": 0.9, "error": ""},
  {"test_name": "test_search", "module": "search", "build_id": 102, "status": "pass", "retried": false, "duration": 1.0, "error": ""},
  {"test_name": "test_search", "module": "search", "build_id": 103, "status": "fail", "retried": true, "duration": 1.8, "error": "503 Service Unavailable"},
  {"test_name": "test_search", "module": "search", "build_id": 104, "status": "pass", "retried": false, "duration": 1.0, "error": ""},
  {"test_name": "test_search", "module": "search", "build_id": 105, "status": "fail", "retried": true, "duration": 1.7, "error": "Connection reset"}
]

第二步：实现脆弱识别脚本

保存为 flaky_detector.py：

import json
from collections import defaultdict, Counter


def load_history(file_path: str):
    with open(file_path, "r", encoding="utf-8") as f:
        return json.load(f)


def group_by_test(records):
    grouped = defaultdict(list)
    for r in records:
        grouped[r["test_name"]].append(r)
    for test_name in grouped:
        grouped[test_name].sort(key=lambda x: x["build_id"])
    return grouped


def calc_volatility(statuses):
    if len(statuses) < 2:
        return 0.0
    switches = 0
    for i in range(1, len(statuses)):
        if statuses[i] != statuses[i - 1]:
            switches += 1
    return switches / (len(statuses) - 1)


def analyze_test(records):
    total = len(records)
    pass_count = sum(1 for r in records if r["status"] == "pass")
    fail_count = total - pass_count
    pass_rate = pass_count / total if total else 0.0

    retried_fail_count = sum(
        1 for r in records if r["status"] == "fail" and r.get("retried", False)
    )
    retry_recovery_signal = retried_fail_count / fail_count if fail_count else 0.0

    statuses = [r["status"] for r in records]
    volatility = calc_volatility(statuses)

    errors = [r["error"] for r in records if r["status"] == "fail" and r["error"]]
    error_diversity = len(set(errors))

    avg_duration = sum(r["duration"] for r in records) / total if total else 0.0
    module = records[0]["module"] if records else "unknown"

    # 简单规则：
    # 1. 波动大
    # 2. 重试失败信号明显
    # 3. 失败原因较分散
    flaky_score = (
        volatility * 0.4
        + retry_recovery_signal * 0.4
        + min(error_diversity / 3, 1.0) * 0.2
    )

    if fail_count == total:
        label = "consistent_failure"
    elif flaky_score >= 0.5:
        label = "suspected_flaky"
    else:
        label = "stable_or_minor_issue"

    return {
        "test_name": records[0]["test_name"] if records else "",
        "module": module,
        "total_runs": total,
        "pass_rate": round(pass_rate, 2),
        "fail_count": fail_count,
        "volatility": round(volatility, 2),
        "retry_recovery_signal": round(retry_recovery_signal, 2),
        "error_diversity": error_diversity,
        "avg_duration": round(avg_duration, 2),
        "flaky_score": round(flaky_score, 2),
        "label": label,
    }


def governance_priority(item):
    impact = 3 if item["module"] in ("auth", "order", "payment") else 2
    cost = 2 if item["avg_duration"] > 1.5 else 1
    return round(item["fail_count"] * impact * cost * max(item["flaky_score"], 0.3), 2)


def main():
    records = load_history("test_history.json")
    grouped = group_by_test(records)

    report = []
    for test_name, items in grouped.items():
        result = analyze_test(items)
        result["priority"] = governance_priority(result)
        report.append(result)

    report.sort(key=lambda x: x["priority"], reverse=True)

    print("=== Stability Governance Report ===")
    for item in report:
        print(
            f'{item["test_name"]:<20} '
            f'module={item["module"]:<8} '
            f'label={item["label"]:<20} '
            f'pass_rate={item["pass_rate"]:<4} '
            f'volatility={item["volatility"]:<4} '
            f'flaky_score={item["flaky_score"]:<4} '
            f'priority={item["priority"]}'
        )

    print("\n=== Suggested Flaky Candidates ===")
    for item in report:
        if item["label"] == "suspected_flaky":
            print(f'- {item["test_name"]} ({item["module"]}), priority={item["priority"]}')


if __name__ == "__main__":
    main()

第三步：运行

python flaky_detector.py

示例输出大致如下：

=== Stability Governance Report ===
test_search          module=search   label=suspected_flaky      pass_rate=0.6  volatility=0.5  flaky_score=0.67 priority=2.68
test_login           module=auth     label=suspected_flaky      pass_rate=0.6  volatility=1.0  flaky_score=0.87 priority=5.22
test_create_order    module=order    label=consistent_failure   pass_rate=0.0  volatility=0.0  flaky_score=0.07 priority=9.0

=== Suggested Flaky Candidates ===
- test_login (auth), priority=5.22
- test_search (search), priority=2.68

这里有个很重要的观察点：

test_login：高波动、失败带重试信号，典型脆弱用例
test_search：依赖服务不稳定，也像脆弱用例
test_create_order：虽然优先级很高，但它是持续稳定失败，应按真实缺陷处理，而不是 flaky

这一步很多团队最容易误判。我当时踩过一个坑：把所有高失败率用例都打进 flaky 池，结果真正的产品缺陷被延后处理，治理方向直接跑偏。

持续反馈闭环怎么搭

识别出来只是第一步，下面要把它接入日常流程。

一个最小闭环的动作链

sequenceDiagram
    participant CI as CI流水线
    participant Detector as 脆弱识别服务
    participant Tracker as 缺陷/任务平台
    participant Owner as 模块责任人
    participant Dashboard as 稳定性看板

    CI->>Detector: 上传测试结果、日志、重试信息
    Detector->>Detector: 计算 flaky_score 与标签
    Detector->>Tracker: 创建/更新治理任务
    Detector->>Owner: 发送告警与摘要
    Detector->>Dashboard: 更新趋势数据
    Owner->>Tracker: 标记修复方式与根因
    CI->>Dashboard: 持续反馈修复后表现

闭环里建议记录的根因分类

为了后面做统计，你最好把根因标准化，不然全靠自由文本，最后谁都看不懂。

建议至少有这几类：

script_issue：脚本定位方式脆弱、断言不稳
test_data_issue：测试数据污染、数据依赖错误
environment_issue：环境抖动、资源不足、服务未就绪
network_issue：接口超时、连接重置、DNS 问题
product_defect：真实产品缺陷
unknown：暂未定位

示例：生成治理任务清单

保存为 governance_ticket_generator.py：

import json
from datetime import datetime


def generate_ticket(item):
    root_cause_hint = "environment_issue" if item["volatility"] > 0.7 else "script_issue"
    return {
        "title": f'[Flaky治理] {item["test_name"]}',
        "module": item["module"],
        "priority": item["priority"],
        "label": item["label"],
        "suggested_root_cause": root_cause_hint,
        "created_at": datetime.utcnow().isoformat() + "Z",
        "description": (
            f'用例 {item["test_name"]} 被识别为疑似脆弱用例。\n'
            f'- pass_rate: {item["pass_rate"]}\n'
            f'- volatility: {item["volatility"]}\n'
            f'- retry_recovery_signal: {item["retry_recovery_signal"]}\n'
            f'- flaky_score: {item["flaky_score"]}\n'
            '请优先检查等待机制、测试数据隔离、外部依赖可用性。'
        )
    }


def main():
    report = [
        {
            "test_name": "test_login",
            "module": "auth",
            "priority": 5.22,
            "label": "suspected_flaky",
            "pass_rate": 0.6,
            "volatility": 1.0,
            "retry_recovery_signal": 1.0,
            "flaky_score": 0.87
        },
        {
            "test_name": "test_search",
            "module": "search",
            "priority": 2.68,
            "label": "suspected_flaky",
            "pass_rate": 0.6,
            "volatility": 0.5,
            "retry_recovery_signal": 1.0,
            "flaky_score": 0.67
        }
    ]

    tickets = [generate_ticket(item) for item in report if item["label"] == "suspected_flaky"]
    print(json.dumps(tickets, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

这个脚本虽然简单，但已经足够表达一个很关键的思路：
识别结果必须转成可分派、可追踪、可回看的治理对象。

逐步验证清单

如果你想把这套方案真正落到团队里，我建议按下面顺序推进，不要一次做太满。

阶段 1：先把数据采上来

每条测试执行都有唯一标识
能拿到 pass/fail、耗时、重试、错误摘要
能按构建、分支、模块聚合

阶段 2：做最小识别

最近 20~50 次执行历史可查询
能算通过率、波动率、重试恢复信号
能区分“持续失败”与“疑似脆弱”

阶段 3：做治理优先级

关键链路有影响系数
阻断主干/发布的用例优先级更高
输出前 10 条治理候选清单

阶段 4：做反馈闭环

自动创建任务或发送通知
记录根因分类和修复动作
修复后连续观察 1~2 周
看板能看到治理前后趋势

常见坑与排查

这部分我尽量说得接地气一些，因为真正做起来，问题几乎都出在细节。

1. 把“重试通过”当作“问题解决”

这是最常见的错觉。

现象：流水线总体通过率不错，但失败告警依旧频繁。
本质：重试掩盖了不稳定，而不是修复不稳定。
排查建议：

看“首轮通过率”而不是只看最终通过率
单独统计“重试后通过”的比例
如果某模块重试恢复率高，优先排查该模块

2. 环境问题和脚本问题混在一起

现象：大家都说是 flaky，但每个人理解不一样。
排查建议：

检查失败是否集中在特定机器、时间段、环境区域
对比同一时段同一环境其他用例表现
如果只有一条用例抖，优先看脚本和数据
如果一批用例一起抖，优先看环境和基础设施

3. 用例粒度太大，定位困难

现象：一条 E2E 用例走完整个主流程，失败原因非常散。
问题：你知道它不稳，但不知道到底哪一步不稳。
建议：

保留关键 E2E，但拆出核心步骤级验证
在关键节点增加埋点和日志
对异步等待点记录实际等待时间

4. 测试数据污染

我个人认为这是很多“伪 flaky”的根源。

典型场景：

同一个账号被多条用例复用
用例执行顺序影响数据状态
清理逻辑不完整，残留脏数据

排查建议：

每次执行生成独立数据命名空间
尽量使用可回收测试数据
用例间避免共享可变状态

5. 过度依赖固定等待

比如前端测试里常见的 sleep(3)。

问题：

环境快时浪费时间
环境慢时依然失败
稳定性和执行效率都差

更好的方式：

显式等待元素状态
等待接口响应或事件完成
轮询+超时+失败上下文记录

安全/性能最佳实践

稳定性治理看起来更偏测试工程，但里面也有不少安全和性能细节，忽略了会出问题。

安全最佳实践

1. 测试日志脱敏

自动化测试日志里很容易带出：

token
cookie
用户手机号
邮箱
测试库连接串

建议在日志采集阶段做统一脱敏。

示例：

import re


def mask_sensitive(text: str) -> str:
    text = re.sub(r'(token=)[A-Za-z0-9\-_\.]+', r'\1***', text)
    text = re.sub(r'(Authorization: Bearer )[A-Za-z0-9\-_\.]+', r'\1***', text)
    text = re.sub(r'(\b1[3-9]\d{9}\b)', '***手机号***', text)
    return text


if __name__ == "__main__":
    sample = "token=abc123xyz Authorization: Bearer secret-token 13812345678"
    print(mask_sensitive(sample))

2. 权限最小化

治理平台、报告系统、任务系统之间打通时：

API Token 只授最小权限
测试结果上传与任务创建账号分离
避免把生产级凭证直接放进 CI 变量

性能最佳实践

1. 识别任务不要拖慢主流水线

脆弱识别可以做成：

主流程只产出原始结果
识别分析异步执行
看板延迟几分钟更新也可以接受

否则你会遇到一个很尴尬的事：
为了提升测试稳定性，反而让 CI 更慢。

2. 保留足够历史，但别无限堆积

建议策略：

近 30 天保留明细
90 天保留聚合统计
超长历史进冷存储

3. 分层治理

不要所有测试一把抓。可以按层次看：

单元测试：重点看确定性失败
接口测试：重点看依赖波动和数据隔离
UI/E2E：重点看等待机制、环境抖动、异步时序

一些可直接落地的治理建议

如果你现在就想开始，不妨先做这 5 件事：

给测试报告增加“首轮通过率”和“重试恢复率”
把最近 2 周波动最大的前 10 条用例拉出来
要求每条 flaky 修复任务必须标注根因分类
把 sleep 类等待逐步替换为显式等待
修复后连续观察至少 10 次执行，不要当天通过就结项

边界条件也要说清楚：

如果你的执行历史数据非常少，识别结果会不稳
如果环境本身高度不一致，先统一环境基线，再谈治理
如果团队没有责任归属机制，闭环很容易断在“通知已发出”

总结

自动化测试稳定性治理，真正难的不是“知道 flaky 很烦”，而是把它做成一个持续运转的系统。

你可以记住这三个核心点：

先区分脆弱失败和真实失败
按影响度排序，不要平均发力
把识别结果接回研发流程，形成闭环

如果只做重试，你得到的是“看起来通过”；
如果做了识别但没有闭环，你得到的是“看到了问题”；
只有把数据、识别、治理、回看串起来，自动化测试才会重新变成团队可信的质量信号。

从实操角度，我建议先从一个最小版本开始：

用最近 20~50 次历史识别脆弱用例
每周治理 top 5
修复后持续观察趋势

别追求一步到位。
稳定性治理本身，也需要用“迭代”的方式来做。