自动化测试中的测试数据管理实战：从数据构造、隔离到回收的可复用方案

做自动化测试时，很多团队一开始把注意力都放在“脚本怎么写”“框架怎么搭”上，结果跑着跑着发现，真正把测试稳定性拖垮的，往往不是断言写错，而是测试数据失控。

比如：

• 用例 A 创建了一个用户，没清掉；
• 用例 B 以为自己拿到的是“新用户”，结果撞上历史脏数据；
• 并发执行时，多个任务抢同一个手机号、同一个订单号；
• 测试环境复用了线上脱敏数据，改一次状态，其他人全挂；
• 数据回收做得随缘，数据库越来越脏，最后没人敢全量回归。

我自己就踩过这个坑：脚本在本地跑得飞起，一上 CI 就随机失败，排查半天，最后发现不是代码不稳定，而是测试数据没有隔离，前一个任务把后一个任务依赖的状态改掉了。

这篇文章不讲空泛原则，我会带你从一个中级团队可落地的角度，搭一套测试数据构造、隔离、回收的可复用方案。重点不是“最完美”，而是“你下周就能开始落地”。

背景与问题

在自动化测试里，测试数据通常会经历这几个阶段：

1. 构造：生成用户、订单、商品、优惠券等测试实体；
2. 使用：用例执行过程中读写这些数据；
3. 隔离：避免不同用例、不同任务、不同环境互相污染；
4. 回收：执行完成后清理，或者标记过期统一回收。

真正难的地方在于：这四件事往往不是分开的，而是彼此牵连。

常见问题画像

1. 数据构造方式混乱

很多项目里，测试数据来源可能同时存在：

• 手工在数据库插入
• 通过接口预置
• 从公共账号池里取
• 从固定 SQL 文件导入
• 在用例里临时拼装

短期看都能用，长期看会出现几个典型问题：

• 数据来源不一致，失败时很难定位；
• 预置逻辑散落在各个用例文件里，难复用；
• 环境一变，SQL/接口全部失效。

2. 数据状态不可预测

自动化最怕“这个账号之前登录过”“这张优惠券已经被领过”“这个订单已经支付过”。

如果一个测试用例依赖“确定状态的数据”，但又没有自己构造，就一定会被别人污染。

3. 并发执行冲突

CI 里一旦开启并发，固定手机号、固定用户名、固定业务单号会立刻撞车：

• 唯一键冲突
• 状态竞争
• 资源锁争抢
• 回收误删别人数据

4. 回收机制缺失

很多团队的回收策略靠“跑完手动删一下”，这在单人调试时还能撑住，一到团队协作就会变成灾难。

前置知识与环境准备

为了把例子写得足够直观，下面我用 Python 演示一套轻量的测试数据管理方案。你可以把它迁移到 Java、TypeScript、Go，本质思路是一样的。

示例环境

• Python 3.10+
• SQLite（方便本地直接运行）
• pytest（示意自动化测试集成思路）

安装依赖：

pip install pytest

项目结构可以先这么放：

test-data-demo/
├── app.py
├── data_manager.py
├── repository.py
├── cleanup.py
└── tests/
    └── test_order_flow.py

核心原理

如果要把测试数据管理做成“可复用方案”，我建议抓住三个原则：

原则一：测试数据要“显式创建”，不要“隐式借用”

也就是说：

• 不要默认依赖某个公共账号；
• 不要默认环境里已经有一条可用商品；
• 不要在断言里偷偷依赖历史数据。

更稳妥的方式是：每个测试只拿自己创建的数据，或者拿平台明确分配给自己的数据。

原则二：隔离优先于共享

共享数据看起来省事，实际上是制造不确定性的来源。

隔离可以分层设计：

• 命名隔离：用户名、订单号、邮箱带唯一前缀；
• 逻辑隔离：按 run_id / test_case_id 打标签；
• 物理隔离：独立 schema、独立库、独立租户；
• 时间隔离：设置 TTL，超过时间统一清理。

中级团队一般不需要一上来就做物理隔离，但至少要做到：

• 唯一标识；
• 可追踪；
• 可回收。

原则三：回收不是补丁，而是设计的一部分

