【接口自动化】10 - 接口自动化框架搭建

2501_90821979

768人浏览 · 2026-09-02 22:45:00

2501_90821979 · 2026-09-02 22:45:00 发布

前言

第 9 篇搭建了骨架，但框架还比较"朴素"——ApiClient 只有基本的请求和日志，没有重试、没有超时配置、没有统一响应校验。本篇在已有基础上做五件事：

增强 ApiClient：重试机制、可配置超时、Session 管理完善
统一响应处理：自动校验 JSON 结构、自动附加日志
完善接口封装：补充遗漏的接口、统一方法命名
JSON Schema 校验：验证返回数据的类型和字段是否符合预期
执行速度优化：Session 共享 + 数据驱动

一、增强 ApiClient

第 9 篇的 ApiClient 能用，但有几个问题：

网络抖动时直接失败，没有重试
超时写死 10 秒，不能按接口配置
没有关闭 session 的安全保证（异常时可能泄漏）

1.1 重写 base_api.py

替换 api/api_objects/base_api.py：

"""
ApiClient 基类（增强版）

增强点：
1. 自动重试（5xx 错误自动重试 2 次，带退避间隔）
2. 可配置超时（默认 10 秒，单个接口可覆盖）
3. Session 安全管理（确保异常时也能关闭）
4. 统一的请求/响应日志格式
"""

import requests
import json
import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from api.common.logger import get_logger

logger = get_logger("api_client")


class ApiClient:
    """
    接口请求基类

    所有业务 API 类（UserApi、CartApi 等）继承此类

    用法：
        api = ApiClient("http://localhost:8000")
        resp = api.get("/api/products")
        resp = api.post("/api/login", json_data={"username": "admin", "password": "123"})
        api.close()
    """

    def __init__(self, base_url="http://localhost:8000"):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()

        # 默认请求头
        self.session.headers.update({
            "Content-Type": "application/json",
            "Accept": "application/json",
        })

        # 配置自动重试策略
        # 只对 GET/PUT/DELETE（幂等请求）重试，POST 不自动重试（避免重复创建）
        retry_strategy = Retry(
            total=2,                          # 最多重试 2 次
            backoff_factor=0.5,               # 退避间隔：第 1 次等 0.5s，第 2 次等 1s
            status_forcelist=[500, 502, 503, 504],  # 这些 HTTP 状态码触发重试
            allowed_methods=["GET", "PUT", "DELETE"],
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)

        # 请求统计
        self._request_count = 0
        self._fail_count = 0

    # ========================================
    # HTTP 请求方法
    # ========================================

    def get(self, path, params=None, timeout=None):
        """
        GET 请求

        参数：
            path: 接口路径
            params: URL 参数
            timeout: 超时秒数（None 用默认 10 秒）
        """
        return self._request("GET", path, params=params, timeout=timeout)

    def post(self, path, json_data=None, timeout=None):
        """POST 请求"""
        return self._request("POST", path, json_data=json_data, timeout=timeout)

    def put(self, path, json_data=None, timeout=None):
        """PUT 请求"""
        return self._request("PUT", path, json_data=json_data, timeout=timeout)

    def delete(self, path, timeout=None):
        """DELETE 请求"""
        return self._request("DELETE", path, timeout=timeout)

    def _request(self, method, path, json_data=None, params=None, timeout=None):
        """
        发送请求的核心方法

        职责：
        1. 拼接完整 URL
        2. 发送请求（自动重试由 urllib3 Retry 处理）
        3. 记录请求/响应日志
        4. 更新请求统计
        """
        url = f"{self.base_url}{path}"
        self._request_count += 1
        req_timeout = timeout or 10

        # ===== 请求日志 =====
        logger.info(f">>> {method} {url}")
        if params:
            logger.info(f"    Params: {json.dumps(params, ensure_ascii=False)}")
        if json_data:
            logger.info(f"    Body: {json.dumps(json_data, ensure_ascii=False)}")

        # ===== 发送请求 =====
        start = time.time()
        try:
            response = self.session.request(
                method=method,
                url=url,
                json=json_data,
                params=params,
                timeout=req_timeout,
            )
            elapsed = time.time() - start

            # ===== 响应日志 =====
            logger.info(f"<<< {response.status_code} ({elapsed:.3f}s)")
            try:
                body = response.json()
                logger.info(f"    {json.dumps(body, ensure_ascii=False)[:300]}")
            except Exception:
                logger.info(f"    {response.text[:200]}")

            # 统计失败
            if response.status_code >= 400:
                self._fail_count += 1

            return response

        except requests.exceptions.ConnectionError:
            self._fail_count += 1
            logger.error(f"连接失败：{url}（请确认 MallLite 是否启动）")
            raise
        except requests.exceptions.Timeout:
            self._fail_count += 1
            logger.error(f"请求超时：{url}（{req_timeout}s）")
            raise
        except Exception as e:
            self._fail_count += 1
            logger.error(f"请求异常：{e}")
            raise

    # ========================================
    # 会话管理
    # ========================================

    def close(self):
        """关闭 session"""
        self.session.close()
        logger.debug(f"Session 已关闭（请求 {self._request_count} 次，失败 {self._fail_count} 次）")

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.close()

