自动化测试中的测试数据治理实战：从数据构造、隔离到回放的中级落地方案

自动化测试推进到一定阶段，真正拖后腿的往往不是脚本本身，而是测试数据。

我见过不少团队，UI 自动化、接口自动化、回归流水线都搭起来了，但一到 CI 环境就开始“玄学失败”：

• 昨天还能跑，今天报“手机号已注册”
• 并发执行时，A 用例把 B 用例的数据删了
• 本地能复现，流水线死活复现不了
• 线上故障想回放一遍，却发现请求有了、数据上下文没了

这些问题看起来零散，根子通常都在一件事上：没有把测试数据当成一等公民来治理。

这篇文章我不讲太虚的理念，而是从一个中级团队最容易落地的路径来展开：
数据构造 → 数据隔离 → 数据回放。
目标是让你的自动化测试从“能跑”走到“稳定、可复现、可排查”。

背景与问题

为什么自动化测试总在数据上翻车

自动化脚本容易写，稳定运行难。尤其在以下场景里，测试数据会变成核心瓶颈：

1. 共享环境

   • 多个测试任务共用一个数据库或服务环境
   • 用例之间互相污染数据

2. 状态型业务

   • 订单、支付、退款、库存、审批流这类流程有明确状态推进
   • 测试前后状态不一致，导致断言失效

3. 依赖外部系统

   • 第三方风控、短信、支付网关、消息队列
   • 一旦依赖系统返回变化，数据初始化和回放都变复杂

4. 并发执行

   • 用例为了提速开启并发
   • 同一份固定数据模板被多个线程抢用

典型坏味道

如果你的项目里出现下面这些现象，基本可以判定测试数据治理还没到位：

• 测试代码里写死账号：test_user_01
• SQL 初始化脚本只有一份，大家手动改来改去
• 清理数据靠“跑完再删”，失败后就留垃圾
• 复现问题要先问同事：“你当时用的是哪条数据？”
• 用例只在固定时间、固定顺序下能通过

一个更靠谱的目标

中级阶段的测试数据治理，不必一上来就搞成平台，但至少要做到：

• 可构造：用例需要什么数据，就能按规则生成出来
• 可隔离：不同用例、不同任务之间互不影响
• 可回放：失败场景可以带着上下文重新执行
• 可清理：产生的数据能追踪、可回收

前置知识与环境准备

这篇文章用一个简化的“用户注册 + 下单”场景来演示，技术栈选择尽量轻量：

• Python 3.10+
• SQLite（演示方便，换成 MySQL/PostgreSQL 思路一样）
• pytest
• 可选：Faker 生成模拟数据

安装依赖：

pip install pytest faker

项目结构如下：

test-data-governance/
├─ app.py
├─ db.py
├─ data_factory.py
├─ replay.py
├─ tests/
│  └─ test_order_flow.py
└─ run_demo.py

核心原理

我建议把测试数据治理拆成三层看，而不是一上来就陷进“建多少张初始化表”的细节。

1. 数据构造：不要手写样例，要“按意图生成”

重点不是造一堆固定数据，而是把业务数据抽象成可组合的构造器：

• 一个“新用户”
• 一个“已实名用户”
• 一个“余额充足的用户”
• 一个“可支付订单”

也就是说，数据构造的单位应该是业务意图，不是数据库行。

2. 数据隔离：隔离的是“作用域”，不只是库表

常见隔离层级有三种：

• 字段级隔离：如 run_id、tenant_id
• 账号级隔离：每条用例独立账号
• 环境级隔离：独立 schema / 独立库 / 独立容器

中级落地最常见、性价比最高的是：

共享环境 + 业务主键唯一化 + run_id 追踪 + 用例级清理

3. 数据回放：回放的不只是请求，还包括“前置上下文”

很多团队说“我们有接口日志，所以能回放”。但真正能复现问题，至少需要三部分：

• 输入请求
• 关联数据快照
• 执行顺序与依赖关系

否则你重放一个“支付成功”请求，但订单状态根本不是待支付，自然复现不出来。

一张全局图：从构造到回放

flowchart LR
    A[测试用例启动] --> B[生成 run_id]
    B --> C[按业务意图构造测试数据]
    C --> D[写入隔离标识]
    D --> E[执行自动化测试]
    E --> F{是否失败}
    F -- 否 --> G[清理 run_id 相关数据]
    F -- 是 --> H[记录请求/响应/关键数据快照]
    H --> I[生成回放包 replay package]
    I --> J[本地或CI复现]

