背景与问题

自动化测试写到一定规模后，真正拖后腿的往往不是断言，也不是框架，而是测试数据。

很多团队一开始都这样：

用一套固定账号跑所有用例
数据靠 SQL 脚本手工灌
环境共享，谁先跑谁先改数据
接口回归只看 HTTP 200，不校验业务结果
线上问题复盘时，没有办法把请求和当时的数据状态还原出来

短期内看，测试“能跑”；一旦用例数量上来，就会出现几个非常典型的问题：

用例相互污染
A 用例创建了订单，B 用例刚好依赖“无订单”状态，结果时好时坏。
数据构造成本高
一条完整业务链路，可能依赖用户、账户、商品、库存、优惠券、订单等多张表。每次都手工插，维护成本极高。
回归结果不可信
返回码正常，不代表数据库落库正确，也不代表下游调用符合预期。
问题难复现
线上 bug 不是没有日志，而是没有“当时那一份数据上下文”。

我自己在做接口自动化时，最早也踩过一个很典型的坑：同一个测试账号在不同流水线里被并发使用，优惠券状态被互相消费，导致回归报告一会儿红一会儿绿。后来才意识到，测试数据管理不是附属工作，而是自动化体系的基础设施。

这篇文章不讲空泛原则，而是带你从一个可落地的角度，把测试数据管理拆成三件事：

数据构造：怎么快速生成可用测试数据
数据隔离：怎么让不同用例、不同任务互不影响
回放校验：怎么把一次测试请求“留档”，并做结果复核

前置知识与环境准备

本文默认你已经有这些基础：

会写基础接口自动化
了解 Python、pytest、requests
能使用 SQLite/MySQL 这类关系型数据库
对测试环境、业务状态机有基本概念

本文示例为了便于直接运行，使用：

Python 3.10+
SQLite
pytest
标准库为主，少依赖

安装依赖：

pip install pytest

目录结构建议如下：

test-data-demo/
├── app.py
├── data_factory.py
├── replay.py
├── conftest.py
├── test_order.py
└── test.db

核心原理

测试数据管理如果想长期稳定，核心不是“多写几个初始化脚本”，而是建立一套可组合、可隔离、可追踪的机制。

我通常把它拆成四层：

数据模板层
定义“一个可测试用户/商品/订单”应该长什么样。
数据工厂层
按模板生成唯一数据，自动补充主键、租户标识、批次号、时间戳。
隔离策略层
控制不同测试执行之间如何隔离，常见有：
- 按账号隔离
- 按租户隔离
- 按批次号隔离
- 按事务回滚隔离
- 按环境/命名空间隔离
回放校验层
记录请求、响应、关键数据库快照，便于：
- 重放请求
- 校验落库结果
- 比较版本变化
- 排查偶发问题

可以先看总流程图。

flowchart TD
    A[测试开始] --> B[生成批次ID batch_id]
    B --> C[数据工厂构造用户/商品/库存]
    C --> D[执行接口请求]
    D --> E[记录请求响应与关键上下文]
    E --> F[数据库断言]
    F --> G{是否需要清理}
    G -->|是| H[按 batch_id 清理数据]
    G -->|否| I[保留用于回放]
    H --> J[测试结束]
    I --> J

1. 数据构造：从“写死数据”变成“声明式生成”

不要在每个用例里手写大量 SQL。更好的做法是：

用一个数据工厂统一生成测试实体
所有数据都带上唯一标记
业务默认值集中维护

比如：

用户名：user_{batch_id}_{seq}
商品编码：sku_{batch_id}_{seq}
订单号：由程序生成
所有记录统一带 batch_id

这样测试失败时，你甚至可以直接按 batch_id 去库里查整批数据。

2. 数据隔离：优先考虑“逻辑隔离”，再考虑“物理隔离”

理想情况当然是每次测试都起一个独立环境，但现实中成本太高。实际落地时，我建议优先级是：