1.2 增强了什么

跟第 9 篇对比：

特性	第 9 篇	本篇
重试	无	5xx 自动重试 2 次，带退避间隔
重试范围	-	只重试 GET/PUT/DELETE（幂等），POST 不重试
超时	写死 10 秒	可按接口配置
上下文管理器	无	支持 `with ApiClient() as api:`
请求统计	无	记录总请求数和失败数

1.3 为什么 POST 不自动重试

GET/PUT/DELETE 是幂等的——调一次和调多次效果一样。但 POST 可能创建资源，重试会导致重复创建。比如 POST /api/cart 加购物车，第一次成功了但响应超时，自动重试会加两次。

二、完善接口封装

2.1 UserApi 补充

第 9 篇的 UserApi 只有 login 和 register。补充两个实用方法：

替换 api/api_objects/user_api.py：

"""
UserApi - 用户相关接口（完善版）

完善点：
1. 新增 login_and_share_session：登录并返回 session（供其他 API 共享）
2. 统一方法命名风格
"""

from api.api_objects.base_api import ApiClient
from api.common.logger import get_logger

logger = get_logger("user_api")


class UserApi(ApiClient):
    """用户接口类"""

    def login(self, username, password):
        """
        用户登录

        MallLite 使用 Session + Cookie 认证
        登录后 session 自动保存 Cookie，后续请求自动携带

        参数：
            username: 用户名
            password: 密码

        返回：
            requests.Response
        """
        logger.info(f"登录：{username}")
        return self.post("/api/login", json_data={
            "username": username,
            "password": password,
        })

    def login_and_share(self, username, password):
        """
        登录并返回 session（供其他 API 类共享）

        这是 fixture 中使用的方法：
        1. 用 UserApi 登录
        2. session 保存了 Cookie
        3. 把 session 赋给 CartApi/OrderApi
        4. 后者就能以登录状态发请求

        返回：
            当前实例的 session（已含 Cookie）

        异常：
            AssertionError: 登录失败时抛出
        """
        resp = self.login(username, password)
        data = resp.json()
        assert data["code"] == 200, f"登录失败：{data['message']}"
        logger.info(f"登录成功：{data['data']['username']}（{data['data']['role']}）")
        return self.session

    def register(self, username, password, nickname=None, phone=None, email=None):
        """
        用户注册

        参数：
            username: 用户名（必须）
            password: 密码（必须）
            nickname: 昵称（可选）
            phone: 手机号（可选）
            email: 邮箱（可选）
        """
        logger.info(f"注册：{username}")
        body = {"username": username, "password": password}
        if nickname:
            body["nickname"] = nickname
        if phone:
            body["phone"] = phone
        if email:
            body["email"] = email
        return self.post("/api/register", json_data=body)

    def get_user_info(self):
        """
        获取当前登录用户信息

        注意：MallLite 没有独立的 /api/user/info 接口
        用户信息在登录响应中返回
        如果 MallLite 后续增加了此接口，可直接使用
        """
        return self.get("/api/user/info")

2.2 ProductApi 完善

api/api_objects/product_api.py 保持不变，第 9 篇的版本已经够用。

2.3 CartApi 完善

替换 api/api_objects/cart_api.py：

"""
CartApi - 购物车相关接口（完善版）

关键点（来自探针脚本验证）：
  - 购物车的增删改查都用 product_id，不是 cart_item_id
  - 未登录访问购物车：code=200 + 返回空数据
  - 未登录添加购物车：code=200 + 添加成功（MallLite 的实际行为）

接口：
  GET    /api/cart            获取购物车
  POST   /api/cart            添加商品     body: {"product_id":1, "quantity":2}
  PUT    /api/cart/{product_id}  修改数量  body: {"quantity":5}
  DELETE /api/cart/{product_id}  删除商品
"""

from api.api_objects.base_api import ApiClient
from api.common.logger import get_logger

logger = get_logger("cart_api")


