背景与问题

做自动化测试时，很多团队一开始关注的是“脚本能不能跑起来”，但跑着跑着就会发现，真正让测试变得不稳定的，往往不是断言本身，而是测试数据。

典型症状很常见：

• 用例昨天能过，今天失败，原因是共享账号状态被别人改了
• 回归测试一跑，数据库里堆满脏数据，后续测试全部串味
• 同一批用例在本地、测试环境、预发环境表现不一致
• 并发执行时，多个任务抢同一条数据，偶发失败很难复现
• 为了“图省事”直接复用线上脱敏数据，结果又慢又难维护，还伴随安全风险

如果把自动化测试比作流水线，那么测试数据管理就是原材料供应系统。原材料不稳定，后面的流程越自动化，放大的问题就越严重。

这篇文章我会从一个偏实战的角度，带你把测试数据管理拆成三件事：

1. 环境隔离：让不同测试任务互不干扰
2. 数据构造：让用例需要什么数据，就能快速、确定地拿到什么数据
3. 数据回收：让测试执行完后，环境还能保持可重复使用

文章会用一个可运行的 Python + SQLite 小示例来演示思路。虽然示例轻量，但方法可以迁移到 MySQL、PostgreSQL、接口自动化、UI 自动化，甚至 CI 流水线里。

前置知识与环境准备

建议你具备以下基础：

• 了解自动化测试基本概念
• 能看懂 Python 基础语法
• 知道数据库中事务、主键、唯一约束的基本含义

本教程使用：

• Python 3.10+
• SQLite（内置，无需额外安装）
• pytest（可选，用于演示自动化测试风格）

安装 pytest：

pip install pytest

项目结构可以很简单：

test-data-demo/
├─ app.py
├─ data_manager.py
├─ test_order_flow.py
└─ demo.db

为什么测试数据管理这么容易失控

先别急着写“造数脚本”，先看问题源头。多数团队的数据问题不是某一段代码写错，而是没有建立清晰的数据生命周期。

常见失控模式

1. 共享环境 + 共享账号

最常见，也最危险。

比如所有测试都用 test_user_01 登录。登录测试刚把密码改了，订单测试又需要它下单，权限测试还会给它加角色。最后结果是谁都能改，谁都不敢删。

2. 用例依赖历史数据

例如：

• “数据库里必须已经有一条待支付订单”
• “用户必须先有一张优惠券”
• “库存表必须预置某 SKU 的剩余量”

这类测试看似省事，实际可维护性很差。环境一重建，测试就废。

3. 只造数据，不回收

刚开始环境干净，跑久了以后：

• 表数据越来越大
• 唯一索引冲突越来越频繁
• 查询越来越慢
• 旧数据干扰新断言

4. 为了快，直接手工改库

这种方式短期有效，长期灾难。因为：

• 不可追踪
• 不可复制
• 不适合并发
• 新人接手几乎无法理解

核心原理

测试数据管理真正有效，靠的不是“多写几个 SQL”，而是以下几个原则。

1. 数据最小化：每条用例只拥有自己需要的数据

不要依赖“大而全”的公共数据池。
更推荐的做法是：

• 每个用例构造自己的用户、订单、券、库存
• 数据命名带唯一标识
• 用例结束后回收或事务回滚

这样测试才能做到独立、可重复、可并发。

2. 环境隔离优先于数据修补

如果环境天然隔离，很多问题根本不会出现。隔离可以分层做：

• 账号隔离：每个测试任务独立账号
• 数据命名空间隔离：比如按 run_id、case_id 打标签
• 数据库隔离：独立 schema / 库 / 容器
• 服务依赖隔离：外部系统 mock 或沙箱化

从成本角度看，通常采用分层策略：

1. 先做命名空间隔离
2. 再做账号隔离
3. 关键链路再做库级或容器级隔离

3. 数据构造必须“可声明”

