背景与问题

自动化测试平台跑到一定规模后，团队通常都会遇到一个很现实的问题：不是用例写不出来，而是写出来后越来越“不可信”。

最典型的表现有几个：

• 同一批代码，今天过、明天挂
• 用例本地通过，CI 环境随机失败
• 失败后手工重跑又好了
• 测试报告里红一片，但真正有价值的缺陷没几个
• 团队逐渐对自动化测试失去信任，最后把它当成“参考信息”

这类问题的根源，往往不是单个测试脚本写得差，而是稳定性治理缺失。如果不治理，自动化测试会逐渐退化成“高噪音报警器”。

我做过几次自动化测试平台改造，最大的体会是：稳定性不是靠“多重跑几次”解决的，而是要把“脆弱用例定位”和“失败重试策略”放在同一个架构里设计。前者负责找出不稳定源头，后者负责在工程上做隔离和止损。两者缺一不可。

本文从架构视角展开，重点回答三个问题：

1. 如何识别脆弱用例，而不是凭感觉猜？
2. 什么样的失败适合重试，什么样的不应该重试？
3. 怎么把治理能力做成可持续运行的机制，而不是一次性运动？

先定义问题：什么叫“脆弱用例”

在自动化测试里，我一般把脆弱用例（Flaky Test）定义为：

在被测代码未发生影响结果的变化时，同一用例在相同或近似环境下执行结果不稳定，表现为通过与失败随机切换。

注意这个定义里有两个关键词：

• 结果不稳定
• 不是由真实产品缺陷导致

这就意味着，我们不能把所有失败都归到“脆弱”。比如：

• 接口契约变更导致断言失败：这是真失败
• 测试数据被污染导致偶发失败：这可能是脆弱
• 网络抖动导致超时：通常是脆弱
• 并发竞争触发真实线程安全问题：可能是真缺陷，不该简单重试掩盖

所以稳定性治理的第一步，不是“加重试”，而是给失败分类。

核心原理

1. 稳定性治理的目标不是“全绿”，而是“结果可信”

很多团队一开始会把目标设成：

• 用例通过率 99%
• 失败率降低到 1% 以下

这没错，但不够准确。更重要的是：

• 失败是否可解释
• 失败是否能归因
• 重试是否会掩盖真实问题
• 平台是否能量化测试可信度

所以从架构上，建议把稳定性治理拆成四层：

1. 采集层：记录每次执行的结果、耗时、错误信息、环境信息
2. 识别层：识别脆弱用例、环境型失败、代码型失败
3. 决策层：决定是否重试、重试几次、是否隔离
4. 治理层：输出排行榜、归因报告、修复闭环

下面这张图描述整体流转：

flowchart TD
    A[测试任务触发] --> B[执行用例]
    B --> C[采集结果与日志]
    C --> D{失败?}
    D -- 否 --> E[记录成功统计]
    D -- 是 --> F[失败分类]
    F --> G{是否可重试}
    G -- 是 --> H[按策略重试]
    G -- 否 --> I[直接标记失败]
    H --> J[汇总重试结果]
    J --> K[更新脆弱度评分]
    I --> K
    E --> K
    K --> L[输出治理报表与修复清单]

2. 脆弱用例定位的核心：不是看一次失败，而是看“模式”

单次失败几乎没有信息量。定位脆弱用例，必须看一个时间窗口内的行为模式。

常见指标包括：

• 失败率：最近 N 次执行中失败占比
• 重试后通过率：首次失败但重试成功的比例
• 环境相关性：是否集中在某个节点、时段、浏览器、区域
• 耗时抖动：执行耗时方差是否异常
• 错误分布：超时、元素找不到、连接失败、断言失败等类型分布
• 变更关联度：是否与代码提交强相关

其中我觉得最实用的是两个组合指标：

指标 A：脆弱度评分

一个简单可落地的公式：

脆弱度 = 0.4 * 失败率
       + 0.3 * 重试后通过率
       + 0.2 * 环境集中失败度
       + 0.1 * 耗时抖动归一化值

含义很直观：

• 失败率高，说明它不稳定
• 重试后通过率高，说明它“像脆弱”
• 如果失败都集中在某个环境，说明有环境依赖
• 耗时抖动大，通常意味着等待机制或资源争用有问题

指标 B：可疑度分层

可以按分数或规则把用例分层：

• S 级：高频随机失败，优先治理
• A 级：特定环境下明显不稳定
• B 级：偶发失败，需要持续观察
• C 级：基本稳定