class CartApi(ApiClient):

    def get_cart(self):
        """获取购物车列表"""
        return self.get("/api/cart")

    def add_to_cart(self, product_id, quantity=1):
        """
        添加商品到购物车

        参数：
            product_id: 商品 ID
            quantity: 数量（默认 1）
        """
        logger.info(f"添加到购物车：商品ID={product_id}，数量={quantity}")
        return self.post("/api/cart", json_data={
            "product_id": product_id,
            "quantity": quantity,
        })

    def update_quantity(self, product_id, quantity):
        """
        修改购物车中某商品的数量

        参数：
            product_id: 商品 ID（不是购物车记录 ID）
            quantity: 新数量
        """
        logger.info(f"修改数量：商品ID={product_id} → {quantity}")
        return self.put(f"/api/cart/{product_id}", json_data={
            "quantity": quantity,
        })

    def delete_from_cart(self, product_id):
        """
        从购物车删除某商品

        参数：
            product_id: 商品 ID
        """
        logger.info(f"删除购物车商品：商品ID={product_id}")
        return self.delete(f"/api/cart/{product_id}")

    def clear_cart(self):
        """
        清空购物车（便捷方法）

        获取所有商品，逐个删除
        """
        resp = self.get_cart()
        data = resp.json()
        if data.get("code") == 200:
            items = data["data"]["items"]
            for item in items:
                self.delete_from_cart(item["product_id"])
            logger.info(f"购物车已清空（删除 {len(items)} 个商品）")

2.4 OrderApi 完善

替换 api/api_objects/order_api.py：

"""
OrderApi - 订单相关接口（完善版）

关键点（来自探针脚本验证）：
  - 订单没有 id 字段，查详情用 order_no
  - 订单列表字段：order_no, user_id, items, total_price, status, created_at, updated_at
  - 空购物车创建订单：code=400, message=购物车为空
"""

from api.api_objects.base_api import ApiClient
from api.common.logger import get_logger

logger = get_logger("order_api")


class OrderApi(ApiClient):

    def create_order(self):
        """
        创建订单（从购物车结算）

        前提：购物车中必须有商品
        否则返回：code=400, message=购物车为空
        """
        logger.info("创建订单")
        return self.post("/api/orders")

    def get_orders(self, page=1, page_size=10):
        """
        获取订单列表

        返回字段：items 中每个元素包含
          order_no, user_id, items, total_price, status, created_at, updated_at
        """
        return self.get("/api/orders", params={"page": page, "page_size": page_size})

    def get_order(self, order_no):
        """
        获取订单详情

        参数：
            order_no: 订单号（如 "ORD20260827-000003"）

        注意：这里用 order_no 查询，不是 id
        """
        return self.get(f"/api/orders/{order_no}")

三、统一响应处理

3.1 响应校验器

创建 api/common/response_validator.py：

"""
统一响应校验器

MallLite 的响应格式固定为：
  {"code": 200, "message": "...", "data": {...}}

这个校验器自动检查基本结构，避免每个用例重复写相同的断言
"""

from api.common.logger import get_logger

logger = get_logger("validator")


class ResponseValidator:
    """
    响应校验器

    用法：
        resp = user_api.login("admin", "admin123")
        validator = ResponseValidator(resp)

        # 检查基本结构
        validator.validate_structure()

        # 或者用快捷方法
        validator.validate_success()
        validator.validate_error()
    """

    def __init__(self, response):
        """
        参数：
            response: requests.Response 对象
        """
        self.resp = response
        self.status_code = response.status_code
        try:
            self.body = response.json()
        except Exception:
            self.body = None

    def validate_structure(self):
        """
        校验响应的基本结构

        校验项：
        1. HTTP 状态码 200
        2. 响应体是有效 JSON
        3. 包含 code 字段
        4. 包含 message 字段
        """
        assert self.status_code == 200, \
            f"HTTP 状态码期望 200，实际 {self.status_code}"
        assert self.body is not None, \
            "响应体不是有效的 JSON"
        assert "code" in self.body, \
            f"响应缺少 code 字段：{self.body}"
        assert "message" in self.body or self.body["code"] == 200, \
            f"响应缺少 message 字段：{self.body}"
        logger.debug("响应结构校验通过")
        return self

    def validate_success(self):
        """
        校验业务成功

        校验项：
        1. 基本结构正确
        2. code == 200
        """
        self.validate_structure()
        assert self.body["code"] == 200, \
            f"业务码期望 200，实际 {self.body['code']}，消息：{self.body.get('message')}"
        return self

    def validate_error(self):
        """
        校验业务失败

        校验项：
        1. 基本结构正确
        2. code != 200
        """
        self.validate_structure()
        assert self.body["code"] != 200, \
            "接口不应该返回成功"
        return self

    def validate_list(self, items_path="data.items"):
        """
        校验列表类接口

        校验项：
        1. 业务成功
        2. data.items 是列表
        """
        self.validate_success()
        parts = items_path.split(".")
        current = self.body
        for part in parts:
            current = current[part]
        assert isinstance(current, list), \
            f"{items_path} 应该是列表，实际为 {type(current)}"
        return self

3.2 JSON Schema 校验

JSON Schema 是一种描述 JSON 数据结构的标准。用它可以自动验证接口返回的数据是否符合预期——字段类型对不对、必填字段有没有、值的范围合不合法。

安装：

pip install jsonschema

创建 api/common/schemas.py：