很多人把回收放在最后考虑，结果就是“先跑起来再说”。但实际上，回收要在数据模型设计阶段就想好：

• 用什么字段追踪测试数据？
• 怎么区分自动化生成和人工数据？
• 删除失败怎么办？
• 是立即删除还是延迟清理？
• 回收任务误删的边界怎么控？

一套可落地的方案设计

这里给出一个实战里很常用、成本也不高的方案：

方案目标

1. 自动生成唯一测试数据；
2. 给每一条测试数据打上 run_id 和 created_by=test_automation；
3. 用例执行完成后优先按 run_id 回收；
4. 即使回收漏了，也能通过定时任务按 TTL 兜底清理。

数据流总览

flowchart TD
    A[测试启动] --> B[生成 run_id]
    B --> C[DataManager 构造测试数据]
    C --> D[写入业务表并记录元信息]
    D --> E[测试用例执行]
    E --> F{执行后是否立即清理}
    F -->|是| G[按 run_id 删除测试数据]
    F -->|否| H[保留到 TTL 回收]
    G --> I[输出清理结果]
    H --> J[定时清理任务扫描过期数据]
    J --> I

关键对象关系

classDiagram
    class TestDataContext {
        +run_id: str
        +case_id: str
        +created_by: str
        +expire_at: datetime
    }

    class DataManager {
        +create_user()
        +create_order()
        +cleanup_by_run_id()
    }

    class Repository {
        +init_db()
        +insert_user()
        +insert_order()
        +find_by_run_id()
        +delete_by_run_id()
    }

    DataManager --> TestDataContext
    DataManager --> Repository

字段设计建议

无论你是数据库表，还是调用后端接口创建数据，都建议统一携带这些元信息：

这些字段不一定都在业务核心表里，也可以存到一张测试数据追踪表中。

实战代码（可运行）

下面我们实现一个最小可运行示例：自动创建用户和订单，并支持按 run_id 回收。

第一步：定义数据访问层

文件：repository.py

import sqlite3
from datetime import datetime
from typing import List, Tuple


class Repository:
    def __init__(self, db_path: str = "test_demo.db"):
        self.db_path = db_path

    def get_conn(self):
        return sqlite3.connect(self.db_path)

    def init_db(self):
        with self.get_conn() as conn:
            cursor = conn.cursor()
            cursor.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                username TEXT UNIQUE NOT NULL,
                email TEXT UNIQUE NOT NULL,
                run_id TEXT NOT NULL,
                case_id TEXT NOT NULL,
                created_by TEXT NOT NULL,
                expire_at TEXT NOT NULL,
                created_at TEXT NOT NULL
            )
            """)
            cursor.execute("""
            CREATE TABLE IF NOT EXISTS orders (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                order_no TEXT UNIQUE NOT NULL,
                user_id INTEGER NOT NULL,
                amount REAL NOT NULL,
                status TEXT NOT NULL,
                run_id TEXT NOT NULL,
                case_id TEXT NOT NULL,
                created_by TEXT NOT NULL,
                expire_at TEXT NOT NULL,
                created_at TEXT NOT NULL,
                FOREIGN KEY(user_id) REFERENCES users(id)
            )
            """)
            conn.commit()

    def insert_user(
        self,
        username: str,
        email: str,
        run_id: str,
        case_id: str,
        created_by: str,
        expire_at: str
    ) -> int:
        now = datetime.utcnow().isoformat()
        with self.get_conn() as conn:
            cursor = conn.cursor()
            cursor.execute("""
            INSERT INTO users (username, email, run_id, case_id, created_by, expire_at, created_at)
            VALUES (?, ?, ?, ?, ?, ?, ?)
            """, (username, email, run_id, case_id, created_by, expire_at, now))
            conn.commit()
            return cursor.lastrowid

    def insert_order(
        self,
        order_no: str,
        user_id: int,
        amount: float,
        status: str,
        run_id: str,
        case_id: str,
        created_by: str,
        expire_at: str
    ) -> int:
        now = datetime.utcnow().isoformat()
        with self.get_conn() as conn:
            cursor = conn.cursor()
            cursor.execute("""
            INSERT INTO orders (order_no, user_id, amount, status, run_id, case_id, created_by, expire_at, created_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (order_no, user_id, amount, status, run_id, case_id, created_by, expire_at, now))
            conn.commit()
            return cursor.lastrowid

    def find_users_by_run_id(self, run_id: str) -> List[Tuple]:
        with self.get_conn() as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT * FROM users WHERE run_id = ?", (run_id,))
            return cursor.fetchall()

    def find_orders_by_run_id(self, run_id: str) -> List[Tuple]:
        with self.get_conn() as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT * FROM orders WHERE run_id = ?", (run_id,))
            return cursor.fetchall()

    def delete_orders_by_run_id(self, run_id: str) -> int:
        with self.get_conn() as conn:
            cursor = conn.cursor()
            cursor.execute("DELETE FROM orders WHERE run_id = ?", (run_id,))
            count = cursor.rowcount
            conn.commit()
            return count

    def delete_users_by_run_id(self, run_id: str) -> int:
        with self.get_conn() as conn:
            cursor = conn.cursor()
            cursor.execute("DELETE FROM users WHERE run_id = ?", (run_id,))
            count = cursor.rowcount
            conn.commit()
            return count

    def delete_expired_orders(self, now_iso: str) -> int:
        with self.get_conn() as conn:
            cursor = conn.cursor()
            cursor.execute("""
            DELETE FROM orders
            WHERE created_by = 'test_automation' AND expire_at < ?
            """, (now_iso,))
            count = cursor.rowcount
            conn.commit()
            return count

    def delete_expired_users(self, now_iso: str) -> int:
        with self.get_conn() as conn:
            cursor = conn.cursor()
            cursor.execute("""
            DELETE FROM users
            WHERE created_by = 'test_automation' AND expire_at < ?
            """, (now_iso,))
            count = cursor.rowcount
            conn.commit()
            return count

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