数据模型设计：先给数据一个“身份证”

先建立最小可运行模型。这里演示两张表：users 和 orders。

实战代码（可运行）

1. 数据库初始化

db.py

import sqlite3
from contextlib import contextmanager

DB_FILE = "demo.db"

def init_db():
    conn = sqlite3.connect(DB_FILE)
    cur = conn.cursor()

    cur.execute("""
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        username TEXT NOT NULL UNIQUE,
        phone TEXT NOT NULL UNIQUE,
        balance INTEGER NOT NULL DEFAULT 0,
        run_id TEXT NOT NULL,
        created_at TEXT NOT NULL
    )
    """)

    cur.execute("""
    CREATE TABLE IF NOT EXISTS orders (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        order_no TEXT NOT NULL UNIQUE,
        username TEXT NOT NULL,
        amount INTEGER NOT NULL,
        status TEXT NOT NULL,
        run_id TEXT NOT NULL,
        created_at TEXT NOT NULL
    )
    """)

    cur.execute("""
    CREATE TABLE IF NOT EXISTS replay_logs (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        run_id TEXT NOT NULL,
        case_name TEXT NOT NULL,
        action TEXT NOT NULL,
        payload TEXT NOT NULL,
        snapshot TEXT NOT NULL,
        created_at TEXT NOT NULL
    )
    """)

    conn.commit()
    conn.close()

@contextmanager
def get_conn():
    conn = sqlite3.connect(DB_FILE)
    try:
        yield conn
        conn.commit()
    finally:
        conn.close()

2. 业务接口模拟

app.py

import json
from datetime import datetime
from db import get_conn

def now_str():
    return datetime.utcnow().isoformat()

def create_user(username: str, phone: str, balance: int, run_id: str):
    with get_conn() as conn:
        cur = conn.cursor()
        cur.execute(
            "INSERT INTO users(username, phone, balance, run_id, created_at) VALUES (?, ?, ?, ?, ?)",
            (username, phone, balance, run_id, now_str())
        )
    return {"username": username, "phone": phone, "balance": balance}

def create_order(order_no: str, username: str, amount: int, run_id: str):
    with get_conn() as conn:
        cur = conn.cursor()
        cur.execute("SELECT balance FROM users WHERE username = ?", (username,))
        row = cur.fetchone()
        if not row:
            raise ValueError("user not found")

        cur.execute(
            "INSERT INTO orders(order_no, username, amount, status, run_id, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            (order_no, username, amount, "CREATED", run_id, now_str())
        )
    return {"order_no": order_no, "username": username, "amount": amount, "status": "CREATED"}

def pay_order(order_no: str, run_id: str):
    with get_conn() as conn:
        cur = conn.cursor()

        cur.execute("SELECT username, amount, status FROM orders WHERE order_no = ?", (order_no,))
        order = cur.fetchone()
        if not order:
            raise ValueError("order not found")

        username, amount, status = order
        if status != "CREATED":
            raise ValueError("order status invalid")

        cur.execute("SELECT balance FROM users WHERE username = ?", (username,))
        user = cur.fetchone()
        if not user:
            raise ValueError("user not found")

        balance = user[0]
        if balance < amount:
            raise ValueError("insufficient balance")

        cur.execute("UPDATE users SET balance = balance - ? WHERE username = ?", (amount, username))
        cur.execute("UPDATE orders SET status = ? WHERE order_no = ?", ("PAID", order_no))

    return {"order_no": order_no, "status": "PAID", "run_id": run_id}

def snapshot_state(username: str, order_no: str):
    with get_conn() as conn:
        cur = conn.cursor()
        cur.execute("SELECT username, phone, balance, run_id FROM users WHERE username = ?", (username,))
        user = cur.fetchone()
        cur.execute("SELECT order_no, username, amount, status, run_id FROM orders WHERE order_no = ?", (order_no,))
        order = cur.fetchone()

    return {
        "user": user,
        "order": order
    }

def record_replay(run_id: str, case_name: str, action: str, payload: dict, snapshot: dict):
    with get_conn() as conn:
        cur = conn.cursor()
        cur.execute(
            "INSERT INTO replay_logs(run_id, case_name, action, payload, snapshot, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            (run_id, case_name, action, json.dumps(payload, ensure_ascii=False), json.dumps(snapshot, ensure_ascii=False), now_str())
        )