"""
MallLite 接口响应 Schema 定义

用途：自动校验响应数据的结构和类型是否正确

用法：
    from api.common.schemas import validate_schema
    validate_schema(resp.json(), "login")         # 校验登录响应
    validate_schema(resp.json(), "product_list")  # 校验商品列表
"""

from api.common.logger import get_logger

logger = get_logger("schemas")


# MallLite 的接口 Schema 定义
# 每个 Schema 对应一个接口的响应格式

SCHEMAS = {}

# ===== 登录响应 =====
# 探针确认返回：
#   code: 200, message: "登录成功",
#   data: {id: 1, username: "admin", phone: "...", email: "...", role: "admin"}
SCHEMAS["login"] = {
    "type": "object",
    "required": ["code", "message", "data"],
    "properties": {
        "code": {"type": "integer"},
        "message": {"type": "string"},
        "data": {
            "type": "object",
            "required": ["username", "role"],
            "properties": {
                "id": {"type": "integer"},
                "username": {"type": "string"},
                "role": {"type": "string"},
                "phone": {"type": ["string", "null"]},
                "email": {"type": ["string", "null"]},
            }
        }
    }
}

# ===== 商品列表 =====
SCHEMAS["product_list"] = {
    "type": "object",
    "required": ["code", "data"],
    "properties": {
        "code": {"type": "integer"},
        "data": {
            "type": "object",
            "required": ["items"],
            "properties": {
                "items": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "required": ["name", "price"],
                        "properties": {
                            "id": {"type": "integer"},
                            "name": {"type": "string"},
                            "price": {"type": "number"},
                        }
                    }
                }
            }
        }
    }
}

# ===== 商品详情 =====
SCHEMAS["product_detail"] = {
    "type": "object",
    "required": ["code", "data"],
    "properties": {
        "code": {"type": "integer"},
        "data": {
            "type": "object",
            "required": ["name", "price"],
            "properties": {
                "id": {"type": "integer"},
                "name": {"type": "string"},
                "price": {"type": "number"},
                "stock": {"type": "integer"},
                "description": {"type": ["string", "null"]},
            }
        }
    }
}

# ===== 购物车 =====
SCHEMAS["cart"] = {
    "type": "object",
    "required": ["code", "data"],
    "properties": {
        "code": {"type": "integer"},
        "data": {
            "type": "object",
            "required": ["items"],
            "properties": {
                "items": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "required": ["product_id", "quantity"],
                        "properties": {
                            "product_id": {"type": "integer"},
                            "quantity": {"type": "integer"},
                        }
                    }
                },
                "total_price": {"type": "number"},
                "total_quantity": {"type": "integer"},
                "item_count": {"type": "integer"},
            }
        }
    }
}

# ===== 订单 =====
SCHEMAS["order_detail"] = {
    "type": "object",
    "required": ["code", "data"],
    "properties": {
        "code": {"type": "integer"},
        "data": {
            "type": "object",
            "required": ["order_no", "status", "items"],
            "properties": {
                "order_no": {"type": "string"},
                "status": {"type": "string"},
                "items": {"type": "array"},
                "total_price": {"type": "number"},
            }
        }
    }
}

# ===== 分类列表 =====
SCHEMAS["category_list"] = {
    "type": "object",
    "required": ["code", "data"],
    "properties": {
        "code": {"type": "integer"},
        "data": {
            "type": "array",
            "items": {
                "type": "object",
                "required": ["id", "name"],
            }
        }
    }
}


def validate_schema(data, schema_name):
    """
    校验数据是否符合 Schema

    参数：
        data: 要校验的数据（通常是 resp.json()）
        schema_name: Schema 名称

    用法：
        validate_schema(resp.json(), "login")
        validate_schema(resp.json(), "product_list")

    通过：无返回
    失败：抛出 jsonschema.ValidationError
    """
    from jsonschema import validate

    schema = SCHEMAS.get(schema_name)
    if schema is None:
        raise ValueError(f"Schema '{schema_name}' 不存在。可用：{list(SCHEMAS.keys())}")

    validate(instance=data, schema=schema)
    logger.debug(f"Schema 校验通过：{schema_name}")

3.3 在断言工具中集成

在 api/common/assertions.py 中追加一个方法：

# 在 ApiAssertions 类中追加

def assert_schema(self, schema_name):
    """
    断言响应体符合 JSON Schema

    用法：
        assert_api(resp).assert_success().assert_schema("login")
    """
    from api.common.schemas import validate_schema
    validate_schema(self.body, schema_name)
    logger.info(f"  ✓ Schema 校验通过：{schema_name}")
    return self

四、执行速度优化

4.1 问题

每条用例的 fixture 都创建新的 API 实例并登录，32 条用例登录 20+ 次，每次约 0.5 秒，光登录就花 10-20 秒。

4.2 解法：Session 共享 + 自动清理

替换 api/conftest.py：