这样做的好处是，测试团队不会陷入“全量排查”的泥潭，而是先处理最影响信任度的那一批。

3. 失败重试策略设计：重试是隔离手段，不是修复手段

重试本质上是工程上的缓冲层，用来吸收暂时性、非确定性失败。

适合重试的失败一般有：

• 网络超时
• 浏览器启动失败
• 页面元素短暂不可见
• 下游依赖短时 5xx
• CI 节点资源瞬时不足

不适合重试的失败包括：

• 明确断言失败
• 数据校验失败
• 接口返回业务错误码
• 权限不足
• 版本兼容问题
• 可稳定复现的脚本 bug

这一点非常重要。我见过最危险的做法是：不分类，所有失败统一重试 3 次。结果是：

• 真实缺陷被“洗绿”
• 构建时长暴涨
• 团队误以为质量变好了
• 后面再查根因时，证据已经被冲掉了

更合理的设计是：基于错误类型、历史行为和环境上下文做条件重试。

4. 推荐的重试决策模型

可以把重试决策看作一个状态机：

stateDiagram-v2
    [*] --> FirstRun
    FirstRun --> Pass: 成功
    FirstRun --> Fail: 失败
    Fail --> Classify: 失败分类
    Classify --> Retryable: 可重试
    Classify --> NonRetryable: 不可重试
    Retryable --> RetryRun: 执行重试
    RetryRun --> PassAfterRetry: 重试成功
    RetryRun --> RetryLimitReached: 达到上限
    RetryLimitReached --> FinalFail
    NonRetryable --> FinalFail
    Pass --> [*]
    PassAfterRetry --> [*]
    FinalFail --> [*]

一个比较稳妥的策略是：

• 最多重试 1~2 次
• 只对明确可重试错误生效
• 使用退避等待（exponential backoff）
• 保留首次失败日志，不允许重试覆盖原始证据
• 统计“首次失败、重试成功”的灰色结果

这里的“灰色结果”非常关键。
不要让重试成功后的用例直接等同于稳定通过。更合理的状态至少有三种：

• PASS
• PASS_AFTER_RETRY
• FAIL

否则你看日报时会误以为一切正常。

方案对比与取舍分析

方案一：无脑全量重试

做法

所有失败统一重试 2~3 次。

优点

• 实现最简单
• 短期内看起来通过率明显提升

缺点

• 掩盖真实缺陷
• 构建耗时显著上升
• 治理数据失真
• 容易形成路径依赖

适用场景

• 临时止血，不建议长期使用

方案二：基于错误类型的规则重试

做法

对 timeout、connection reset、浏览器启动失败等异常做白名单重试。

优点

• 风险可控
• 实现成本低
• 比全量重试更合理

缺点

• 规则维护成本会逐渐增加
• 新型失败模式需要持续补充

适用场景

• 大多数中型团队都适用
• 是比较推荐的起步方案

方案三：基于历史数据的动态重试

做法

结合错误类型、历史脆弱度评分、执行环境、最近变更信息动态决定是否重试。

优点

• 更精细
• 能兼顾效率和真实性
• 可逐步演进为智能治理能力

缺点

• 需要较完整的数据采集链路
• 实现复杂度较高

适用场景

• 自动化测试规模较大
• 已经有统一测试平台和结果存储系统

容量估算：为什么重试策略会影响平台成本

重试不仅影响结果，也直接影响资源成本。

假设：

• 每天执行用例数：10,000
• 平均单用例耗时：30 秒
• 首次失败率：8%
• 其中可重试失败占比：50%
• 平均重试 1 次

则额外执行量约为：

10,000 * 8% * 50% * 1 = 400 次

额外耗时约为：

400 * 30 秒 = 12,000 秒 = 200 分钟

如果你有 20 个并发 worker，看起来问题不大；但如果失败率涨到 20%，或者重试次数设成 3 次，资源成本会迅速放大。

所以重试策略设计必须带上两个约束：

• 重试预算：每天、每任务、每用例允许的重试上限
• 重试收益：通过率提升与额外耗时之间的平衡

实战代码（可运行）

下面我用 Python 写一个最小可运行示例，模拟：

1. 用例执行结果采集
2. 脆弱度评分计算
3. 基于失败类型的重试策略
4. 最终生成治理报告

你可以直接保存为 stability_governance.py 运行。

import random
import time
from collections import defaultdict
from statistics import pstdev

RETRYABLE_ERRORS = {
    "TimeoutError",
    "ConnectionResetError",
    "BrowserStartError",
    "ElementNotReadyError",
}