好的数据构造，不是堆很多 SQL，而是让测试代码表达业务意图：

• 我要一个“新注册用户”
• 我要一个“余额充足且实名通过的用户”
• 我要一笔“待支付订单”
• 我要一个“库存不足商品”

这类接口通常叫 Test Data Builder（测试数据构造器） 或 Factory（工厂）。

4. 数据回收要有兜底机制

理想情况是每次测试结束都能清理干净，但现实里经常会遇到：

• 测试中途失败
• 清理逻辑未执行
• 流水线被强制中断
• 外部系统状态无法事务回滚

所以回收策略通常要两层：

• 即时回收：测试完成即清理
• 延迟回收：定时任务按标签扫描并清理过期测试数据

5. 优先使用“可预测”而不是“真实复杂”

很多人会执着于“数据越像线上越好”，但自动化测试最重要的是：

• 确定性
• 低耦合
• 易定位

不是不能用真实分布数据，而是要区分场景：

• 功能自动化：优先小而稳定的数据集
• 性能/压测：才需要大规模拟真数据
• 风控/推荐类场景：可以结合采样和脱敏数据

一张图看清整体策略

flowchart TD
    A[测试开始] --> B[生成 run_id / case_id]
    B --> C[按环境策略选择隔离级别]
    C --> D[通过 Data Builder 构造测试数据]
    D --> E[执行自动化测试]
    E --> F{是否支持事务回滚?}
    F -- 是 --> G[回滚事务]
    F -- 否 --> H[按标签执行即时清理]
    G --> I[记录审计日志]
    H --> I
    I --> J[定时任务兜底回收过期数据]

环境隔离的落地方式

测试数据管理中，环境隔离不是“要不要做”，而是“做到哪一层”。

方案一：逻辑隔离（推荐先做）

在数据中加入测试标识，例如：

• run_id
• case_id
• created_by = 'automation'
• expires_at

优点：

• 成本低
• 容易快速落地
• 适合大多数接口自动化项目

缺点：

• 需要每张关键表都支持标签字段，或者至少主记录支持标签
• 清理时要注意级联关系

方案二：账号隔离

例如每个并发 worker 拿一个独立账号池：

• worker-1 -> user_a
• worker-2 -> user_b
• worker-3 -> user_c

优点：

• 对 UI 自动化非常有效
• 适合登录态、权限态、购物车等强状态场景

缺点：

• 账号池要维护
• 容易因账号耗尽或状态漂移出问题

方案三：Schema / 数据库隔离

每次测试任务启动独立 schema，甚至独立数据库实例。

优点：

• 隔离最彻底
• 并发稳定性最好
• 清理简单，直接 drop

缺点：

• 成本高
• 初始化速度和资源消耗更大

方案四：容器级环境隔离

CI 里常见：每次 MR / PR 拉起独立测试环境。

优点：

• 最接近真实环境
• 适合集成测试

缺点：

• 资源昂贵
• 构建时间长
• 运维复杂度高

隔离策略选型图

flowchart LR
    A[测试场景] --> B{主要问题是什么?}
    B -->|数据串用| C[逻辑隔离 + 标签清理]
    B -->|账号状态冲突| D[账号池隔离]
    B -->|高并发集成测试| E[Schema/库隔离]
    B -->|跨服务强依赖验证| F[容器级隔离]

数据构造策略：从“插库”升级为“构造器”

这里是实践中的关键转折点。

很多团队的写法是这样：

insert into users ...
insert into coupons ...
insert into orders ...

能用，但很快会散落在各个测试文件里，维护成本越来越高。

更好的做法是统一封装成构造器，让测试只描述“我要什么状态”。

一个好的数据构造器应具备什么能力

• 生成唯一数据，避免冲突
• 支持默认值，减少样板代码
• 能表达业务状态
• 自动记录构造的数据，方便回收
• 提供组合能力，而不是死板模板

例如：