"""
接口自动化 - conftest.py（增强版）

增强点：
1. 登录 session 改为 session scope，整个会话只登录一次
2. 每个用例执行前自动清空购物车，避免数据污染
3. 新增 admin_user_api fixture
"""

import pytest
from api.api_objects.user_api import UserApi
from api.api_objects.product_api import ProductApi
from api.api_objects.cart_api import CartApi
from api.api_objects.order_api import OrderApi
from api.common.assertions import ApiAssertions
from api.common.logger import get_logger

logger = get_logger("conftest")

BASE_URL = "http://localhost:8000"


# ========================================
# session 级共享（只登录一次）
# ========================================

@pytest.fixture(scope="session")
def admin_session():
    """
    管理员登录后的 session（整个测试会话只创建一次）

    原理：
      UserApi 登录后 session 保存了 Cookie
      所有需要登录的 fixture 共享这个 session
      不需要重复登录，节省大量时间
    """
    user = UserApi(BASE_URL)
    session = user.login_and_share("admin", "admin123")
    return session


# ========================================
# 未登录的 API 客户端
# ========================================

@pytest.fixture
def user_api():
    """用户 API（未登录）"""
    api = UserApi(BASE_URL)
    yield api
    api.close()


@pytest.fixture
def product_api():
    """商品 API（不需要登录）"""
    api = ProductApi(BASE_URL)
    yield api
    api.close()


@pytest.fixture
def cart_api():
    """购物车 API（未登录）"""
    api = CartApi(BASE_URL)
    yield api
    api.close()


@pytest.fixture
def order_api():
    """订单 API（未登录）"""
    api = OrderApi(BASE_URL)
    yield api
    api.close()


# ========================================
# 已登录的 API 客户端（共享 session）
# ========================================

@pytest.fixture
def admin_user_api(admin_session):
    """已登录的用户 API"""
    api = UserApi(BASE_URL)
    api.session = admin_session
    return api


@pytest.fixture
def admin_cart_api(admin_session):
    """已登录的购物车 API"""
    api = CartApi(BASE_URL)
    api.session = admin_session
    return api


@pytest.fixture
def admin_order_api(admin_session):
    """已登录的订单 API"""
    api = OrderApi(BASE_URL)
    api.session = admin_session
    return api


# ========================================
# 自动清理（避免用例间数据污染）
# ========================================

@pytest.fixture(autouse=True)
def clean_cart(admin_session):
    """
    每个用例执行前自动清空购物车

    为什么需要这个？
      因为 admin_session 是共享的，用例 A 往购物车加了商品，
      用例 B 执行时购物车不是空的，会导致用例 B 的断言失败。
      通过 autouse=True 自动清理，每个用例开始时购物车都是空的。
    """
    cart = CartApi(BASE_URL)
    cart.session = admin_session
    cart.clear_cart()
    yield


# ========================================
# 工具和数据
# ========================================

@pytest.fixture
def assert_api():
    """断言工具"""
    return ApiAssertions


@pytest.fixture
def admin_account():
    return {"username": "admin", "password": "admin123"}

4.3 速度对比

项目	第 9 篇	本篇
登录次数	每条用例登录一次（20+ 次）	整个会话登录一次
购物车清理	无（用例间可能互相污染）	每个用例前自动清理
预计耗时	~89 秒	~30-40 秒

五、数据驱动

5.1 测试数据文件

创建 api/test_data/login_cases.json：

{
  "success_cases": [
    {"case_name": "管理员", "username": "admin", "password": "admin123", "expected_role": "admin"},
    {"case_name": "普通用户", "username": "testuser", "password": "test123", "expected_role": "user"},
    {"case_name": "VIP用户", "username": "vipuser", "password": "vip123", "expected_role": "vip"}
  ],
  "fail_cases": [
    {"case_name": "密码错误", "username": "admin", "password": "wrong", "expected_code": 401, "expected_msg": "密码错误"},
    {"case_name": "用户不存在", "username": "nobody_xyz", "password": "123", "expected_code": 401, "expected_msg": "用户不存在"},
    {"case_name": "用户名为空", "username": "", "password": "admin123", "expected_code": 401, "expected_msg": "请输入用户名"},
    {"case_name": "密码为空", "username": "admin", "password": "", "expected_code": 401, "expected_msg": "请输入密码"}
  ]
}

创建 api/test_data/product_cases.json：

{
  "search_cases": [
    {"keyword": "iPhone", "min_count": 1},
    {"keyword": "Pro", "min_count": 2},
    {"keyword": "华为", "min_count": 1},
    {"keyword": "AirPods", "min_count": 1}
  ],
  "category_cases": [
    {"category_id": 1, "name": "手机", "min_count": 2},
    {"category_id": 2, "name": "笔记本", "min_count": 1},
    {"category_id": 3, "name": "平板", "min_count": 1},
    {"category_id": 4, "name": "配件", "min_count": 2}
  ]
}

5.2 数据读取工具

创建 api/common/data_reader.py：

