基于 Pytest 的数据驱动 API 自动化测试框架项目详解
·
项目详解：基于 Pytest 的数据驱动 API 自动化测试框架
项目路径：
https://github.com/yl8631739-jpg/pythonproject
目录
1. 项目概述
这是一个面向电商/物流项目的接口自动化测试框架，基于 pytest + requests + Allure 构建。核心思想是：
| 思想 | 实现方式 |
|---|---|
| 数据驱动 | YAML 文件承载全部测试数据，测试代码无硬编码 |
| 关键字驱动 | YAML 中 ${函数名(参数)} 通过反射动态调用 Python 函数 |
| 接口关联 | extract.yaml 文件作为"数据总线"，在接口间传递参数 |
| 多种断言 | 5 种断言模式（包含、相等、不相等、任意值、数据库） |
| 多数据源 | 支持 MySQL/Redis/ClickHouse/MongoDB/SSH |
2. 整体架构与模块依赖关系
分层架构图
┌──────────────────────────────────────────────────────────────────────┐
│ run.py (启动器) │
│ pytest.main() → 收集所有 test_*.py 并执行 │
├──────────────────────────────────────────────────────────────────────┤
│ │
│ testcase/ (测试用例层) │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ test_productList.py test_debug_api.py │ │
│ │ test_business_scenario.py │ │
│ │ │ │
│ │ 核心: @pytest.mark.parametrize → get_testcase_yaml(yaml) │ │
│ │ RequestBase().specification_yaml(base_info, testcase) │ │
│ └──────────┬──────────────────────────────────────────────────┘ │
│ │ 调用 │
│ ▼ │
│ base/ (核心引擎层) │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ apiutil.py apiutil_business.py │ │
│ │ generateId.py removefile.py │ │
│ │ │ │
│ │ 核心方法: specification_yaml() │ │
│ │ ├─ replace_load() ← 反射 ${...} │ │
│ │ ├─ 提取 extract │ │
│ │ └─ 断言 validation │ │
│ └──┬─────────┬──────────┬──────────┬─────────┬──────────────┘ │
│ │ │ │ │ │ │
│ ▼ ▼ ▼ ▼ ▼ │
│ common/ (公共能力层) │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ sendrequest.py assertions.py readyaml.py │ │
│ │ debugtalk.py connection.py recordlog.py │ │
│ │ dingRobot.py semail.py Pjenkins.py │ │
│ │ handleExcel.py operxml.py operationcsv.py │ │
│ │ two_dimension_data.py │ │
│ └──┬─────────┬──────────┬──────────┬──────────────────────────┘ │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ conf/ (配置层) data/ (数据文件) 外部服务 │
│ ┌──────────┐ ┌────────────────┐ ┌──────────────────┐ │
│ │config.ini│ │ YAML / CSV / │ │ 数据库 / Redis / │ │
│ │setting.py│ │ XML / Excel │ │ 钉钉 / Jenkins │ │
│ │operation │ └────────────────┘ └──────────────────┘ │
│ │Config.py │ │
│ └──────────┘ │
└──────────────────────────────────────────────────────────────────────┘
依赖方向
依赖是单向向下的，严格遵循分层：
testcase/ → base/ → common/ → conf/
data/
外部服务
conftest.py → common/ (readyaml, dingRobot, removefile)
conf/ (setting)
testcase/只依赖base/和common/readyamlbase/依赖common/和conf/common/依赖conf/conf/无内部依赖
3. 模块逐层详解
3.1 入口层：run.py
文件位置：项目根目录 run.py
if __name__ == '__main__':
if REPORT_TYPE == 'allure':
pytest.main(
['-s', '-v', '--alluredir=./report/temp', './testcase',
'--clean-alluredir', '--junitxml=./report/results.xml'])
shutil.copy('./environment.xml', './report/temp')
os.system('allure serve ./report/temp')
elif REPORT_TYPE == 'tm':
pytest.main(['-vs', '--pytest-tmreport-name=testReport.html',
'--pytest-tmreport-path=./report/tmreport'])
webbrowser.open_new_tab(...)
职责：
- 读取
setting.REPORT_TYPE决定报告类型（allure或tm） - 调用
pytest.main()执行全部测试用例 - 将
environment.xml复制到 Allure 数据目录（作为环境信息） - 启动 Allure 本地服务查看报告
关键点：--clean-alluredir 会清空 report/temp，所以必须在 pytest 执行完成后才复制 environment.xml 进去。
3.2 测试用例层：testcase/
目录结构：
testcase/
├── Single interface/ # 单接口测试（CRUD 模式）
│ ├── test_debug_api.py
│ ├── addUser.yaml
│ ├── updateUser.yaml
│ ├── deleteUser.yaml
│ └── queryUser.yaml
├── ProductManager/ # 商品管理模块（单接口）
│ ├── test_productList.py
│ ├── login_dw.yaml
│ ├── getProductList.yaml
│ ├── productDetail.yaml
│ ├── commitOrder.yaml
│ ├── orderPay.yaml
│ └── apiType.yaml
└── Business interface/ # 业务场景（多接口串联）
├── test_business_scenario.py
└── BusinessScenario.yml
测试类的标准模板
以 test_productList.py 为例：
@allure.feature(next(m_id) + '商品管理（单接口）')
class TestLogin:
@allure.story(next(c_id) + "获取商品列表")
@pytest.mark.run(order=1)
@pytest.mark.parametrize('base_info,testcase',
get_testcase_yaml('./testcase/ProductManager/getProductList.yaml'))
def test_get_product_list(self, base_info, testcase):
allure.dynamic.title(testcase['case_name'])
RequestBase().specification_yaml(base_info, testcase)
关键模式：
-
m_id/c_id：来自base/generateId.py，是两个无限生成器：m_id→M01_,M02_,M03_…（模块编号）c_id→C01_,C02_,C03_…（用例编号）
每次
next(m_id)取出下一个序号，确保 Allure 报告中模块和用例按编写顺序展示。 -
@pytest.mark.run(order=n)：来自pytest-ordering插件，控制用例执行顺序。 -
@pytest.mark.parametrize+get_testcase_yaml()：从 YAML 文件加载测试数据，每个testCase条目生成一条 pytest 测试用例。 -
RequestBase().specification_yaml(base_info, testcase)：将测试数据交给引擎执行。
YAML 文件的通用结构
- baseInfo: # ← 接口通用信息（URL、方法、请求头）
api_name: 用户登录
url: /dar/user/login
method: post
header:
Content-Type: application/x-www-form-urlencoded;charset=UTF-8
testCase: # ← 测试用例列表（可多个）
- case_name: 正常登录
data: # ← 请求参数（data/json/params 三选一）
user_name: test01
passwd: admin123
validation: # ← 断言列表
- contains: { 'error_code': none }
- eq: { 'msg': '登录成功' }
extract: # ← 参数提取
token: $.data.token
- case_name: 错误密码登录 # ← 第二条用例
data:
user_name: test01
passwd: wrongpwd
validation:
- contains: { 'msg': '登录失败' }
三种参数类型（在 YAML 的 testCase 中只能选一种）：
| YAML 关键字 | 对应 HTTP | Content-Type |
|---|---|---|
data |
POST 表单 | application/x-www-form-urlencoded |
json |
POST JSON | application/json |
params |
GET 查询字符串 | URL 后 ?key=value |
3.3 核心引擎层：base/
文件清单
| 文件 | 职责 |
|---|---|
apiutil.py |
单接口测试引擎（旧版，接收 (base_info, testcase) 分开传入） |
apiutil_business.py |
业务场景引擎（新版，接收 case_info 整体 dict） |
generateId.py |
Allure 报告编号生成器 |
removefile.py |
清理 temp 目录 |
new_testcase_tools.py |
GUI 工具（生成 yaml 用例文件） |
两个 RequestBase 的区别
| 方面 | base/apiutil.py |
base/apiutil_business.py |
|---|---|---|
| 构造 | 含 self.asserts = Assertions() |
使用模块级 assert_res = Assertions() |
| 方法签名 | specification_yaml(base_info, testcase) |
specification_yaml(case_info) |
| YAML 结构 | [{baseInfo},{testCase}] (列表) |
[{baseInfo,testCase},...] (列表中每个元素包含二者) |
| 适用场景 | 单接口多用例参数化 | 多接口业务场景串联 |
| list 参数处理 | 无特殊处理 | 有 handler_yaml_list() 方法 |
3.4 公共能力层：common/
完整文件清单及职责：
| 文件 | 类/函数 | 职责 |
|---|---|---|
sendrequest.py |
SendRequest |
封装 requests 库，提供 get/post/send_request/run_main |
assertions.py |
Assertions |
5 种断言模式统一调度 |
readyaml.py |
ReadYamlData |
YAML 文件读写，含 extract.yaml 的读写 |
readyaml.py |
get_testcase_yaml() |
加载测试用例 YAML 并解析为 parametrize 可用的格式 |
debugtalk.py |
DebugTalk |
辅助函数库，被反射调用，提供时间/加密/随机/CSV读取等 20+ 方法 |
connection.py |
ConnectMysql/Redis/ClickHouse/Mongo/SSH |
多种数据源连接 |
recordlog.py |
RecordLog → logs |
日志系统，按大小滚动备份，保留 7 个文件，30 天清理 |
handleExcel.py |
OperationExcel |
读写 .xls 格式 Excel |
operxml.py |
OperXML |
读取 XML 文件（SQL 配置） |
operationcsv.py |
read_csv() |
读取 CSV 文件 |
dingRobot.py |
send_dd_msg() |
钉钉机器人推送 |
semail.py |
SendEmail / BuildEmail |
SMTP 邮件发送 |
Pjenkins.py |
PJenkins |
Jenkins API 集成 |
two_dimension_data.py |
print_table() |
终端表格格式化打印（仅工具函数） |
3.5 配置层：conf/
| 文件 | 关键内容 |
|---|---|
setting.py |
全局常量：路径字典 FILE_PATH、日志级别、超时时间、报告类型、默认请求头 |
config.ini |
环境配置：API host、MySQL/Redis/ClickHouse/MongoDB/SSH 连接参数、邮箱、报告类型 |
operationConfig.py |
OperationConfig 类，封装 ConfigParser，提供各数据库的 get 方法 |
setting.py 的核心：FILE_PATH 字典定义了所有重要文件的路径，是整个项目的"路径中心"：
FILE_PATH = {
'CONFIG': os.path.join(DIR_BASE, 'conf/config.ini'),
'LOG': os.path.join(DIR_BASE, 'logs'),
'YAML': os.path.join(DIR_BASE), # 根目录（用于 loginName.yaml）
'TEMP': os.path.join(DIR_BASE, 'report/temp'),
'EXTRACT': os.path.join(DIR_BASE, 'extract.yaml'),
'XML': os.path.join(DIR_BASE, 'data/sql'),
'RESULTXML': os.path.join(DIR_BASE, 'report'),
'EXCEL': os.path.join(DIR_BASE, 'data', '测试数据.xls')
}
3.6 数据层：data/
| 文件 | 用途 |
|---|---|
loginName.yaml |
登录测试数据（含 baseInfo 和 testCase） |
login_data.csv |
CSV 格式的登录数据 |
vehicleNo.csv |
车牌号列表（供 debugtalk.vehicle_random() 随机读取） |
sql/homePage.xml |
首页相关 SQL 语句（XML 格式） |
sql/newVehicleAddShare.xml |
车辆新增相关 SQL 语句 |
测试数据.xls |
Excel 格式测试数据 |
3.7 全局钩子：conftest.py
pytest 的全局配置，定义了 3 个钩子：
pytest_sessionstart：记录会话开始时间clear_extract(fixture, session, autouse)：- 禁用 ResourceWarning
- 清空
extract.yaml（清除上次运行的接口关联数据） - 删除
report/temp下的中间文件
pytest_terminal_summary：测试结束后统计结果，如果dd_msg=True则推送钉钉消息
4. 模块间的关联机制
4.1 关联总图
┌─────────────────────────────────────────────────────────────────┐
│ run.py │
│ pytest.main(['./testcase', ...]) │
└──────────────────────────┬──────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ conftest.py │
│ pytest_sessionstart → 记录时间 │
│ clear_extract → 清理 extract.yaml 和 temp 目录 │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ testcase/test_*.py │
│ │
│ ① get_testcase_yaml(YAML_FILE) ←── 读取 data 目录下的 YAML │
│ └─ yaml.safe_load(file) → 返回 [(baseInfo,testCase),...] │
│ │
│ ② @pytest.mark.parametrize → 参数化注入 │
│ │
│ ③ RequestBase().specification_yaml(base_info, testcase) │
│ ↑ │
│ └─ 调用 base/apiutil.py 或 apiutil_business.py │
└──────────────────────────┬──────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ base/apiutil.py: specification_yaml() │
│ │
│ ① 读取 conf/config.ini → 获取 host │
│ conf.operationConfig: get_section_for_data('api_envi','host')│
│ │
│ ② replace_load(header) ←── 反射替换 ${...} 表达式 │
│ ↓ │
│ DebugTalk.xxx() → 调用 common/debugtalk.py 的方法 │
│ ↓ │
│ getattr(DebugTalk(), func_name)(*args) │
│ → 可调用: get_extract_data() 读取 extract.yaml │
│ start_time() / timestamp() / md5_encryption() 等 │
│ │
│ ③ self.run.run_main(...) ←── 调用 common/sendrequest.py │
│ ↓ │
│ requests.session.request(method, url, headers, ...) │
│ → 发送 HTTP 请求 │
│ │
│ ④ extract_data(extract, response) │
│ ↓ │
│ 正则 或 jsonpath 提取 → 写入 extract.yaml │
│ read.write_yaml_data({key: value}) │
│ │
│ ⑤ self.asserts.assert_result(validation, response, status_code)│
│ ↓ │
│ common/assertions.py → 5种断言模式 │
│ → contains / eq / ne / rv / db │
└──────────────────────────┬──────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ conftest.py: pytest_terminal_summary │
│ → 统计 passed/failed/error/skipped │
│ → 如果 dd_msg=True → 调用 dingRobot.send_dd_msg() │
└─────────────────────────────────────────────────────────────────┘
4.2 关联实现一：pytest 参数化 + YAML 加载
代码位置：test_*.py + common/readyaml.py
实现原理：
# 测试文件中：
@pytest.mark.parametrize('base_info,testcase',
get_testcase_yaml('./testcase/ProductManager/getProductList.yaml'))
def test_get_product_list(self, base_info, testcase):
RequestBase().specification_yaml(base_info, testcase)
get_testcase_yaml() 解析 YAML：
def get_testcase_yaml(file):
testcase_list = []
with open(file, 'r', encoding='utf-8') as f:
data = yaml.safe_load(f)
# 分支1: 单个 baseInfo + 多个 testCase（单接口多场景）
if len(data) <= 1:
yam_data = data[0]
base_info = yam_data.get('baseInfo')
for ts in yam_data.get('testCase'):
param = [base_info, ts]
testcase_list.append(param)
return testcase_list
# 返回: [(baseInfo, testCase1), (baseInfo, testCase2), ...]
# 分支2: 多个独立接口（业务场景），保持原样返回
else:
return data
关联效果：一个 YAML 文件中有 1 个 baseInfo + N 个 testCase，pytest 会生成 N 条测试用例，每条共享相同的 baseInfo 但使用不同的 testCase 数据。
4.3 关联实现二：反射替换 ${...} 表达式
代码位置：base/apiutil.py 的 replace_load() 方法
这是框架的核心机制，也是实现"关键字驱动"的关键。
实现原理：
def replace_load(self, data):
str_data = data
if not isinstance(data, str):
str_data = json.dumps(data, ensure_ascii=False)
# 统计 ${ 出现次数，逐个替换
for i in range(str_data.count('${')):
if '${' in str_data and '}' in str_data:
start_index = str_data.index('$')
end_index = str_data.index('}', start_index)
ref_all_params = str_data[start_index:end_index + 1]
# ref_all_params = "${get_extract_data(token)}"
# 提取函数名: "get_extract_data"
func_name = ref_all_params[2:ref_all_params.index("(")]
# 提取参数: "token"
func_params = ref_all_params[
ref_all_params.index("(") + 1:
ref_all_params.index(")")
]
# 反射调用：DebugTalk().get_extract_data("token")
extract_data = getattr(
DebugTalk(), func_name
)(*func_params.split(',') if func_params else "")
# 替换回原字符串
str_data = str_data.replace(ref_all_params, str(extract_data))
# 还原为 dict
if data and isinstance(data, dict):
data = json.loads(str_data)
else:
data = str_data
return data
关联效果：
| YAML 中的写法 | 反射调用的函数 | 实际效果 |
|---|---|---|
${get_extract_data(token)} |
DebugTalk().get_extract_data("token") |
从 extract.yaml 读取 token 值 |
${get_extract_data(goodsId,0)} |
DebugTalk().get_extract_data("goodsId","0") |
随机读取 goodsId |
${get_extract_data(goodsIds,1)} |
DebugTalk().get_extract_data("goodsIds","1") |
取 goodsIds 列表第 1 个 |
${start_time()} |
DebugTalk().start_time() |
获取昨天此刻时间 |
${timestamp()} |
DebugTalk().timestamp() |
获取 10 位时间戳 |
${md5_encryption(password)} |
DebugTalk().md5_encryption("password") |
MD5 加密 |
${vehicle_random()} |
DebugTalk().vehicle_random() |
从 CSV 随机取车牌号 |
${get_baseurl(host)} |
DebugTalk().get_baseurl("host") |
从 config.ini 读取 URL |
DebugTalk 类的完整工具清单（common/debugtalk.py）：
时间相关: timestamp() / timestamp_thirteen() / start_time() / end_time()
start_forward_time() / start_after_time() / end_year_time()
today_zero_tenstamp() / today_zero_stamp() / today_end_stamp()
specified_zero_tamp(days) / specified_end_tamp(days)
month_start_time() / month_end_time() / month_first_time()
加密相关: md5_encryption(params) / sha1_encryption(params) / base64_encryption(params)
数据读取: get_extract_data(node_name, randoms) / get_extract_order_data(data, randoms)
vehicle_random() / read_csv_data(file_name, index)
网络相关: get_baseurl(host)
随机相关: fenceAlarm_alarmType_random() / fatigueAlarm_alarmType_random()
jurisdictionAlarm_random()
4.4 关联实现三：extract.yaml 接口依赖文件
代码位置：common/readyaml.py + 项目根目录 extract.yaml
这是框架实现"接口关联"的核心机制。一个完整的接口关联流程如下：
步骤1: 登录接口返回 {"data": {"token": "abc123", "userId": "999"}}
步骤2: extract_data() 从响应中提取
→ extract: { token: $.data.token, userId: $.data.userId }
→ write_yaml_data({"token": "abc123"})
→ write_yaml_data({"userId": "999"})
↓
extract.yaml 内容变为:
token: abc123
userId: '999'
步骤3: 下一个接口的 YAML 中引用
header:
token: ${get_extract_data(token)}
步骤4: replace_load() 遇到 ${get_extract_data(token)}
→ 反射调用 DebugTalk().get_extract_data("token")
→ 读取 extract.yaml → 返回 "abc123"
→ 替换后 header 变为: token: abc123
extract.yaml 的读写细节：
# 写入（追加模式，每个接口提取的数据都 append 进去）
def write_yaml_data(self, value):
file = open(FILE_PATH['EXTRACT'], 'a', encoding='utf-8')
write_data = yaml.dump(value, allow_unicode=True, sort_keys=False)
file.write(write_data)
# 读取
def get_extract_yaml(self, node_name, second_node_name=None):
with open(FILE_PATH['EXTRACT'], 'r', encoding='utf-8') as rf:
ext_data = yaml.safe_load(rf)
if second_node_name is None:
return ext_data[node_name]
else:
return ext_data[node_name][second_node_name]
注意：每次测试运行前，conftest.py 的 clear_extract fixture 会清空 extract.yaml，确保每次运行从头开始，没有脏数据。
4.5 关联实现四：断言引擎
代码位置：common/assertions.py
assert_result() 是断言调度中心，接收 YAML 中 validation 字段的列表：
def assert_result(self, expected, response, status_code):
all_flag = 0
for yq in expected: # 遍历 validation 列表
for key, value in yq.items(): # 每个 dict 一个断言模式
if key == "contains":
flag = self.contains_assert(value, response, status_code)
elif key == "eq":
flag = self.equal_assert(value, response)
elif key == 'ne':
flag = self.not_equal_assert(value, response)
elif key == 'rv':
flag = self.assert_response_any(value, response)
elif key == 'db':
flag = self.assert_mysql_data(value)
all_flag += flag
# all_flag == 0 表示所有断言通过
if all_flag == 0:
assert True
else:
assert False
all_flag 机制：flag 初始为 0，每个断言失败则 flag += 1，最后 all_flag 是所有 flag 的累加。只有 all_flag == 0 时测试才通过。这使得 YAML 中可以混合使用多种断言模式：
validation:
- contains: { 'status_code': 200 } # 状态码断言
- eq: { 'error_code': '0000' } # 相等断言
- contains: { 'msg': '查询成功' } # 字符串包含断言
5. 核心代码深度解析
5.1 replace_load——框架的"心脏"
位置：base/apiutil.py 第 25-53 行
这是整个框架最重要的函数，它实现了 YAML 中的动态表达式替换。让我们逐行解析：
def replace_load(self, data):
# 步骤1: 统一转为字符串处理
str_data = data
if not isinstance(data, str):
str_data = json.dumps(data, ensure_ascii=False)
# 为什么转 JSON？因为 dict 无法直接做字符串替换操作，
# 序列化为 JSON 后，${...} 就可以用字符串方法处理了。
# 例如: {"token": "${get_extract_data(token)}"}
# → '{"token": "${get_extract_data(token)}"}'
# 步骤2: 循环找出所有 ${...} 表达式
for i in range(str_data.count('${')): # 统计 ${ 出现次数，逐个处理
if '${' in str_data and '}' in str_data:
# 定位 ${ 和 } 的位置
start_index = str_data.index('$')
end_index = str_data.index('}', start_index)
ref_all_params = str_data[start_index:end_index + 1]
# 例如: "${get_extract_data(token)}"
# 步骤3: 解析函数名和参数
func_name = ref_all_params[2:ref_all_params.index("(")]
# "get_extract_data(token)" → 从索引2到"("前 → "get_extract_data"
func_params = ref_all_params[
ref_all_params.index("(") + 1:
ref_all_params.index(")")
]
# 提取括号内的内容 → "token"
# 步骤4: 反射调用 — 这是最关键的一行！
extract_data = getattr(
DebugTalk(), func_name
)(*func_params.split(',') if func_params else "")
# getattr(DebugTalk(), "get_extract_data") → 获取方法对象
# ("token") → 调用该方法
# 如果 func_params 为空字符串则不传参
# 步骤5: 替换回原字符串
if extract_data and isinstance(extract_data, list):
extract_data = ','.join(e for e in extract_data)
str_data = str_data.replace(ref_all_params, str(extract_data))
# 步骤6: 还原数据类型
if data and isinstance(data, dict):
data = json.loads(str_data) # 反序列化为 dict
else:
data = str_data
return data
为什么每次只替换一个 ${} 而不是用正则一次性替换？
因为替换后的值可能本身也包含 ${}（虽然实际场景少见），用逐个查找+替换的方式更安全。另外，str_data.count('${') 限制了循环次数，避免无限循环。
5.2 specification_yaml——完整的测试执行流水线
位置：base/apiutil.py 第 55-119 行
这个方法串联了从 YAML 解析到 HTTP 请求到提取到断言的完整流程：
def specification_yaml(self, base_info, test_case):
# ─────── 阶段1: 准备请求 ───────
params_type = ['data', 'json', 'params'] # 允许的参数类型
# ① 获取 host → 拼接完整 URL
url_host = self.conf.get_section_for_data('api_envi', 'host')
url = url_host + base_info['url']
# ② 反射替换请求头中的动态表达式
header = self.replace_load(base_info['header'])
# ③ 处理 Cookie（可选）
cookie = None
if base_info.get('cookies') is not None:
cookie = eval(self.replace_load(base_info['cookies']))
# ─────── 阶段2: 处理 testCase ───────
case_name = test_case.pop('case_name')
# ④ 反射替换断言值
val = self.replace_load(test_case.get('validation'))
test_case['validation'] = val
validation = eval(test_case.pop('validation')) # 字符串→list
# ⑤ 提取 extract/extract_list（如果有）
extract = test_case.pop('extract', None)
extract_list = test_case.pop('extract_list', None)
# ⑥ 反射替换请求参数（data/json/params）
for key, value in test_case.items():
if key in params_type:
test_case[key] = self.replace_load(value)
# ⑦ 处理文件上传
file, files = test_case.pop('files', None), None
if file is not None:
files = {fk: open(fv, mode='rb') for fk, fv in file.items()}
# ─────── 阶段3: 发送请求 ───────
res = self.run.run_main(
name=api_name, url=url, case_name=case_name,
header=header, method=method, file=files,
cookies=cookie, **test_case
)
# run_main 内部调用 send_request，最终使用 requests.session.request()
status_code = res.status_code
# ─────── 阶段4: 提取响应参数 ───────
res_json = json.loads(res.text)
if extract is not None:
self.extract_data(extract, res.text)
if extract_list is not None:
self.extract_data_list(extract_list, res.text)
# ─────── 阶段5: 断言 ───────
self.asserts.assert_result(validation, res_json, status_code)
关键设计点：
-
test_case.pop('case_name')等 pop 操作：因为test_case中除了请求参数（data/json/params），还混合了case_name、validation、extract等元数据。pop 将它们从 dict 中移除，剩下的键值对正好作为**test_case传给run_main()，不会污染请求参数。 -
eval(test_case.pop('validation'))：replace_load替换后返回值是 JSON 字符串（如'[{"contains": {"msg": "成功"}}]'），需要eval转成真正的 Python 列表。 -
self.run.run_main(...)传递**test_case：此时 test_case 中只剩data/json/params等请求参数，直接解包传给requests.session.request()。
5.3 get_testcase_yaml——YAML 加载器
位置：common/readyaml.py 第 11-30 行
def get_testcase_yaml(file):
testcase_list = []
with open(file, 'r', encoding='utf-8') as f:
data = yaml.safe_load(f)
# 分支1: 单个 baseInfo + 多个 testCase（单接口多场景）
if len(data) <= 1:
yam_data = data[0]
base_info = yam_data.get('baseInfo')
for ts in yam_data.get('testCase'):
param = [base_info, ts]
testcase_list.append(param)
return testcase_list
# 返回: [(baseInfo, testCase1), (baseInfo, testCase2), ...]
# 分支2: 多个独立接口（业务场景串联）
else:
return data
# 返回原始列表, apiutil_business.py 自行处理
两种返回格式决定了两种测试模式：
分支1 (单接口文件):
返回: [(baseInfo, testCase1), (baseInfo, testCase2)]
↓
pytest 生成 2 条用例: test_get_product_list[base_info0-testcase0]
test_get_product_list[base_info0-testcase1]
↓
调用: RequestBase().specification_yaml(base_info, testcase)
分支2 (业务场景文件):
返回: [{baseInfo, testCase}, {baseInfo, testCase}, ...]
↓
pytest 生成 1 条用例: test_business_scenario[case_info0]
↓
调用: RequestBase().specification_yaml(case_info) # 整体传入
5.4 assert_result——断言调度中心
位置：common/assertions.py 第 160-202 行
def assert_result(self, expected, response, status_code):
all_flag = 0
for yq in expected: # 遍历 validation 列表
for key, value in yq.items(): # 每个 dict 一个模式键值对
if key == "contains":
flag = self.contains_assert(value, response, status_code)
elif key == "eq":
flag = self.equal_assert(value, response)
elif key == 'ne':
flag = self.not_equal_assert(value, response)
elif key == 'rv':
flag = self.assert_response_any(value, response)
elif key == 'db':
flag = self.assert_mysql_data(value)
else:
logs.error("不支持此种断言方式")
all_flag += flag
if all_flag == 0:
logs.info("测试成功")
assert True
else:
logs.error("测试失败")
assert False
五种断言模式详解：
| 模式 | YAML 写法 | 实现逻辑 | 用途 |
|---|---|---|---|
contains |
{'msg':'调用成功'} |
JSONPath 提取字段 → in 判断包含 |
验证字符串片段 |
contains |
{'status_code':200} |
直接比对状态码 | 验证 HTTP 状态码 |
eq |
{'msg':'登录成功'} |
取交集 key → operator.eq 比对字典 |
精确匹配 |
ne |
{'msg':'登录失败'} |
operator.ne |
反向验证 |
rv |
{"data":2} |
取第一个 key 比对 value | 验证任意字段值 |
db |
SELECT * FROM user... |
执行 SQL，结果非 None 即通过 | 数据库断言 |
contains 断言的特别之处——它还能断言 status_code：
def contains_assert(self, value, response, status_code):
for assert_key, assert_value in value.items():
if assert_key == "status_code":
# 特殊处理：直接比较状态码
if assert_value != status_code:
flag += 1
else:
# 普通字段：用 JSONPath 提取后再用 in 判断
resp_list = jsonpath.jsonpath(response, "$..%s" % assert_key)
if assert_value in resp_list:
logs.info("字符串包含断言成功")
else:
flag += 1
return flag
5.5 run_main——HTTP 请求的统一入口
位置：common/sendrequest.py 第 119-163 行
def run_main(self, name, url, case_name, header, method,
cookies=None, file=None, **kwargs):
# 收集 Allure 报告信息
logs.info('接口名称：%s' % name)
logs.info('请求地址：%s' % url)
logs.info('请求方式：%s' % method)
logs.info('测试用例名称：%s' % case_name)
# 将请求参数 attach 到 Allure 报告
req_params = json.dumps(kwargs, ensure_ascii=False)
if "data" in kwargs.keys():
allure.attach(req_params, '请求参数', allure.attachment_type.TEXT)
# ... 同理 json 和 params
# 发送请求
response = self.send_request(
method=method, url=url, headers=header,
cookies=cookies, files=file,
timeout=setting.API_TIMEOUT,
verify=False, # 忽略 SSL 证书验证
**kwargs
)
return response
send_request 内部（第 96-117 行）：
def send_request(self, **kwargs):
session = requests.session() # 使用 session 保持 Cookie
result = session.request(**kwargs)
# 自动保存 Cookie 到 extract.yaml
set_cookie = requests.utils.dict_from_cookiejar(result.cookies)
if set_cookie:
cookie = {'Cookie': set_cookie}
self.read.write_yaml_data(cookie) # 持久化 cookie
return result
5.6 extract_data / extract_data_list——响应参数提取
位置：base/apiutil.py 第 129-186 行
这是接口关联的具体实现，支持两种提取方式：
方式一：JSONPath 提取
YAML 写法：
extract:
token: $.data.token # JSONPath 表达式
goodsId: $.goodsList[*].goodsId
代码实现（第 150-157 行）：
if '$' in value:
ext_json = jsonpath.jsonpath(json.loads(response), value)[0]
if ext_json:
extract_data = {key: ext_json}
else:
extract_data = {key: '未提取到数据，请检查接口返回值是否为空！'}
self.read.write_yaml_data(extract_data)
方式二：正则表达式提取
YAML 写法：
extract:
status: '"status":"(.*?)"' # 正则表达式
data: '"data":(\d*)'
代码实现（第 140-149 行）：
pattern_lst = ['(.*?)', '(.+?)', r'(\d+)', r'(\d*)']
for pat in pattern_lst:
if pat in value: # 检测是否为正则模式
ext_lst = re.search(value, response)
if pat in [r'(\d+)', r'(\d*)']:
extract_data = {key: int(ext_lst.group(1))} # 数字类型
else:
extract_data = {key: ext_lst.group(1)} # 字符串类型
self.read.write_yaml_data(extract_data)
检测策略：通过检查 value 中是否包含 (.+?)、(.*?)、(\d+)、(\d*) 这些特征字符串来区分提取方式。如果包含这些特征且也包含 $，则两种方法都会尝试（但一般不同时出现）。
extract_list 与 extract 的区别：
extract: 提取单个值，用re.search()，找到第一个即停止extract_list: 提取多个值，用re.findall()，找到所有匹配项，返回列表
5.7 generateId——Allure 报告排序
位置：base/generateId.py
def generate_module_id():
for i in range(1, 1000):
module_id = 'M' + str(i).zfill(2) + '_' # M01_, M02_, ...
yield module_id
def generate_testcase_id():
for i in range(1, 10000):
case_id = 'C' + str(i).zfill(2) + '_' # C01_, C02_, ...
yield case_id
m_id = generate_module_id()
c_id = generate_testcase_id()
使用方式：
@allure.feature(next(m_id) + '商品管理（单接口）')
# → "@allure.feature('M01_商品管理（单接口）')"
@allure.story(next(c_id) + "获取商品列表")
# → "@allure.story('C01_获取商品列表')"
为什么需要这个：Allure 报告的 feature/story 默认按字母顺序排序。加上 M01_、C01_ 前缀后，字母排序就等于编写顺序，保证了报告的可读性。
注意：由于 m_id 和 c_id 是模块级生成器，在多线程执行时会有问题。但 pytest 默认是单线程顺序执行，所以没问题。
6. 完整请求流程追溯
以"商品管理"模块的一条完整测试为例，从头到尾追溯数据流：
步骤序列:
┌─────────────────────────────────────────────────────────────────┐
│ 1. run.py │
│ pytest.main(['-s','-v','--alluredir=./report/temp', │
│ './testcase', '--clean-alluredir']) │
├─────────────────────────────────────────────────────────────────┤
│ 2. conftest.py │
│ clear_extract fixture (autouse) │
│ → clear_yaml_data() → 清空 extract.yaml │
│ → remove_file() → 清空 report/temp 目录 │
├─────────────────────────────────────────────────────────────────┤
│ 3. test_productList.py::TestLogin::test_get_product_list │
│ → next(m_id) → "M01_" │
│ → next(c_id) → "C01_" │
│ → get_testcase_yaml("./testcase/ProductManager/ │
│ getProductList.yaml") │
│ → yaml.safe_load(file) │
│ → 返回 [(baseInfo, testCase1), (baseInfo, testCase2)] │
│ → parametrize 生成 2 条 pytest 用例 │
│ │
│ 用例 testcase[base_info0-testcase0]: │
│ → RequestBase().specification_yaml(base_info, testcase) │
├─────────────────────────────────────────────────────────────────┤
│ 4. base/apiutil.py::specification_yaml │
│ │
│ 4.1 准备请求: │
│ → host = "http://127.0.0.1:8787" (from config.ini) │
│ → url = "http://127.0.0.1:8787/coupApply/cms/goodsList" │
│ │
│ 4.2 replace_load(header): │
│ → header = {"Content-Type":"...", │
│ "token":"${get_extract_data(cookie)}"} │
│ → 反射调用 DebugTalk().get_extract_data("cookie") │
│ → 读取 extract.yaml → 返回 "abc123..." │
│ → header = {"Content-Type":"...", │
│ "token":"abc123..."} │
│ │
│ 4.3 replace_load(validation): │
│ → [{'contains': {'error_code': '0000'}}] (无${}，不变) │
│ → eval() → [{'contains': {'error_code': '0000'}}] │
│ │
│ 4.4 replace_load(params): │
│ → {"msgType":"getHandsetListOfCust","page":1,"size":20} │
│ → 无 ${}，不变 │
│ │
│ 4.5 run_main(...): │
│ → send_request(method="GET", url=..., params=..., │
│ headers=..., verify=False) │
│ → requests.session().request(...) │
│ → 返回 Response │
│ │
│ 4.6 extract_data_list: │
│ → extract_list: { goodsId: $.goodsList[*].goodsId } │
│ → jsonpath 提取 → ["18382788819", "33809635011", ...] │
│ → write_yaml_data({"goodsId": ["18382788819", ...]}) │
│ → extract.yaml 新增: │
│ goodsId: │
│ - '18382788819' │
│ - '33809635011' │
│ ... │
│ │
│ 4.7 assert_result: │
│ → contains_assert({'error_code':'0000'}, response, 200) │
│ → jsonpath 提取 error_code → "0000" │
│ → "0000" in "0000" → True → flag=0 │
│ → all_flag=0 → assert True → 测试通过 │
├─────────────────────────────────────────────────────────────────┤
│ 5. conftest.py::pytest_terminal_summary │
│ → 统计 all 用例通过/失败/错误/跳过数量 │
│ → 如果 dd_msg=True → send_dd_msg(summary) → 钉钉推送 │
├─────────────────────────────────────────────────────────────────┤
│ 6. run.py │
│ → shutil.copy('./environment.xml','./report/temp') │
│ → os.system('allure serve ./report/temp') │
│ → 启动本地 Allure 报告服务 │
└─────────────────────────────────────────────────────────────────┘
多接口串联流程（以 BusinessScenario.yml 为例）：
接口1: 获取商品列表
→ extract_list: goodsIds: $.goodsList[*].goodsId
→ extract.yaml: goodsIds: ['111','222','333']
接口2: 获取商品详情
→ json: pro_id: ${get_extract_data(goodsIds,1)}
→ replace_load → DebugTalk().get_extract_data("goodsIds","1")
→ read extract.yaml → goodsIds[1-1] = '111'
接口3: 提交订单
→ json: goods_id: ${get_extract_data(goodsIds,1)}
→ replace_load → '111'
→ extract: orderNumber: $.orderNumber, userId: $.userId
→ extract.yaml: orderNumber: '12345', userId: '999'
接口4: 订单支付
→ json: orderNumber: ${get_extract_data(orderNumber)}
→ replace_load → '12345'
→ 支付成功
7. 两种 apiutil 的差异
虽然 base/apiutil.py 和 base/apiutil_business.py 功能高度相似，但存在以下关键差异：
| 项目 | base/apiutil.py |
base/apiutil_business.py |
|---|---|---|
| YAML 结构 | 分离式：[{baseInfo},{testCase}] |
一体式：[{baseInfo, testCase}] |
| 方法签名 | specification_yaml(base_info, test_case) |
specification_yaml(case_info) |
| 断言对象 | self.asserts = Assertions() (实例级) |
assert_res = Assertions() (模块级) |
| list 处理 | 无 | handler_yaml_list() 方法将 list 转为逗号分隔 |
| Allure 参数 | 直接用 test_case 中剩余参数 |
用 tc 遍历 case_info['testCase'] |
| 请求参数 | **test_case 直接解包 |
同样方式但变量名为 tc |
业务场景 YAML 的结构特点：
# 业务场景文件包含多个接口
- baseInfo: # ← 接口1
api_name: 商品列表
url: /api/goodsList
header: ...
testCase:
- case_name: ...
params: ...
extract_list: ...
- baseInfo: # ← 接口2（依赖接口1的提取结果）
api_name: 商品详情
url: /api/productDetail
header: ...
testCase:
- case_name: ...
json:
pro_id: ${get_extract_data(goodsIds,1)} # ← 引用接口1的提取
8. 总结
架构总览
┌──────────┐ ┌──────────────────┐ ┌──────────────┐
│ run.py │ → │ conftest.py │ → │ testcase/ │
│ (启动器) │ │ (全局钩子) │ │ (测试用例) │
└──────────┘ └──────────────────┘ └──────┬───────┘
│
┌─────────────────────────┘
▼
┌──────────────────┐
│ base/apiutil.py │ ← 核心引擎
│ (specification_ │
│ yaml) │
└──────┬───────────┘
│
┌────────────┼────────────┬──────────────┐
▼ ▼ ▼ ▼
┌────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐
│common/ │ │common/ │ │common/ │ │ conf/ │
│sendreq │ │readyaml │ │assertions│ │ config.ini │
│uest.py │ │.py │ │.py │ │ (host/DB) │
│(HTTP) │ │(YAML读写)│ │(断言引擎)│ └──────────────┘
└────────┘ └────┬─────┘ └──────────┘
│
┌───────┴───────┐
▼ ▼
┌──────────────┐ ┌──────────────┐
│ extract.yaml │ │ debugtalk.py │
│ (接口数据总线)│ │ (20+辅助函数)│
└──────────────┘ └──────────────┘
核心设计思想
- 数据与代码分离：测试数据全部在 YAML 文件中，测试代码只做"框架"工作
- 用反射实现"关键字驱动"：YAML 中的
${函数名(args)}通过getattr(DebugTalk(), name)(*args)动态解析 - extract.yaml 作为"接口数据总线"：前一个接口的输出经过提取写入 YAML，后一个接口通过反射读取，实现接口关联
- all_flag 断言累加机制：5 种断言模式可自由组合，任一失败则全用例失败
- Allure 报告集成：每一步操作（URL、参数、响应、断言）都 attach 到报告，便于排查
电商企业物流数字化转型必备！快递鸟 API 接口，72 小时快速完成物流系统集成。全流程实战1V1指导，营造开放的API技术生态圈。
更多推荐
7
0- 0
已为社区贡献1条内容
所有评论(0)