背景与问题

做接口自动化时，最容易在一开始“跑起来”，最难的是半年后还“跑得稳”。

很多团队的接口测试都有过类似阶段：

• 一开始直接在测试脚本里写 URL、Header、断言、数据构造
• 用例数量一多，复制粘贴越来越严重
• 某个公共鉴权逻辑变了，要改几十上百个文件
• 测试环境数据不稳定，导致同一套用例今天过、明天挂
• CI 一跑就是一片红，但真正的产品缺陷没几个，更多是测试自身不稳定

我当时踩过一个典型坑：把“业务断言”“通用协议校验”“数据准备”“重试策略”全揉在一层里。短期看开发快，长期看几乎不可维护。后来回头重构，核心思路其实就两件事：

1. 接口用例要分层
2. 稳定性要系统治理，而不是靠“失败重跑”掩盖问题

这篇文章就从架构视角，把这两件事串起来讲清楚，并给出一套可运行的 Python 示例。

为什么接口自动化容易失稳

先别急着上代码，先看失稳来源。接口测试“不稳定”，通常不是一个原因，而是几类问题叠加：

• 脚本层失稳：等待不足、参数硬编码、断言写得脆弱
• 数据层失稳：共享测试数据被污染、环境脏数据不可控
• 依赖层失稳：外部服务超时、鉴权服务抖动、消息链路延迟
• 环境层失稳：测试环境部署频繁、配置漂移、限流策略不一致
• 流程层失稳：用例耦合、执行顺序依赖、失败后无法快速定位

如果不分层，这些问题会以“随机失败”的形式一起冒出来。你看到的是一个红色用例，背后其实可能是架构问题。

核心原理

1. 用例分层不是“多写几层类”，而是拆责任

我比较推荐把接口自动化拆成下面几层：

1. 基础设施层

   • HTTP Client
   • 配置管理
   • 日志、重试、超时、鉴权
   • 测试报告、上下文管理

2. 接口封装层

   • 面向接口能力，而不是面向测试步骤
   • 例如 create_user()、get_user()、delete_user()

3. 业务编排层

   • 将多个接口调用组合成稳定可复用的业务动作
   • 例如“注册一个新用户并登录”

4. 断言层

   • 通用断言：状态码、响应结构、耗时
   • 业务断言：字段值、状态流转、幂等性

5. 测试用例层

   • 只描述场景，不关心底层实现细节
   • 例如“新用户注册后可被查询到”

这样做的最大价值，不是“优雅”，而是：

• 变更集中
• 复用清晰
• 故障定位更快
• 稳定性治理有抓手

2. 让“用例”只关心意图，不关心实现细节

理想中的用例应该像这样：

def test_user_can_be_created_and_queried(user_flow):
    user = user_flow.register_random_user()
    detail = user_flow.query_user(user["id"])
    assert detail["name"] == user["name"]

如果一个测试里充满了这些内容：

• 拼 URL
• 手动组 Header
• 写 token 获取逻辑
• 拼接 JSON
• 处理重试
• sleep(3) 等数据落库

那它大概率已经越层了。

3. 稳定性优化的核心是“消除非确定性”

一个稳定的自动化体系，不是“执行时永不失败”，而是：

• 真正有缺陷时能失败
• 没有缺陷时尽量不误报
• 失败后能快速知道失败在哪一层

本质上是在控制非确定性。常见手段包括：

• 数据隔离
• 接口幂等设计
• 统一超时与重试策略
• 更稳健的断言边界
• 异步场景用轮询代替盲等
• 对外部依赖做 Mock/Stub 或分层兜底

分层架构设计

下面这张图可以帮助理解整体结构。

flowchart TD
    A[测试用例层] --> B[业务编排层]
    B --> C[接口封装层]
    C --> D[基础设施层]
    D --> E[被测服务]
    B --> F[断言层]
    A --> F

再细一点，看一次典型调用链：