"""
测试数据读取工具

从 test_data/ 目录读取 JSON / YAML 文件
"""

import json
from pathlib import Path
from api.common.logger import get_logger

logger = get_logger("data_reader")

DATA_DIR = Path(__file__).parent.parent / "test_data"


def read_json(filename):
    """读取 JSON 文件"""
    filepath = DATA_DIR / filename
    if not filepath.exists():
        raise FileNotFoundError(f"数据文件不存在：{filepath}")
    with open(filepath, "r", encoding="utf-8") as f:
        data = json.load(f)
    logger.debug(f"读取数据：{filename}")
    return data


def read_yaml(filename):
    """读取 YAML 文件"""
    import yaml
    filepath = DATA_DIR / filename
    if not filepath.exists():
        raise FileNotFoundError(f"数据文件不存在：{filepath}")
    with open(filepath, "r", encoding="utf-8") as f:
        data = yaml.safe_load(f)
    logger.debug(f"读取数据：{filename}")
    return data

5.3 数据提取器

创建 api/common/extractor.py：

"""
响应数据提取器

从 JSON 响应中按路径提取数据，用于接口关联场景

用法：
    body = resp.json()
    name = Extractor.get(body, "data.username")          # "admin"
    first = Extractor.get(body, "data.items.0.name")     # "iPhone 15 Pro"
    names = Extractor.get_all(body, "data.items.*.name") # ["iPhone", "MacBook", ...]
    n = Extractor.count(body, "data.items")               # 15
"""

from api.common.logger import get_logger

logger = get_logger("extractor")


class Extractor:

    @staticmethod
    def get(data, path, default=None):
        """
        按路径提取值

        路径语法：
            data.username       → 取 data 下的 username
            data.items.0.name   → 取 items 列表第 1 个元素的 name
            data.items.-1.name  → 取最后一个元素的 name
        """
        try:
            parts = path.split(".")
            current = data
            for part in parts:
                if current is None:
                    return default
                if isinstance(current, list):
                    current = current[int(part)]
                elif isinstance(current, dict):
                    current = current[part]
                else:
                    return default
            return current
        except (KeyError, IndexError, TypeError, ValueError):
            return default

    @staticmethod
    def get_all(data, path):
        """
        提取列表中每个元素的某个字段

        路径中 * 表示遍历列表

        用法：
            Extractor.get_all(body, "data.items.*.name")
            返回 ["iPhone 15 Pro", "MacBook Pro", ...]
        """
        parts = path.split(".")
        star_idx = parts.index("*")

        # 定位到列表
        list_path = ".".join(parts[:star_idx])
        items = Extractor.get(data, list_path) if list_path else data

        if not isinstance(items, list):
            return []

        # 从每个元素提取
        remaining = ".".join(parts[star_idx + 1:])
        if not remaining:
            return items

        return [Extractor.get(item, remaining) for item in items
                if Extractor.get(item, remaining) is not None]

    @staticmethod
    def count(data, path):
        """获取列表长度"""
        items = Extractor.get(data, path)
        return len(items) if isinstance(items, list) else 0

    @staticmethod
    def first(data, list_path):
        """获取列表第一个元素"""
        items = Extractor.get(data, list_path)
        return items[0] if isinstance(items, list) and items else None

六、用数据驱动重写用例

6.1 test_user.py（数据驱动版）

"""
用户接口测试（数据驱动版 + Schema 校验）
"""

import pytest
import time
from api.common.data_reader import read_json
from api.common.schemas import validate_schema

login_data = read_json("login_cases.json")
success_cases = login_data["success_cases"]
fail_cases = login_data["fail_cases"]


class TestLogin:

    @pytest.mark.smoke
    @pytest.mark.login
    @pytest.mark.p0
    @pytest.mark.parametrize("case", success_cases,
                             ids=[c["case_name"] for c in success_cases])
    def test_login_success(self, user_api, assert_api, case):
        """数据驱动：多种用户类型登录"""
        resp = user_api.login(case["username"], case["password"])

        a = assert_api(resp)
        a.assert_success()
        a.assert_field_value("data.username", case["username"])
        a.assert_field_value("data.role", case["expected_role"])
        a.assert_schema("login")

    @pytest.mark.regression
    @pytest.mark.login
    @pytest.mark.negative
    @pytest.mark.parametrize("case", fail_cases,
                             ids=[c["case_name"] for c in fail_cases])
    def test_login_fail(self, user_api, assert_api, case):
        """数据驱动：各种登录失败场景"""
        resp = user_api.login(case["username"], case["password"])

        a = assert_api(resp)
        a.assert_biz_fail()
        a.assert_code(case["expected_code"])
        a.assert_message(case["expected_msg"])

    @pytest.mark.smoke
    @pytest.mark.login
    def test_login_session_persists(self, user_api, assert_api, admin_account):
        """登录后 Cookie 能保持"""
        user_api.login(admin_account["username"], admin_account["password"])
        resp = user_api.get("/api/cart")
        assert resp.json()["code"] == 200

    @pytest.mark.regression
    @pytest.mark.login
    def test_login_roles(self, user_api, assert_api):
        """所有角色都能登录"""
        for case in success_cases:
            resp = user_api.login(case["username"], case["password"])
            assert_api(resp).assert_success().assert_field_value("data.role", case["expected_role"])

    @pytest.mark.regression
    @pytest.mark.login
    @pytest.mark.negative
    @pytest.mark.parametrize("username,password", [
        ("admin' OR '1'='1", "123"),
        ("<script>alert(1)</script>", "123"),
        ("a" * 256, "123"),
    ], ids=["SQL注入", "XSS", "超长用户名"])
    def test_login_malicious(self, user_api, username, password):
        """恶意输入不导致服务端崩溃"""
        resp = user_api.login(username, password)
        assert resp.status_code == 200

    @pytest.mark.regression
    @pytest.mark.login
    @pytest.mark.schema
    def test_login_response_schema(self, user_api, admin_account):
        """Schema 校验：登录响应结构"""
        resp = user_api.login(admin_account["username"], admin_account["password"])
        validate_schema(resp.json(), "login")


