背景与问题

做自动化测试的人，通常都会在两个地方反复摔跤：

1. 接口测试依赖固定数据，今天能跑，明天因为数据被改掉就红了。
2. 端到端用例链路长，注册、下单、支付、发货一串下来，只要某个前置数据状态不对，整条用例就废。

很多团队一开始会这么做：

• 在测试环境里准备一批“公共测试账号”
• 用 SQL 脚本预置一些订单、商品、优惠券
• 把这些 ID 直接写进测试代码里

这套办法在用例少的时候看起来没问题，但一旦并发执行、环境共享、版本频繁迭代，问题会很快暴露：

• 数据被复用后状态污染
• 不同用例互相踩数据
• 环境重置后主键、配置、权限全变
• 接口测试和 E2E 测试对同一批数据有不同诉求
• 排查失败时，不知道是代码问题还是数据问题

我之前在一个订单系统项目里就踩过这个坑：UI 自动化用一个“标准已实名用户”做下单，接口回归也用这个用户做优惠券核销。结果某天夜里并发跑回归，账号余额和券状态被多套用例改乱，第二天看报告一片红，但真正的产品逻辑其实没坏。

这时候，测试数据工厂（Test Data Factory） 就不是“锦上添花”，而是让自动化体系可持续运行的基础设施。

什么是测试数据工厂

简单说，测试数据工厂不是一堆 SQL 脚本，而是一层按业务语义创建测试数据的能力。
它的核心目标有三个：

• 按需创建：测试要什么状态的数据，就生成什么状态
• 隔离复用：每条用例尽量拥有独立数据，避免互相污染
• 可追踪清理：知道数据从哪来、给谁用、何时销毁

比如，不再在代码里写死：

使用用户 ID=10086、商品 ID=20001、优惠券 ID=30009

而是写成：

创建一个“已实名、已绑卡、账户余额充足”的用户
创建一个“可售、库存充足、支持优惠券”的商品
创建一张“对该用户可用、未过期、满减门槛满足”的券

这样做的变化非常大：测试代码开始描述业务条件，而不是依赖偶然存在的数据。

核心原理

测试数据工厂要真正落地，建议把它拆成四层来看。

1. 领域语义层：数据不是表记录，而是业务状态

测试不应该关心数据库有几张表，而应该关心业务对象处于什么状态。
例如“可下单用户”背后可能涉及：

• 用户表
• 实名信息表
• 钱包账户表
• 风控白名单
• 默认收货地址

测试代码不需要知道这些细节，它只需要表达“给我一个可下单用户”。

flowchart LR
    A[测试用例] --> B[数据工厂 API]
    B --> C[领域构造器<br/>如 create_orderable_user]
    C --> D[仓储/服务适配层]
    D --> E[(DB)]
    D --> F[内部服务 API]
    D --> G[消息/缓存]

2. 组合式构建：基础数据 + 状态装配

好的测试数据工厂通常不是“一把梭”式的大函数，而是可以拼装：

• 创建基础用户
• 实名
• 绑卡
• 充值余额
• 分配优惠券
• 调整风控状态

这样既能复用，也能针对特殊场景做精细控制。

3. 幂等与唯一性：避免并发测试撞车

接口测试和 E2E 测试最怕共享数据。
所以工厂创建的数据最好具备：

• 唯一业务标识，如带时间戳/UUID 的手机号、订单号、商品编码
• 可重复执行，重复调用不会产生不可控副作用
• 隔离标签，如 run_id、case_id、suite_name，便于追踪和清理

4. 生命周期管理：创建、使用、清理

测试数据如果只会创建不会清理，环境很快会变成垃圾场。
一套完整方案至少要回答：

• 谁创建的？
• 给哪条用例使用？
• 用完是否要删？
• 失败中断后怎么回收？
• 哪些数据适合短生命周期，哪些适合保留用于排障？

适合接口测试与端到端测试的落地模式

接口测试和 E2E 测试虽然都依赖数据，但关注点不一样。

接口测试更关注

• 前置状态可控
• 构造速度快
• 数据粒度精细
• 易于参数化覆盖边界

E2E 测试更关注

• 链路上下文完整
• 业务对象关系真实
• UI 可见状态一致
• 一次创建，多步骤复用