sequenceDiagram
    participant T as 测试用例
    participant F as 业务编排
    participant A as 接口封装
    participant C as HTTP Client
    participant S as 被测服务

    T->>F: 注册随机用户
    F->>A: create_user(payload)
    A->>C: POST /users
    C->>S: HTTP 请求
    S-->>C: 响应
    C-->>A: 标准化响应
    A-->>F: 业务结果
    F-->>T: user_info
    T->>F: 查询用户
    F->>A: get_user(user_id)
    A->>C: GET /users/{id}
    C->>S: HTTP 请求
    S-->>C: 响应
    C-->>A: 标准化响应
    A-->>F: 业务结果
    F-->>T: detail

方案对比：平铺脚本 vs 分层设计

我的建议是：只要你预期用例规模会超过几十条，并进入 CI 常态执行，就应该做分层。

核心分层职责拆解

基础设施层

这一层负责解决“横切问题”：

• 统一请求入口
• 默认超时
• 重试机制
• 请求日志与响应日志
• 环境配置读取
• 鉴权 token 注入
• 错误标准化

它不关心业务。

接口封装层

这一层负责把 HTTP 接口转成清晰的方法。比如：

• create_user(payload)
• get_user(user_id)

注意它也不应该承载测试场景逻辑。它做的是“接口能力暴露”。

业务编排层

这一层最容易被忽视，但恰恰很关键。它负责：

• 生成唯一测试数据
• 把多个接口组合成稳定流程
• 为用例提供场景级能力

例如：

• register_random_user()
• create_and_activate_user()

断言层

断言建议分两类：

通用断言

• 状态码是否正确
• 返回体是否符合 schema
• 错误码是否符合预期
• 耗时是否在阈值内

业务断言

• 用户状态是否从 CREATED 变成 ACTIVE
• 重复提交是否幂等
• 查询结果是否包含新建记录

这样拆的好处是：当接口响应结构统一变化时，只改通用断言；当业务规则变化时，只改业务断言。

实战代码（可运行）

下面用一个可运行的 Python 示例演示分层结构。为了保证示例可以本地跑通，我用 Flask 启一个简单服务，再用 pytest + requests 写自动化测试。

目录结构建议

project/
├── app.py
├── tests/
│   ├── test_user_api.py
│   └── conftest.py
├── framework/
│   ├── client.py
│   ├── assertions.py
│   └── utils.py
├── apis/
│   └── user_api.py
├── flows/
│   └── user_flow.py
└── requirements.txt

安装依赖

Flask==2.2.5
pytest==7.4.0
requests==2.31.0

你也可以保存为 requirements.txt，然后执行：

pip install -r requirements.txt

1）被测服务：app.py

from flask import Flask, request, jsonify
import uuid
import threading

app = Flask(__name__)

db = {}
lock = threading.Lock()

@app.route("/health", methods=["GET"])
def health():
    return jsonify({"status": "ok"}), 200

@app.route("/users", methods=["POST"])
def create_user():
    data = request.get_json() or {}
    name = data.get("name")
    if not name:
        return jsonify({"code": "INVALID_PARAM", "message": "name is required"}), 400

    user_id = str(uuid.uuid4())
    user = {
        "id": user_id,
        "name": name,
        "status": "CREATED"
    }
    with lock:
        db[user_id] = user
    return jsonify({"code": "SUCCESS", "data": user}), 201

@app.route("/users/<user_id>", methods=["GET"])
def get_user(user_id):
    user = db.get(user_id)
    if not user:
        return jsonify({"code": "NOT_FOUND", "message": "user not found"}), 404
    return jsonify({"code": "SUCCESS", "data": user}), 200

@app.route("/users/<user_id>/activate", methods=["POST"])
def activate_user(user_id):
    with lock:
        user = db.get(user_id)
        if not user:
            return jsonify({"code": "NOT_FOUND", "message": "user not found"}), 404
        user["status"] = "ACTIVE"
    return jsonify({"code": "SUCCESS", "data": user}), 200