文件：data_manager.py

import uuid
from dataclasses import dataclass
from datetime import datetime, timedelta
from repository import Repository


@dataclass
class TestDataContext:
    run_id: str
    case_id: str
    created_by: str = "test_automation"
    ttl_hours: int = 6

    @property
    def expire_at(self) -> str:
        return (datetime.utcnow() + timedelta(hours=self.ttl_hours)).isoformat()


class DataManager:
    def __init__(self, repo: Repository, context: TestDataContext):
        self.repo = repo
        self.context = context

    @staticmethod
    def new_run_id() -> str:
        return f"run_{uuid.uuid4().hex[:8]}"

    def create_user(self) -> dict:
        suffix = uuid.uuid4().hex[:8]
        username = f"u_{self.context.run_id}_{suffix}"
        email = f"{username}@test.local"

        user_id = self.repo.insert_user(
            username=username,
            email=email,
            run_id=self.context.run_id,
            case_id=self.context.case_id,
            created_by=self.context.created_by,
            expire_at=self.context.expire_at
        )
        return {
            "id": user_id,
            "username": username,
            "email": email
        }

    def create_order(self, user_id: int, amount: float = 99.0) -> dict:
        suffix = uuid.uuid4().hex[:10]
        order_no = f"O{self.context.run_id[-4:]}{suffix}"

        order_id = self.repo.insert_order(
            order_no=order_no,
            user_id=user_id,
            amount=amount,
            status="CREATED",
            run_id=self.context.run_id,
            case_id=self.context.case_id,
            created_by=self.context.created_by,
            expire_at=self.context.expire_at
        )
        return {
            "id": order_id,
            "order_no": order_no,
            "amount": amount,
            "status": "CREATED"
        }

    def cleanup_by_run_id(self) -> dict:
        deleted_orders = self.repo.delete_orders_by_run_id(self.context.run_id)
        deleted_users = self.repo.delete_users_by_run_id(self.context.run_id)
        return {
            "run_id": self.context.run_id,
            "deleted_orders": deleted_orders,
            "deleted_users": deleted_users
        }

第三步：模拟业务入口

文件：app.py

from repository import Repository
from data_manager import DataManager, TestDataContext