逻辑隔离：tenant_id / batch_id / 唯一账号
事务隔离：单测场景可回滚
物理隔离：独立库、独立 schema、独立命名空间

对于中级规模团队，最实用的办法通常是：

每次测试运行生成一个 batch_id
每个用例生成自己的 case_id
所有构造数据都挂在 batch_id/case_id 下
清理时按标识删除，而不是全表 truncate

3. 回放校验：不是“存日志”，而是“存可验证上下文”

很多人理解的回放，只是把请求 body 记下来。其实这不够。

要支持真正的问题复现和回归比较，至少要记录：

请求方法、URL、headers、body
响应状态码、响应体
业务关键参数
关键数据库状态快照
执行时间、批次号、用例名

下面这个时序图能看得更清楚：

sequenceDiagram
    participant T as TestCase
    participant F as DataFactory
    participant A as App/API
    participant D as DB
    participant R as ReplayStore

    T->>F: 创建测试用户/商品
    F->>D: 插入基础数据(batch_id)
    T->>A: 发起下单请求
    A->>D: 写订单/扣库存
    A-->>T: 返回响应
    T->>D: 查询订单和库存
    T->>R: 记录请求/响应/DB快照
    T-->>T: 断言业务正确

方案设计：一套简单但够用的落地模型

为了把概念落地，我们做一个最小示例业务：下单接口。

业务规则：

用户可以下单购买商品
库存足够时创建订单，库存减一
库存不足时返回失败
所有测试数据都带 batch_id

数据库实体关系可以简化成这样：

classDiagram
    class users {
        +id: int
        +username: str
        +balance: int
        +batch_id: str
    }

    class products {
        +id: int
        +sku: str
        +name: str
        +stock: int
        +price: int
        +batch_id: str
    }

    class orders {
        +id: int
        +order_no: str
        +user_id: int
        +product_id: int
        +amount: int
        +status: str
        +batch_id: str
    }

    class replay_logs {
        +id: int
        +case_name: str
        +request_json: str
        +response_json: str
        +db_snapshot_json: str
        +batch_id: str
        +created_at: str
    }

    users --> orders
    products --> orders

实战代码（可运行）

下面的代码是完整可运行示例。你可以直接保存成对应文件并执行 pytest -s。

1）应用与数据库初始化：`app.py`

import json
import sqlite3
import time
import uuid
from pathlib import Path

DB_FILE = Path("test.db")


def get_conn():
    conn = sqlite3.connect(DB_FILE)
    conn.row_factory = sqlite3.Row
    return conn