if __name__ == "__main__":
    app.run(port=5001)

启动服务：

python app.py

2）基础设施层：framework/client.py

import time
import requests

class HttpClient:
    def __init__(self, base_url, timeout=3, retry=1):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.retry = retry

    def request(self, method, path, **kwargs):
        url = f"{self.base_url}{path}"
        last_exc = None

        for attempt in range(self.retry + 1):
            start = time.time()
            try:
                resp = requests.request(
                    method=method,
                    url=url,
                    timeout=self.timeout,
                    **kwargs
                )
                elapsed_ms = int((time.time() - start) * 1000)
                return {
                    "status_code": resp.status_code,
                    "json": resp.json() if resp.text else {},
                    "elapsed_ms": elapsed_ms
                }
            except requests.RequestException as e:
                last_exc = e
                if attempt < self.retry:
                    time.sleep(0.2)
                else:
                    raise RuntimeError(f"HTTP request failed: {method} {url}, error={e}") from e

        raise RuntimeError(f"Unexpected request failure: {last_exc}")

这个 HttpClient 做了几件事：

• 统一 URL 拼接
• 统一超时
• 简单重试
• 统一返回结构，方便后续断言

3）接口封装层：apis/user_api.py

class UserApi:
    def __init__(self, client):
        self.client = client

    def health(self):
        return self.client.request("GET", "/health")

    def create_user(self, name):
        return self.client.request(
            "POST",
            "/users",
            json={"name": name}
        )

    def get_user(self, user_id):
        return self.client.request(
            "GET",
            f"/users/{user_id}"
        )

    def activate_user(self, user_id):
        return self.client.request(
            "POST",
            f"/users/{user_id}/activate"
        )

4）工具与测试数据：framework/utils.py

import time
import uuid

def random_name(prefix="user"):
    return f"{prefix}_{int(time.time() * 1000)}_{uuid.uuid4().hex[:6]}"

5）业务编排层：flows/user_flow.py

from framework.utils import random_name

class UserFlow:
    def __init__(self, user_api):
        self.user_api = user_api

    def register_random_user(self):
        name = random_name()
        resp = self.user_api.create_user(name)
        return {
            "request_name": name,
            "response": resp
        }

    def create_and_activate_user(self):
        created = self.register_random_user()
        user_id = created["response"]["json"]["data"]["id"]
        activated = self.user_api.activate_user(user_id)
        return {
            "created": created["response"],
            "activated": activated
        }

    def query_user(self, user_id):
        return self.user_api.get_user(user_id)

6）断言层：framework/assertions.py

def assert_status_code(resp, expected_code):
    assert resp["status_code"] == expected_code, \
        f"expected status={expected_code}, actual={resp['status_code']}, resp={resp}"

def assert_success_code(resp):
    assert resp["json"].get("code") == "SUCCESS", \
        f"expected code=SUCCESS, actual={resp['json'].get('code')}, resp={resp}"

def assert_response_time_less_than(resp, threshold_ms):
    assert resp["elapsed_ms"] < threshold_ms, \
        f"response too slow: actual={resp['elapsed_ms']}ms, threshold={threshold_ms}ms"

def assert_user_name(resp, expected_name):
    actual = resp["json"]["data"]["name"]
    assert actual == expected_name, \
        f"expected name={expected_name}, actual={actual}"

def assert_user_status(resp, expected_status):
    actual = resp["json"]["data"]["status"]
    assert actual == expected_status, \
        f"expected status={expected_status}, actual={actual}"

7）pytest 夹具：tests/conftest.py

import pytest
from framework.client import HttpClient
from apis.user_api import UserApi
from flows.user_flow import UserFlow

@pytest.fixture(scope="session")
def client():
    return HttpClient(base_url="http://127.0.0.1:5001", timeout=2, retry=1)