MAX_RETRY = 2
BACKOFF_SECONDS = [1, 2]

class TestCase:
    def __init__(self, name, failure_pattern):
        self.name = name
        self.failure_pattern = failure_pattern

    def run(self):
        """
        failure_pattern 返回:
        {
            "status": "PASS" or "FAIL",
            "error_type": str or None,
            "duration": float,
            "env": str
        }
        """
        return self.failure_pattern(self.name)

def flaky_network_case(_):
    env = random.choice(["ci-node-1", "ci-node-2", "ci-node-3"])
    duration = round(random.uniform(0.8, 2.5), 2)
    if random.random() < 0.35:
        return {
            "status": "FAIL",
            "error_type": random.choice(["TimeoutError", "ConnectionResetError"]),
            "duration": duration,
            "env": env
        }
    return {"status": "PASS", "error_type": None, "duration": duration, "env": env}

def real_assertion_bug(_):
    env = random.choice(["ci-node-1", "ci-node-2"])
    duration = round(random.uniform(0.5, 1.5), 2)
    if random.random() < 0.7:
        return {
            "status": "FAIL",
            "error_type": "AssertionError",
            "duration": duration,
            "env": env
        }
    return {"status": "PASS", "error_type": None, "duration": duration, "env": env}

def env_sensitive_case(_):
    env = random.choice(["ci-node-1", "ci-node-2", "ci-node-3"])
    duration = round(random.uniform(1.0, 3.0), 2)
    if env == "ci-node-3" and random.random() < 0.6:
        return {
            "status": "FAIL",
            "error_type": "BrowserStartError",
            "duration": duration,
            "env": env
        }
    return {"status": "PASS", "error_type": None, "duration": duration, "env": env}

def stable_case(_):
    env = random.choice(["ci-node-1", "ci-node-2"])
    duration = round(random.uniform(0.4, 1.0), 2)
    return {"status": "PASS", "error_type": None, "duration": duration, "env": env}

def run_with_retry(test_case):
    attempts = []
    for attempt in range(MAX_RETRY + 1):
        result = test_case.run()
        result["attempt"] = attempt + 1
        attempts.append(result)

        if result["status"] == "PASS":
            return {
                "final_status": "PASS" if attempt == 0 else "PASS_AFTER_RETRY",
                "attempts": attempts
            }

        if result["error_type"] not in RETRYABLE_ERRORS:
            return {
                "final_status": "FAIL",
                "attempts": attempts
            }

        if attempt < MAX_RETRY:
            time.sleep(BACKOFF_SECONDS[min(attempt, len(BACKOFF_SECONDS) - 1)])

    return {
        "final_status": "FAIL",
        "attempts": attempts
    }

def calc_flaky_score(history):
    total = len(history)
    fail_count = sum(1 for x in history if x["final_status"] == "FAIL")
    retry_pass_count = sum(1 for x in history if x["final_status"] == "PASS_AFTER_RETRY")

    env_fail_counter = defaultdict(int)
    durations = []

    for record in history:
        for attempt in record["attempts"]:
            durations.append(attempt["duration"])
        if record["final_status"] == "FAIL":
            last_env = record["attempts"][-1]["env"]
            env_fail_counter[last_env] += 1

    failure_rate = fail_count / total if total else 0
    retry_pass_rate = retry_pass_count / total if total else 0

    env_concentration = 0
    if fail_count > 0:
        env_concentration = max(env_fail_counter.values()) / fail_count

    duration_jitter = pstdev(durations) if len(durations) > 1 else 0
    duration_jitter_norm = min(duration_jitter / 2.0, 1)

    flaky_score = (
        0.4 * failure_rate +
        0.3 * retry_pass_rate +
        0.2 * env_concentration +
        0.1 * duration_jitter_norm
    )

    return {
        "failure_rate": round(failure_rate, 3),
        "retry_pass_rate": round(retry_pass_rate, 3),
        "env_concentration": round(env_concentration, 3),
        "duration_jitter_norm": round(duration_jitter_norm, 3),
        "flaky_score": round(flaky_score, 3),
    }

def level_from_score(score):
    if score >= 0.6:
        return "S"
    if score >= 0.4:
        return "A"
    if score >= 0.2:
        return "B"
    return "C"