class TestRegister:

    @pytest.mark.regression
    @pytest.mark.register
    def test_register_success(self, user_api, assert_api):
        resp = user_api.register(f"auto_{int(time.time())}", "Test123456")
        assert_api(resp).assert_success()

    @pytest.mark.regression
    @pytest.mark.register
    @pytest.mark.negative
    def test_register_duplicate(self, user_api, assert_api):
        resp = user_api.register("admin", "123456")
        assert_api(resp).assert_biz_fail()

6.2 test_product.py（数据驱动版）

"""
商品接口测试（数据驱动版 + Schema 校验）
"""

import pytest
from api.common.data_reader import read_json
from api.common.schemas import validate_schema

product_data = read_json("product_cases.json")


class TestProductList:

    @pytest.mark.smoke
    @pytest.mark.product
    @pytest.mark.p0
    def test_get_product_list(self, product_api, assert_api):
        resp = product_api.get_products()
        a = assert_api(resp)
        a.assert_success()
        a.assert_list_not_empty("data.items")
        a.assert_schema("product_list")

    @pytest.mark.smoke
    @pytest.mark.product
    def test_get_product_detail(self, product_api, assert_api):
        resp = product_api.get_products(page_size=1)
        first_id = assert_api(resp).get_field("data.items")[0]["id"]

        resp = product_api.get_product(first_id)
        a = assert_api(resp)
        a.assert_success()
        a.assert_field_gt("data.price", 0)
        a.assert_schema("product_detail")

    @pytest.mark.regression
    @pytest.mark.product
    @pytest.mark.parametrize("case", product_data["search_cases"],
                             ids=[c["keyword"] for c in product_data["search_cases"]])
    def test_search_by_keyword(self, product_api, assert_api, case):
        resp = product_api.get_products(keyword=case["keyword"])
        a = assert_api(resp)
        a.assert_success()
        a.assert_list_min_length("data.items", case["min_count"])

    @pytest.mark.regression
    @pytest.mark.product
    @pytest.mark.negative
    def test_search_no_result(self, product_api, assert_api):
        resp = product_api.get_products(keyword="xyz_not_exist_999")
        a = assert_api(resp)
        a.assert_success()
        assert len(a.get_field("data.items")) == 0

    @pytest.mark.regression
    @pytest.mark.product
    @pytest.mark.parametrize("case", product_data["category_cases"],
                             ids=[c["name"] for c in product_data["category_cases"]])
    def test_filter_by_category(self, product_api, assert_api, case):
        resp = product_api.get_products(category_id=case["category_id"])
        a = assert_api(resp)
        a.assert_success()
        a.assert_list_min_length("data.items", case["min_count"])

    @pytest.mark.regression
    @pytest.mark.product
    @pytest.mark.p2
    def test_pagination(self, product_api, assert_api):
        r1 = product_api.get_products(page=1, page_size=3)
        r2 = product_api.get_products(page=2, page_size=3)
        ids1 = {i["id"] for i in assert_api(r1).get_field("data.items")}
        ids2 = {i["id"] for i in assert_api(r2).get_field("data.items")}
        if ids1 and ids2:
            assert ids1 != ids2

    @pytest.mark.smoke
    @pytest.mark.category
    def test_get_categories(self, product_api, assert_api):
        resp = product_api.get_categories()
        a = assert_api(resp)
        a.assert_success()
        a.assert_schema("category_list")
        assert len(a.get_data()) >= 4

cart 和 order 用例保持第 9 篇修正版不变。

七、运行验证

pytest api\ -v

预期输出（用例数增加，但速度更快）：

============================= test session starts ==============================
collected 52 items

