Pytest 的接口自动化封装实战
1. 什么是接口自动化封装
1.1 封装的概念
封装（Encapsulation） 是面向对象编程的核心概念之一，在接口自动化测试中，封装指的是将重复的代码逻辑、通用的功能模块进行抽象和封装，形成可复用的组件。
简单来说，封装就是把复杂的操作简化成简单的调用，把重复的代码提取成公共的方法。
1.2 为什么需要封装
在接口自动化测试中，如果不进行封装，我们可能会写出这样的代码：
# 不封装的代码示例 - 问题很多
def test_login():
import requests
url = "https://api.example.com/login"
headers = {
"Content-Type": "application/json",
"User-Agent": "Mozilla/5.0"
}
data = {
"username": "admin",
"password": "123456"
}
response = requests.post(url, json=data, headers=headers)
assert response.status_code == 200
result = response.json()
assert result["code"] == 0
assert result["message"] == "登录成功"
token = result["data"]["token"]
# 下一个接口测试又需要重复写类似的代码
url2 = "https://api.example.com/user/info"
headers2 = {
"Content-Type": "application/json",
"Authorization": f"Bearer {token}"
}
response2 = requests.get(url2, headers=headers2)
assert response2.status_code == 200
# ... 更多重复代码这种写法的问题：
- 代码重复：每个测试用例都要写请求代码、断言代码
- 难以维护：如果接口地址改变，需要修改很多地方
- 错误处理不统一：每个地方都要单独处理异常
- 日志记录不统一：每个地方都要单独记录日志
- 可读性差：测试用例中混杂了大量技术细节
封装后的代码：
# 封装后的代码示例 - 简洁清晰
def test_login(api_client):
"""测试登录接口"""
# 发送登录请求
response = api_client.login(username="admin", password="123456")
# 验证响应
response.assert_status_code(200)
response.assert_json_path("code", 0)
response.assert_json_path("message", "登录成功")
# 提取 token
token = response.extract("data.token")
# 使用 token 获取用户信息
user_info = api_client.get_user_info(token=token)
user_info.assert_status_code(200)封装的优点：
- 代码复用：一次封装，多处使用
- 易于维护：修改一处，全局生效
- 统一管理：错误处理、日志记录统一管理
- 提高可读性：测试用例更清晰，专注于业务逻辑
- 降低学习成本：新手只需要调用封装好的方法
1.3 封装的核心原则
在进行接口自动化封装时，需要遵循以下原则：
1.3.1 单一职责原则
每个封装类或方法只负责一个功能，不要在一个方法中做太多事情。
# ❌ 不好的封装 - 职责过多
class ApiClient:
def login_and_get_user_info(self, username, password):
# 登录
login_response = requests.post(...)
token = login_response.json()["data"]["token"]
# 获取用户信息
user_response = requests.get(..., headers={"Authorization": token})
return user_response.json()
# 问题：这个方法做了两件事，不够灵活
# ✅ 好的封装 - 职责单一
class ApiClient:
def login(self, username, password):
"""只负责登录"""
response = requests.post(...)
return response
def get_user_info(self, token):
"""只负责获取用户信息"""
response = requests.get(..., headers={"Authorization": token})
return response1.3.2 DRY 原则（Don’t Repeat Yourself）
不要重复代码，将重复的逻辑提取成公共方法。
# ❌ 不好的封装 - 代码重复
def test_login():
headers = {"Content-Type": "application/json"}
response = requests.post(url, json=data, headers=headers)
def test_register():
headers = {"Content-Type": "application/json"}
response = requests.post(url2, json=data2, headers=headers)
# ✅ 好的封装 - 消除重复
class ApiClient:
def _build_headers(self):
"""构建公共请求头"""
return {"Content-Type": "application/json"}
def post(self, url, data):
"""统一的 POST 请求方法"""
headers = self._build_headers()
return requests.post(url, json=data, headers=headers)1.3.3 开闭原则
对扩展开放，对修改关闭。封装应该易于扩展，而不需要修改原有代码。
# ✅ 好的封装 - 易于扩展
class ApiClient:
def __init__(self, base_url):
self.base_url = base_url
self.session = requests.Session()
def _request(self, method, endpoint, **kwargs):
"""统一的请求方法，易于扩展"""
url = f"{self.base_url}{endpoint}"
return self.session.request(method, url, **kwargs)
# 可以轻松添加新的请求方法，不需要修改 _request
def get(self, endpoint, **kwargs):
return self._request("GET", endpoint, **kwargs)
def post(self, endpoint, **kwargs):
return self._request("POST", endpoint, **kwargs)1.3.4 配置与代码分离
将配置信息（如 URL、超时时间等）与代码分离，便于不同环境使用。
# ✅ 好的封装 - 配置分离
# config.yaml
base_url: https://api.example.com
timeout: 30
# api_client.py
class ApiClient:
def __init__(self, config):
self.base_url = config["base_url"]
self.timeout = config["timeout"]2. 接口自动化封装的层次结构
2.1 封装层次概览
接口自动化封装通常分为以下几个层次：
┌─────────────────────────────────────┐
│ 测试用例层（Test Cases） │ ← 最上层，编写测试用例
├─────────────────────────────────────┤
│ 业务接口层（Business API） │ ← 封装业务接口（登录、注册等）
├─────────────────────────────────────┤
│ HTTP 请求层（HTTP Client） │ ← 封装 HTTP 请求（GET、POST等）
├─────────────────────────────────────┤
│ 工具层（Utils） │ ← 封装工具方法（断言、日志等）
├─────────────────────────────────────┤
│ 配置层（Config） │ ← 最底层，管理配置信息
└─────────────────────────────────────┘2.2 各层次详细说明
2.2.1 配置层（Config Layer）
作用：管理所有配置信息，包括环境配置、数据库配置、接口地址等。
示例：
# config/config.py
import os
import yaml
from pathlib import Path
class Config:
"""配置管理类"""
def __init__(self, env="test"):
"""
初始化配置
:param env: 环境名称（test、dev、prod）
"""
self.env = env
self.base_dir = Path(__file__).parent.parent
self.config_file = self.base_dir / "config" / f"{env}_config.yaml"
self._load_config()
def _load_config(self):
"""加载配置文件"""
with open(self.config_file, "r", encoding="utf-8") as f:
self.config = yaml.safe_load(f)
@property
def base_url(self):
"""获取基础 URL"""
return self.config["api"]["base_url"]
@property
def timeout(self):
"""获取超时时间"""
return self.config["api"]["timeout"]
@property
def db_config(self):
"""获取数据库配置"""
return self.config["database"]配置文件示例：
# config/test_config.yaml
api:
base_url: https://api.test.example.com
timeout: 30
database:
host: localhost
port: 3306
username: test_user
password: test_pass
database: test_db
logging:
level: DEBUG
file: logs/test.log2.2.2 工具层（Utils Layer）
作用：封装通用的工具方法，如断言、日志、数据解析等。
示例：
# utils/assertion.py
import json
import jsonpath
class Assertion:
"""断言工具类"""
@staticmethod
def assert_status_code(response, expected_code):
"""
断言状态码
:param response: 响应对象
:param expected_code: 期望的状态码
"""
actual_code = response.status_code
assert actual_code == expected_code,
f"状态码断言失败：期望 {expected_code}，实际 {actual_code}"
@staticmethod
def assert_json_path(response, json_path, expected_value):
"""
断言 JSON 路径的值
:param response: 响应对象
:param json_path: JSON 路径表达式
:param expected_value: 期望的值
"""
try:
data = response.json()
actual_value = jsonpath.jsonpath(data, json_path)
if not actual_value:
raise AssertionError(f"JSON 路径 {json_path} 不存在")
actual_value = actual_value[0]
assert actual_value == expected_value,
f"JSON 路径断言失败：路径 {json_path}，期望 {expected_value}，实际 {actual_value}"
except json.JSONDecodeError:
raise AssertionError(f"响应不是有效的 JSON 格式：{response.text}")
@staticmethod
def assert_contains(response, text):
"""
断言响应包含指定文本
:param response: 响应对象
:param text: 要查找的文本
"""
assert text in response.text,
f"响应中不包含文本 '{text}'：{response.text}"
# utils/logger.py
import logging
import os
from pathlib import Path
class Logger:
"""日志工具类"""
def __init__(self, name="test", level=logging.DEBUG):
"""
初始化日志器
:param name: 日志器名称
:param level: 日志级别
"""
self.logger = logging.getLogger(name)
self.logger.setLevel(level)
# 避免重复添加处理器
if not self.logger.handlers:
# 控制台处理器
console_handler = logging.StreamHandler()
console_handler.setLevel(level)
# 文件处理器
log_dir = Path(__file__).parent.parent / "logs"
log_dir.mkdir(exist_ok=True)
file_handler = logging.FileHandler(
log_dir / f"{name}.log",
encoding="utf-8"
)
file_handler.setLevel(level)
# 格式化器
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
console_handler.setFormatter(formatter)
file_handler.setFormatter(formatter)
self.logger.addHandler(console_handler)
self.logger.addHandler(file_handler)
def debug(self, message):
"""记录 DEBUG 级别日志"""
self.logger.debug(message)
def info(self, message):
"""记录 INFO 级别日志"""
self.logger.info(message)
def warning(self, message):
"""记录 WARNING 级别日志"""
self.logger.warning(message)
def error(self, message):
"""记录 ERROR 级别日志"""
self.logger.error(message)2.2.3 HTTP 请求层（HTTP Client Layer）
作用：封装底层的 HTTP 请求，提供统一的请求接口。
示例：
# clients/http_client.py
import requests
import json
from typing import Dict, Optional, Any
from utils.logger import Logger
class HttpClient:
"""HTTP 客户端封装类"""
def __init__(self, base_url: str, timeout: int = 30):
"""
初始化 HTTP 客户端
:param base_url: 基础 URL
:param timeout: 超时时间（秒）
"""
self.base_url = base_url.rstrip("/")
self.timeout = timeout
self.session = requests.Session()
self.logger = Logger("http_client")
# 设置默认请求头
self.session.headers.update({
"Content-Type": "application/json",
"User-Agent": "Python-Automation-Test/1.0"
})
def _build_url(self, endpoint: str) -> str:
"""
构建完整 URL
:param endpoint: 接口路径
:return: 完整 URL
"""
endpoint = endpoint.lstrip("/")
return f"{self.base_url}/{endpoint}"
def _log_request(self, method: str, url: str, **kwargs):
"""记录请求日志"""
self.logger.info(f"[请求] {method} {url}")
if "json" in kwargs:
self.logger.debug(f"[请求体] {json.dumps(kwargs['json'], ensure_ascii=False, indent=2)}")
if "params" in kwargs:
self.logger.debug(f"[请求参数] {kwargs['params']}")
if "headers" in kwargs:
self.logger.debug(f"[请求头] {kwargs['headers']}")
def _log_response(self, response: requests.Response):
"""记录响应日志"""
self.logger.info(f"[响应] 状态码: {response.status_code}")
try:
self.logger.debug(f"[响应体] {json.dumps(response.json(), ensure_ascii=False, indent=2)}")
except:
self.logger.debug(f"[响应体] {response.text[:500]}") # 只记录前500个字符
def request(
self,
method: str,
endpoint: str,
params: Optional[Dict] = None,
json_data: Optional[Dict] = None,
data: Optional[Dict] = None,
headers: Optional[Dict] = None,
**kwargs
) -> requests.Response:
"""
发送 HTTP 请求
:param method: 请求方法（GET、POST、PUT、DELETE等）
:param endpoint: 接口路径
:param params: URL 参数
:param json_data: JSON 格式的请求体
:param data: 表单格式的请求体
:param headers: 请求头
:param kwargs: 其他 requests 参数
:return: 响应对象
"""
url = self._build_url(endpoint)
# 合并请求头
request_headers = self.session.headers.copy()
if headers:
request_headers.update(headers)
# 记录请求日志
self._log_request(method, url, params=params, json=json_data, headers=request_headers)
try:
# 发送请求
response = self.session.request(
method=method,
url=url,
params=params,
json=json_data,
data=data,
headers=request_headers,
timeout=self.timeout,
**kwargs
)
# 记录响应日志
self._log_response(response)
return response
except requests.exceptions.Timeout:
self.logger.error(f"请求超时：{method} {url}")
raise
except requests.exceptions.ConnectionError:
self.logger.error(f"连接错误：{method} {url}")
raise
except Exception as e:
self.logger.error(f"请求异常：{method} {url}，错误：{str(e)}")
raise
def get(self, endpoint: str, params: Optional[Dict] = None, **kwargs) -> requests.Response:
"""发送 GET 请求"""
return self.request("GET", endpoint, params=params, **kwargs)
def post(self, endpoint: str, json_data: Optional[Dict] = None, **kwargs) -> requests.Response:
"""发送 POST 请求"""
return self.request("POST", endpoint, json_data=json_data, **kwargs)
def put(self, endpoint: str, json_data: Optional[Dict] = None, **kwargs) -> requests.Response:
"""发送 PUT 请求"""
return self.request("PUT", endpoint, json_data=json_data, **kwargs)
def delete(self, endpoint: str, **kwargs) -> requests.Response:
"""发送 DELETE 请求"""
return self.request("DELETE", endpoint, **kwargs)
def set_header(self, key: str, value: str):
"""设置请求头"""
self.session.headers[key] = value
def remove_header(self, key: str):
"""移除请求头"""
if key in self.session.headers:
del self.session.headers[key]2.2.4 业务接口层（Business API Layer）
作用：封装具体的业务接口，如登录、注册、查询用户等。
示例：
# api/user_api.py
from clients.http_client import HttpClient
from utils.response_handler import ResponseHandler
class UserAPI:
"""用户相关接口"""
def __init__(self, http_client: HttpClient):
"""
初始化用户 API
:param http_client: HTTP 客户端实例
"""
self.client = http_client
def login(self, username: str, password: str) -> ResponseHandler:
"""
用户登录
:param username: 用户名
:param password: 密码
:return: 响应处理器
"""
endpoint = "/api/user/login"
data = {
"username": username,
"password": password
}
response = self.client.post(endpoint, json_data=data)
return ResponseHandler(response)
def register(self, username: str, password: str, email: str) -> ResponseHandler:
"""
用户注册
:param username: 用户名
:param password: 密码
:param email: 邮箱
:return: 响应处理器
"""
endpoint = "/api/user/register"
data = {
"username": username,
"password": password,
"email": email
}
response = self.client.post(endpoint, json_data=data)
return ResponseHandler(response)
def get_user_info(self, token: str) -> ResponseHandler:
"""
获取用户信息
:param token: 认证 token
:return: 响应处理器
"""
endpoint = "/api/user/info"
headers = {
"Authorization": f"Bearer {token}"
}
response = self.client.get(endpoint, headers=headers)
return ResponseHandler(response)
def update_user_info(self, token: str, **kwargs) -> ResponseHandler:
"""
更新用户信息
:param token: 认证 token
:param kwargs: 要更新的用户信息字段
:return: 响应处理器
"""
endpoint = "/api/user/info"
headers = {
"Authorization": f"Bearer {token}"
}
response = self.client.put(endpoint, json_data=kwargs, headers=headers)
return ResponseHandler(response)2.2.5 响应处理器（Response Handler）
作用：封装响应对象的常用操作，如断言、数据提取等。
示例：
# utils/response_handler.py
import requests
import json
import jsonpath
from typing import Any, Optional
class ResponseHandler:
"""响应处理器"""
def __init__(self, response: requests.Response):
"""
初始化响应处理器
:param response: requests 响应对象
"""
self.response = response
self._json_data = None
@property
def status_code(self) -> int:
"""获取状态码"""
return self.response.status_code
@property
def text(self) -> str:
"""获取响应文本"""
return self.response.text
@property
def json(self) -> dict:
"""获取 JSON 数据（缓存）"""
if self._json_data is None:
try:
self._json_data = self.response.json()
except json.JSONDecodeError:
raise ValueError(f"响应不是有效的 JSON 格式：{self.response.text}")
return self._json_data
def assert_status_code(self, expected_code: int):
"""
断言状态码
:param expected_code: 期望的状态码
"""
actual_code = self.status_code
assert actual_code == expected_code,
f"状态码断言失败：期望 {expected_code}，实际 {actual_code}"
return self
def assert_json_path(self, json_path: str, expected_value: Any):
"""
断言 JSON 路径的值
:param json_path: JSON 路径表达式
:param expected_value: 期望的值
"""
actual_value = self.extract(json_path)
assert actual_value == expected_value,
f"JSON 路径断言失败：路径 {json_path}，期望 {expected_value}，实际 {actual_value}"
return self
def assert_contains(self, text: str):
"""
断言响应包含指定文本
:param text: 要查找的文本
"""
assert text in self.text,
f"响应中不包含文本 '{text}'：{self.text[:200]}"
return self
def extract(self, json_path: str) -> Any:
"""
提取 JSON 路径的值
:param json_path: JSON 路径表达式
:return: 提取的值
"""
try:
data = self.json
result = jsonpath.jsonpath(data, json_path)
if not result:
raise ValueError(f"JSON 路径 {json_path} 不存在")
return result[0]
except Exception as e:
raise ValueError(f"提取 JSON 路径失败：{json_path}，错误：{str(e)}")
def extract_all(self, json_path: str) -> list:
"""
提取 JSON 路径的所有匹配值
:param json_path: JSON 路径表达式
:return: 提取的值列表
"""
try:
data = self.json
result = jsonpath.jsonpath(data, json_path)
return result if result else []
except Exception as e:
raise ValueError(f"提取 JSON 路径失败：{json_path}，错误：{str(e)}")2.2.6 测试用例层（Test Cases Layer）
作用：编写具体的测试用例，调用封装好的接口。
示例：
# tests/test_user_api.py
import pytest
from config.config import Config
from clients.http_client import HttpClient
from api.user_api import UserAPI
@pytest.fixture(scope="module")
def config():
"""配置 fixture"""
return Config(env="test")
@pytest.fixture(scope="module")
def http_client(config):
"""HTTP 客户端 fixture"""
return HttpClient(
base_url=config.base_url,
timeout=config.timeout
)
@pytest.fixture(scope="module")
def user_api(http_client):
"""用户 API fixture"""
return UserAPI(http_client)
class TestUserAPI:
"""用户接口测试类"""
def test_login_success(self, user_api):
"""测试登录成功"""
# 发送登录请求
response = user_api.login(username="admin", password="123456")
# 断言
response.assert_status_code(200)
response.assert_json_path("code", 0)
response.assert_json_path("message", "登录成功")
# 提取 token
token = response.extract("data.token")
assert token is not None
def test_login_failed(self, user_api):
"""测试登录失败"""
response = user_api.login(username="wrong", password="wrong")
response.assert_status_code(200)
response.assert_json_path("code", 1001)
response.assert_json_path("message", "用户名或密码错误")
def test_get_user_info(self, user_api):
"""测试获取用户信息"""
# 先登录获取 token
login_response = user_api.login(username="admin", password="123456")
token = login_response.extract("data.token")
# 获取用户信息
user_info_response = user_api.get_user_info(token=token)
user_info_response.assert_status_code(200)
user_info_response.assert_json_path("code", 0)
# 验证用户信息
username = user_info_response.extract("data.username")
assert username == "admin"3. 完整的项目结构
3.1 项目目录结构
一个完整的接口自动化测试项目应该有以下目录结构：
project/
├── config/ # 配置目录
│ ├── __init__.py
│ ├── config.py # 配置管理类
│ ├── test_config.yaml # 测试环境配置
│ ├── dev_config.yaml # 开发环境配置
│ └── prod_config.yaml # 生产环境配置
│
├── clients/ # HTTP 客户端目录
│ ├── __init__.py
│ └── http_client.py # HTTP 客户端封装
│
├── api/ # 业务接口目录
│ ├── __init__.py
│ ├── user_api.py # 用户相关接口
│ ├── order_api.py # 订单相关接口
│ └── product_api.py # 商品相关接口
│
├── utils/ # 工具类目录
│ ├── __init__.py
│ ├── assertion.py # 断言工具
│ ├── logger.py # 日志工具
│ ├── response_handler.py # 响应处理器
│ └── data_helper.py # 数据辅助工具
│
├── tests/ # 测试用例目录
│ ├── __init__.py
│ ├── conftest.py # pytest 配置文件
│ ├── test_user_api.py # 用户接口测试
│ ├── test_order_api.py # 订单接口测试
│ └── test_product_api.py # 商品接口测试
│
├── data/ # 测试数据目录
│ ├── test_data.yaml # 测试数据
│ └── test_cases.yaml # 测试用例数据
│
├── logs/ # 日志目录（自动生成）
│ └── test.log
│
├── reports/ # 测试报告目录（自动生成）
│ └── report.html
│
├── requirements.txt # 依赖文件
└── README.md # 项目说明3.2 各文件详细说明
3.2.1 requirements.txt
# 测试框架
pytest==7.4.3
pytest-html==4.1.1
pytest-xdist==3.5.0
pytest-rerunfailures==12.0
pytest-timeout==2.2.0
# HTTP 请求
requests==2.31.0
# 数据处理
pyyaml==6.0.1
jsonpath==0.82
# 日志
loguru==0.7.2
# 其他工具
python-dotenv==1.0.03.2.2 conftest.py
# tests/conftest.py
import pytest
from config.config import Config
from clients.http_client import HttpClient
from api.user_api import UserAPI
from api.order_api import OrderAPI
@pytest.fixture(scope="session")
def config():
"""全局配置 fixture"""
env = pytest.config.getoption("--env", default="test")
return Config(env=env)
@pytest.fixture(scope="session")
def http_client(config):
"""全局 HTTP 客户端 fixture"""
return HttpClient(
base_url=config.base_url,
timeout=config.timeout
)
@pytest.fixture(scope="session")
def user_api(http_client):
"""用户 API fixture"""
return UserAPI(http_client)
@pytest.fixture(scope="session")
def order_api(http_client):
"""订单 API fixture"""
return OrderAPI(http_client)
def pytest_addoption(parser):
"""添加命令行参数"""
parser.addoption(
"--env",
action="store",
default="test",
help="测试环境：test、dev、prod"
)4. 实战案例：完整的登录流程测试
4.1 需求分析
假设我们要测试一个用户登录系统，需要测试以下场景：
- 正常登录：用户名和密码正确
- 密码错误：用户名正确，密码错误
- 用户名不存在：用户名不存在
- 登录后获取用户信息：登录成功后，使用 token 获取用户信息
- 登录后修改用户信息：登录成功后，修改用户信息
4.2 实现步骤
步骤 1：创建配置文件
# config/test_config.yaml
api:
base_url: https://api.test.example.com
timeout: 30
database:
host: localhost
port: 3306
username: test_user
password: test_pass
database: test_db
logging:
level: DEBUG
file: logs/test.log
test_data:
valid_user:
username: admin
password: 123456
invalid_user:
username: wrong_user
password: wrong_pass步骤 2：实现配置管理类
# config/config.py
import yaml
from pathlib import Path
from typing import Dict, Any
class Config:
"""配置管理类"""
def __init__(self, env: str = "test"):
"""
初始化配置
:param env: 环境名称
"""
self.env = env
self.base_dir = Path(__file__).parent.parent
self.config_file = self.base_dir / "config" / f"{env}_config.yaml"
self._config = self._load_config()
def _load_config(self) -> Dict[str, Any]:
"""加载配置文件"""
with open(self.config_file, "r", encoding="utf-8") as f:
return yaml.safe_load(f)
@property
def base_url(self) -> str:
"""获取基础 URL"""
return self._config["api"]["base_url"]
@property
def timeout(self) -> int:
"""获取超时时间"""
return self._config["api"]["timeout"]
@property
def test_data(self) -> Dict[str, Any]:
"""获取测试数据"""
return self._config.get("test_data", {})步骤 3：实现 HTTP 客户端
# clients/http_client.py
import requests
import json
from typing import Dict, Optional
from utils.logger import Logger
class HttpClient:
"""HTTP 客户端"""
def __init__(self, base_url: str, timeout: int = 30):
self.base_url = base_url.rstrip("/")
self.timeout = timeout
self.session = requests.Session()
self.logger = Logger("http_client")
# 设置默认请求头
self.session.headers.update({
"Content-Type": "application/json"
})
def _build_url(self, endpoint: str) -> str:
"""构建完整 URL"""
endpoint = endpoint.lstrip("/")
return f"{self.base_url}/{endpoint}"
def _log_request(self, method: str, url: str, **kwargs):
"""记录请求日志"""
self.logger.info(f"[请求] {method} {url}")
if "json_data" in kwargs:
self.logger.debug(f"[请求体] {json.dumps(kwargs['json_data'], ensure_ascii=False, indent=2)}")
def _log_response(self, response: requests.Response):
"""记录响应日志"""
self.logger.info(f"[响应] 状态码: {response.status_code}")
try:
self.logger.debug(f"[响应体] {json.dumps(response.json(), ensure_ascii=False, indent=2)}")
except:
self.logger.debug(f"[响应体] {response.text[:500]}")
def request(
self,
method: str,
endpoint: str,
json_data: Optional[Dict] = None,
headers: Optional[Dict] = None,
**kwargs
) -> requests.Response:
"""发送 HTTP 请求"""
url = self._build_url(endpoint)
request_headers = self.session.headers.copy()
if headers:
request_headers.update(headers)
self._log_request(method, url, json_data=json_data)
try:
response = self.session.request(
method=method,
url=url,
json=json_data,
headers=request_headers,
timeout=self.timeout,
**kwargs
)
self._log_response(response)
return response
except Exception as e:
self.logger.error(f"请求异常：{method} {url}，错误：{str(e)}")
raise
def post(self, endpoint: str, json_data: Optional[Dict] = None, **kwargs) -> requests.Response:
"""POST 请求"""
return self.request("POST", endpoint, json_data=json_data, **kwargs)
def get(self, endpoint: str, **kwargs) -> requests.Response:
"""GET 请求"""
return self.request("GET", endpoint, **kwargs)
def put(self, endpoint: str, json_data: Optional[Dict] = None, **kwargs) -> requests.Response:
"""PUT 请求"""
return self.request("PUT", endpoint, json_data=json_data, **kwargs)步骤 4：实现响应处理器
# utils/response_handler.py
import requests
import json
import jsonpath
from typing import Any
class ResponseHandler:
"""响应处理器"""
def __init__(self, response: requests.Response):
self.response = response
self._json_data = None
@property
def status_code(self) -> int:
return self.response.status_code
@property
def json(self) -> dict:
if self._json_data is None:
try:
self._json_data = self.response.json()
except json.JSONDecodeError:
raise ValueError(f"响应不是有效的 JSON：{self.response.text}")
return self._json_data
def assert_status_code(self, expected_code: int):
"""断言状态码"""
assert self.status_code == expected_code,
f"状态码断言失败：期望 {expected_code}，实际 {self.status_code}"
return self
def assert_json_path(self, json_path: str, expected_value: Any):
"""断言 JSON 路径"""
actual_value = self.extract(json_path)
assert actual_value == expected_value,
f"JSON 路径断言失败：{json_path}，期望 {expected_value}，实际 {actual_value}"
return self
def extract(self, json_path: str) -> Any:
"""提取 JSON 路径的值"""
data = self.json
result = jsonpath.jsonpath(data, json_path)
if not result:
raise ValueError(f"JSON 路径不存在：{json_path}")
return result[0]步骤 5：实现用户 API
# api/user_api.py
from clients.http_client import HttpClient
from utils.response_handler import ResponseHandler
class UserAPI:
"""用户相关接口"""
def __init__(self, http_client: HttpClient):
self.client = http_client
def login(self, username: str, password: str) -> ResponseHandler:
"""用户登录"""
endpoint = "/api/user/login"
data = {
"username": username,
"password": password
}
response = self.client.post(endpoint, json_data=data)
return ResponseHandler(response)
def get_user_info(self, token: str) -> ResponseHandler:
"""获取用户信息"""
endpoint = "/api/user/info"
headers = {"Authorization": f"Bearer {token}"}
response = self.client.get(endpoint, headers=headers)
return ResponseHandler(response)
def update_user_info(self, token: str, **kwargs) -> ResponseHandler:
"""更新用户信息"""
endpoint = "/api/user/info"
headers = {"Authorization": f"Bearer {token}"}
response = self.client.put(endpoint, json_data=kwargs, headers=headers)
return ResponseHandler(response)步骤 6：编写测试用例
# tests/test_user_login.py
import pytest
from config.config import Config
from clients.http_client import HttpClient
from api.user_api import UserAPI
@pytest.fixture(scope="module")
def config():
"""配置 fixture"""
return Config(env="test")
@pytest.fixture(scope="module")
def http_client(config):
"""HTTP 客户端 fixture"""
return HttpClient(
base_url=config.base_url,
timeout=config.timeout
)
@pytest.fixture(scope="module")
def user_api(http_client):
"""用户 API fixture"""
return UserAPI(http_client)
@pytest.fixture(scope="module")
def test_data(config):
"""测试数据 fixture"""
return config.test_data
class TestUserLogin:
"""用户登录测试类"""
def test_login_success(self, user_api, test_data):
"""测试登录成功"""
# 获取测试数据
valid_user = test_data["valid_user"]
# 发送登录请求
response = user_api.login(
username=valid_user["username"],
password=valid_user["password"]
)
# 断言响应
response.assert_status_code(200)
response.assert_json_path("code", 0)
response.assert_json_path("message", "登录成功")
# 提取 token
token = response.extract("data.token")
assert token is not None
assert len(token) > 0
def test_login_password_error(self, user_api, test_data):
"""测试密码错误"""
valid_user = test_data["valid_user"]
response = user_api.login(
username=valid_user["username"],
password="wrong_password"
)
response.assert_status_code(200)
response.assert_json_path("code", 1001)
response.assert_json_path("message", "密码错误")
def test_login_username_not_exist(self, user_api):
"""测试用户名不存在"""
response = user_api.login(
username="not_exist_user",
password="123456"
)
response.assert_status_code(200)
response.assert_json_path("code", 1002)
response.assert_json_path("message", "用户名不存在")
def test_login_and_get_user_info(self, user_api, test_data):
"""测试登录后获取用户信息"""
# 先登录
valid_user = test_data["valid_user"]
login_response = user_api.login(
username=valid_user["username"],
password=valid_user["password"]
)
token = login_response.extract("data.token")
# 获取用户信息
user_info_response = user_api.get_user_info(token=token)
user_info_response.assert_status_code(200)
user_info_response.assert_json_path("code", 0)
# 验证用户信息
username = user_info_response.extract("data.username")
assert username == valid_user["username"]
def test_login_and_update_user_info(self, user_api, test_data):
"""测试登录后更新用户信息"""
# 先登录
valid_user = test_data["valid_user"]
login_response = user_api.login(
username=valid_user["username"],
password=valid_user["password"]
)
token = login_response.extract("data.token")
# 更新用户信息
update_data = {
"nickname": "新昵称",
"email": "newemail@example.com"
}
update_response = user_api.update_user_info(token=token, **update_data)
update_response.assert_status_code(200)
update_response.assert_json_path("code", 0)
update_response.assert_json_path("message", "更新成功")
# 验证更新后的信息
user_info_response = user_api.get_user_info(token=token)
assert user_info_response.extract("data.nickname") == update_data["nickname"]
assert user_info_response.extract("data.email") == update_data["email"]5. 高级封装技巧
5.1 会话管理（Session Management）
在接口测试中，很多接口需要先登录获取 token，然后在后续请求中使用。我们可以封装一个会话管理器来统一管理。
# utils/session_manager.py
from typing import Optional
from clients.http_client import HttpClient
from api.user_api import UserAPI
class SessionManager:
"""会话管理器"""
def __init__(self, http_client: HttpClient, user_api: UserAPI):
"""
初始化会话管理器
:param http_client: HTTP 客户端
:param user_api: 用户 API
"""
self.http_client = http_client
self.user_api = user_api
self._token: Optional[str] = None
self._user_info: Optional[dict] = None
def login(self, username: str, password: str):
"""
登录并保存 token
:param username: 用户名
:param password: 密码
"""
response = self.user_api.login(username=username, password=password)
response.assert_status_code(200)
response.assert_json_path("code", 0)
self._token = response.extract("data.token")
# 自动设置 Authorization 头
self.http_client.set_header("Authorization", f"Bearer {self._token}")
# 获取并保存用户信息
user_info_response = self.user_api.get_user_info(token=self._token)
self._user_info = user_info_response.extract("data")
return response
@property
def token(self) -> Optional[str]:
"""获取当前 token"""
return self._token
@property
def user_info(self) -> Optional[dict]:
"""获取当前用户信息"""
return self._user_info
def logout(self):
"""登出，清除 token"""
if self._token:
self.http_client.remove_header("Authorization")
self._token = None
self._user_info = None
def is_logged_in(self) -> bool:
"""检查是否已登录"""
return self._token is not None使用示例：
# tests/test_with_session.py
import pytest
from utils.session_manager import SessionManager
@pytest.fixture(scope="function")
def session(user_api, http_client):
"""会话 fixture"""
session = SessionManager(http_client, user_api)
yield session
session.logout() # 测试结束后清理
def test_with_session(session):
"""使用会话管理器的测试"""
# 登录
session.login(username="admin", password="123456")
# 后续请求自动携带 token
# 不需要手动传递 token
assert session.is_logged_in()
assert session.token is not None
assert session.user_info is not None5.2 数据驱动测试封装
将测试数据与测试逻辑分离，使用 YAML 或 JSON 文件存储测试数据。
# utils/data_driver.py
import yaml
from pathlib import Path
from typing import List, Dict, Any
class DataDriver:
"""数据驱动工具"""
@staticmethod
def load_yaml(file_path: str) -> List[Dict[str, Any]]:
"""
从 YAML 文件加载测试数据
:param file_path: YAML 文件路径
:return: 测试数据列表
"""
file_path = Path(file_path)
if not file_path.exists():
raise FileNotFoundError(f"测试数据文件不存在：{file_path}")
with open(file_path, "r", encoding="utf-8") as f:
data = yaml.safe_load(f)
# 如果数据是字典，转换为列表
if isinstance(data, dict):
return [data]
return data if isinstance(data, list) else []
@staticmethod
def load_test_cases(file_path: str) -> List[Dict[str, Any]]:
"""
加载测试用例数据
:param file_path: 测试用例文件路径
:return: 测试用例列表
"""
data = DataDriver.load_yaml(file_path)
return data测试数据文件：
# data/test_login_cases.yaml
- name: 登录成功
username: admin
password: 123456
expected:
status_code: 200
code: 0
message: 登录成功
- name: 密码错误
username: admin
password: wrong_password
expected:
status_code: 200
code: 1001
message: 密码错误
- name: 用户名不存在
username: not_exist_user
password: 123456
expected:
status_code: 200
code: 1002
message: 用户名不存在使用数据驱动的测试用例：
# tests/test_login_data_driven.py
import pytest
from utils.data_driver import DataDriver
from pathlib import Path
@pytest.fixture(scope="module")
def test_cases():
"""加载测试用例数据"""
data_file = Path(__file__).parent.parent / "data" / "test_login_cases.yaml"
return DataDriver.load_test_cases(str(data_file))
@pytest.mark.parametrize("case", [
pytest.param(case, id=case["name"])
for case in DataDriver.load_test_cases(
str(Path(__file__).parent.parent / "data" / "test_login_cases.yaml")
)
])
def test_login_data_driven(user_api, case):
"""数据驱动的登录测试"""
# 发送请求
response = user_api.login(
username=case["username"],
password=case["password"]
)
# 断言
expected = case["expected"]
response.assert_status_code(expected["status_code"])
response.assert_json_path("code", expected["code"])
response.assert_json_path("message", expected["message"])5.3 重试机制封装
对于不稳定的接口，可以封装重试机制。
# utils/retry.py
import time
from functools import wraps
from typing import Callable, Any
from utils.logger import Logger
class Retry:
"""重试装饰器"""
def __init__(self, max_attempts: int = 3, delay: float = 1.0, backoff: float = 2.0):
"""
初始化重试配置
:param max_attempts: 最大重试次数
:param delay: 初始延迟时间（秒）
:param backoff: 延迟时间倍数
"""
self.max_attempts = max_attempts
self.delay = delay
self.backoff = backoff
self.logger = Logger("retry")
def __call__(self, func: Callable) -> Callable:
"""装饰器实现"""
@wraps(func)
def wrapper(*args, **kwargs) -> Any:
attempt = 0
current_delay = self.delay
while attempt < self.max_attempts:
try:
return func(*args, **kwargs)
except Exception as e:
attempt += 1
if attempt >= self.max_attempts:
self.logger.error(f"重试 {self.max_attempts} 次后仍然失败：{str(e)}")
raise
self.logger.warning(
f"第 {attempt} 次尝试失败：{str(e)}，"
f"{current_delay} 秒后重试..."
)
time.sleep(current_delay)
current_delay *= self.backoff
return None
return wrapper
# 使用示例
class UserAPI:
@Retry(max_attempts=3, delay=1.0)
def login(self, username: str, password: str) -> ResponseHandler:
"""登录接口（带重试）"""
# ... 实现代码
pass5.4 接口依赖管理
在测试中，很多接口之间有依赖关系，我们可以封装一个依赖管理器。
# utils/dependency_manager.py
from typing import Dict, Any, Optional
from utils.logger import Logger
class DependencyManager:
"""依赖管理器"""
def __init__(self):
"""初始化依赖管理器"""
self._dependencies: Dict[str, Any] = {}
self.logger = Logger("dependency")
def set(self, key: str, value: Any):
"""
设置依赖值
:param key: 依赖键
:param value: 依赖值
"""
self._dependencies[key] = value
self.logger.debug(f"设置依赖：{key} = {value}")
def get(self, key: str, default: Any = None) -> Any:
"""
获取依赖值
:param key: 依赖键
:param default: 默认值
:return: 依赖值
"""
value = self._dependencies.get(key, default)
if value is None and default is None:
self.logger.warning(f"依赖不存在：{key}")
return value
def has(self, key: str) -> bool:
"""
检查依赖是否存在
:param key: 依赖键
:return: 是否存在
"""
return key in self._dependencies
def clear(self):
"""清除所有依赖"""
self._dependencies.clear()
self.logger.debug("清除所有依赖")
# 使用示例
@pytest.fixture(scope="function")
def dependency_manager():
"""依赖管理器 fixture"""
manager = DependencyManager()
yield manager
manager.clear()
def test_with_dependency(user_api, dependency_manager):
"""使用依赖管理器的测试"""
# 登录并保存 token
response = user_api.login(username="admin", password="123456")
token = response.extract("data.token")
dependency_manager.set("token", token)
# 后续测试可以使用保存的 token
saved_token = dependency_manager.get("token")
user_info = user_api.get_user_info(token=saved_token)
# ...5.5 接口 Mock 封装
在测试中，有时候需要 Mock 某些接口的响应，我们可以封装一个 Mock 工具。
# utils/mock_helper.py
from unittest.mock import Mock, patch
from typing import Dict, Any, Optional
import json
class MockHelper:
"""Mock 辅助工具"""
@staticmethod
def mock_response(
status_code: int = 200,
json_data: Optional[Dict[str, Any]] = None,
text: Optional[str] = None,
headers: Optional[Dict[str, str]] = None
) -> Mock:
"""
创建 Mock 响应对象
:param status_code: 状态码
:param json_data: JSON 数据
:param text: 文本数据
:param headers: 响应头
:return: Mock 响应对象
"""
mock_response = Mock()
mock_response.status_code = status_code
mock_response.headers = headers or {}
if json_data:
mock_response.json.return_value = json_data
mock_response.text = json.dumps(json_data, ensure_ascii=False)
elif text:
mock_response.text = text
else:
mock_response.text = ""
mock_response.json.return_value = {}
return mock_response
@staticmethod
def patch_request(return_value: Mock):
"""
创建请求 Mock 装饰器
:param return_value: Mock 响应对象
:return: 装饰器
"""
def decorator(func):
def wrapper(*args, **kwargs):
with patch('requests.Session.request', return_value=return_value):
return func(*args, **kwargs)
return wrapper
return decorator
# 使用示例
def test_with_mock(user_api):
"""使用 Mock 的测试"""
# 创建 Mock 响应
mock_response = MockHelper.mock_response(
status_code=200,
json_data={
"code": 0,
"message": "登录成功",
"data": {
"token": "mock_token_12345"
}
}
)
# Mock 请求
with patch('requests.Session.request', return_value=mock_response):
response = user_api.login(username="admin", password="123456")
assert response.extract("data.token") == "mock_token_12345"6. 最佳实践和注意事项
6.1 命名规范
6.1.1 类命名
- 配置类：使用
Config后缀，如Config、DatabaseConfig - 客户端类：使用
Client后缀，如HttpClient、DatabaseClient - API 类：使用
API后缀，如UserAPI、OrderAPI - 工具类：使用描述性名称，如
Logger、Assertion、DataDriver
6.1.2 方法命名
- 请求方法：使用动词，如
get、post、put、delete - 业务方法：使用业务动词，如
login、register、get_user_info - 断言方法：使用
assert_前缀，如assert_status_code、assert_json_path - 提取方法：使用
extract、get等动词，如extract、get_token
6.1.3 变量命名
- 配置变量：使用小写字母和下划线，如
base_url、timeout - 响应变量：使用描述性名称，如
login_response、user_info_response - 测试数据：使用
test_data、case等
6.2 错误处理
6.2.1 统一异常处理
# utils/exceptions.py
class APIException(Exception):
"""API 异常基类"""
pass
class HTTPException(APIException):
"""HTTP 异常"""
def __init__(self, status_code: int, message: str):
self.status_code = status_code
self.message = message
super().__init__(f"HTTP {status_code}: {message}")
class ValidationException(APIException):
"""验证异常"""
pass
# 在 HTTP 客户端中使用
class HttpClient:
def request(self, method: str, endpoint: str, **kwargs):
try:
response = self.session.request(...)
response.raise_for_status() # 自动抛出 HTTP 错误
return response
except requests.exceptions.HTTPError as e:
raise HTTPException(e.response.status_code, str(e))
except requests.exceptions.RequestException as e:
raise APIException(f"请求失败：{str(e)}")6.2.2 优雅的错误处理
# 在测试用例中使用 try-except
def test_login_with_error_handling(user_api):
"""带错误处理的登录测试"""
try:
response = user_api.login(username="admin", password="123456")
response.assert_status_code(200)
except HTTPException as e:
pytest.fail(f"HTTP 错误：{e.status_code} - {e.message}")
except ValidationException as e:
pytest.fail(f"验证错误：{str(e)}")
except Exception as e:
pytest.fail(f"未知错误：{str(e)}")6.3 日志记录最佳实践
6.3.1 日志级别使用
# DEBUG：详细的调试信息
logger.debug(f"请求参数：{params}")
# INFO：重要的业务流程信息
logger.info(f"开始执行登录测试")
# WARNING：警告信息，不影响测试
logger.warning(f"响应时间较长：{response_time}ms")
# ERROR：错误信息
logger.error(f"请求失败：{str(e)}")6.3.2 日志格式
# 统一的日志格式
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s'
)
# 输出示例：
# 2024-01-27 10:30:45 - http_client - INFO - [http_client.py:45] - [请求] POST /api/user/login6.4 测试数据管理
6.4.1 测试数据分离
# ✅ 好的做法：测试数据在配置文件中
# config/test_config.yaml
test_data:
valid_user:
username: admin
password: 123456
# ❌ 不好的做法：测试数据硬编码在代码中
def test_login():
response = user_api.login(username="admin", password="123456") # 硬编码6.4.2 测试数据清理
@pytest.fixture(scope="function", autouse=True)
def cleanup_test_data():
"""自动清理测试数据"""
yield
# 测试结束后清理
# 例如：删除测试创建的用户、订单等
pass6.5 性能优化
6.5.1 使用 Session
# ✅ 好的做法：使用 Session 复用连接
class HttpClient:
def __init__(self):
self.session = requests.Session() # 复用连接
# ❌ 不好的做法：每次请求都创建新连接
def request(self):
response = requests.get(url) # 每次都创建新连接6.5.2 合理设置超时
# 根据接口特点设置合理的超时时间
class HttpClient:
def __init__(self, timeout=30): # 默认 30 秒
self.timeout = timeout
def quick_request(self, endpoint):
# 快速接口使用短超时
return self.request(endpoint, timeout=5)
def slow_request(self, endpoint):
# 慢速接口使用长超时
return self.request(endpoint, timeout=60)6.6 可维护性
6.6.1 代码注释
class UserAPI:
def login(self, username: str, password: str) -> ResponseHandler:
"""
用户登录接口
:param username: 用户名
:param password: 密码
:return: 响应处理器对象
:raises HTTPException: 当 HTTP 请求失败时
:raises ValidationException: 当响应验证失败时
示例：
>>> api = UserAPI(http_client)
>>> response = api.login("admin", "123456")
>>> response.assert_status_code(200)
"""
# 实现代码
pass6.6.2 类型提示
from typing import Dict, Optional, List
def get_user_info(
self,
token: str,
fields: Optional[List[str]] = None
) -> ResponseHandler:
"""
获取用户信息
:param token: 认证 token
:param fields: 要获取的字段列表，None 表示获取所有字段
:return: 响应处理器
"""
pass7. 完整实战项目示例
7.1 项目结构
api_test_framework/
├── config/
│ ├── __init__.py
│ ├── config.py
│ └── test_config.yaml
│
├── clients/
│ ├── __init__.py
│ └── http_client.py
│
├── api/
│ ├── __init__.py
│ ├── base_api.py
│ ├── user_api.py
│ └── order_api.py
│
├── utils/
│ ├── __init__.py
│ ├── logger.py
│ ├── assertion.py
│ ├── response_handler.py
│ ├── session_manager.py
│ └── data_driver.py
│
├── tests/
│ ├── __init__.py
│ ├── conftest.py
│ ├── test_user_api.py
│ └── test_order_api.py
│
├── data/
│ ├── test_users.yaml
│ └── test_orders.yaml
│
├── logs/
│ └── .gitkeep
│
├── requirements.txt
└── README.md7.2 核心代码实现
7.2.1 基础 API 类
# api/base_api.py
from clients.http_client import HttpClient
from utils.response_handler import ResponseHandler
class BaseAPI:
"""API 基类"""
def __init__(self, http_client: HttpClient):
"""
初始化 API
:param http_client: HTTP 客户端实例
"""
self.client = http_client
def _get(self, endpoint: str, **kwargs) -> ResponseHandler:
"""GET 请求"""
response = self.client.get(endpoint, **kwargs)
return ResponseHandler(response)
def _post(self, endpoint: str, json_data: dict = None, **kwargs) -> ResponseHandler:
"""POST 请求"""
response = self.client.post(endpoint, json_data=json_data, **kwargs)
return ResponseHandler(response)
def _put(self, endpoint: str, json_data: dict = None, **kwargs) -> ResponseHandler:
"""PUT 请求"""
response = self.client.put(endpoint, json_data=json_data, **kwargs)
return ResponseHandler(response)
def _delete(self, endpoint: str, **kwargs) -> ResponseHandler:
"""DELETE 请求"""
response = self.client.delete(endpoint, **kwargs)
return ResponseHandler(response)7.2.2 用户 API 实现
# api/user_api.py
from api.base_api import BaseAPI
from utils.response_handler import ResponseHandler
class UserAPI(BaseAPI):
"""用户相关接口"""
def login(self, username: str, password: str) -> ResponseHandler:
"""用户登录"""
endpoint = "/api/user/login"
data = {
"username": username,
"password": password
}
return self._post(endpoint, json_data=data)
def register(self, username: str, password: str, email: str) -> ResponseHandler:
"""用户注册"""
endpoint = "/api/user/register"
data = {
"username": username,
"password": password,
"email": email
}
return self._post(endpoint, json_data=data)
def get_user_info(self, token: str) -> ResponseHandler:
"""获取用户信息"""
endpoint = "/api/user/info"
headers = {"Authorization": f"Bearer {token}"}
return self._get(endpoint, headers=headers)
def update_user_info(self, token: str, **kwargs) -> ResponseHandler:
"""更新用户信息"""
endpoint = "/api/user/info"
headers = {"Authorization": f"Bearer {token}"}
return self._put(endpoint, json_data=kwargs, headers=headers)
def logout(self, token: str) -> ResponseHandler:
"""用户登出"""
endpoint = "/api/user/logout"
headers = {"Authorization": f"Bearer {token}"}
return self._post(endpoint, headers=headers)7.2.3 订单 API 实现
# api/order_api.py
from api.base_api import BaseAPI
from utils.response_handler import ResponseHandler
class OrderAPI(BaseAPI):
"""订单相关接口"""
def create_order(self, token: str, product_id: int, quantity: int) -> ResponseHandler:
"""创建订单"""
endpoint = "/api/order/create"
headers = {"Authorization": f"Bearer {token}"}
data = {
"product_id": product_id,
"quantity": quantity
}
return self._post(endpoint, json_data=data, headers=headers)
def get_order_list(self, token: str, page: int = 1, page_size: int = 10) -> ResponseHandler:
"""获取订单列表"""
endpoint = "/api/order/list"
headers = {"Authorization": f"Bearer {token}"}
params = {
"page": page,
"page_size": page_size
}
return self._get(endpoint, params=params, headers=headers)
def get_order_detail(self, token: str, order_id: int) -> ResponseHandler:
"""获取订单详情"""
endpoint = f"/api/order/{order_id}"
headers = {"Authorization": f"Bearer {token}"}
return self._get(endpoint, headers=headers)
def cancel_order(self, token: str, order_id: int) -> ResponseHandler:
"""取消订单"""
endpoint = f"/api/order/{order_id}/cancel"
headers = {"Authorization": f"Bearer {token}"}
return self._post(endpoint, headers=headers)7.2.4 conftest.py 配置
# tests/conftest.py
import pytest
from config.config import Config
from clients.http_client import HttpClient
from api.user_api import UserAPI
from api.order_api import OrderAPI
def pytest_addoption(parser):
"""添加命令行参数"""
parser.addoption(
"--env",
action="store",
default="test",
help="测试环境：test、dev、prod"
)
@pytest.fixture(scope="session")
def config(request):
"""全局配置 fixture"""
env = request.config.getoption("--env")
return Config(env=env)
@pytest.fixture(scope="session")
def http_client(config):
"""全局 HTTP 客户端 fixture"""
return HttpClient(
base_url=config.base_url,
timeout=config.timeout
)
@pytest.fixture(scope="session")
def user_api(http_client):
"""用户 API fixture"""
return UserAPI(http_client)
@pytest.fixture(scope="session")
def order_api(http_client):
"""订单 API fixture"""
return OrderAPI(http_client)
@pytest.fixture(scope="function")
def login_token(user_api, config):
"""登录并返回 token fixture"""
test_data = config.test_data
valid_user = test_data["valid_user"]
response = user_api.login(
username=valid_user["username"],
password=valid_user["password"]
)
response.assert_status_code(200)
token = response.extract("data.token")
return token7.2.5 测试用例示例
# tests/test_user_api.py
import pytest
from api.user_api import UserAPI
class TestUserAPI:
"""用户接口测试"""
def test_login_success(self, user_api, config):
"""测试登录成功"""
test_data = config.test_data
valid_user = test_data["valid_user"]
response = user_api.login(
username=valid_user["username"],
password=valid_user["password"]
)
response.assert_status_code(200)
response.assert_json_path("code", 0)
response.assert_json_path("message", "登录成功")
token = response.extract("data.token")
assert token is not None
def test_get_user_info(self, user_api, login_token):
"""测试获取用户信息"""
response = user_api.get_user_info(token=login_token)
response.assert_status_code(200)
response.assert_json_path("code", 0)
username = response.extract("data.username")
assert username is not None
# tests/test_order_api.py
import pytest
from api.order_api import OrderAPI
class TestOrderAPI:
"""订单接口测试"""
def test_create_order(self, order_api, login_token):
"""测试创建订单"""
response = order_api.create_order(
token=login_token,
product_id=1,
quantity=2
)
response.assert_status_code(200)
response.assert_json_path("code", 0)
order_id = response.extract("data.order_id")
assert order_id is not None
def test_get_order_list(self, order_api, login_token):
"""测试获取订单列表"""
response = order_api.get_order_list(
token=login_token,
page=1,
page_size=10
)
response.assert_status_code(200)
response.assert_json_path("code", 0)
orders = response.extract("data.orders")
assert isinstance(orders, list)7.3 运行测试
7.3.1 基本运行
# 运行所有测试
pytest
# 运行指定测试文件
pytest tests/test_user_api.py
# 运行指定测试类
pytest tests/test_user_api.py::TestUserAPI
# 运行指定测试方法
pytest tests/test_user_api.py::TestUserAPI::test_login_success7.3.2 带参数运行
# 指定测试环境
pytest --env=test
# 生成 HTML 报告
pytest --html=reports/report.html
# 并行运行测试
pytest -n auto
# 显示详细输出
pytest -v -s8. 常见问题和解决方案
8.1 问题：如何处理接口依赖？
问题描述：测试订单接口需要先登录获取 token，如何处理这种依赖关系？
解决方案 1：使用 fixture
@pytest.fixture(scope="function")
def login_token(user_api, config):
"""登录并返回 token"""
test_data = config.test_data
response = user_api.login(
username=test_data["valid_user"]["username"],
password=test_data["valid_user"]["password"]
)
return response.extract("data.token")
def test_order(order_api, login_token):
"""使用 login_token fixture"""
response = order_api.create_order(token=login_token, ...)解决方案 2：使用会话管理器
@pytest.fixture(scope="function")
def session(user_api, http_client):
"""会话管理器"""
session = SessionManager(http_client, user_api)
session.login(username="admin", password="123456")
yield session
session.logout()
def test_order(order_api, session):
"""使用会话管理器"""
# session.token 自动可用
response = order_api.create_order(token=session.token, ...)8.2 问题：如何管理不同环境的配置？
问题描述：测试环境、开发环境、生产环境的配置不同，如何管理？
解决方案：使用配置文件 + 环境变量
# config/config.py
import os
import yaml
class Config:
def __init__(self, env=None):
# 优先使用环境变量，其次使用参数
self.env = env or os.getenv("TEST_ENV", "test")
self._load_config()
def _load_config(self):
config_file = f"config/{self.env}_config.yaml"
with open(config_file, "r") as f:
self.config = yaml.safe_load(f)
# 环境变量可以覆盖配置文件
self.config["api"]["base_url"] = os.getenv(
"API_BASE_URL",
self.config["api"]["base_url"]
)使用方式：
# 方式 1：命令行参数
pytest --env=test
# 方式 2：环境变量
export TEST_ENV=test
pytest
# 方式 3：环境变量覆盖配置
export API_BASE_URL=https://api.custom.com
pytest --env=test8.3 问题：如何处理接口返回的数据格式不一致？
问题描述：有些接口返回 {"code": 0, "data": {...}}，有些返回 {"status": "success", "result": {...}}，如何处理？
解决方案：使用适配器模式
# utils/response_adapter.py
class ResponseAdapter:
"""响应适配器"""
@staticmethod
def normalize(response_data: dict) -> dict:
"""
标准化响应数据格式
:param response_data: 原始响应数据
:return: 标准化后的数据
"""
# 统一转换为标准格式
normalized = {
"code": response_data.get("code") or response_data.get("status_code", 0),
"message": response_data.get("message") or response_data.get("msg", ""),
"data": response_data.get("data") or response_data.get("result", {})
}
return normalized
# 在 ResponseHandler 中使用
class ResponseHandler:
def __init__(self, response):
self.response = response
self._normalized_data = None
@property
def normalized_json(self):
"""获取标准化后的 JSON"""
if self._normalized_data is None:
raw_data = self.json
self._normalized_data = ResponseAdapter.normalize(raw_data)
return self._normalized_data
def assert_code(self, expected_code):
"""断言 code（适配不同格式）"""
normalized = self.normalized_json
assert normalized["code"] == expected_code
return self8.4 问题：如何实现接口的批量测试？
问题描述：需要对多个接口进行批量测试，如何实现？
解决方案：使用数据驱动 + 参数化
# data/batch_test_cases.yaml
- name: 用户登录
api: user_api
method: login
params:
username: admin
password: 123456
expected:
status_code: 200
code: 0
- name: 创建订单
api: order_api
method: create_order
params:
product_id: 1
quantity: 2
expected:
status_code: 200
code: 0
# tests/test_batch.py
import pytest
from utils.data_driver import DataDriver
@pytest.fixture
def api_instances(user_api, order_api):
"""API 实例字典"""
return {
"user_api": user_api,
"order_api": order_api
}
@pytest.mark.parametrize("case", DataDriver.load_test_cases("data/batch_test_cases.yaml"))
def test_batch_api(api_instances, case, login_token):
"""批量测试接口"""
# 获取 API 实例
api = api_instances[case["api"]]
# 获取方法
method = getattr(api, case["method"])
# 准备参数
params = case["params"].copy()
if "token" in str(case.get("params", {})):
params["token"] = login_token
# 调用方法
response = method(**params)
# 断言
expected = case["expected"]
response.assert_status_code(expected["status_code"])
response.assert_json_path("code", expected["code"])8.5 问题：如何处理接口的异步调用？
问题描述：某些接口是异步的，需要等待一段时间才能获取结果，如何处理？
解决方案：使用轮询机制
# utils/polling.py
import time
from typing import Callable, Optional
from utils.logger import Logger
class Polling:
"""轮询工具"""
def __init__(self, timeout: int = 30, interval: float = 1.0):
"""
初始化轮询配置
:param timeout: 超时时间（秒）
:param interval: 轮询间隔（秒）
"""
self.timeout = timeout
self.interval = interval
self.logger = Logger("polling")
def wait_for(
self,
condition: Callable[[], bool],
message: str = "等待条件满足"
) -> bool:
"""
等待条件满足
:param condition: 条件函数，返回 True 表示条件满足
:param message: 等待消息
:return: 是否满足条件
"""
start_time = time.time()
while time.time() - start_time < self.timeout:
if condition():
self.logger.info(f"{message} - 条件已满足")
return True
self.logger.debug(f"{message} - 继续等待...")
time.sleep(self.interval)
self.logger.error(f"{message} - 超时")
return False
# 使用示例
def test_async_task(user_api, login_token):
"""测试异步任务"""
# 提交异步任务
response = user_api.submit_task(token=login_token, task_type="export")
task_id = response.extract("data.task_id")
# 轮询任务状态
polling = Polling(timeout=60, interval=2)
is_completed = polling.wait_for(
condition=lambda: user_api.get_task_status(token=login_token, task_id=task_id)
.extract("data.status") == "completed",
message=f"等待任务 {task_id} 完成"
)
assert is_completed, "任务未在指定时间内完成"9. 总结
9.1 封装的核心价值
通过本文的学习，你应该理解接口自动化封装的核心价值：
- 提高代码复用性：一次封装，多处使用
- 降低维护成本：修改一处，全局生效
- 提高测试效率：简化测试用例编写
- 增强可读性：测试用例更清晰，专注于业务逻辑
- 便于团队协作：统一的封装规范，便于团队协作
9.2 封装的关键点
- 分层设计：配置层 → 工具层 → HTTP 层 → 业务层 → 测试层
- 单一职责：每个类和方法只负责一个功能
- DRY 原则：消除重复代码
- 配置分离：配置与代码分离
- 错误处理：统一的错误处理机制
- 日志记录：完善的日志记录
9.3 学习路径建议
对于新手，建议按以下顺序学习：
- 第一步：理解封装的概念和必要性
- 第二步：学习基础的 HTTP 客户端封装
- 第三步：学习响应处理和断言封装
- 第四步：学习业务接口封装
- 第五步：学习高级技巧（会话管理、数据驱动等）
- 第六步：实践完整项目
9.4 持续改进
封装不是一次性的工作，需要根据实际使用情况持续改进：
- 收集反馈：从使用者的反馈中发现问题
- 优化重构：定期优化和重构代码
- 文档更新：及时更新文档和注释
- 最佳实践：总结和分享最佳实践
9.5 扩展阅读
- pytest 官方文档：https://docs.pytest.org/
- requests 官方文档：https://requests.readthedocs.io/
- Python 设计模式：了解更多的设计模式
- 测试框架设计：学习如何设计测试框架
附录 A：完整代码清单
A.1 项目依赖文件
# requirements.txt
pytest==7.4.3
pytest-html==4.1.1
pytest-xdist==3.5.0
requests==2.31.0
pyyaml==6.0.1
jsonpath==0.82A.2 配置文件示例
# config/test_config.yaml
api:
base_url: https://api.test.example.com
timeout: 30
test_data:
valid_user:
username: admin
password: 123456
invalid_user:
username: wrong_user
password: wrong_passA.3 README.md 示例
# 接口自动化测试框架
## 项目简介
这是一个基于 pytest 的接口自动化测试框架，提供了完整的封装和工具支持。
## 快速开始
### 安装依赖
```bash
pip install -r requirements.txt运行测试
# 运行所有测试
pytest
# 运行指定测试
pytest tests/test_user_api.py
# 指定环境运行
pytest --env=test项目结构
config/: 配置文件clients/: HTTP 客户端api/: 业务接口封装utils/: 工具类tests/: 测试用例data/: 测试数据