@pytest.fixture(scope="session")
def user_api(client):
    return UserApi(client)

@pytest.fixture(scope="session")
def user_flow(user_api):
    return UserFlow(user_api)

8）测试用例层：tests/test_user_api.py

from framework.assertions import (
    assert_status_code,
    assert_success_code,
    assert_response_time_less_than,
    assert_user_name,
    assert_user_status
)

def test_health(user_api):
    resp = user_api.health()
    assert_status_code(resp, 200)
    assert resp["json"]["status"] == "ok"

def test_create_user_success(user_flow):
    result = user_flow.register_random_user()
    resp = result["response"]

    assert_status_code(resp, 201)
    assert_success_code(resp)
    assert_response_time_less_than(resp, 1000)
    assert_user_name(resp, result["request_name"])
    assert_user_status(resp, "CREATED")

def test_get_user_success(user_flow):
    created = user_flow.register_random_user()
    create_resp = created["response"]
    user_id = create_resp["json"]["data"]["id"]
    expected_name = create_resp["json"]["data"]["name"]

    query_resp = user_flow.query_user(user_id)
    assert_status_code(query_resp, 200)
    assert_success_code(query_resp)
    assert_user_name(query_resp, expected_name)
    assert_user_status(query_resp, "CREATED")

def test_activate_user_success(user_flow):
    result = user_flow.create_and_activate_user()
    activated_resp = result["activated"]

    assert_status_code(activated_resp, 200)
    assert_success_code(activated_resp)
    assert_user_status(activated_resp, "ACTIVE")

def test_get_user_not_found(user_api):
    resp = user_api.get_user("not-exist-id")
    assert_status_code(resp, 404)
    assert resp["json"]["code"] == "NOT_FOUND"

执行：

pytest -q

这套分层设计为什么更稳

看上面的代码，测试用例层已经很“薄”了。真正容易变动的东西，被分别放进了不同层：

• URL 变化：改接口封装层
• 超时/重试策略变化：改基础设施层
• 创建测试数据方式变化：改业务编排层
• 统一响应码变化：改断言层
• 业务流程变化：改 flow，不用批量改测试

再看一个类关系图：

classDiagram
    class HttpClient {
        +base_url
        +timeout
        +retry
        +request(method, path, **kwargs)
    }

    class UserApi {
        +health()
        +create_user(name)
        +get_user(user_id)
        +activate_user(user_id)
    }

    class UserFlow {
        +register_random_user()
        +create_and_activate_user()
        +query_user(user_id)
    }

    class Assertions {
        +assert_status_code(resp, expected_code)
        +assert_success_code(resp)
        +assert_response_time_less_than(resp, threshold_ms)
        +assert_user_name(resp, expected_name)
        +assert_user_status(resp, expected_status)
    }

    HttpClient <.. UserApi
    UserApi <.. UserFlow
    UserFlow ..> Assertions

稳定性优化实战

分层只是基础，真正让体系稳下来，还要做下面这些事。

1. 测试数据唯一化与隔离

最常见的不稳定原因之一，就是多个用例共享数据。

比如：

• 大家都创建 test_user
• 某个清理任务把公共数据删了
• 并发执行时互相覆盖

解决思路：

• 用户名、手机号、订单号都做唯一化
• 每条用例尽量自建自销，不依赖预置共享数据
• 对必须共享的数据，显式标记并加保护机制

上面示例里的 random_name() 就是最简单的一种唯一化手段。

2. 异步场景不要盲等，用轮询

很多接口调用后，结果不是立刻可见，比如：

• 写入 MQ 后异步消费
• 搜索索引异步更新
• 审批流状态延迟同步

这时最忌讳的是：

time.sleep(5)

因为它有两个问题：

• 等少了会失败
• 等多了拖慢整体执行

更好的做法是轮询：

import time