所以一个实用方案通常是：

• 接口测试：偏向直接通过服务层/数据库/API 组合创建状态
• E2E 测试：偏向创建更贴近真实业务流程的数据种子

sequenceDiagram
    participant T as 测试用例
    participant F as 测试数据工厂
    participant S as 领域服务
    participant D as 数据库
    participant U as 被测系统接口/UI

    T->>F: 请求创建“可支付订单”
    F->>S: 创建用户/商品/库存/优惠券
    S->>D: 写入基础数据
    S-->>F: 返回订单上下文
    F-->>T: user_id/order_id/token
    T->>U: 执行支付接口或 UI 操作
    U->>D: 更新订单状态
    T-->>F: 记录用例结束

设计一套最小可用的数据工厂

下面我用 Python 演示一个“够用且可运行”的最小实现。它不是生产级框架，但能说明落地思路。

我们模拟一个电商系统中的三个对象：

• 用户
• 商品
• 订单

目标是让测试可以一句话拿到：

• 可下单用户
• 可支付订单

实战代码（可运行）

目录结构示意

test_data_factory_demo/
├── app.py
├── factory.py
└── tests.py

1）数据工厂实现

# factory.py
import sqlite3
import uuid
import time
from dataclasses import dataclass


@dataclass
class User:
    id: int
    username: str
    is_verified: bool
    balance: int


@dataclass
class Product:
    id: int
    name: str
    stock: int
    price: int
    is_active: bool


@dataclass
class Order:
    id: int
    user_id: int
    product_id: int
    amount: int
    status: str


class TestDataFactory:
    def __init__(self, conn: sqlite3.Connection, run_id: str = None):
        self.conn = conn
        self.run_id = run_id or f"run_{int(time.time())}_{uuid.uuid4().hex[:6]}"

    def init_schema(self):
        cur = self.conn.cursor()
        cur.execute("""
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT UNIQUE NOT NULL,
            is_verified INTEGER NOT NULL,
            balance INTEGER NOT NULL,
            run_id TEXT NOT NULL
        )
        """)
        cur.execute("""
        CREATE TABLE IF NOT EXISTS products (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT UNIQUE NOT NULL,
            stock INTEGER NOT NULL,
            price INTEGER NOT NULL,
            is_active INTEGER NOT NULL,
            run_id TEXT NOT NULL
        )
        """)
        cur.execute("""
        CREATE TABLE IF NOT EXISTS orders (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id INTEGER NOT NULL,
            product_id INTEGER NOT NULL,
            amount INTEGER NOT NULL,
            status TEXT NOT NULL,
            run_id TEXT NOT NULL,
            FOREIGN KEY(user_id) REFERENCES users(id),
            FOREIGN KEY(product_id) REFERENCES products(id)
        )
        """)
        self.conn.commit()

    def create_user(self, verified: bool = False, balance: int = 0) -> User:
        username = f"u_{uuid.uuid4().hex[:8]}"
        cur = self.conn.cursor()
        cur.execute(
            "INSERT INTO users(username, is_verified, balance, run_id) VALUES (?, ?, ?, ?)",
            (username, int(verified), balance, self.run_id)
        )
        self.conn.commit()
        user_id = cur.lastrowid
        return User(id=user_id, username=username, is_verified=verified, balance=balance)

    def create_product(self, stock: int = 10, price: int = 100, active: bool = True) -> Product:
        name = f"p_{uuid.uuid4().hex[:8]}"
        cur = self.conn.cursor()
        cur.execute(
            "INSERT INTO products(name, stock, price, is_active, run_id) VALUES (?, ?, ?, ?, ?)",
            (name, stock, price, int(active), self.run_id)
        )
        self.conn.commit()
        product_id = cur.lastrowid
        return Product(id=product_id, name=name, stock=stock, price=price, is_active=active)

    def create_order(self, user_id: int, product_id: int, amount: int, status: str = "CREATED") -> Order:
        cur = self.conn.cursor()
        cur.execute(
            "INSERT INTO orders(user_id, product_id, amount, status, run_id) VALUES (?, ?, ?, ?, ?)",
            (user_id, product_id, amount, status, self.run_id)
        )
        self.conn.commit()
        order_id = cur.lastrowid
        return Order(id=order_id, user_id=user_id, product_id=product_id, amount=amount, status=status)

    def create_orderable_user(self) -> User:
        return self.create_user(verified=True, balance=1000)

    def create_sellable_product(self, price: int = 100) -> Product:
        return self.create_product(stock=100, price=price, active=True)

    def create_payable_order(self) -> dict:
        user = self.create_orderable_user()
        product = self.create_sellable_product(price=120)
        order = self.create_order(user.id, product.id, amount=120, status="CREATED")
        return {
            "user": user,
            "product": product,
            "order": order
        }

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