def main():
    repo = Repository()
    repo.init_db()

    run_id = DataManager.new_run_id()
    context = TestDataContext(run_id=run_id, case_id="test_create_order")
    manager = DataManager(repo, context)

    user = manager.create_user()
    order = manager.create_order(user_id=user["id"], amount=199.0)

    print("创建用户:", user)
    print("创建订单:", order)

    users = repo.find_users_by_run_id(run_id)
    orders = repo.find_orders_by_run_id(run_id)

    print("当前 run_id 下用户数量:", len(users))
    print("当前 run_id 下订单数量:", len(orders))

    result = manager.cleanup_by_run_id()
    print("清理结果:", result)


if __name__ == "__main__":
    main()

运行：

python app.py

你会看到完整流程：

• 初始化数据库
• 生成唯一 run_id
• 创建测试用户、测试订单
• 查询验证
• 按 run_id 回收

第四步：增加定时兜底回收

文件：cleanup.py

from datetime import datetime
from repository import Repository


def cleanup_expired():
    repo = Repository()
    repo.init_db()

    now_iso = datetime.utcnow().isoformat()
    deleted_orders = repo.delete_expired_orders(now_iso)
    deleted_users = repo.delete_expired_users(now_iso)

    print({
        "deleted_orders": deleted_orders,
        "deleted_users": deleted_users,
        "cleanup_time": now_iso
    })


if __name__ == "__main__":
    cleanup_expired()

这个脚本适合挂到：

• CI 后置任务
• Jenkins 定时任务
• GitLab Scheduler
• K8s CronJob

第五步：接入 pytest

文件：tests/test_order_flow.py

from repository import Repository
from data_manager import DataManager, TestDataContext


def test_create_order_and_cleanup():
    repo = Repository()
    repo.init_db()

    run_id = DataManager.new_run_id()
    context = TestDataContext(run_id=run_id, case_id="test_create_order_and_cleanup")
    manager = DataManager(repo, context)

    user = manager.create_user()
    order = manager.create_order(user_id=user["id"], amount=88.8)

    users = repo.find_users_by_run_id(run_id)
    orders = repo.find_orders_by_run_id(run_id)

    assert len(users) == 1
    assert len(orders) == 1
    assert order["amount"] == 88.8

    result = manager.cleanup_by_run_id()
    assert result["deleted_orders"] == 1
    assert result["deleted_users"] == 1

执行：

pytest -q

逐步验证清单

如果你准备把这套思路搬进真实项目，我建议按下面顺序验证，不要一口气大改。

第 1 步：先统一唯一命名规则

确保所有自动化创建的数据都带唯一前缀，比如：

• 用户名：u_{run_id}_{random}
• 手机号：预留号段 + 随机数
• 订单号：测试前缀 + 时间戳/随机串

第 2 步：补充追踪字段

至少保证能知道：

• 这条数据是不是测试生成的
• 是哪次任务生成的
• 什么时候过期

第 3 步：先实现按 run_id 精准回收

不要一上来就“清库”，容易误伤。

第 4 步：再补 TTL 兜底

因为 CI 中断、脚本异常退出、网络闪断都可能导致即时清理没执行到。

第 5 步：最后再考虑更高阶隔离

例如：

• 独立租户
• 独立测试 schema
• 按分支创建临时环境

隔离设计再深入一点：三种常见层级

不同团队复杂度不同，不需要都用最重的方案。

1. 轻量级：标识隔离

适合中小团队、单环境自动化。

手段：

• 唯一命名
• run_id
• created_by
• TTL 清理

优点：

• 成本低
• 快速落地

缺点：

• 仍共享同一套环境资源
• 会受环境整体稳定性影响

2. 中量级：租户/命名空间隔离

适合有多项目并发测试的团队。

手段：

• 每条测试任务使用独立租户
• 不同任务共享应用，但业务数据逻辑隔离

优点：

• 数据污染显著下降

缺点：

• 需要业务系统支持租户维度隔离

3. 重量级：临时环境隔离

适合集成测试、端到端回归、发布前验证。

手段：

• 每次流水线临时创建数据库/schema/服务环境
• 用完整体销毁

优点：

• 隔离最彻底

缺点：

• 成本高
• 环境编排复杂