def wait_until(assert_func, timeout=5, interval=0.5):
    start = time.time()
    last_error = None
    while time.time() - start < timeout:
        try:
            return assert_func()
        except AssertionError as e:
            last_error = e
            time.sleep(interval)
    raise AssertionError(f"wait_until timeout, last_error={last_error}")

用法示例：

def test_async_example(user_flow):
    created = user_flow.register_random_user()
    user_id = created["response"]["json"]["data"]["id"]

    def check():
        resp = user_flow.query_user(user_id)
        assert resp["status_code"] == 200
        return resp

    final_resp = wait_until(check, timeout=3, interval=0.2)
    assert final_resp["json"]["data"]["id"] == user_id

3. 重试只处理“瞬时故障”，不要掩盖真实缺陷

重试是把双刃剑。

适合重试的场景：

• 网络瞬时抖动
• 下游偶发超时
• 环境短暂不可用

不适合重试的场景：

• 参数错误
• 业务逻辑错误
• 明确的 4xx 响应
• 断言失败

也就是说，请求可以按规则重试，业务断言不要无脑重跑。否则 CI 虽然“绿”了，但问题被藏起来了。

4. 断言不要写得过脆

脆弱断言长这样：

• 精确比较整个 JSON 串
• 把无关字段都纳入断言
• 连动态时间戳、traceId 也要求一致

更稳妥的做法是分层断言：

• 核心字段精确断言
• 非核心字段做存在性或类型断言
• 动态字段做格式或范围断言

例如，不要这样：

assert resp["json"] == {
    "code": "SUCCESS",
    "data": {
        "id": "固定值",
        "name": "abc",
        "status": "CREATED"
    }
}

而是这样：

assert resp["json"]["code"] == "SUCCESS"
assert "id" in resp["json"]["data"]
assert resp["json"]["data"]["name"] == "abc"
assert resp["json"]["data"]["status"] == "CREATED"

常见坑与排查

坑 1：把鉴权、数据准备、断言全写进用例

现象

测试脚本很长，每个用例都像在重新写一遍接口调用。

排查

看一个测试文件里是否同时存在这些内容：

• token 获取逻辑
• URL 拼接
• 数据生成
• 业务流程编排
• 断言细节

如果都在一个函数里，基本就是越层了。

处理建议

按“基础设施层 → 接口层 → flow 层 → 用例层”逐步拆，不要一次性大重构。先抽公共请求方法，再抽 API，再抽 flow。

坑 2：环境不稳定时，所有问题都被归咎于脚本

现象

CI 经常失败，但本地偶尔能过；失败原因五花八门。

排查路径

建议按下面顺序查：

1. 服务是否健康
2. 配置是否一致
3. 依赖服务是否超时
4. 数据是否被其他任务污染
5. 是否存在执行顺序依赖
6. 是否断言过严

一个简单状态模型可以帮助团队统一认知：

stateDiagram-v2
    [*] --> 待执行
    待执行 --> 执行中
    执行中 --> 通过
    执行中 --> 失败
    失败 --> 环境问题
    失败 --> 脚本问题
    失败 --> 产品缺陷
    失败 --> 数据问题

处理建议

给失败分类，不要把所有失败都叫“用例失败”。至少区分：

• 产品缺陷
• 自动化脚本问题
• 环境问题
• 外部依赖问题
• 测试数据问题

坑 3：用例之间有顺序依赖

现象

单独跑没问题，整套跑就失败；换个执行顺序结果又不同。

根因

通常是某条用例依赖上一条用例创建的数据，或者共享全局状态。

处理建议

• 用例自包含
• 不依赖执行顺序
• setup/teardown 明确化
• 必须串行的场景单独归类，不混在主干冒烟集里

坑 4：为了“稳定”，塞满 sleep

现象

整套回归越来越慢，但稳定性仍然一般。

根因

sleep 只是拖延，不是同步机制。

处理建议