3. 数据工厂：按业务意图构造数据

data_factory.py

import uuid
from faker import Faker
from app import create_user, create_order

fake = Faker("zh_CN")

def new_run_id():
    return uuid.uuid4().hex[:12]

class TestDataFactory:
    def __init__(self, run_id: str):
        self.run_id = run_id

    def unique_username(self):
        return f"u_{self.run_id}_{uuid.uuid4().hex[:6]}"

    def unique_phone(self):
        # 生成一个大概率唯一的手机号字符串
        return "13" + uuid.uuid4().hex[:9]

    def user_with_balance(self, balance: int = 1000):
        username = self.unique_username()
        phone = self.unique_phone()
        create_user(username=username, phone=phone, balance=balance, run_id=self.run_id)
        return {"username": username, "phone": phone, "balance": balance}

    def created_order(self, username: str, amount: int = 100):
        order_no = f"O{self.run_id}{uuid.uuid4().hex[:8]}"
        create_order(order_no=order_no, username=username, amount=amount, run_id=self.run_id)
        return {"order_no": order_no, "username": username, "amount": amount}

这里有两个关键点：

1. 唯一键不要写死
   • username、phone、order_no 全部动态生成
2. 所有数据都带 run_id
   • 这样后续清理、追踪、回放才有抓手

4. 测试用例：构造、执行、回放记录一体化

tests/test_order_flow.py

from db import init_db, get_conn
from data_factory import TestDataFactory, new_run_id
from app import pay_order, snapshot_state, record_replay

def cleanup_by_run_id(run_id: str):
    with get_conn() as conn:
        cur = conn.cursor()
        cur.execute("DELETE FROM orders WHERE run_id = ?", (run_id,))
        cur.execute("DELETE FROM users WHERE run_id = ?", (run_id,))

def test_pay_order_success():
    init_db()
    run_id = new_run_id()
    case_name = "test_pay_order_success"
    factory = TestDataFactory(run_id)

    user = factory.user_with_balance(balance=500)
    order = factory.created_order(username=user["username"], amount=200)

    result = pay_order(order_no=order["order_no"], run_id=run_id)
    assert result["status"] == "PAID"

    snapshot = snapshot_state(user["username"], order["order_no"])
    record_replay(
        run_id=run_id,
        case_name=case_name,
        action="pay_order",
        payload={"order_no": order["order_no"], "run_id": run_id},
        snapshot=snapshot
    )

    cleanup_by_run_id(run_id)

def test_pay_order_insufficient_balance():
    init_db()
    run_id = new_run_id()
    case_name = "test_pay_order_insufficient_balance"
    factory = TestDataFactory(run_id)

    user = factory.user_with_balance(balance=50)
    order = factory.created_order(username=user["username"], amount=200)

    try:
        pay_order(order_no=order["order_no"], run_id=run_id)
        assert False, "expected insufficient balance"
    except ValueError as e:
        assert str(e) == "insufficient balance"

    snapshot = snapshot_state(user["username"], order["order_no"])
    record_replay(
        run_id=run_id,
        case_name=case_name,
        action="pay_order",
        payload={"order_no": order["order_no"], "run_id": run_id},
        snapshot=snapshot
    )

    cleanup_by_run_id(run_id)

执行：

pytest -q

用时序图看一次测试执行

sequenceDiagram
    participant T as TestCase
    participant F as DataFactory
    participant A as App
    participant DB as Database
    participant R as ReplayLog

    T->>F: 生成 run_id
    T->>F: 构造 user_with_balance
    F->>A: create_user(...)
    A->>DB: insert users
    T->>F: 构造 created_order
    F->>A: create_order(...)
    A->>DB: insert orders
    T->>A: pay_order(order_no)
    A->>DB: 查询订单/用户并更新状态
    T->>A: snapshot_state(...)
    T->>R: 记录 payload + snapshot
    T->>DB: cleanup by run_id

5. 回放脚本：把失败现场重新拉起来

很多人做日志记录时，只记一个请求体。这个不够。
这里我们演示一个简单版回放器：读取 replay_logs 中某个 run_id 的记录，再重新准备上下文并执行。

replay.py

import json
from db import get_conn, init_db
from app import create_user, create_order, pay_order

