接口自动化数据管理与环境隔离¶
"本地全过，CI 一半红"的根因基本集中在三件事：数据被别人动了、环境配置不一样、并行时抢同一个资源。这篇按这三个问题给方案，代码可以直接落地。
一、先把环境差异变成配置¶
1.1 配置矩阵¶
| 环境 | 用途 | 数据策略 | 允许写操作 |
|---|---|---|---|
| local | 本地调试 | 容器内独立库 | 是 |
| test | 团队联调 | 共享库，按租户隔离 | 是 |
| staging | 发布前验证 | 接近生产，只读为主 | 少量 |
| prod | 只做巡检 | 只读 | 否 |
配置从环境变量注入，代码里不允许出现域名：
# config.py
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
base_url: str
db_dsn: str
api_user: str
api_password: str
tenant_prefix: str = "qa"
timeout: float = 10.0
class Config:
env_prefix = "QALAB_"
env_file = ".env"
settings = Settings()
QALAB_BASE_URL=https://staging.example.com \
QALAB_DB_DSN=postgresql://qa:pass@db:5432/app \
QALAB_API_USER=qa_bot \
QALAB_API_PASSWORD="$SECRET" \
pytest -m regression
好处：切换环境只改环境变量；CI 里用 Secret 注入；本地用 .env 且 .env 进 .gitignore。
1.2 不要在用例里写 if env ==¶
# 错误
if settings.base_url.endswith("staging.com"):
payload["cardNo"] = "6222020000000000000"
else:
payload["cardNo"] = "4111111111111111"
# 正确：把差异放进配置
payload["cardNo"] = settings.test_card_no
用例里出现环境分支，等于把环境差异固化进代码，后续每加一个环境就要改所有用例。
二、造数据：优先走接口，兜底走 SQL¶
2.1 四种方式的适用边界¶
| 方式 | 速度 | 真实性 | 适用 |
|---|---|---|---|
| 接口构造 | 慢（走业务） | 高 | 绝大多数用例 |
| DB 直插 | 快 | 低（绕过校验） | 造"接口到不了"的状态（冻结、逾期） |
| SQL 脚本/快照 | 最快 | 中 | 批量基础数据、环境初始化 |
| 固定夹具 | 快 | 只读 | 字典、SPU、SKU |
原则：被测接口用接口造，前置状态用 SQL 造。用 DB 直插造被测对象，会掩盖接口层校验缺陷。
2.2 数据工厂¶
import time, uuid, itertools
class UserFactory:
_seq = itertools.count(1)
@classmethod
def create(cls, client, role="user"):
name = f"qa_{int(time.time())}_{next(cls._seq)}_{uuid.uuid4().hex[:6]}"
resp = client.post("/api/users", json={
"username": name, "role": role, "password": "Passw0rd!",
})
assert resp.status_code in (200, 201), resp.text
data = resp.json()["data"]
assert data["username"] == name
return data
唯一性三段式：时间戳 + 自增 + 随机后缀。只用时间戳在并行下会撞（同一秒多个 worker）；只用自增在多进程下会撞。
2.3 用 SQL 造状态（明确限定范围）¶
def freeze_account(db, user_id):
with db.cursor() as cur:
cur.execute(
"UPDATE accounts SET status = %s, updated_at = now() WHERE id = %s",
("FROZEN", user_id))
assert cur.rowcount == 1, f"user {user_id} not found"
db.commit()
一定要断言 rowcount，否则 SQL 静默失败会让用例在错误的前提下继续跑。
三、清理：三步兜底¶
3.1 fixture teardown¶
@pytest.fixture
def temp_user(client):
user = UserFactory.create(client)
yield user
client.delete(f"/api/users/{user['id']}")
teardown 里的删除不要 assert，用日志记录：
def cleanup(client, user):
resp = client.delete(f"/api/users/{user['id']}")
if resp.status_code not in (200, 204, 404):
logging.warning("cleanup failed: %s %s", resp.status_code, resp.text)
清理失败让用例变红，会掩盖真正的失败原因。
3.2 批次标记 + 批量清理¶
所有测试数据带上批次号，事后统一按批次清：
-- 兜底清理（每天跑一次）
DELETE FROM accounts WHERE username LIKE 'qa\_2026%' AND created_at < now() - interval '1 day';
注意 LIKE 里的 _ 是通配符，要转义，否则会命中更多数据。
3.3 清理钩子挂在 pytest 上¶
def pytest_sessionfinish(session, exitstatus):
if exitstatus in (0, 1): # 无论通过与否都清理
cleanup_by_batch(session.config.getoption("--batch-id"))
四、并行隔离：四种方案¶
4.1 账号/数据按 worker 隔离¶
@pytest.fixture(scope="session")
def worker_prefix(worker_id):
return "master" if worker_id == "master" else f"w{worker_id}"
@pytest.fixture(scope="session")
def client(worker_prefix):
return ApiClient(user=f"qa_{worker_prefix}", password=settings.api_password)
4.2 数据库按 worker 分 schema（最彻底）¶
@pytest.fixture(scope="session", autouse=True)
def schema(worker_id, db):
name = f"test_{worker_id}".replace("-", "_")
db.execute(f"DROP SCHEMA IF EXISTS {name} CASCADE")
db.execute(f"CREATE SCHEMA {name}")
db.execute(f"SET search_path TO {name}")
yield name
db.execute(f"DROP SCHEMA IF EXISTS {name} CASCADE")
worker_id 可能返回 gw0、gw1，也可能是 master（单 worker），必须处理两种值。
4.3 资源池（库存、优惠券）¶
抢同一库存是并行下最典型的偶发失败。方案：
- 每个 worker 预分配独立库存（
stock_w0、stock_w1） - 或用"创建即独占"的资源，避免共享
@pytest.fixture(scope="function")
def exclusive_sku(client, worker_prefix):
sku = client.post("/api/skus", json={
"name": f"sku_{worker_prefix}_{uuid.uuid4().hex[:8]}",
"stock": 1,
}).json()["data"]
yield sku
4.4 时间与随机数¶
- 依赖"今天"的用例，用
freezegun固定时间，别用系统时钟。 - 依赖随机数的断言，注入固定 seed。
- 时区问题：CI 容器默认 UTC，本地是 Asia/Shanghai，跨天边界用例必挂。
from freezegun import freeze_time
@freeze_time("2026-10-22 23:59:00")
def test_cross_day_settlement(client):
...
五、"本地过、CI 挂"逐项排查¶
按命中率排序：
| 现象 | 根因 | 验证方法 | 修复 |
|---|---|---|---|
| 偶发 409/库存不足 | 共享数据被并行消费 | pytest -n 4 本地复现 |
资源池/独立 SKU |
| 时区相关断言失败 | 容器 UTC | date 对比 |
固定 TZ 或 freeze_time |
| 连接超时 | CI 网络策略、DNS | curl -v 目标域名 |
显式代理/重试 |
| 401 全挂 | token 过期时间按本地时钟算 | 打印 token exp | 每次 session 重新登录 |
| 金额差 0.01 | 浮点在不同平台舍入 | 打印原始响应 | 用 Decimal 比较 |
| 上传失败 | CI 无对应文件/权限 | ls 路径 |
用 tmp_path 造文件 |
| 顺序相关 | 用例间共享状态 | pytest -p no:randomly 对照 |
用例独立 |
| 慢 10 倍 | CI CPU 少、无缓存 | 看 --durations |
减少 sleep，改显式等待 |
一个很实用的命令，直接定位慢与不稳定：
六、敏感数据¶
- 测试账号密码走 CI Secret，不进仓库。
- 日志里对
password、token、idCard打码：
SENSITIVE = ("password", "token", "authorization", "idCard", "phone")
def mask(payload: dict) -> dict:
return {k: ("***" if k.lower() in SENSITIVE else v) for k, v in payload.items()}
- Allure 附件里不要附完整请求体，附打码后的版本。
- 生产环境只跑只读用例，用独立只读账号，并在代码层加开关：
七、检查清单¶
- 所有环境相关值来自配置，用例里无硬编码域名
- 数据唯一性 = 时间戳 + 自增 + 随机
- 造数据优先接口，SQL 仅用于状态
- teardown 不 assert，失败只告警
- 有按批次的兜底清理脚本
-
pytest -n 4本地能稳定跑通 - 依赖时间的用例已 freeze
- 日志与 Allure 附件已脱敏
- 生产环境有写操作拦截