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

前言

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

一、增强 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 篇对比：

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 速度对比

五、数据驱动

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/
│ ├── __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

十一、今日成果

• 增强了 ApiClient：自动重试（5xx）、可配置超时、上下文管理器、请求统计
• 完善了 4 个业务 API 类：新增 login_and_share、clear_cart 等方法
• 实现了 ResponseValidator 统一响应校验器
• 实现了 JSON Schema 校验（6 个 Schema 定义）
• 实现了 Extractor 数据提取器
• 实现了 data_reader 测试数据读取
• 优化了执行速度：session 共享 + 自动清购物车
• 用数据驱动重写了用户和商品用例
• 项目累计 52 条用例，执行时间从 ~89 秒降到 ~30 秒

十二、下篇预告

11 – 接口自动化实战

下一篇编写完整的 CRUD 测试、链路测试（注册→登录→搜索→加购→下单→查单）、接口间数据关联、以及 Allure 报告集成。

接口自动化篇进度：10/13。