• create_user()
• create_user(balance=100, verified=True)
• create_order(status="PENDING")
• create_paid_order(user_id=xxx)

实战代码（可运行）

下面我们实现一个轻量版本的测试数据管理器，演示：

• 环境标签隔离
• 数据构造
• 即时清理
• 兜底清理

第一步：初始化数据库

创建 app.py：

import sqlite3

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,
        balance INTEGER NOT NULL DEFAULT 0,
        verified INTEGER NOT NULL DEFAULT 0,
        run_id TEXT,
        created_at DATETIME DEFAULT CURRENT_TIMESTAMP
    )
    """)

    cur.execute("""
    CREATE TABLE IF NOT EXISTS orders (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        user_id INTEGER NOT NULL,
        amount INTEGER NOT NULL,
        status TEXT NOT NULL,
        run_id TEXT,
        created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
        FOREIGN KEY(user_id) REFERENCES users(id)
    )
    """)

    conn.commit()
    conn.close()


if __name__ == "__main__":
    init_db()
    print("database initialized.")

执行：

python app.py

第二步：实现测试数据管理器

创建 data_manager.py：

import sqlite3
import uuid
from contextlib import contextmanager

DB_FILE = "demo.db"


class TestDataManager:
    def __init__(self, db_file=DB_FILE, run_id=None):
        self.db_file = db_file
        self.run_id = run_id or f"run_{uuid.uuid4().hex[:8]}"
        self.created_entities = {
            "orders": [],
            "users": []
        }

    def connect(self):
        return sqlite3.connect(self.db_file)

    def create_user(self, balance=0, verified=False, username=None):
        username = username or f"user_{self.run_id}_{uuid.uuid4().hex[:6]}"
        verified_int = 1 if verified else 0

        with self.connect() as conn:
            cur = conn.cursor()
            cur.execute("""
                INSERT INTO users (username, balance, verified, run_id)
                VALUES (?, ?, ?, ?)
            """, (username, balance, verified_int, self.run_id))
            user_id = cur.lastrowid
            conn.commit()

        self.created_entities["users"].append(user_id)
        return {
            "id": user_id,
            "username": username,
            "balance": balance,
            "verified": verified
        }

    def create_order(self, user_id, amount, status="PENDING"):
        with self.connect() as conn:
            cur = conn.cursor()
            cur.execute("""
                INSERT INTO orders (user_id, amount, status, run_id)
                VALUES (?, ?, ?, ?)
            """, (user_id, amount, status, self.run_id))
            order_id = cur.lastrowid
            conn.commit()

        self.created_entities["orders"].append(order_id)
        return {
            "id": order_id,
            "user_id": user_id,
            "amount": amount,
            "status": status
        }

    def cleanup(self):
        with self.connect() as conn:
            cur = conn.cursor()
            cur.execute("DELETE FROM orders WHERE run_id = ?", (self.run_id,))
            cur.execute("DELETE FROM users WHERE run_id = ?", (self.run_id,))
            conn.commit()

    def cleanup_expired_runs(self, run_ids):
        with self.connect() as conn:
            cur = conn.cursor()
            for run_id in run_ids:
                cur.execute("DELETE FROM orders WHERE run_id = ?", (run_id,))
                cur.execute("DELETE FROM users WHERE run_id = ?", (run_id,))
            conn.commit()


@contextmanager
def managed_test_data(run_id=None):
    manager = TestDataManager(run_id=run_id)
    try:
        yield manager
    finally:
        manager.cleanup()

第三步：写一个简单的业务函数

为了演示，我们做一个“支付订单”的业务逻辑。
创建 test_order_flow.py：

import sqlite3
from data_manager import managed_test_data, DB_FILE


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

    cur.execute("""
        SELECT o.id, o.user_id, o.amount, o.status, u.balance
        FROM orders o
        JOIN users u ON o.user_id = u.id
        WHERE o.id = ?
    """, (order_id,))
    row = cur.fetchone()

    if not row:
        conn.close()
        raise ValueError("order not found")

    _, user_id, amount, status, balance = row

    if status != "PENDING":
        conn.close()
        raise ValueError("order status invalid")

    if balance < amount:
        conn.close()
        raise ValueError("insufficient balance")

    cur.execute("UPDATE users SET balance = balance - ? WHERE id = ?", (amount, user_id))
    cur.execute("UPDATE orders SET status = 'PAID' WHERE id = ?", (order_id,))
    conn.commit()
    conn.close()


def get_order(order_id):
    conn = sqlite3.connect(DB_FILE)
    cur = conn.cursor()
    cur.execute("SELECT id, status FROM orders WHERE id = ?", (order_id,))
    row = cur.fetchone()
    conn.close()
    return row


def get_user(user_id):
    conn = sqlite3.connect(DB_FILE)
    cur = conn.cursor()
    cur.execute("SELECT id, balance FROM users WHERE id = ?", (user_id,))
    row = cur.fetchone()
    conn.close()
    return row


def test_pay_order_success():
    with managed_test_data() as dm:
        user = dm.create_user(balance=100, verified=True)
        order = dm.create_order(user_id=user["id"], amount=30, status="PENDING")

        pay_order(order["id"])

        saved_order = get_order(order["id"])
        saved_user = get_user(user["id"])

        assert saved_order[1] == "PAID"
        assert saved_user[1] == 70


def test_pay_order_insufficient_balance():
    with managed_test_data() as dm:
        user = dm.create_user(balance=10, verified=True)
        order = dm.create_order(user_id=user["id"], amount=30, status="PENDING")

        try:
            pay_order(order["id"])
            assert False, "should raise ValueError"
        except ValueError as e:
            assert str(e) == "insufficient balance"

        saved_order = get_order(order["id"])
        saved_user = get_user(user["id"])

        assert saved_order[1] == "PENDING"
        assert saved_user[1] == 10

运行测试：

pytest -q

如果一切正常，你会看到两个测试通过，并且测试结束后对应数据被自动清理。

代码里真正值得学的点

上面的代码不复杂，但已经体现了测试数据管理中的几个关键动作。

1. run_id 是隔离的核心抓手

每轮测试都带一个唯一 run_id：

self.run_id = run_id or f"run_{uuid.uuid4().hex[:8]}"

这样你可以：

• 精准找到本轮测试造的数据
• 清理时不误删别人的数据
• 在日志里追踪测试任务与数据关联

2. 测试只关心业务状态，不关心插库细节

测试代码里写的是：

user = dm.create_user(balance=100, verified=True)
order = dm.create_order(user_id=user["id"], amount=30, status="PENDING")

而不是把 SQL 散落在用例中。
这样后面即使表结构变化，你只需要改构造器。

3. 用上下文管理器保证回收

with managed_test_data() as dm:
    ...

即使测试中间断言失败，finally 仍会执行清理。
这是我很推荐的基础做法，简单但有效。

如果要支持事务回滚，应该怎么做

对于纯数据库型测试，事务回滚是最高性价比的清理手段之一。
它的思路是：

1. 测试开始时开启事务
2. 所有造数和业务操作都在同一事务内
3. 测试结束直接 rollback

不过它有边界：

• 如果应用代码内部自己提交事务，外部回滚未必能兜住
• 如果测试经过消息队列、缓存、第三方系统，事务只能覆盖数据库，不能覆盖外部副作用
• 分布式服务调用时通常做不到单事务包裹全链路

下面是一个简化示意图：

sequenceDiagram
    participant T as 测试用例
    participant M as DataManager
    participant DB as Database
    participant S as 业务服务

    T->>M: 开启事务
    M->>DB: BEGIN
    T->>M: 创建用户/订单
    M->>DB: INSERT users/orders
    T->>S: 调用支付逻辑
    S->>DB: UPDATE users/orders
    T->>M: 测试结束
    M->>DB: ROLLBACK
    DB-->>T: 数据恢复

如果你的系统适合事务包裹，那就优先用它；如果不适合，再退回到“标签清理 + 定时兜底回收”。

逐步验证清单

如果你打算把本文方法迁移到自己的项目，我建议按下面顺序推进，而不是一口气重构全部测试。

第 1 步：先给测试数据打标签

至少做到：

• 每条主测试数据可追踪到 run_id
• 所有自动化造的数据可区分于人工数据

第 2 步：封装 3~5 个高频构造器

优先抽出最常用的对象，比如：

• 用户
• 订单
• 支付记录
• 优惠券
• 商品库存

第 3 步：接入即时清理

先确保大多数测试结束后能自动清理。

第 4 步：增加定时兜底任务

比如每小时清理一次超过 24 小时的测试数据。

第 5 步：按冲突点增强隔离级别

如果还会串数据，再按需补：

• 账号池
• schema 隔离
• 容器隔离

常见坑与排查

下面这些坑，我基本都见过，甚至有些还亲手踩过。

坑一：唯一键冲突频发

现象

测试并发执行时，经常报：

• 用户名重复
• 订单号重复
• 手机号重复

根因

• 用固定前缀但没有足够随机性
• 时间戳精度不够
• 多 worker 并发时命名规则冲突

处理建议

• 使用 uuid、雪花 ID 或“worker_id + 时间 + 随机串”
• 不要只依赖秒级时间戳
• 唯一字段统一由工厂生成，不要每个测试自己拼

示例：

import uuid

def unique_username(run_id):
    return f"user_{run_id}_{uuid.uuid4().hex[:8]}"

坑二：清理顺序错误导致外键删除失败

现象

删用户时报外键约束错误。

根因

先删父表，再删子表。

处理建议

按依赖顺序删除：

1. 订单
2. 支付记录
3. 优惠券使用记录
4. 用户

或者在测试环境中使用受控级联删除，但要谨慎。

示例：

DELETE FROM orders WHERE run_id = ?;
DELETE FROM users WHERE run_id = ?;

坑三：测试失败后没执行清理

现象

环境里出现大量历史测试数据。

根因

• 清理逻辑写在测试最后一行，断言失败就跳过
• 进程被 kill，finally 也来不及执行

处理建议

• 本地即时清理：try/finally 或 fixture teardown
• 服务端兜底清理：定时任务扫描 run_id + created_at
• 最好增加过期时间字段 expires_at

坑四：依赖异步任务，清理时机过早

现象

主流程结束后马上删数据，异步消费者稍后处理时找不到数据。

根因

测试把“同步事务完成”和“全链路最终完成”混为一谈。

处理建议

• 为异步链路增加完成标记
• 等待事件完成后再清理
• 或者对异步依赖使用 mock / stub

坑五：跨服务数据不一致

现象

数据库清了，但 Redis、ES、对象存储、消息队列里的数据还在。

根因

只清理了主库，没有定义“全链路测试数据边界”。

处理建议

建立测试数据资产清单，明确每类数据的归属和清理方法：

• DB：按 run_id 删除
• Redis：按 key 前缀清理
• ES：按索引字段过滤删除
• MQ：使用独立 topic / consumer group
• 文件：放测试专用 bucket/prefix

安全/性能最佳实践

测试数据管理不是只看“测得过”，还要考虑安全和性能。

安全方面

1. 不要直接使用生产数据，即使脱敏也要谨慎

原因包括：

• 字段间关联关系复杂，脱敏后不一定真实可用
• 数据体量太大，不适合日常自动化
• 合规要求越来越严格

更推荐：

• 基于模板生成测试数据
• 仅在必要场景下使用最小化脱敏样本

2. 测试账号权限最小化

测试造数账号不要直接给 DBA 级别权限。建议只开放：

• 指定 schema 的增删改查
• 测试所需存储过程权限
• 有范围限制的清理权限

3. 清理脚本必须带作用域限制

这是高危点。
删除 SQL 必须附带测试标识，不要写出这种语句：

DELETE FROM users;

正确思路：

DELETE FROM users WHERE run_id = 'run_xxx';

或者：

DELETE FROM users
WHERE created_by = 'automation'
  AND created_at < CURRENT_TIMESTAMP - INTERVAL '1 day';

性能方面

1. 避免每条用例都全量初始化环境

如果每次测试都重建整库，速度会非常慢。更高效的方法是：

• 固定基础数据快照
• 每条测试只增量造业务数据
• 测试后局部回收

2. 给清理条件建索引

如果你的删除条件经常是：

• run_id
• created_at
• created_by

那这些字段需要索引，否则测试一多，清理反而变成负担。

示例：

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);

3. 大批量造数用批处理，不要单条提交

例如压测准备数据时，优先：

• 批量插入
• 减少事务提交次数
• 分页清理

而不是 10 万条数据每条都单独 commit。

4. 区分“功能测试数据”和“性能测试数据”

两者目标不同：

• 功能测试：小、准、稳
• 性能测试：多、真、可扩展

不要拿性能测试那套巨量数据去拖慢日常回归。

一个更实用的回收策略模板

真实项目里，我通常建议使用“两阶段回收”。

阶段一：测试结束即时清理

适合大多数功能测试。

阶段二：定时兜底清理

比如每小时执行一次：

• 清理 created_by = automation
• 且 created_at 超过阈值
• 且状态不在保留名单中

可以抽象成下面这个状态流：

stateDiagram-v2
    [*] --> Created
    Created --> InUse: 测试执行
    InUse --> CleanupNow: 用例结束
    InUse --> Abandoned: 异常中断
    CleanupNow --> Deleted
    Abandoned --> CleanupLater: 定时任务扫描
    CleanupLater --> Deleted
    Deleted --> [*]

迁移到真实项目时的落地建议

如果你现在的项目还比较混乱，不建议“全量推翻重做”。更务实的方式是渐进演进。

推荐落地顺序

阶段 1：建立最小闭环

先做到：

• 自动化造数统一入口
• 所有测试数据带 run_id
• 用例结束自动清理

阶段 2：封装领域构造器

例如电商项目里，逐步沉淀：

• UserBuilder
• OrderBuilder
• CouponBuilder
• InventoryBuilder

阶段 3：把环境冲突点专项治理

针对失败率最高的地方重点处理：

• 登录态冲突：账号池
• 并发串库：schema 隔离
• 外部依赖不稳定：mock/sandbox

阶段 4：接入 CI 可观测性

建议记录这些信息：

• 本次 run_id
• 构造了哪些实体
• 清理是否成功
• 遗留数据数量
• 清理耗时

这样测试失败时，排查成本会明显下降。

总结

自动化测试中的测试数据管理，本质上是在解决三个问题：

1. 如何避免互相污染：靠环境隔离
2. 如何稳定得到所需状态：靠数据构造器
3. 如何保持环境可重复使用：靠即时回收 + 兜底回收

如果你只记住几条最实用的建议，我建议是这几条：

• 所有自动化测试数据都要可追踪，至少带 run_id
• 不要让测试依赖历史数据，要让数据“按需构造”
• 清理必须自动化，不能靠人肉收尾
• 先做逻辑隔离，再按痛点升级到账号/库/容器隔离
• 优先保证确定性，不要盲目追求“像线上”

最后说个很现实的边界条件：
如果你的系统是强分布式、强异步、多存储混合架构，那么“完全清理”和“完全隔离”的成本会很高。这时候不要追求一步到位，而是先把可追踪、可声明、可回收这三件事做好。只要这三点站稳，自动化测试的稳定性通常就会提升一个量级。