def replay_by_run_id(run_id: str):
    init_db()
    with get_conn() as conn:
        cur = conn.cursor()
        cur.execute("""
            SELECT case_name, action, payload, snapshot
            FROM replay_logs
            WHERE run_id = ?
            ORDER BY id ASC
        """, (run_id,))
        rows = cur.fetchall()

    if not rows:
        print(f"no replay logs found for run_id={run_id}")
        return

    for case_name, action, payload_str, snapshot_str in rows:
        payload = json.loads(payload_str)
        snapshot = json.loads(snapshot_str)

        user = snapshot.get("user")
        order = snapshot.get("order")

        if user:
            username, phone, balance, snapshot_run_id = user
            try:
                create_user(username, phone, balance, run_id)
            except Exception:
                pass

        if order:
            order_no, username, amount, status, snapshot_run_id = order
            try:
                create_order(order_no, username, amount, run_id)
            except Exception:
                pass

        if action == "pay_order":
            try:
                result = pay_order(payload["order_no"], run_id)
                print(f"[REPLAY SUCCESS] {case_name}: {result}")
            except Exception as e:
                print(f"[REPLAY FAIL] {case_name}: {e}")

运行示例：

from replay import replay_by_run_id

replay_by_run_id("你的run_id")

生产环境里当然不会这么简化。真实回放通常需要：

• 脱敏后的请求参数
• 关联数据库快照或事件快照
• MQ 消息、缓存状态
• 外部依赖的桩响应

但这套演示代码至少说明一个核心观点：
可回放的前提是测试执行期间保留“足够上下文”。

隔离策略怎么选：别追求绝对隔离，先追求稳定性价比

很多团队会问：到底该用独立库，还是共享库加前缀？

我一般这么建议。

方案对比

我的经验结论

如果你还没有平台团队支撑，最值得先做的是这三件事：

1. 所有测试数据加唯一标识
2. 所有写入数据带 run_id
3. 所有用例具备独立清理能力

这三件做好，已经能解决 70% 以上的“数据脏、复现难、并发冲突”问题。

常见坑与排查

这一节我尽量讲得接地气一点，因为很多坑真不是原理问题，而是执行细节问题。

坑 1：只隔离创建，不隔离查询

比如你创建用户时带了唯一前缀，但查询订单时却按“最新一条订单”取：

SELECT * FROM orders ORDER BY id DESC LIMIT 1;

这在并发场景里几乎必炸。

正确思路是：
创建和查询都必须使用同一组可追踪标识，例如 order_no、username、run_id。

坑 2：清理放在用例最后，异常时没执行

很多人喜欢这么写：

# 不推荐
create data
run test
delete data

但一旦中间抛异常，删除步骤根本跑不到。

更稳妥的方式是用 try/finally 或测试框架 fixture 托管清理。

run_id = new_run_id()
try:
    # arrange + act + assert
    pass
finally:
    cleanup_by_run_id(run_id)

坑 3：回放只记请求，不记依赖状态

比如“支付失败”其实是由余额不足引起的。
你只回放支付请求，不恢复用户余额，就没法稳定重现。

排查建议：

• 看失败接口前的关键实体状态
• 记录状态流转前后快照
• 对外部依赖返回值做存档

坑 4：随机数据过度随机，导致不可断言

有些团队迷信 Faker，什么都随机。结果断言时反而不知道该校验什么。

经验上要区分两类字段：

• 需要唯一，但不参与断言：可随机
• 参与断言或流程判断：要可控

比如金额、状态、用户等级这些字段，最好明确指定，不要全靠随机。

坑 5：复用生产脱敏数据时违反最小权限原则

拿线上数据做测试样本是常见操作，但一定要小心：

• 不要直接拉原始手机号、身份证号
• 不要把敏感字段复制到共享测试库
• 不要让自动化日志输出隐私数据

这个问题不只是“规范”，而是真可能带来安全事故。

一个状态视角：测试数据在生命周期里怎么流转

stateDiagram-v2
    [*] --> Defined: 定义业务意图
    Defined --> Generated: 工厂生成数据
    Generated --> Isolated: 写入唯一标识/run_id
    Isolated --> Consumed: 被用例消费
    Consumed --> Recorded: 记录日志与快照
    Recorded --> Replayed: 回放复现
    Recorded --> Cleaned: 清理归档
    Replayed --> Cleaned
    Cleaned --> [*]

安全/性能最佳实践

测试数据治理不是只管“能不能跑”，还要考虑安全和效率。

安全最佳实践

1. 对敏感字段统一脱敏

如果要记录回放日志，建议对以下字段做脱敏或哈希：