def main():
    tests = [
        TestCase("test_flaky_network", flaky_network_case),
        TestCase("test_real_assertion_bug", real_assertion_bug),
        TestCase("test_env_sensitive", env_sensitive_case),
        TestCase("test_stable", stable_case),
    ]

    history_store = defaultdict(list)

    rounds = 20
    for _ in range(rounds):
        for test in tests:
            record = run_with_retry(test)
            history_store[test.name].append(record)

    print("=" * 80)
    print("稳定性治理报告")
    print("=" * 80)

    for name, history in history_store.items():
        score_info = calc_flaky_score(history)
        level = level_from_score(score_info["flaky_score"])

        final_status_counter = defaultdict(int)
        error_counter = defaultdict(int)

        for record in history:
            final_status_counter[record["final_status"]] += 1
            for attempt in record["attempts"]:
                if attempt["error_type"]:
                    error_counter[attempt["error_type"]] += 1

        print(f"\n用例: {name}")
        print(f"分级: {level}")
        print(f"最终结果分布: {dict(final_status_counter)}")
        print(f"错误类型分布: {dict(error_counter)}")
        print(f"评分详情: {score_info}")

if __name__ == "__main__":
    main()

代码说明

这个示例故意做了几件事：

• AssertionError 不进入重试
• TimeoutError、BrowserStartError 进入重试
• 重试成功会记为 PASS_AFTER_RETRY
• 统计的是完整尝试历史，不是只看最终结果

这几点在真实平台里都很重要。

一种更贴近 CI 的执行时序

如果把上述逻辑接到 CI/CD 平台里，典型时序如下：

sequenceDiagram
    participant CI as CI任务
    participant Runner as Test Runner
    participant Policy as Retry Policy
    participant Store as Result Store
    participant Report as Governance Report

    CI->>Runner: 执行测试任务
    Runner->>Store: 写入首次执行结果
    Runner->>Policy: 请求失败分类
    Policy-->>Runner: 返回是否可重试/重试次数
    alt 可重试
        Runner->>Runner: 执行重试
        Runner->>Store: 写入重试结果
    else 不可重试
        Runner->>Store: 保留失败结论
    end
    Store->>Report: 聚合历史记录
    Report-->>CI: 输出脆弱用例排行与任务结论

脆弱用例定位的落地步骤

如果你现在手上还没有完整平台，不用一上来就做得很重。我建议按下面四步落地。

第一步：先把原始数据采全

至少保证每次执行都记录：

• 用例 ID
• 执行时间
• 分支/提交号
• 环境节点
• 浏览器/设备信息
• 首次结果
• 每次重试结果
• 错误类型
• 日志与截图链接
• 执行耗时

很多团队失败就失败在这里：只有“通过/失败”两个字段，后面什么都分析不了。

第二步：做出最基础的失败分类

推荐先分成这几类：

• 断言类失败
• 等待/超时类失败
• 环境启动类失败
• 网络/依赖类失败
• 数据污染类失败
• 未知类失败

先别追求特别精细，能把不可重试和可重试分开，已经能解决一大半问题。

第三步：建立脆弱度排行榜

每周固定输出：

• Top 10 脆弱用例
• Top 5 环境问题节点
• Top 5 高频错误类型
• 首次失败后重试成功率
• 平均重试成本

排行榜的价值很大，因为它把“大家都觉得有点不稳”变成了有证据的改进清单。

第四步：治理闭环，而不是只报表不修

治理闭环至少要有这些动作：

• 脆弱用例自动打标
• S/A 级用例进入专项修复池
• 连续多周高脆弱度用例禁止作为发布门禁
• 某环境节点集中异常时自动摘除
• 修复后观察一段时间再取消标记

这一步是区分“有个统计页面”和“真的在治理”的关键。

常见坑与排查

坑一：把所有失败都当成测试脚本问题

实际情况往往更复杂，失败可能来自：

• 被测系统不稳定
• 环境节点不稳定
• 测试数据竞争
• 外部依赖抖动
• 脚本等待机制有缺陷

排查建议

从三个维度交叉看：

• 时间：是否集中在某个时间段
• 环境：是否集中在某台机器
• 变更：是否与某次代码提交同步出现

如果某个用例只有在 ci-node-3 失败，那优先查环境，不要急着改脚本。

坑二：重试后绿了，就当没事发生

这是最常见也最误导人的做法。

问题

• 真实稳定性问题被掩盖
• 报表看起来很好看，但团队体感很差
• 用例可信度持续下降

排查建议

单独统计：

• 首次通过率
• 首次失败率
• 重试转绿率
• 最终失败率

如果某个用例“最终通过率”很高，但“首次通过率”很低，它依然是问题用例。

坑三：重试日志覆盖首次失败证据

有些执行框架会把重试后的截图、日志覆盖第一次失败信息。等你回头排查时，最关键的现场已经没了。

排查建议

