接口自动化测试教程

api-tests/
├─ pytest.ini                 # 标记、超时与报告配置
├─ requirements.txt           # pytest、requests 或 httpx
├─ clients/
│  ├─ base_client.py          # URL、请求、日志、超时
│  └─ order_client.py         # 下单接口动作
├─ data/
│  └─ order_cases.py          # 参数化数据
├─ helpers/
│  ├─ assertions.py           # 可读的业务断言
│  └─ db.py                   # 只读数据核对
├─ tests/
│  └─ test_create_order.py    # 用例只表达业务意图
└─ conftest.py                # fixture 与清理策略

# clients/base_client.py
import requests

class BaseClient:
    def __init__(self, base_url: str, token: str):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
        })

    def request(self, method: str, path: str, **kwargs):
        # 所有请求必须有超时，避免 CI 一直等待
        kwargs.setdefault("timeout", 10)
        return self.session.request(
            method, f"{self.base_url}{path}", **kwargs
        )

# clients/order_client.py
from .base_client import BaseClient

class OrderClient(BaseClient):
    def create_order(self, payload: dict, idempotency_key: str):
        return self.request(
            "POST",
            "/api/orders",
            json=payload,
            headers={"Idempotency-Key": idempotency_key},
        )

# 如果项目采用异步调用，可将 requests.Session 替换为
# httpx.AsyncClient，并使用 await client.post(...)。

# conftest.py 中的测试数据工厂
import uuid

def build_order_payload(**overrides):
    payload = {
        "sku_id": "sku_1001",
        "quantity": 2,
        "address_id": "address_a",
        "coupon_id": "coupon_20",
        "case_id": f"auto-{uuid.uuid4().hex[:10]}",
    }
    payload.update(overrides)
    return payload

def test_create_order(order_client, order_payload):
    response = order_client.create_order(
        order_payload, idempotency_key=order_payload["case_id"]
    )

    # 1. 协议层
    assert response.status_code == 200, response.text
    assert response.headers["Content-Type"].startswith("application/json")

    body = response.json()
    # 2. 结构层
    assert isinstance(body["order_id"], str)
    assert body["order_id"]

    # 3. 业务层
    assert body["status"] == "PENDING_PAYMENT"
    assert body["amount"] == 80

    # 数据层断言放在独立 helper 中，便于复用与诊断

import pytest

@pytest.mark.parametrize(
    "quantity, expected_status, expected_code",
    [
        (0, 400, "INVALID_QUANTITY"),
        (1, 200, None),
        (5, 200, None),
        (6, 400, "INVALID_QUANTITY"),
    ],
    ids=["below-min", "min", "max", "above-max"],
)
def test_order_quantity(
    order_client, build_payload, quantity,
    expected_status, expected_code,
):
    payload = build_payload(quantity=quantity)
    response = order_client.create_order(payload, payload["case_id"])
    assert response.status_code == expected_status
    if expected_code:
        assert response.json()["code"] == expected_code

def test_same_key_creates_only_one_order(
    order_client, build_payload, order_repository,
):
    payload = build_payload(quantity=2)
    key = payload["case_id"]

    first = order_client.create_order(payload, key)
    second = order_client.create_order(payload, key)

    assert first.status_code == 200
    assert second.status_code == 200
    assert second.json()["order_id"] == first.json()["order_id"]
    assert order_repository.count_by_case_id(key) == 1
    assert order_repository.stock_change(key) == -2

# helpers/db.py：测试账号只授予只读权限
class OrderRepository:
    def __init__(self, connection):
        self.connection = connection

    def find_by_id(self, order_id: str):
        with self.connection.cursor() as cursor:
            cursor.execute(
                """SELECT order_id, user_id, amount, status
                   FROM orders WHERE order_id = %s""",
                (order_id,),
            )
            return cursor.fetchone()

def assert_order_saved(repository, body, expected_user_id):
    row = repository.find_by_id(body["order_id"])
    assert row is not None
    assert row["user_id"] == expected_user_id
    assert row["amount"] == body["amount"]
    assert row["status"] == body["status"]

# conftest.py
import pytest
from clients.order_client import OrderClient

@pytest.fixture(scope="session")
def order_client(settings, user_token):
    return OrderClient(settings.base_url, user_token)

@pytest.fixture
def build_payload(request):
    created_case_ids = []

    def _build(**overrides):
        payload = make_order_payload(**overrides)
        created_case_ids.append(payload["case_id"])
        return payload

    yield _build

    # 清理动作必须按 case_id 精确定位，不能全表清空
    cleanup_test_orders(created_case_ids)

# .github/workflows/api-tests.yml（示意）
name: api-tests
on: [pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install -r requirements.txt
      - run: pytest -m smoke --junitxml=reports/junit.xml
        env:
          BASE_URL: ${{ secrets.TEST_BASE_URL }}
          TEST_TOKEN: ${{ secrets.TEST_TOKEN }}
      - uses: actions/upload-artifact@v4
        if: always()
        with:
          name: api-test-report
          path: reports/