• 异步一致性场景改为轮询
• 针对状态变更设置最大等待时间
• 记录每次轮询返回，便于定位延迟在哪一步

安全/性能最佳实践

接口自动化虽然主要目标是质量保障，但如果测试框架本身不注意安全和性能，也会制造新问题。

安全最佳实践

1. 不要把敏感信息写死在代码里

例如：

• token
• 数据库密码
• API key

建议放到环境变量或 CI 密钥管理里。

import os

BASE_URL = os.getenv("BASE_URL", "http://127.0.0.1:5001")
API_TOKEN = os.getenv("API_TOKEN", "")

2. 日志脱敏

很多测试框架喜欢打印完整请求响应，这对排查很方便，但也容易泄露：

• 手机号
• 身份证号
• token
• cookie

建议对敏感字段脱敏后再输出。

3. 测试账号权限最小化

自动化账号只给必要权限，不要直接拿生产级超级管理员权限在测试环境乱跑。

4. 严格区分环境

• 测试环境、预发环境、生产环境要有明确隔离
• 避免误把 destructive case 跑到高价值环境

性能最佳实践

1. 设置合理超时

超时太短会误报，太长会拖慢反馈。中级团队常见建议：

• connect timeout：1~3 秒
• read timeout：2~5 秒

具体还要结合系统基线。

2. 冒烟集与全量集分层

不是所有接口都要每次提交跑一遍。建议拆成：

• 冒烟集：核心链路，分钟级反馈
• 回归集：全量业务覆盖，定时或发版前执行
• 专项集：异常、幂等、权限、兼容性等

3. 控制并发度

并发执行能提升效率，但也会引入：

• 限流
• 数据冲突
• 依赖服务雪崩式失败

建议逐步压测执行框架，而不是一上来把并发开很大。

4. 记录耗时并做基线观察

接口自动化不仅能测对错，也能顺手做轻量性能守护。比如：

• P50/P95 响应时间
• 最近 7 天波动趋势
• 某核心接口耗时明显退化报警

但边界要明确：这不替代专业性能测试。

容量与落地取舍建议

如果团队刚开始做接口自动化，我建议按下面节奏推进，而不是一次到位。

阶段 1：先统一基础设施层

先把下面这几个能力统一：

• HTTP 请求入口
• 日志
• 超时
• 重试
• 配置管理

这一步投入小，收益大。

阶段 2：抽接口封装层

把最常用的一批接口抽成方法，减少复制粘贴。

阶段 3：沉淀业务 flow

当你发现多个用例都在重复“注册-登录-查询”这类流程时，就该抽 flow 了。

阶段 4：建设稳定性治理指标

可以从最简单的几个指标开始：

• 用例通过率
• 假失败率
• 平均执行时长
• Top 10 易失败用例
• 环境类失败占比

如果没有这些数据，稳定性优化很容易停留在“感觉上”。

总结

接口自动化真正难的，不是写出第一批用例，而是让它们在持续交付里长期可用。

这篇文章的核心观点可以收敛成三句：

1. 用例分层，本质是拆责任，不是堆抽象
2. 稳定性优化，本质是减少非确定性
3. CI 里的“绿”，必须建立在可解释、可定位、可维护的架构上

最后给几个可直接执行的建议：

• 用例层不要直接拼 URL、写鉴权、写数据构造
• 至少拆出：基础设施层、接口封装层、业务编排层、断言层
• 对异步场景用轮询，不要堆 sleep
• 让测试数据唯一化，减少共享状态
• 重试只兜底瞬时问题，不要掩盖业务缺陷
• 给失败分类，单独统计环境问题和脚本问题

如果你的团队当前还处在“脚本能跑，但经常红”的阶段，不需要一口气做大重构。最实际的路径是：先统一请求层，再抽 API，再沉淀 flow，最后治理稳定性指标。 这样落地阻力最小，也最容易看到效果。