必须为每次尝试保留独立证据，例如：

• attempt_1.log
• attempt_1.png
• attempt_2.log
• attempt_2.png

并在报告中显式展示“最终结论”和“首次失败证据”。

坑四：等待策略写死，导致高抖动

例如 UI 自动化里常见的：

• 固定 sleep(5)
• 页面渲染没完成就断言
• 后端异步任务没结束就校验结果

排查建议

优先替换为：

• 显式等待
• 轮询等待
• 条件达成后继续，而不是固定睡眠

固定睡眠不但慢，而且不稳定。我自己早年就踩过这个坑，看起来“加了等待更稳了”，其实只是把偶发现象向后拖了一点。

坑五：数据隔离不彻底

例如多个并发用例共享：

• 同一个账号
• 同一份订单数据
• 同一个租户
• 同一个缓存 key

排查建议

检查是否存在：

• 用例并发修改同一资源
• 测试数据未清理
• 数据构造不可重复
• 环境残留状态

如果失败跟并发量升高强相关，优先怀疑数据隔离问题。

安全/性能最佳实践

虽然这是测试平台话题，但安全和性能也不能忽略。

安全最佳实践

1. 日志脱敏

失败日志中可能包含：

• token
• cookie
• 手机号
• 邮箱
• 用户隐私数据

建议在采集和存储前统一脱敏。

import re

def mask_sensitive(text):
    text = re.sub(r'Bearer\s+[A-Za-z0-9\-\._]+', 'Bearer ***', text)
    text = re.sub(r'(\b1[3-9]\d{9}\b)', '1**********', text)
    text = re.sub(r'([a-zA-Z0-9_.+-]+)@([a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+)', '***@***', text)
    return text

2. 最小权限原则

测试账号不要默认给全量生产级权限，尤其是联调或仿真环境中。

3. 限制重试风暴

如果某个下游服务已经异常，大量重试可能变成放大器，反而把服务压垮。
建议加入：

• 每任务最大重试数
• 某类错误全局熔断
• 某依赖异常时暂停相关测试

性能最佳实践

1. 区分重试粒度

不是所有场景都要“重跑整个测试类”或“重跑整个任务”。

可优先级如下：

1. 重试单次操作
2. 重试单条用例
3. 重试单个测试集
4. 重跑整任务

粒度越粗，成本越高。

2. 退避而不是立即连打

建议采用指数退避，避免瞬时资源争用持续放大。

3. 限制高脆弱用例的门禁权重

如果某用例长期处于 S 级脆弱状态，在修复前不宜直接作为强门禁条件，否则发布流程会被噪音劫持。

4. 环境健康检查前置

在执行大批量 UI 或集成测试前，先做：

• 节点 CPU/内存检查
• 浏览器驱动可用性检查
• 网络连通性检查
• 关键依赖服务探活

这类预检查很便宜，但能挡掉一批环境型假失败。

一个推荐的治理基线

如果你想尽快开始，我建议先定一个“够用”的治理基线：

数据基线

• 保留最近 30 天执行历史
• 每次执行保留完整尝试记录
• 用例、环境、错误类型三维可查询

策略基线

• 仅对白名单错误类型重试
• 最大重试 1~2 次
• 重试结果单独标记为 PASS_AFTER_RETRY
• 保留首次失败证据

报表基线

• 用例脆弱度 Top N
• 环境异常 Top N
• 首次失败率与重试转绿率
• 重试带来的额外执行成本

流程基线

• 每周清理一批 S/A 级脆弱用例
• 高脆弱用例在修复前降级门禁权重
• 环境问题与脚本问题分流处理

这套基线不复杂，但已经能让大多数团队从“凭感觉治理”进入“基于数据治理”。

总结

自动化测试稳定性治理，核心不是“把报表洗绿”，而是让测试结果重新变得可信。要做到这一点，我建议抓住两条主线：

1. 脆弱用例定位要看模式，不看单次结果
2. 失败重试策略要做分类控制，不能无脑全量重试

如果你只能立刻做一件事，我建议优先落地下面这三项：

• 给每次执行补齐完整历史数据
• 把失败分成“可重试”和“不可重试”
• 把 PASS_AFTER_RETRY 从普通 PASS 中单独拆出来

等这三项建立起来，后面的脆弱度评分、环境归因、治理闭环才有真实基础。

最后再强调一个边界条件：
重试只能缓解暂时性失败，不能代替根因修复。
如果一个用例长期靠重试维持通过，那它不是“稳定”，只是“暂时没炸”。这两者在工程上差别非常大。