2）模拟被测业务接口

# app.py
import sqlite3


def pay_order(conn: sqlite3.Connection, order_id: int) -> dict:
    cur = conn.cursor()

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

    if not row:
        return {"ok": False, "error": "ORDER_NOT_FOUND"}

    _, user_id, product_id, amount, status, balance, stock, is_active = row

    if status != "CREATED":
        return {"ok": False, "error": "INVALID_ORDER_STATUS"}

    if not is_active:
        return {"ok": False, "error": "PRODUCT_INACTIVE"}

    if stock <= 0:
        return {"ok": False, "error": "OUT_OF_STOCK"}

    if balance < amount:
        return {"ok": False, "error": "INSUFFICIENT_BALANCE"}

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

    return {"ok": True, "order_id": order_id, "status": "PAID"}

3）测试用例示例

# tests.py
import sqlite3
from factory import TestDataFactory
from app import pay_order


def test_pay_order_success():
    conn = sqlite3.connect(":memory:")
    factory = TestDataFactory(conn)
    factory.init_schema()

    data = factory.create_payable_order()
    result = pay_order(conn, data["order"].id)

    assert result["ok"] is True
    assert result["status"] == "PAID"

    factory.cleanup()
    conn.close()


def test_pay_order_insufficient_balance():
    conn = sqlite3.connect(":memory:")
    factory = TestDataFactory(conn)
    factory.init_schema()

    user = factory.create_user(verified=True, balance=10)
    product = factory.create_product(stock=10, price=100, active=True)
    order = factory.create_order(user.id, product.id, amount=100, status="CREATED")

    result = pay_order(conn, order.id)

    assert result["ok"] is False
    assert result["error"] == "INSUFFICIENT_BALANCE"

    factory.cleanup()
    conn.close()


if __name__ == "__main__":
    test_pay_order_success()
    test_pay_order_insufficient_balance()
    print("All tests passed.")

4）如何运行

python tests.py

如果输出：

All tests passed.

说明这个最小版测试数据工厂已经能支持：

• 根据业务状态创建数据
• 给测试返回明确上下文
• 用 run_id 进行隔离和清理

这套写法为什么比“手写 SQL”更稳

关键不在于“封装了一层”，而在于它解决了自动化测试中的几个结构性问题。

1. 测试意图更清晰

对比一下：

手写 SQL/硬编码方式

user_id = 12
product_id = 98
order_id = 305

数据工厂方式

data = factory.create_payable_order()

后者读起来就是业务语言，别人接手时也更容易理解。

2. 更容易适配接口和 E2E 的不同粒度

接口测试可以只拿某个对象：

user = factory.create_orderable_user()

E2E 测试可以拿一整套上下文：

ctx = factory.create_payable_order()

3. 更利于环境迁移

如果测试环境从直连数据库改成只能调内部服务，测试代码可以不变，只替换工厂底层实现。

进阶设计：把工厂做成“状态模板”

当系统越来越复杂时，我更推荐把工厂抽象成“状态模板”而不是“对象创建器”。

比如：

• create_user_with_risk_blocked()
• create_order_waiting_refund()
• create_shipped_order_with_invoice()

这些模板不是偷懒，反而是对业务状态的显式沉淀。
它能帮助测试团队把“经常要用的场景状态”固化下来，减少重复拼装。

classDiagram
    class TestDataFactory {
        +create_user()
        +create_product()
        +create_order()
        +create_orderable_user()
        +create_payable_order()
        +cleanup()
    }

    class UserTemplate {
        +verified_user()
        +blocked_user()
        +low_balance_user()
    }

    class OrderTemplate {
        +created_order()
        +paid_order()
        +refunding_order()
    }

    TestDataFactory --> UserTemplate
    TestDataFactory --> OrderTemplate