test_user.py::TestLogin::test_login_success[管理员] PASSED
test_user.py::TestLogin::test_login_success[普通用户] PASSED
test_user.py::TestLogin::test_login_success[VIP用户] PASSED
test_user.py::TestLogin::test_login_fail[密码错误] PASSED
test_user.py::TestLogin::test_login_fail[用户不存在] PASSED
test_user.py::TestLogin::test_login_fail[用户名为空] PASSED
test_user.py::TestLogin::test_login_fail[密码为空] PASSED
test_user.py::TestLogin::test_login_session_persists PASSED
test_user.py::TestLogin::test_login_roles PASSED
test_user.py::TestLogin::test_login_malicious[SQL注入] PASSED
test_user.py::TestLogin::test_login_malicious[XSS] PASSED
test_user.py::TestLogin::test_login_malicious[超长用户名] PASSED
test_user.py::TestLogin::test_login_response_schema PASSED
test_user.py::TestRegister::test_register_success PASSED
test_user.py::TestRegister::test_register_duplicate PASSED
test_product.py::...（12 条）PASSED
test_cart.py::...（6 条）PASSED
test_order.py::...（6 条）PASSED

======================= 52 passed in ~30s ========================

Schema 校验加了，数据驱动加了，用例数从 32 涨到 52，但执行时间从 89 秒降到约 30 秒。

# 只跑冒烟
pytest api\ -v -m smoke

# 只跑 Schema 校验
pytest api\ -v -m schema

# 只跑反向用例
pytest api\ -v -m negative

八、更新配置文件

8.1 pytest.ini

第九篇的 pytest.ini 缺少 schema marker，测试用例中用了 @pytest.mark.schema，--strict-markers 模式下会报错。更新 api/pytest.ini：

[pytest]
testpaths = test_cases
addopts = -v --tb=short --strict-markers

markers =
   smoke: 冒烟测试
   regression: 回归测试
   login: 登录接口
   register: 注册接口
   product: 商品接口
   category: 分类接口
   cart: 购物车接口
   order: 订单接口
   positive: 正向用例
   negative: 反向用例
   schema: Schema 校验用例
   p0: 最高优先级
   p1: 高优先级
   p2: 中优先级

log_cli = true
log_cli_level = INFO
log_cli_format = %(asctime)s | %(levelname)-8s | %(name)-12s | %(message)s
log_cli_date_format = %H:%M:%S

只新增了一行：schema: Schema 校验用例

8.2 requirements.txt

requests==2.32.3
pytest==8.3.4
allure-pytest==2.13.5
jsonschema==4.23.0
pyyaml==6.0.2

九、本篇新增和修改的文件

文件	操作	说明
`api/api_objects/base_api.py`	重写	增加重试、可配置超时、上下文管理器
`api/api_objects/user_api.py`	重写	新增 login_and_share 方法
`api/api_objects/cart_api.py`	重写	新增 clear_cart 方法
`api/api_objects/order_api.py`	重写	get_order 改为 order_no 参数
`api/common/response_validator.py`	新增	统一响应校验器
`api/common/schemas.py`	新增	JSON Schema 定义（6 个 Schema）
`api/common/data_reader.py`	新增	测试数据读取
`api/common/extractor.py`	新增	响应数据提取器
`api/common/assertions.py`	修改	追加 assert_schema 和 assert_code
`api/test_data/login_cases.json`	新增	登录测试数据
`api/test_data/product_cases.json`	新增	商品测试数据
`api/test_cases/test_user.py`	重写	数据驱动 + Schema 校验
`api/test_cases/test_product.py`	重写	数据驱动 + Schema 校验
`api/conftest.py`	重写	session 共享 + 自动清理
`api/pytest.ini`	修改	新增 schema marker

十、当前项目结构

api/
├── api_objects/
│   ├── __init__.py
│   ├── base_api.py         ← 增强版（重试 + 超时 + 统计）
│   ├── user_api.py         ← 完善版（login_and_share）
│   ├── product_api.py
│   ├── cart_api.py         ← 完善版（clear_cart）
│   └── order_api.py        ← 完善版（order_no 查询）
│
├── common/
│   ├── __init__.py
│   ├── logger.py
│   ├── assertions.py       ← 加了 assert_code + assert_schema
│   ├── response_validator.py  ← 新增
│   ├── schemas.py             ← 新增（6 个 Schema）
│   ├── extractor.py           ← 新增
│   └── data_reader.py         ← 新增
│
├── test_cases/
│   ├── __init__.py
│   ├── test_user.py        ← 数据驱动 + Schema
│   ├── test_product.py     ← 数据驱动 + Schema
│   ├── test_cart.py
│   └── test_order.py
│
├── test_data/
│   ├── login_cases.json    ← 新增
│   └── product_cases.json  ← 新增
│
├── conftest.py             ← session 共享 + 自动清理
├── pytest.ini
└── requirements.txt        ← 加了 jsonschema