用例执行中的时序关系

下面这张图能帮助你理解，为什么“即时清理 + TTL 兜底”通常比只靠一种方式更稳。

sequenceDiagram
    participant T as TestRunner
    participant D as DataManager
    participant DB as Database
    participant C as CleanupJob

    T->>D: 初始化 context(run_id, case_id)
    D->>DB: 创建用户/订单（带 run_id, expire_at）
    T->>DB: 执行业务操作与断言
    alt 用例正常结束
        T->>D: cleanup_by_run_id()
        D->>DB: 按 run_id 删除数据
    else 用例异常中断
        Note over T,D: 即时清理未执行
    end
    C->>DB: 扫描 expire_at < now 的测试数据
    C->>DB: 兜底删除

常见坑与排查

这一节我尽量写得接地气一点，因为这些问题在真实项目里太常见了。

坑 1：唯一 ID 看似唯一，实际并发下仍冲突

比如你用了：

username = f"test_{int(time.time())}"

如果多个任务在同一秒执行，还是会撞。

建议

使用以下组合之一：

• UUID
• 时间戳 + 随机串
• CI job id + case id + 随机串

更稳妥一点：

import uuid
username = f"u_{uuid.uuid4().hex[:12]}"

坑 2：回收顺序不对，外键删除失败

如果订单依赖用户，先删用户就会失败，或者需要级联删除。

排查思路

1. 看数据库外键约束；
2. 看删除顺序；
3. 确认有没有部分业务表漏清理。

建议

统一维护删除拓扑，子表先删，主表后删。

坑 3：只清业务表，没清索引表/日志表/扩展表

一个业务实体在系统里往往不是一张表。比如用户可能关联：

• 用户主表
• 用户画像表
• 认证表
• 地址表
• 权限关系表

只删一张表，表面上数据没了，实际上唯一索引或者关联状态还在，下次继续失败。

建议

做一份“测试数据影响面清单”，至少列出：

• 主实体表
• 关系表
• 异步任务表
• 缓存键
• 搜索索引

坑 4：接口创建了数据，但异步链路还没完成

例如创建订单后，库存、营销、消息、搜索索引都是异步更新。你立刻断言，很可能时好时坏。

排查思路

• 看消息队列消费延迟；
• 看异步任务调度时间；
• 看最终一致性窗口。

建议

不要盲目 sleep(3)，优先使用带超时的轮询等待。

示例：

import time

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

坑 5：TTL 清理误删了非测试数据

这是非常危险的一类问题。

比如你只根据“创建时间早于 6 小时”删除，却没限制 created_by=test_automation，就可能删到正常业务数据。

建议

回收条件至少包含两个维度：

• created_by = test_automation
• expire_at < now

必要时再加环境前缀或租户限制。

坑 6：用例失败后没清理，越跑越脏

最常见原因是清理代码写在测试逻辑最后，但异常时根本不会执行。

建议

在测试框架层做兜底，比如 pytest fixture 的 yield 后清理。

示例：

import pytest
from repository import Repository
from data_manager import DataManager, TestDataContext


@pytest.fixture
def data_manager():
    repo = Repository()
    repo.init_db()
    run_id = DataManager.new_run_id()
    context = TestDataContext(run_id=run_id, case_id="fixture_case")
    manager = DataManager(repo, context)
    yield manager
    manager.cleanup_by_run_id()

安全/性能最佳实践

测试数据管理不只是“能跑”，还要考虑安全和性能边界。

安全最佳实践

1. 不要直接使用生产数据做自动化

哪怕做了脱敏，也要谨慎。原因包括：

• 真实关系复杂，状态不可控；
• 脱敏不彻底可能带来合规风险；
• 测试修改可能破坏样本一致性。

更好的方式是：

• 用模板数据生成器生成；
• 用最小业务闭环数据集；
• 仅在专用测试环境使用。

2. 给自动化账号最小权限

自动化如果需要清理数据，不代表它应该拥有全库删除权限。

建议：

• 限制到测试环境；
• 限制到特定 schema/table；
• 限制删除条件必须带测试标识。

3. 回收脚本要有防误删保护

我会建议至少加这些保护：