在真实项目中的接入方式

实际项目里，测试数据工厂常见有三种接入方式。

方案一：测试代码内嵌工厂

适合：

• 单体项目
• 团队规模不大
• 测试环境可直连数据库

优点：

• 起步快
• 维护成本低

缺点：

• 与具体服务耦合较重
• 跨项目复用差

方案二：独立测试数据服务

把数据工厂做成内部服务，例如：

• POST /test-data/users/orderable
• POST /test-data/orders/payable

适合：

• 微服务架构
• 多团队共享测试环境
• 接口测试、E2E、性能测试都要复用数据能力

优点：

• 统一标准
• 便于权限控制和审计

缺点：

• 初期建设成本更高
• 需要版本治理

方案三：Fixture + 工厂混合模式

这是很多团队比较容易落地的折中方案：

• 固定大对象用 fixture，比如基础组织、默认商户、地区配置
• 易污染对象用工厂动态创建，比如用户、订单、优惠券

这个组合通常最实用。

常见坑与排查

这一部分很重要，因为测试数据工厂“写出来”和“长期稳定运行”完全是两回事。

坑 1：工厂创建的数据和真实业务链路不一致

表现：

• 接口测通过了，UI/E2E 却失败
• 数据库里状态看起来对，但页面上不显示

原因通常是：

• 只插了数据库，没触发消息、缓存、索引刷新
• 业务读的是衍生表、搜索索引或缓存

排查建议

1. 明确被测功能的数据读取路径
2. 区分“源数据”和“可见数据”
3. 对必须经过业务流程的数据，优先通过内部服务创建，而不是直接写库

我个人的经验是：凡是涉及搜索、推荐、报表、异步状态聚合，直接写库几乎一定会踩坑。

坑 2：数据唯一性不足，并发执行互相污染

表现：

• 本地单跑稳定，CI 并发跑偶发失败
• 某些用户名、订单号、手机号重复

排查建议

• 所有业务唯一字段统一加随机后缀
• 每次执行分配 run_id
• 清理时按 run_id 精准删除，不要全表扫

示例：

run_id = f"build_1024_case_18"

坑 3：清理机制不可靠，环境越来越脏

表现：

• 测试环境数据量暴涨
• 某些旧数据被新用例误命中
• 查询越来越慢

排查建议

• 创建时写入 created_by/run_id/expire_at
• 增加兜底定时清理任务
• 对无法删除的数据做归档或软删除标记

可以在数据库层面准备一个清理 SQL：

DELETE FROM orders
WHERE run_id LIKE 'run_%'
  AND id IN (
    SELECT id FROM orders
    WHERE run_id LIKE 'run_%'
  );

真实项目里，建议按业务表依赖顺序清理，避免外键报错。

坑 4：工厂 API 越写越大，最后没人敢改

表现：

• 一个 create_full_business_context() 包打天下
• 参数几十个，默认值复杂到没人看得懂

排查建议

• 保持基础构建器小而稳
• 常用场景抽模板
• 稀有场景在测试代码中局部拼装
• 每新增一个模板，都问一句：它是“高频业务状态”还是“临时方便”？

坑 5：测试失败后无法复现现场

表现：

• 报告里只写“下单失败”
• 但不知道当时创建了哪个用户、商品、订单

排查建议

• 在日志里输出工厂返回的关键 ID
• 保留失败场景的上下文快照
• 失败时可配置“不自动清理”，方便复盘

例如：

print({
    "run_id": factory.run_id,
    "user_id": data["user"].id,
    "product_id": data["product"].id,
    "order_id": data["order"].id
})

安全/性能最佳实践

测试数据工厂虽然主要服务测试，但它本身也是系统的一部分，安全和性能不能忽略。

安全最佳实践

1. 不要在工厂里绕过所有鉴权边界

很多团队为了方便，会给测试数据服务开“超级权限”。短期省事，长期危险。
建议：

• 仅在测试环境开放
• 通过白名单或 VPN 访问
• 记录调用人、调用来源、创建内容
• 对高风险操作加开关，例如批量删库、强制改订单状态

2. 脱敏与伪造数据要分清