• 手机号
• 身份证号
• 邮箱
• 地址
• token / session / cookie

示例：

def mask_phone(phone: str) -> str:
    if len(phone) < 7:
        return phone
    return phone[:3] + "****" + phone[-4:]

2. 回放日志不要直接存密钥

外部系统调用相关的：

• Access Token
• API Key
• Cookie
• 鉴权签名原文

不要直接落库。建议：

• 仅保留必要字段
• 用引用 ID 指向安全存储
• 设置日志过期清理

3. 区分“测试数据”和“生产影子数据”

测试环境里混入生产脱敏数据时，建议打标签：

• data_origin = synthetic
• data_origin = masked_prod

这样排查和权限控制都更清晰。

性能最佳实践

1. 优先做“最小数据集”构造

别一上来就灌全量初始化脚本。
一个支付用例只需要：

• 1 个用户
• 1 个订单
• 必要账户余额

就不要把营销、优惠券、地址簿、发票抬头都初始化出来。

2. 避免每个用例都重建全库

全量重建数据库虽然干净，但 CI 时间会很难看。更现实的方案是：

• 套件启动时初始化结构
• 用例级按 run_id 构造/清理数据
• 高风险套件单独跑独立环境

3. 给 run_id 和业务唯一键建索引

如果正式数据库里要做清理和检索，run_id 没索引会非常慢。

示例 SQL：

CREATE INDEX IF NOT EXISTS idx_users_run_id ON users(run_id);
CREATE INDEX IF NOT EXISTS idx_orders_run_id ON orders(run_id);
CREATE INDEX IF NOT EXISTS idx_orders_order_no ON orders(order_no);

4. 回放包不要无限增长

回放日志很有用，但如果每次都存完整快照，很容易膨胀。建议：

• 只记录关键实体
• 给日志设置 TTL
• 失败场景全量保留，成功场景抽样保留

逐步验证清单

如果你准备把这套方案落到现有项目里，可以按下面顺序推进。

第 1 步：先统一唯一键策略

确认所有核心实体都有可控唯一标识：

• 用户名
• 手机号
• 订单号
• 请求幂等号

第 2 步：引入 run_id

在测试启动时生成 run_id，并贯穿：

• 数据构造
• 业务调用
• 日志记录
• 清理脚本

第 3 步：封装数据工厂

把散落在各个测试里的 SQL、接口调用、造数逻辑收进工厂类：

• user_with_balance
• created_order
• paid_order（必要时）
• refund_ready_order

第 4 步：增加失败回放能力

先别追求完美，把最关键两类信息存起来：

• 请求入参
• 关键实体快照

第 5 步：接入 CI 清理与归档

至少做到：

• 测试结束后按 run_id 清理
• 异常失败时保留回放日志
• 定时清理过期日志

适用边界与取舍

这套方案很适合：

• 接口自动化为主的团队
• 共享测试环境较多的项目
• 需要并发执行、又暂时没有独立环境资源的团队

但它也有边界：

1. 强一致复杂链路
   • 涉及缓存、MQ、异步任务、外部支付时，单靠数据库快照不够
2. 大规模并发 CI
   • 如果几百条用例同时跑，共享环境隔离压力会变大
3. 强监管数据场景
   • 金融、医疗等场景对数据回放和脱敏要求更严格

在这些情况下，你可能需要逐步升级到：

• 环境级隔离
• 服务虚拟化
• 事件回放平台
• 数据快照编排系统

但请注意，不要一开始就为未来 3 年的复杂度买单。
多数团队先把本文这套“中级版”跑稳，收益已经很高。

总结

测试数据治理，说白了就是回答三个问题：

1. 数据怎么来？
   用业务意图驱动的数据工厂来构造，而不是手填固定样例。

2. 数据怎么不互相污染？
   用唯一键 + run_id + 用例级清理，先实现低成本隔离。

3. 问题怎么复现？
   记录请求、关键快照和执行上下文，形成最小可用回放能力。

如果你现在就想开始落地，我建议按这个优先级执行：

• 第一优先级：统一唯一数据生成规则
• 第二优先级：给所有测试写入加 run_id
• 第三优先级：用例失败时保留回放包
• 第四优先级：补齐清理、索引、脱敏策略

最后送一句很实在的话：
自动化测试的稳定性，很多时候不是脚本技巧比拼，而是谁先把测试数据管明白。

当你的测试数据可构造、可隔离、可回放时，自动化体系才真正开始“工程化”。