• 只能在测试环境运行；
• 必须检查环境变量；
• 必须限制 created_by=test_automation；
• 删除前输出影响行数；
• 超过阈值时中断执行。

示例：

import os

def ensure_test_env():
    env = os.getenv("APP_ENV", "")
    if env != "test":
        raise RuntimeError(f"危险操作被拦截，当前环境不是 test，而是: {env}")

性能最佳实践

1. 批量构造、批量回收

如果你一次回归要构造上千条数据，不要逐条 HTTP 调接口再逐条删。

优先考虑：

• 批量插入
• 批量删除
• 分页清理

2. 不要为每个测试都创建整套重资产数据

比如商品目录、门店、仓库这种基础数据成本高，可以采用：

• 基础主数据共享只读
• 易变交易数据按 run_id 独立创建

这个边界很重要：共享可以，但共享的必须是稳定且只读的。

3. 避免把清理做成全表扫描

如果回收依赖 run_id、created_by、expire_at，建议给这些字段加索引。

示例：

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_users_expire_at ON users(expire_at);
CREATE INDEX IF NOT EXISTS idx_orders_expire_at ON orders(expire_at);

4. 把“造数”从测试断言逻辑中抽出来

如果每个用例自己写一遍造数逻辑，会导致：

• 重复代码多
• 接口调用多
• 构造过程难统一优化

正确做法是统一收敛到 DataManager 或工厂层。

一个更实用的分层建议

如果你的项目规模比示例再大一点，我建议按下面分层：

1. Test Data Factory

负责“造什么数据”。

例如：

• create_new_user()
• create_paid_order()
• create_coupon_with_stock()

2. State Builder

负责“把数据推进到什么状态”。

例如：

• 新订单 -> 已支付
• 已支付 -> 已发货
• 用户 -> 已实名

3. Data Tracker

负责记录：

• 哪次任务造了哪些数据
• 这些数据分布在哪些表/服务

4. Cleanup Executor

负责清理和兜底回收。

这样分层后，自动化脚本写起来会很清晰：

paid_order = order_factory.create_paid_order()
assert paid_order["status"] == "PAID"

而不是每个用例都手写：

• 创建用户
• 绑卡
• 创建商品
• 下单
• 支付
• 推消息
• 改库存
• 清理一堆关联表

边界条件：什么时候这套方案不够用？

这套“唯一标识 + run_id + TTL 回收”的方案很好用，但它不是万能的。

以下场景可能需要升级：

1. 强依赖异步链路且副作用广

比如一个订单会触发：

• 多服务状态联动
• 搜索索引更新
• 风控侧写
• 财务流水

这时仅靠数据库清理可能不够，需要把外部副作用也纳入回收设计。

2. 环境多人共享且资源竞争激烈

例如库存、优惠券额度、配额资源等天然是共享竞争资源，逻辑隔离不足以解决问题，可能需要租户级甚至环境级隔离。

3. 合规场景要求极严

如果测试环境也受审计要求约束，必须对测试数据的创建、访问、删除做完整留痕，那就要进一步加入审计日志和权限控制。

总结

把自动化测试做稳，测试数据管理几乎是绕不过去的一关。

如果你只记住一句话，我希望是这句：

测试数据不是脚本的附属品，而是自动化系统的一等公民。

一套实用、可复用的方案，至少应该做到：

• 构造可控：测试数据由统一入口创建；
• 隔离清晰：每次运行有唯一 run_id；
• 状态可追踪：带 case_id、created_by、expire_at；
• 回收可靠：即时清理 + TTL 兜底；
• 风险可控：只清测试数据，不误删业务数据。

如果你准备在团队里推进，我建议从这三个动作开始，最容易见效：

1. 给自动化创建的数据统一加 run_id 和 created_by；
2. 把常用造数逻辑收敛到一个 DataManager；
3. 落一个定时清理任务，先把环境“止脏”。

别小看这三步，它们通常就能消灭掉一大半“随机失败”“环境被污染”“回归不敢并发”的问题。

当这套基础打稳之后，你再往租户隔离、临时环境、状态工厂这些方向升级，成本会低很多，也更容易被团队接受。