def init_db():
    conn = get_conn()
    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,
        batch_id TEXT NOT NULL
    )
    """)

    cur.execute("""
    CREATE TABLE IF NOT EXISTS products (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        sku TEXT NOT NULL UNIQUE,
        name TEXT NOT NULL,
        stock INTEGER NOT NULL,
        price INTEGER NOT NULL,
        batch_id TEXT NOT NULL
    )
    """)

    cur.execute("""
    CREATE TABLE IF NOT EXISTS orders (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        order_no TEXT NOT NULL UNIQUE,
        user_id INTEGER NOT NULL,
        product_id INTEGER NOT NULL,
        amount INTEGER NOT NULL,
        status TEXT NOT NULL,
        batch_id TEXT NOT NULL,
        FOREIGN KEY(user_id) REFERENCES users(id),
        FOREIGN KEY(product_id) REFERENCES products(id)
    )
    """)

    cur.execute("""
    CREATE TABLE IF NOT EXISTS replay_logs (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        case_name TEXT NOT NULL,
        request_json TEXT NOT NULL,
        response_json TEXT NOT NULL,
        db_snapshot_json TEXT NOT NULL,
        batch_id TEXT NOT NULL,
        created_at TEXT NOT NULL
    )
    """)

    conn.commit()
    conn.close()


def create_order(user_id: int, product_id: int, batch_id: str):
    conn = get_conn()
    cur = conn.cursor()

    user = cur.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
    product = cur.execute("SELECT * FROM products WHERE id = ?", (product_id,)).fetchone()

    if not user:
        conn.close()
        return {"code": 404, "message": "user not found"}

    if not product:
        conn.close()
        return {"code": 404, "message": "product not found"}

    if product["stock"] <= 0:
        conn.close()
        return {"code": 400, "message": "out of stock"}

    order_no = f"ORD-{int(time.time() * 1000)}-{uuid.uuid4().hex[:6]}"

    cur.execute(
        "UPDATE products SET stock = stock - 1 WHERE id = ?",
        (product_id,)
    )

    cur.execute(
        """
        INSERT INTO orders(order_no, user_id, product_id, amount, status, batch_id)
        VALUES (?, ?, ?, ?, ?, ?)
        """,
        (order_no, user_id, product_id, product["price"], "CREATED", batch_id)
    )

    conn.commit()
    conn.close()

    return {
        "code": 200,
        "message": "success",
        "data": {
            "order_no": order_no,
            "amount": product["price"]
        }
    }


def snapshot_order_state(order_no: str):
    conn = get_conn()
    cur = conn.cursor()

    order = cur.execute(
        "SELECT * FROM orders WHERE order_no = ?",
        (order_no,)
    ).fetchone()

    if not order:
        conn.close()
        return {}

    product = cur.execute(
        "SELECT * FROM products WHERE id = ?",
        (order["product_id"],)
    ).fetchone()

    result = {
        "order": dict(order),
        "product": dict(product) if product else None
    }
    conn.close()
    return result


def save_replay(case_name: str, request_data: dict, response_data: dict, db_snapshot: dict, batch_id: str):
    conn = get_conn()
    cur = conn.cursor()

    cur.execute(
        """
        INSERT INTO replay_logs(case_name, request_json, response_json, db_snapshot_json, batch_id, created_at)
        VALUES (?, ?, ?, ?, ?, ?)
        """,
        (
            case_name,
            json.dumps(request_data, ensure_ascii=False),
            json.dumps(response_data, ensure_ascii=False),
            json.dumps(db_snapshot, ensure_ascii=False),
            batch_id,
            time.strftime("%Y-%m-%d %H:%M:%S")
        )
    )
    conn.commit()
    conn.close()

2）测试数据工厂：`data_factory.py`

import itertools
import uuid
from app import get_conn

_counter = itertools.count(1)


class DataFactory:
    def __init__(self, batch_id: str):
        self.batch_id = batch_id

    def create_user(self, balance: int = 1000):
        conn = get_conn()
        cur = conn.cursor()

        seq = next(_counter)
        username = f"user_{self.batch_id}_{seq}_{uuid.uuid4().hex[:4]}"
        cur.execute(
            "INSERT INTO users(username, balance, batch_id) VALUES (?, ?, ?)",
            (username, balance, self.batch_id)
        )
        user_id = cur.lastrowid
        conn.commit()
        conn.close()

        return {
            "id": user_id,
            "username": username,
            "balance": balance,
            "batch_id": self.batch_id
        }

    def create_product(self, stock: int = 10, price: int = 100, name: str = "demo_product"):
        conn = get_conn()
        cur = conn.cursor()

        seq = next(_counter)
        sku = f"sku_{self.batch_id}_{seq}_{uuid.uuid4().hex[:4]}"
        cur.execute(
            "INSERT INTO products(sku, name, stock, price, batch_id) VALUES (?, ?, ?, ?, ?)",
            (sku, name, stock, price, self.batch_id)
        )
        product_id = cur.lastrowid
        conn.commit()
        conn.close()

        return {
            "id": product_id,
            "sku": sku,
            "name": name,
            "stock": stock,
            "price": price,
            "batch_id": self.batch_id
        }

    def cleanup(self):
        conn = get_conn()
        cur = conn.cursor()

        cur.execute("DELETE FROM orders WHERE batch_id = ?", (self.batch_id,))
        cur.execute("DELETE FROM products WHERE batch_id = ?", (self.batch_id,))
        cur.execute("DELETE FROM users WHERE batch_id = ?", (self.batch_id,))
        cur.execute("DELETE FROM replay_logs WHERE batch_id = ?", (self.batch_id,))

        conn.commit()
        conn.close()

3）回放工具：`replay.py`

import json
from app import get_conn, create_order, snapshot_order_state


def get_latest_replay(case_name: str):
    conn = get_conn()
    cur = conn.cursor()
    row = cur.execute(
        """
        SELECT * FROM replay_logs
        WHERE case_name = ?
        ORDER BY id DESC
        LIMIT 1
        """,
        (case_name,)
    ).fetchone()
    conn.close()
    return dict(row) if row else None


def replay_case(case_name: str):
    row = get_latest_replay(case_name)
    if not row:
        return {"ok": False, "reason": "replay log not found"}

    request_data = json.loads(row["request_json"])
    response = create_order(
        user_id=request_data["user_id"],
        product_id=request_data["product_id"],
        batch_id=request_data["batch_id"]
    )

    snapshot = {}
    if response.get("code") == 200:
        snapshot = snapshot_order_state(response["data"]["order_no"])

    return {
        "ok": True,
        "request": request_data,
        "response": response,
        "db_snapshot": snapshot
    }

4）pytest 固件：`conftest.py`

import uuid
import pytest
from app import init_db
from data_factory import DataFactory


@pytest.fixture(scope="session", autouse=True)
def setup_db():
    init_db()


@pytest.fixture()
def batch_id():
    return uuid.uuid4().hex[:8]


@pytest.fixture()
def factory(batch_id):
    fac = DataFactory(batch_id)
    yield fac
    fac.cleanup()

5）测试用例：`test_order.py`

from app import create_order, save_replay, snapshot_order_state, get_conn
from replay import replay_case


def test_create_order_success(factory, batch_id):
    user = factory.create_user()
    product = factory.create_product(stock=2, price=199)

    request_data = {
        "user_id": user["id"],
        "product_id": product["id"],
        "batch_id": batch_id
    }

    response = create_order(**request_data)

    assert response["code"] == 200
    assert response["data"]["amount"] == 199

    order_no = response["data"]["order_no"]
    snapshot = snapshot_order_state(order_no)

    assert snapshot["order"]["status"] == "CREATED"
    assert snapshot["product"]["stock"] == 1

    save_replay(
        case_name="test_create_order_success",
        request_data=request_data,
        response_data=response,
        db_snapshot=snapshot,
        batch_id=batch_id
    )


def test_create_order_out_of_stock(factory, batch_id):
    user = factory.create_user()
    product = factory.create_product(stock=0, price=99)

    request_data = {
        "user_id": user["id"],
        "product_id": product["id"],
        "batch_id": batch_id
    }

    response = create_order(**request_data)

    assert response["code"] == 400
    assert response["message"] == "out of stock"

    save_replay(
        case_name="test_create_order_out_of_stock",
        request_data=request_data,
        response_data=response,
        db_snapshot={},
        batch_id=batch_id
    )


def test_replay_last_success_case(factory, batch_id):
    user = factory.create_user()
    product = factory.create_product(stock=2, price=88)

    request_data = {
        "user_id": user["id"],
        "product_id": product["id"],
        "batch_id": batch_id
    }

    response = create_order(**request_data)
    assert response["code"] == 200

    snapshot = snapshot_order_state(response["data"]["order_no"])
    save_replay(
        case_name="replay_demo_case",
        request_data=request_data,
        response_data=response,
        db_snapshot=snapshot,
        batch_id=batch_id
    )

    replay_result = replay_case("replay_demo_case")
    assert replay_result["ok"] is True
    assert replay_result["response"]["code"] == 200


def test_data_isolation_by_batch(factory, batch_id):
    user = factory.create_user()
    product = factory.create_product(stock=1, price=50)

    response = create_order(user["id"], product["id"], batch_id)
    assert response["code"] == 200

    conn = get_conn()
    cur = conn.cursor()
    count = cur.execute(
        "SELECT COUNT(*) AS c FROM orders WHERE batch_id = ?",
        (batch_id,)
    ).fetchone()["c"]
    conn.close()

    assert count == 1

6）运行方式

pytest -s

如果你想临时保留回放数据用于排查，可以先把 conftest.py 中的清理逻辑注释掉，或者只在失败时保留。

逐步验证清单

如果你想边做边确认，我建议按这个顺序验证：

第一步：只验证数据工厂是否可用

检查：

是否能创建唯一用户
是否能创建唯一商品
batch_id 是否正确写入

你可以执行：

from app import init_db, get_conn
from data_factory import DataFactory

init_db()
factory = DataFactory("demo001")
print(factory.create_user())
print(factory.create_product())

第二步：验证隔离是否生效

思路是：

用两个不同 batch_id
分别创建订单
查询时按 batch_id 过滤
确认互不影响

第三步：验证回放记录是否完整

重点不是“有日志”，而是：

请求参数是否完整
成功时是否记录数据库快照
失败时是否能记录失败现场

第四步：验证失败后的排查效率

你可以手动把一个断言写错，例如把库存断言改成 0，观察自己是否能快速通过 batch_id + replay_logs 找到问题。

说实话，这一步很重要。很多方案在 happy path 上都没问题，一到失败排查就原形毕露。

隔离策略怎么选：给中级团队的实用建议

不同场景，隔离方式不一样。这里给一个比较务实的选择表。

场景	推荐策略	原因
单接口自动化	batch_id + 数据工厂	成本低、易维护
并发流水线回归	batch_id + 独立账号池	避免资源冲突
单元/DAO 层测试	事务回滚	执行快、清理简单
微服务联调环境	tenant_id + 命名空间隔离	可跨服务追踪
高风险回归	保留失败现场 + 回放日志	便于问题复现

有一个经验我很想强调：不要迷信“全量清库”。
在共享环境里，truncate 常常是事故源头。更稳妥的办法是：

所有测试数据都可标记
清理只删自己那一批
失败数据允许保留一段时间

常见坑与排查

这部分很接地气，因为大多数问题并不是框架 bug，而是细节没处理好。

1. 唯一键冲突

现象

用户名重复
商品 SKU 重复
订单号重复

原因

用固定测试数据
并发执行时未加随机后缀
清理不及时，旧数据残留

排查方法

查重复键：

SELECT username, COUNT(*) 
FROM users 
GROUP BY username
HAVING COUNT(*) > 1;

解决建议

所有唯一字段都加 batch_id + seq + 随机串
不要用 test_user_001 这种固定命名

2. 用例之间相互污染

现象

单独跑通过，一起跑失败
本地跑通过，CI 跑失败
重试一次又好了

原因

共享账号
共享商品库存
清理逻辑误删、漏删

排查方法

重点查这三类字段：

batch_id
created_at
资源是否被多个 case 复用

可执行查询：

SELECT * FROM orders
ORDER BY id DESC
LIMIT 20;

解决建议

每个测试任务独立 batch_id
高冲突资源改成每例独享
清理时只清理本批次

3. 回放失败，但原测试成功

现象

第一次跑成功，拿记录回放失败。

原因

常见有：

回放使用的数据已被清理
外部依赖是时效性数据
请求记录不完整，比如漏了 header、上下文 token

排查思路

先确认回放依赖的数据是否还存在：

SELECT * FROM replay_logs WHERE case_name = 'replay_demo_case';

再确认请求上下文是否完整。

解决建议

回放数据和清理策略解耦
对外部依赖做 mock 或录制
记录关键 headers、身份信息、版本号

4. 断言只校验接口返回，不校验落库

现象

接口返回成功，但业务其实错了。

原因

很多自动化只做了：

assert response.status_code == 200

这在真实项目里远远不够。

正确做法

至少做三层断言：

HTTP/接口层
业务字段层
DB/消息/下游副作用层

例如：

assert response["code"] == 200
assert response["data"]["amount"] == 199
assert snapshot["order"]["status"] == "CREATED"
assert snapshot["product"]["stock"] == 1

安全/性能最佳实践

测试数据管理不仅是“能不能跑”，还涉及安全和执行效率。

安全最佳实践

1. 不要在回放日志中落敏感信息明文

例如：

手机号
身份证号
token
密码
支付卡信息

建议在保存回放前做脱敏：

def mask_sensitive(data: dict):
    cloned = dict(data)
    if "token" in cloned:
        cloned["token"] = "***"
    if "password" in cloned:
        cloned["password"] = "***"
    return cloned

2. 测试环境也要做权限隔离

很多团队会忽略这一点：测试库不等于“谁都能删”。

建议：

自动化账号只授予必要表权限
清理账号与读写账号分离
高风险表禁用全表删除

3. 回放数据设置保留周期

失败现场有价值，但无限保存会带来风险。

建议：

成功记录保留 3~7 天
失败记录保留 14~30 天
定时归档或删除

性能最佳实践

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

坏做法：

每个 case 都重建所有基础字典数据
每次都重复造很重的业务前置

更好的方式：

静态基础数据预置
动态交易数据按需生成
重资源数据做分层缓存

2. 批量造数优于逐条插入

如果一个场景要造 1000 条测试数据，不要循环一条一条 insert。

示例：

import sqlite3

conn = sqlite3.connect("test.db")
cur = conn.cursor()

rows = [
    (f"user_batch_x_{i}", 1000, "batch_x")
    for i in range(1000)
]

cur.executemany(
    "INSERT INTO users(username, balance, batch_id) VALUES (?, ?, ?)",
    rows
)

conn.commit()
conn.close()

3. 查询和清理都要命中索引

如果你依赖 batch_id 做隔离，那它最好有索引。

CREATE INDEX IF NOT EXISTS idx_users_batch_id ON users(batch_id);
CREATE INDEX IF NOT EXISTS idx_products_batch_id ON products(batch_id);
CREATE INDEX IF NOT EXISTS idx_orders_batch_id ON orders(batch_id);
CREATE INDEX IF NOT EXISTS idx_replay_logs_batch_id ON replay_logs(batch_id);

否则数据量一上来，清理和查询都会变慢。

4. 失败保留，成功快速清理

这是一个很实用的策略：

成功用例：测试结束立即清理
失败用例：保留现场，供排查和回放
定时任务统一清理过期数据

这个策略在 CI 上尤其有效，既控制环境膨胀，也不牺牲定位效率。

可扩展思路：从示例走向真实项目

上面的示例是最小闭环，真实项目里通常还会继续扩展。

1. 引入“业务对象构建器”

比如电商下单可能不仅有用户和商品，还依赖：

地址
优惠券
会员等级
支付账户
风控白名单

这时可以把数据工厂继续拆分：

UserBuilder
ProductBuilder
CouponBuilder
OrderSceneBuilder

让“创建一套可下单场景”成为一个高级接口，而不是一堆散乱 SQL。

2. 引入状态快照比对

回放时不仅重跑请求，还比较：

返回结构是否变化
关键字段是否变化
落库状态是否一致

这非常适合做灰度回归或版本升级验证。

3. 外部依赖录制与重放

如果下单依赖库存服务、优惠服务、支付网关，那么单靠数据库快照还不够。
可以继续扩展为：

HTTP mock
消息队列事件录制
下游响应桩数据管理

一套推荐的落地步骤

如果你准备在团队里推进，我建议不要一次性搞很大，按下面节奏来。

第 1 阶段：先统一数据标识

目标：

所有自动化数据都带 batch_id
所有唯一字段可追踪
清理脚本按批次删除

这是收益最大、阻力最小的一步。

第 2 阶段：抽出数据工厂

目标：

从用例中移除散乱 SQL
建立统一默认值
一份代码维护多种业务数据模板

第 3 阶段：补齐回放记录

目标：

请求、响应、关键快照入库
失败现场可保留
问题可复现

第 4 阶段：做分层隔离

目标：

高冲突资源独享
普通资源逻辑隔离
高价值链路支持环境级隔离

这四步走下来，自动化测试的稳定性通常会明显提升。

总结

测试数据管理这件事，真正的重点不是“造点数据”，而是建立一条可控链路：

构造：用数据工厂把业务前置标准化
隔离：用 batch_id / case_id / tenant_id 避免互相污染
回放校验：把请求、响应、数据库快照串起来，支持复现和复核

如果你只记住三条，我建议是：

所有测试数据都必须可标记、可追踪、可清理
断言不能只看接口返回，至少补一层落库或副作用校验
失败现场比“跑得快一点”更值钱，回放能力一定要留

边界条件也要说清楚：

如果是纯单元测试，事务回滚通常比批次清理更高效
如果是跨服务联调，仅靠数据库隔离不够，需要配合 mock 或命名空间隔离
如果环境多人共享，绝不要依赖全表清理作为日常手段

最后，测试自动化稳定不稳定，很多时候不是脚本写得好不好，而是测试数据有没有“被当成系统设计的一部分”。
把这件事做好，用例才会从“偶尔能跑”变成“长期可信”。

《自动化测试中的测试数据管理实战：从数据构造、隔离到回放校验的落地方案》

背景与问题

前置知识与环境准备

核心原理

1. 数据构造：从“写死数据”变成“声明式生成”

2. 数据隔离：优先考虑“逻辑隔离”，再考虑“物理隔离”

3. 回放校验：不是“存日志”，而是“存可验证上下文”

方案设计：一套简单但够用的落地模型

实战代码（可运行）

1）应用与数据库初始化：app.py

2）测试数据工厂：data_factory.py

3）回放工具：replay.py

4）pytest 固件：conftest.py

5）测试用例：test_order.py

6）运行方式

逐步验证清单

第一步：只验证数据工厂是否可用

第二步：验证隔离是否生效

第三步：验证回放记录是否完整

第四步：验证失败后的排查效率

隔离策略怎么选：给中级团队的实用建议

常见坑与排查

1. 唯一键冲突

现象

原因

排查方法

解决建议

2. 用例之间相互污染

现象

原因

排查方法

解决建议

3. 回放失败，但原测试成功

现象

原因

排查思路

解决建议

4. 断言只校验接口返回，不校验落库

现象

原因

正确做法

安全/性能最佳实践

安全最佳实践

1. 不要在回放日志中落敏感信息明文

2. 测试环境也要做权限隔离

3. 回放数据设置保留周期

性能最佳实践

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

2. 批量造数优于逐条插入

3. 查询和清理都要命中索引

4. 失败保留，成功快速清理

可扩展思路：从示例走向真实项目

1. 引入“业务对象构建器”

2. 引入状态快照比对

3. 外部依赖录制与重放

一套推荐的落地步骤

第 1 阶段：先统一数据标识

第 2 阶段：抽出数据工厂

第 3 阶段：补齐回放记录

第 4 阶段：做分层隔离

总结

1）应用与数据库初始化：`app.py`

2）测试数据工厂：`data_factory.py`

3）回放工具：`replay.py`

4）pytest 固件：`conftest.py`

5）测试用例：`test_order.py`