• 用户名、手机号、身份证号等应使用专用测试号段或伪造规则
• 不要导入生产真实用户数据做自动化回归
• 日志输出避免泄露敏感字段

3. 给测试数据打标签

至少打上这些元数据：

• run_id
• suite_name
• case_name
• created_at
• created_by

后面做审计、清理、排障都非常有用。

性能最佳实践

1. 分层创建，避免每条用例都从零起环境

可以把数据分成两类：

• 稳定基础数据：组织、类目、支付渠道、仓库等，预置
• 易变业务数据：用户、订单、优惠券等，动态创建

这样既快，又能减少污染。

2. 能批量创建就不要循环单条创建

例如一次要准备 100 个用户时，优先批量插入或批量调用内部服务。

users = [factory.create_user(verified=True, balance=100) for _ in range(10)]

如果底层支持，进一步做成真正批量接口会更高效。

3. 为清理和查询设计索引

如果你按 run_id 清理数据，那就应该给 run_id 建索引。
否则测试跑得越多，清理越慢。

CREATE INDEX IF NOT EXISTS idx_users_run_id ON users(run_id);
CREATE INDEX IF NOT EXISTS idx_products_run_id ON products(run_id);
CREATE INDEX IF NOT EXISTS idx_orders_run_id ON orders(run_id);

4. 减少不必要的跨服务构造

有些测试只验证订单金额计算，就别每次都真的走会员、积分、营销、风控全链路。
原则是：

• 验证哪里，就把数据精确构造到哪里
• 只有 E2E 场景才追求整链真实

一套可执行的落地步骤

如果你所在团队现在还没有测试数据工厂，我建议不要一口气做成“大平台”，可以按下面步骤推进。

第一步：先找最痛的 3 类脆弱用例

例如：

• 注册登录
• 下单支付
• 退款售后

挑最常红、最依赖状态、最难排查的场景先做。

第二步：提炼业务状态，而不是表结构

不要从“有哪几张表”开始设计，而要从“测试常需要什么状态”开始：

• 可登录用户
• 可下单用户
• 已支付订单
• 待退款订单

第三步：先统一元数据规范

哪怕还没做平台，也先统一：

• 每条数据都有 run_id
• 每条用例输出关键对象 ID
• 每次执行有清理策略

第四步：先服务接口测试，再扩到 E2E

接口测试通常更容易接入，因为状态边界更清晰。
等接口层稳定后，再补足 E2E 需要的长链路模板。

flowchart TD
    A[选出高频脆弱场景] --> B[抽象业务状态模板]
    B --> C[实现最小数据工厂]
    C --> D[接入接口测试]
    D --> E[增加run_id与清理机制]
    E --> F[扩展到E2E场景]
    F --> G[沉淀为共享能力]

边界条件：不是所有场景都适合数据工厂

测试数据工厂很有用，但它也不是万能钥匙。

以下场景要谨慎：

1. 强依赖异步链路最终一致性的场景

比如搜索索引、风控决策、推荐召回。
这类场景如果你强行直接造“最终状态”，可能测不到真实问题。

2. 第三方强绑定场景

例如真实支付、短信、外部物流。
通常需要沙箱、Mock 或契约测试配合，而不是纯靠工厂造数据。

3. 非常短平快的一次性验证

如果只是临时验证一个字段映射，没必要为了规范而过度设计工厂。

说白了，数据工厂适合高频、重复、依赖复杂状态的自动化场景。
低频临时场景，简单处理反而更划算。

总结

测试数据工厂真正解决的，不只是“造数据方便”，而是自动化测试体系里的三个老问题：

• 稳定性：减少对公共脏数据和固定 ID 的依赖
• 可维护性：让测试代码表达业务状态，而不是表细节
• 可扩展性：同时服务接口测试和端到端测试

如果你要开始落地，我建议记住这几个最实用的原则：

1. 先抽象业务状态，再写工厂代码
2. 所有动态数据都带唯一标识和 run_id
3. 优先支持高频脆弱场景，不要一上来做大而全
4. 能走业务服务就别盲目直写数据库
5. 失败现场要可追踪，清理机制要可兜底

最后给一个很现实的判断标准：
如果你的自动化用例还在大量依赖“固定账号、固定商品、固定订单号”，那测试数据工厂基本已经值得投入了。越早做，后面补救的成本越低。