Python API 自动化测试实战——requests + pytest 构建完整接口测试框架
2026/4/30大约 6 分钟
Python API 自动化测试实战——requests + pytest 构建完整接口测试框架
适读人群：Python 后端开发 / 接口测试工程师 | 阅读时长：约 17 分钟 | 核心价值：从零搭建一套可用于生产的 Python 接口自动化测试框架
一个 BUG 在生产跑了三个月才被发现
这是我们团队的真实案例。
产品有个"批量导入用户"接口，接受 CSV 文件，最多支持一次导入 1000 条。我们的代码里有一行：
MAX_IMPORT_COUNT = 1000有一天，运营同学导入了 1001 条，接口没有报错，返回了 200，但第 1001 条数据被静默丢弃了。
运营同学以为导入成功了，就没有再检查。三个月后，有个用户投诉"我明明上传了用户数据，为什么查不到"，才发现这个问题。
这个 BUG 如果有接口测试，测试"边界值 1001 条导入"这个场景，第一天就能发现。
框架整体架构
api_tests/
├── conftest.py # pytest 配置和 fixture
├── pytest.ini # pytest 配置文件
├── requirements.txt # 依赖
├── clients/
│ ├── base_client.py # HTTP 客户端基类
│ ├── user_client.py # 用户接口客户端
│ ├── order_client.py # 订单接口客户端
│ └── product_client.py # 商品接口客户端
├── models/
│ ├── request_models.py # 请求数据模型
│ └── response_models.py # 响应数据模型
├── tests/
│ ├── test_user_api.py
│ ├── test_order_api.py
│ └── test_product_api.py
└── utils/
├── data_factory.py # 测试数据工厂
└── assertions.py # 自定义断言核心：HTTP 客户端封装
# clients/base_client.py
import requests
import logging
from typing import Optional, Dict, Any
from dataclasses import dataclass
logger = logging.getLogger(__name__)
@dataclass
class APIResponse:
"""标准化的 API 响应对象"""
status_code: int
headers: Dict
body: Any
raw_response: requests.Response
@property
def is_success(self) -> bool:
return 200 <= self.status_code < 300
def get(self, key: str, default=None):
"""便捷访问响应 JSON 字段"""
if isinstance(self.body, dict):
return self.body.get(key, default)
return default
class BaseAPIClient:
"""
HTTP 客户端基类
封装了：认证、日志、错误处理、响应解析
"""
def __init__(self, base_url: str, token: Optional[str] = None):
self.base_url = base_url.rstrip("/")
self.session = requests.Session()
self.request_history = [] # 保存请求历史，供测试失败时排查
# 设置默认请求头
self.session.headers.update({
"Content-Type": "application/json",
"Accept": "application/json",
"User-Agent": "APITest/1.0",
})
if token:
self.session.headers["Authorization"] = f"Bearer {token}"
def _make_request(
self,
method: str,
path: str,
**kwargs
) -> APIResponse:
url = f"{self.base_url}{path}"
# 记录请求
logger.debug(f"→ {method.upper()} {url}")
if "json" in kwargs:
logger.debug(f" Body: {kwargs['json']}")
response = self.session.request(method, url, **kwargs)
# 解析响应体
try:
body = response.json()
except ValueError:
body = response.text
api_response = APIResponse(
status_code=response.status_code,
headers=dict(response.headers),
body=body,
raw_response=response,
)
# 记录响应
logger.debug(f"← {response.status_code} {url}")
logger.debug(f" Response: {body}")
# 保存到历史
self.request_history.append({
"method": method,
"url": url,
"request_body": kwargs.get("json"),
"response_status": response.status_code,
"response_body": body,
})
return api_response
def get(self, path: str, params=None, **kwargs) -> APIResponse:
return self._make_request("GET", path, params=params, **kwargs)
def post(self, path: str, json=None, **kwargs) -> APIResponse:
return self._make_request("POST", path, json=json, **kwargs)
def put(self, path: str, json=None, **kwargs) -> APIResponse:
return self._make_request("PUT", path, json=json, **kwargs)
def patch(self, path: str, json=None, **kwargs) -> APIResponse:
return self._make_request("PATCH", path, json=json, **kwargs)
def delete(self, path: str, **kwargs) -> APIResponse:
return self._make_request("DELETE", path, **kwargs)
def set_token(self, token: str):
self.session.headers["Authorization"] = f"Bearer {token}"
def clear_token(self):
self.session.headers.pop("Authorization", None)用户接口客户端
# clients/user_client.py
from .base_client import BaseAPIClient, APIResponse
from typing import Optional, List
class UserAPIClient(BaseAPIClient):
"""用户相关接口"""
def register(self, email: str, password: str, **kwargs) -> APIResponse:
return self.post("/api/users/register", json={
"email": email,
"password": password,
**kwargs
})
def login(self, email: str, password: str) -> APIResponse:
response = self.post("/api/auth/login", json={
"email": email,
"password": password,
})
if response.is_success and response.get("token"):
self.set_token(response.get("token"))
return response
def get_profile(self) -> APIResponse:
return self.get("/api/users/me")
def update_profile(self, **fields) -> APIResponse:
return self.patch("/api/users/me", json=fields)
def get_user_by_id(self, user_id: str) -> APIResponse:
return self.get(f"/api/users/{user_id}")
def list_users(self, page: int = 1, page_size: int = 20, **filters) -> APIResponse:
return self.get("/api/users", params={
"page": page,
"page_size": page_size,
**filters,
})
def delete_user(self, user_id: str) -> APIResponse:
return self.delete(f"/api/users/{user_id}")
def bulk_import(self, csv_data: bytes, filename: str = "users.csv") -> APIResponse:
return self._make_request("POST", "/api/users/bulk-import",
files={"file": (filename, csv_data, "text/csv")},
headers={"Content-Type": None} # 让 requests 自动设置 multipart content-type
)conftest.py 配置
# conftest.py
import pytest
import os
from clients.user_client import UserAPIClient
from clients.order_client import OrderAPIClient
BASE_URL = os.environ.get("API_BASE_URL", "http://localhost:8080")
ADMIN_TOKEN = os.environ.get("ADMIN_TOKEN", "dev-admin-token-here")
@pytest.fixture(scope="session")
def admin_client() -> UserAPIClient:
"""管理员权限的 HTTP 客户端（整个会话共享）"""
return UserAPIClient(base_url=BASE_URL, token=ADMIN_TOKEN)
@pytest.fixture
def api_client() -> UserAPIClient:
"""未登录的 HTTP 客户端（每个测试独立）"""
return UserAPIClient(base_url=BASE_URL)
@pytest.fixture
def logged_in_client(api_client) -> UserAPIClient:
"""已登录普通用户的 HTTP 客户端"""
# 创建测试用户
email = f"test_{uuid.uuid4().hex[:8]}@test.com"
password = "Test@123456"
api_client.register(email=email, password=password)
api_client.login(email=email, password=password)
yield api_client
# 清理测试用户
# (通过管理员接口删除)
@pytest.fixture
def test_user_data():
"""生成随机测试用户数据"""
return {
"email": f"test_{uuid.uuid4().hex[:8]}@test.com",
"password": "StrongPass@123",
"name": f"测试用户_{uuid.uuid4().hex[:4]}",
}测试用例
# tests/test_user_api.py
import pytest
import uuid
class TestUserRegistration:
"""用户注册接口测试"""
def test_register_success(self, api_client, test_user_data):
response = api_client.register(**test_user_data)
assert response.status_code == 201
assert response.get("id") is not None
assert response.get("email") == test_user_data["email"]
assert "password" not in response.body # 密码不应该返回
def test_register_duplicate_email(self, api_client, test_user_data):
api_client.register(**test_user_data) # 先注册一次
response = api_client.register(**test_user_data) # 再注册一次
assert response.status_code == 409
assert response.get("error_code") == "DUPLICATE_EMAIL"
@pytest.mark.parametrize("invalid_email", [
"",
"not-an-email",
"user@",
"@domain.com",
"user@domain",
])
def test_register_invalid_email(self, api_client, invalid_email):
response = api_client.register(
email=invalid_email,
password="StrongPass@123"
)
assert response.status_code == 422
assert "email" in response.get("errors", {})
@pytest.mark.parametrize("weak_password,expected_error", [
("123", "密码长度至少 8 位"),
("password", "密码必须包含数字"),
("12345678", "密码必须包含字母"),
("Password1", "密码必须包含特殊字符"),
])
def test_register_weak_password(self, api_client, weak_password, expected_error):
response = api_client.register(
email=f"test_{uuid.uuid4().hex[:8]}@test.com",
password=weak_password
)
assert response.status_code == 422
errors = response.get("errors", {})
assert any(expected_error in str(e) for e in errors.values())
class TestUserAuthentication:
"""用户认证接口测试"""
def test_login_success(self, api_client, test_user_data):
# 先注册
api_client.register(**test_user_data)
# 再登录
response = api_client.login(
email=test_user_data["email"],
password=test_user_data["password"]
)
assert response.status_code == 200
assert response.get("token") is not None
assert len(response.get("token")) > 20 # Token 有一定长度
def test_login_wrong_password(self, api_client, test_user_data):
api_client.register(**test_user_data)
response = api_client.login(
email=test_user_data["email"],
password="WrongPassword@123"
)
assert response.status_code == 401
assert response.get("error_code") == "INVALID_CREDENTIALS"
def test_get_profile_without_auth(self, api_client):
response = api_client.get_profile()
assert response.status_code == 401
def test_get_profile_with_auth(self, logged_in_client, test_user_data):
response = logged_in_client.get_profile()
assert response.status_code == 200
assert response.get("email") == test_user_data["email"]
class TestBulkImport:
"""批量导入测试——覆盖边界值场景"""
def test_import_1000_users(self, logged_in_client):
"""边界值：恰好 1000 条"""
csv_data = self._generate_csv(count=1000)
response = logged_in_client.bulk_import(csv_data)
assert response.status_code == 200
assert response.get("imported_count") == 1000
assert response.get("failed_count") == 0
def test_import_1001_users_should_fail(self, logged_in_client):
"""边界值：超过 1000 条应该报错"""
csv_data = self._generate_csv(count=1001)
response = logged_in_client.bulk_import(csv_data)
assert response.status_code == 422
assert response.get("error_code") == "IMPORT_LIMIT_EXCEEDED"
assert "1000" in response.get("message", "")
def _generate_csv(self, count: int) -> bytes:
lines = ["email,name"]
for i in range(count):
lines.append(f"import_test_{i}@test.com,用户{i}")
return "\n".join(lines).encode("utf-8")踩坑实录
坑一：requests.Session 在多线程测试中不安全
现象： 并行测试时，不同测试的 HTTP 请求互相干扰，出现 token 混乱。
原因： requests.Session 不是线程安全的，多个线程共享同一个 session 会导致 header（包括 token）互相覆盖。
解法： 每个测试函数使用独立的 session 实例（function scope 的 fixture）：
@pytest.fixture
def api_client() -> UserAPIClient:
# function scope，每个测试有独立的 Session
return UserAPIClient(base_url=BASE_URL)坑二：响应体不是 JSON 时报错
现象： 某些 API 在特殊情况下返回 HTML 错误页面而不是 JSON，response.json() 报错。
解法： 在 BaseAPIClient 中处理非 JSON 响应：
try:
body = response.json()
except (ValueError, requests.exceptions.JSONDecodeError):
body = response.text # 退回到纯文本坑三：测试顺序依赖导致间歇性失败
现象： 单独跑某个测试通过，全套跑时失败——前面的测试创建的用户影响了后续测试。
解法： 每个测试用带 UUID 的唯一邮箱，且用 try/finally 确保清理：
@pytest.fixture
def unique_user(api_client, admin_client):
email = f"test_{uuid.uuid4().hex[:8]}@test.com"
register_response = api_client.register(email=email, password="Test@123")
user_id = register_response.get("id")
yield {"email": email, "password": "Test@123", "id": user_id}
try:
admin_client.delete_user(user_id)
except Exception:
pass # 清理失败不影响测试结果小结
用 requests + pytest 构建接口测试框架的核心是分层设计：
- 客户端层（clients）：封装 HTTP 细节，提供业务方法
- 测试层（tests）：用业务语言描述测试场景，不关心 HTTP 细节
- 数据层（fixtures + conftest）：管理测试数据的创建和清理
这样的架构，API 路径改了只改客户端，测试逻辑不动；数据结构变了只改模型，测试逻辑不动。可维护性极高。