实战笔记 2026-09-07 11 分钟阅读 #自动化测试 #接口自动化 #pytest 复制链接

pytest + requests 接口自动化框架：工程化细节

能跑通用例只是开始。真正决定这套框架能不能长期用的，是超时、连接复用、重试、日志和并行隔离。

1. 目录结构

api-tests/
├── core/
│   ├── config.py
│   ├── client.py
│   └── log.py
├── apis/
│   └── order_api.py
├── tests/
│   ├── conftest.py
│   └── test_order.py
├── data/
├── pytest.ini
└── requirements.txt

2. 配置：环境变量 + 默认值

# core/config.py
import os
from dataclasses import dataclass

@dataclass(frozen=True)
class Settings:
    base_url: str = os.getenv("BASE_URL", "https://test.example.com")
    user: str = os.getenv("API_USER", "")
    password: str = os.getenv("API_PASSWORD", "")
    timeout_connect: float = float(os.getenv("TIMEOUT_CONNECT", "3"))
    timeout_read: float = float(os.getenv("TIMEOUT_READ", "10"))

settings = Settings()

不要在代码里写死域名和密码。CI 用凭据注入，本地用 .env（加入 .gitignore）。

3. 客户端：连接池 + 超时 + 重试

# core/client.py
import logging
import time
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

log = logging.getLogger("api")

class ApiClient:
    def __init__(self, base_url: str, connect=3, read=10):
        self.base_url = base_url.rstrip("/")
        self.timeout = (connect, read)
        self.session = requests.Session()

        retry = Retry(
            total=2,
            connect=2,
            read=1,
            backoff_factor=0.3,
            status_forcelist=(502, 503, 504),
            allowed_methods=frozenset({"GET", "HEAD"}),  # 写接口不自动重试
            raise_on_status=False,
        )
        adapter = HTTPAdapter(
            max_retries=retry,
            pool_connections=20,
            pool_maxsize=20,
            pool_block=False,
        )
        self.session.mount("https://", adapter)
        self.session.mount("http://", adapter)

    def request(self, method: str, path: str, **kwargs):
        url = f"{self.base_url}{path}"
        kwargs.setdefault("timeout", self.timeout)
        kwargs.setdefault("headers", {}).setdefault("X-Request-Id", str(int(time.time() * 1000)))

        started = time.perf_counter()
        resp = self.session.request(method, url, **kwargs)
        cost = (time.perf_counter() - started) * 1000

        log.info("%s %s -> %s %.0fms req=%s resp=%s",
                 method, url, resp.status_code, cost,
                 str(kwargs.get("json"))[:300], resp.text[:300])
        return resp

四个关键点：

1. 超时必须是元组：timeout=10 只限制连接，读可能无限等。(3, 10) 分别限制连接与读取。
2. 写接口不自动重试：POST 重试会造成重复下单。要重试必须有幂等键。
3. 连接池要有上限：默认 pool_maxsize=10，并行 20 时会出现 Connection pool is full, discarding connection。
4. raise_on_status=False：让业务层自己判断 4xx/5xx，而不是在 HTTP 层抛异常掩盖响应内容。

4. 登录：会话复用与过期处理

# tests/conftest.py
import pytest
from core.client import ApiClient
from core.config import settings

@pytest.fixture(scope="session")
def client():
    c = ApiClient(settings.base_url)
    resp = c.request("POST", "/api/login",
                     json={"username": settings.user, "password": settings.password})
    body = resp.json()
    assert body["code"] == 0, f"登录失败: {body}"
    c.session.headers.update({"Authorization": f"Bearer {body['data']['token']}"})
    return c

常见问题：

• 401 集中出现：token 过期时间短于测试时长。加自动续期钩子或在 session 级 fixture 里记录签发时间。
• Cookie 污染：多个用例共用 Session，前一个用例退出登录导致后面全 401。退出登录用例用独立 client。
• 并行时串号：scope="session" 与 pytest-xdist 组合时，每个 worker 有自己的 session，但如果你固定用同一个账号，仍会互相踢下线。

5. 并行隔离

@pytest.fixture(scope="session")
def client(worker_id):
    suffix = "master" if worker_id == "master" else worker_id
    user = f"qa_ci_{suffix}"
    c = ApiClient(settings.base_url)
    # 每个 worker 独立账号，避免互踢与库存争抢
    ...
    return c

需要 pytest-xdist 提供 worker_id。串行执行时值恒为 "master"。

数据唯一性：

import time, uuid
def unique(prefix="qa"):
    return f"{prefix}_{int(time.time()*1000)}_{uuid.uuid4().hex[:6]}"

6. 断言：三层 + 契约

from jsonschema import validate

ORDER_SCHEMA = {
    "type": "object",
    "required": ["code", "data"],
    "properties": {
        "code": {"type": "integer"},
        "data": {
            "type": "object",
            "required": ["orderId", "status", "amount"],
            "properties": {
                "orderId": {"type": "integer", "minimum": 1},
                "status": {"enum": ["PENDING", "PAID", "CANCELLED"]},
                "amount": {"type": "string", "pattern": r"^\d+\.\d{2}$"},
            },
        },
    },
}

def test_create_order(client, order_api):
    resp = order_api.create(sku_id=1001, count=2)
    body = resp.json()

    assert resp.status_code == 200
    assert body["code"] == 0, f"业务失败: {body}"
    validate(body, ORDER_SCHEMA)
    assert body["data"]["status"] == "PENDING"

不要断言：

• 整个响应字符串（字段顺序会变）
• 自增 ID 的精确值
• 时间戳的精确值
• 错误提示的完整文案（断言错误码）

7. 清理：失败也要执行

@pytest.fixture
def created_order(client):
    resp = client.request("POST", "/api/orders", json={"skuId": 1001, "count": 1})
    order_id = resp.json()["data"]["orderId"]
    yield order_id
    client.request("DELETE", f"/api/orders/{order_id}")   # 幂等删除

删除接口要幂等（已删除返回成功），否则断言失败时清理会抛异常，掩盖原始错误。

8. 失败信息：能自己解释

def test_amount(client):
    body = client.request("GET", "/api/orders/1").json()
    assert body["data"]["amount"] == "100.00", \
        f"金额不符: 实际={body['data']['amount']} 响应={body}"

配合日志中的请求 ID，可以直接拿同一 ID 去服务端查日志。

9. 常用插件

# pytest.ini
[pytest]
markers =
    smoke: 冒烟
    regression: 全量回归
addopts = -ra --strict-markers

pip install pytest-xdist pytest-rerunfailures allure-pytest jsonschema
pytest -m smoke -n 4 --alluredir=allure-results

10. 验收

□ 断网时 3 秒内失败，而不是挂 5 分钟
□ 单条用例失败不影响其他用例的清理
□ 并行 4 个 worker 跑两次，结果一致
□ 日志里能看到每个请求的 URL、状态码、耗时、请求 ID
□ 服务端返回 500 时，断言失败信息包含响应片段
□ 删除任意一条基础数据，用例报错而不是假绿

相关阅读

• 接口自动化断言与 JSON Schema 校验
• 接口自动化数据管理与环境隔离
• Git + Jenkins 自动化测试流水线