妙妙屋商城自动化项目架构介绍
·
一.项目介绍
技术栈：Python+pytest+sqlalchemy+requests+allure+jsonpath+yaml+Jenkins+Linux
项目源码(gitee):https://gitee.com/xiaomalou/shoppingpytest.git
该项目是一个在线购物的商城网站，包括用户注册，登录，下单，上架/下架商品，下单支付等相关功能。
二.项目结构说明
pythonproject/
├── __init__.py # 包标识
├── run.py # ★ 主程序入口
├── conftest.py # ★ 全局 fixture（清数据、汇总通知）
├── pytest.ini # pytest 配置（文件名固定，不能改）
├── environment.xml # allure 报告"环境信息"配置
├── extract.yaml # ★ 接口依赖参数临时存放文件（自动读写）
├── requirements.txt # 第三方库清单
├── 使用前请阅读此文件.md # 官方使用说明（框架自带的文档）
│
├── base/ # ★ 基础层：用例执行核心 + 工具
│ ├── apiutil.py # ★ 单接口用例执行器（RequestBase）
│ ├── apiutil_business.py # ★ 业务场景用例执行器（RequestBase）
│ ├── generateId.py # 生成用例编号（allure 报告排序用）
│ ├── removefile.py # 删除临时文件工具
│ ├── new_testcase_tools.py # PyQt5 GUI 工具：可视化生成 yaml 用例
│ └── new_tools.ui # GUI 工具的界面文件（PyQt5 Designer 生成）
│
├── common/ # ★ 公共方法封装层
│ ├── readyaml.py # ★ YAML 读写封装
│ ├── sendrequest.py # ★ 发送 HTTP 请求封装
│ ├── assertions.py # ★ 断言封装（contains/eq/ne/rv/db）
│ ├── debugtalk.py # ★ 动态表达式函数库（${函数名()}）
│ ├── connection.py # 数据库连接（MySQL/Redis/ClickHouse/Mongo/SSH）
│ ├── recordlog.py # 日志模块
│ ├── operationcsv.py # 读 CSV 文件
│ ├── handleExcel.py # 读/写 xls 文件
│ ├── operxml.py # 读 XML 文件（SQL 模板）
│ ├── two_dimension_data.py # 打印二维表格（终端美化）
│ ├── dingRobot.py # 钉钉机器人通知
│ ├── semail.py # 邮件发送
│ └── Pjenkins.py # Jenkins 对接
│
├── conf/ # ★ 配置层
│ ├── config.ini # ★ 全局配置（服务地址/数据库/邮箱等）
│ ├── operationConfig.py # 读取 config.ini 的封装
│ └── setting.py # ★ 全局常量（路径、报告类型等）
│
├── data/ # ★ 测试数据
│ ├── loginName.yaml # ★ 登录用例（session 级 fixture 使用）
│ ├── login_data.csv # 登录数据
│ ├── vehicleNo.csv # 车牌号数据
│ ├── 测试数据.xls # Excel 测试数据
│ └── sql/
│ ├── homePage.xml # SQL 模板（首页报表）
│ └── newVehicleAddShare.xml
│
├── testcase/ # ★ 测试用例层
│ ├── conftest.py # 用例层 fixture（登录、数据清理）
│ ├── Single interface/ # 单接口测试
│ │ ├── test_debug_api.py # 用户管理模块测试脚本
│ │ ├── login_dw.yaml # 电网登录用例（未启用）
│ │ ├── addUser.yaml # 新增用户用例
│ │ ├── updateUser.yaml # 修改用户用例
│ │ ├── deleteUser.yaml # 删除用户用例
│ │ └── queryUser.yaml # 查询用户用例
│ ├── ProductManager/ # 电商商品管理测试
│ │ ├── test_productList.py # 商品管理测试脚本
│ │ ├── getProductList.yaml # 商品列表
│ │ ├── productDetail.yaml # 商品详情
│ │ ├── commitOrder.yaml # 提交订单
│ │ ├── orderPay.yaml # 订单支付
│ │ ├── apiType.yaml # 接口状态（未启用）
│ │ └── login_dw.yaml # 电网登录（未启用）
│ └── Business interface/ # 业务场景（多接口串联）测试
│ ├── test_business_scenario.py # 业务场景测试脚本
│ └── BusinessScenario.yml # ★ 完整下单流程（列表→详情→下单→支付→校验）
│
├── report/ # ★ 测试报告
│ ├── temp/ # allure 原始数据（json/附件）
│ ├── allureReport/ # 已生成的 allure 静态报告
│ ├── tmreport/ # tm 风格报告
│ └── results.xml # JUnit 格式结果（CI 用）
│
├── logs/ # ★ 运行日志（按天滚动，保留30天）
│ └── test.20260701.log
│
└── venv/ # 项目虚拟环境三、核心代码块
主程序允许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(f'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(os.getcwd() + '/report/tmreport/testReport.html')REPORT_TYPE在conf/setting.py里配置,切换「出 allure 报告」还是「tm 报告」;--alluredir=./report/temp指定 allure 原始数据存哪;--clean-alluredir跑前清掉上次的原始数据;allure serve把原始数据渲染成网页并启动本地服务给你看
测试用例的数据驱动BusinessScenario.yml
- baseInfo:
description: 业务场景测试用例编写示范
api_name: 商品列表
url: /coupApply/cms/goodsList
method: Get
header:
Content-Type: application/x-www-form-urlencoded;charset=UTF-8
token: ${get_extract_data(token)}
testCase:
- case_name: 获取商品列表
params:
msgType: getHandsetListOfCust
page: 1
size: 20
validation:
- contains: { 'error_code': '0000' }
extract_list:
goodsIds: $.goodsList[*].goodsId
- baseInfo:
api_name: 商品详情
url: /coupApply/cms/productDetail
method: post
header:
Content-Type: application/json;charset=UTF-8
testCase:
- case_name: 获取商品详情
json:
pro_id: ${get_extract_data(goodsIds,1)}
page: 1
size: 20
validation:
- eq: { 'error_code': '0000' }
- baseInfo:
api_name: 提交订单
url: /coupApply/cms/placeAnOrder
method: post
header:
Content-Type: application/json;charset=UTF-8
testCase:
- case_name: 详情页面选择规格，提交订单
json:
goods_id: ${get_extract_data(goodsIds,1)}
number: 2
propertyChildIds: "2:9"
inviter_id: 127839112
price: "128"
freight_insurance: "0.00"
discount_code: "002399"
consignee_info:
{ "name": "张三","phone": 13800000000,"address": "北京市海淀区西三环北路74号院4栋3单元1008" }
validation:
- eq: { 'message': '提交订单成功' }
extract:
orderNumber: $.orderNumber
userId: $.userId
- baseInfo:
api_name: 订单支付
url: /coupApply/cms/orderPay
method: post
header:
Content-Type: application/json;charset=UTF-8
testCase:
- case_name: 订单支付
json:
orderNumber: ${get_extract_data(orderNumber)}
userId: ${get_extract_data(userId)}
timeStamp: ${timestamp()}
validation:
- eq: { 'message': '订单支付成功' }
- baseInfo:
api_name: 校验订单状态
url: /coupApply/cms/checkOrderStatus
method: post
header:
Content-Type: application/json;charset=UTF-8
testCase:
- case_name: 校验商品订单状态
json:
orderNumber: ${get_extract_data(orderNumber)}
timeStamp: ${timestamp()}
validation:
- eq: { 'status': '0' }单接口的 YAML 是一个 baseInfo + 多条 testCase，这个是多个 baseInfo(5 个接口),每个下面只有 1 条 testCase。
接口关联：
① 商品列表(GET)
extract_list: goodsIds: $.goodsList[*].goodsId → 所有商品ID写入 extract.yaml
② 商品详情(POST json)
pro_id: ${get_extract_data(goodsIds,1)} → 取列表第1个商品
③ 提交订单(POST json)
goods_id: ${get_extract_data(goodsIds,1)}
extract: orderNumber: $.orderNumber, userId: $.userId → 写订单号/用户ID
④ 订单支付(POST json)
orderNumber: ${get_extract_data(orderNumber)}
userId: ${get_extract_data(userId)}
timeStamp: ${timestamp()} → 当前时间戳
⑤ 校验订单状态(POST json)
orderNumber: ${get_extract_data(orderNumber)}
validation: - eq: {'status': '0'} → 断言状态=已支付
三个关键动作反复出现,串成一条链:
- 提取:③ 把
orderNumber提取出来; - 读:④⑤ 用
${get_extract_data(orderNumber)}读出来; - 生成:④ 用
${timestamp()}现场生成一个参数。
测试用例数据驱动addUser.yaml（单接口多teastCase）
dates: '2023-12-31'
phone: 13800000000
token: ${get_extract_data(token)}
validation:
- contains: { 'status_code': 200 }
- contains: { 'msg': '新增成功' }
- case_name: 无效新增·缺少token
data:
username: testadduser
password: tset6789890
role_id: 123456789
dates: '2023-12-31'
phone: 13800000000
token:
validation:
- contains: { 'status_code': 200 }
- contains: { 'msg': '新增失败' }
- case_name: 无效新增·缺少必填参数username
data:
password: tset6789890
role_id: 123456789
dates: '2023-12-31'
phone: 13800000000
token:
validation:
- contains: { 'status_code': 200 }
- contains: { 'msg': '新增失败' }
- case_name: 无效新增·缺少必填参数role_id
data:
username: testadduser
password: tset6789890
dates: '2023-12-31'
phone: 13800000000
token:
validation:
- contains: { 'status_code': 200 }
- contains: { 'msg': '新增失败' }
这个是一个单接口然后四个用例，一个正向三个反向。
| 用例名 | 数据特点 | 预期 |
|---|---|---|
| 正常新增用户 | 所有字段都填,带 token | 新增成功 |
| 无效新增·缺少token | token 留空 | 新增失败 |
| 无效新增·缺少username | username 缺失 | 新增失败 |
| 无效新增·缺少role_id | role_id 缺失 | 新增失败 |
@pytest.mark.parametrize('base_info,testcase',get_testcase_yaml("./testcase/Single interface/addUser.yaml"))
YAML 存用例数据,parametrize 展开成多次调用,代码与数据分离
对这个拓展接口用例设计的四个维度：
1. 正向(正常路径)
所有参数合法、完整,验证「接口能正常完成任务」。例如「正常新增用户」。
2. 反向(异常路径)—— 逐字段缺失/非法
把每个必填参数依次去掉或填错,验证「缺了它,接口会正确拒绝并返回明确错误」。 这是接口测试里工作量最大、最能暴露 bug的部分。
3. 边界值
参数取「刚好合法」和「刚好非法」的分界。例如用户名长度「最大长度刚好通过 / 最大长度+1 报错」、页码「第 0 页 / 负数页」。「边界值怎么测」,核心口诀:min-1、min、max、max+1。
4. 业务/数据状态
依赖外部状态的场景:重复提交、数据已存在、数据不存在、权限不足、token 过期等。
断言模式判断assertions.py
def assert_result(self, expected, response, status_code):
"""
断言，通过断言all_flag标记，all_flag==0表示测试通过，否则为失败
:param expected: 预期结果
:param response: 实际响应结果
:param status_code: 响应code码
:return:
"""
all_flag = 0
try:
logs.info("yaml文件预期结果：%s" % expected)
# logs.info("实际结果：%s" % response)
# all_flag = 0
for yq in expected:
for key, value in yq.items():
if key == "contains":
flag = self.contains_assert(value, response, status_code)
all_flag = all_flag + flag
elif key == "eq":
flag = self.equal_assert(value, response)
all_flag = all_flag + flag
elif key == 'ne':
flag = self.not_equal_assert(value, response)
all_flag = all_flag + flag
elif key == 'rv':
flag = self.assert_response_any(actual_results=response, expected_results=value)
all_flag = all_flag + flag
elif key == 'db':
flag = self.assert_mysql_data(value)
all_flag = all_flag + flag
else:
logs.error("不支持此种断言方式")
except Exception as exceptions:
logs.error('接口断言异常，请检查yaml预期结果值是否正确填写!')
raise exceptions
if all_flag == 0:
logs.info("测试成功")
assert True
else:
logs.error("测试失败")
assert False关键机制: 每个断言方法返回一个 flag(0 = 通过,失败就 +1)。 所有断言的结果累加到 all_flag,最后只有 all_flag == 0(一条都没失败)才算通过。 多条断言是「全过才过」,不是「多数过」。
五个断言：
1. contains —— 字符串包含(最常用)
- contains: {'msg': '新增成功'}
逻辑:在响应里找 msg 字段,判断它的值是否包含「新增成功」。
它有个特殊分支——当 key 是 status_code 时,不去解析响应体,直接比较状态码:
- contains: {'status_code': 200} # 等价于 res.status_code == 200
2. eq —— 严格相等
- eq: {'msg': '登录成功'}
逻辑:找「预期和实际共有的 key」,只比这一项是否严格相等(operator.eq)。不是「包含」,是「一字不差」。
3. ne —— 不相等
- ne: {'state': '已入网'}
4. rv —— 任意值相等
- rv: {'data': 2}
断言响应体里 data 字段的值 == 2。和 eq 的区别:rv 更直接地「取某个字段比一个具体值」,eq 是「比预期字典和实际字典的共同项」。
5. db —— 数据库断言
- db: select * from sys_user where login_name = 'test999'
直接写一条 SQL,到数据库查,查得到数据就算通过。用来验证「接口操作真的落库了」。
| 关键字 | 含义 | 判定 | 典型用途 |
|---|---|---|---|
contains |
字符串包含 | 值里含预期 | 返回码、msg 文案 |
eq |
严格相等 | 值 == 预期 | 精确字段值 |
ne |
不相等 | 值 != 预期 | 排除某状态 |
rv |
任意值相等 | 某字段 == 预期 | 具体数值 |
db |
数据库断言 | SQL 查到数据 | 验证落库 |
注意：contains 必须写在最前面:项目文档明确要求「有多种断言时,contains 必须在前面」。 原因是 assert_result 按顺序执行,而 contains 通常用来先拦 status_code——如果状态码都错了,后面的 eq/ne 就没意义。
接口测试 apituil.py
def specification_yaml(self, base_info, test_case):
"""
接口请求处理基本方法
:param base_info: yaml文件里面的baseInfo
:param test_case: yaml文件里面的testCase
:return:
"""
try:
params_type = ['data', 'json', 'params']
url_host = self.conf.get_section_for_data('api_envi', 'host')
api_name = base_info['api_name']
allure.attach(api_name, f'接口名称：{api_name}', allure.attachment_type.TEXT)
url = url_host + base_info['url']
allure.attach(api_name, f'接口地址：{url}', allure.attachment_type.TEXT)
method = base_info['method']
allure.attach(api_name, f'请求方法：{method}', allure.attachment_type.TEXT)
header = self.replace_load(base_info['header'])
allure.attach(api_name, f'请求头：{header}', allure.attachment_type.TEXT)
# 处理cookie
cookie = None
if base_info.get('cookies') is not None:
cookie = eval(self.replace_load(base_info['cookies']))
case_name = test_case.pop('case_name')
allure.attach(api_name, f'测试用例名称：{case_name}', allure.attachment_type.TEXT)
# 处理断言
val = self.replace_load(test_case.get('validation'))
test_case['validation'] = val
validation = eval(test_case.pop('validation'))
# 处理参数提取
extract = test_case.pop('extract', None)
extract_list = test_case.pop('extract_list', None)
# 处理接口的请求参数
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:
for fk, fv in file.items():
allure.attach(json.dumps(file), '导入文件')
files = {fk: open(fv, mode='rb')}
res = self.run.run_main(name=api_name, url=url, case_name=case_name, header=header, method=method,
file=files, cookies=cookie, **test_case)
status_code = res.status_code
allure.attach(self.allure_attach_response(res.json()), '接口响应信息', allure.attachment_type.TEXT)
try:
res_json = json.loads(res.text) # 把json格式转换成字典字典
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)
# 处理断言
self.asserts.assert_result(validation, res_json, status_code)
except JSONDecodeError as js:
logs.error('系统异常或接口未请求！')
raise js
except Exception as e:
logs.error(e)
raise e
except Exception as e:
raise e
步骤拆分：
1.从 baseInfo 取出接口名、路径、方法、请求头，把路径和 host 拼成完整 URL，全部 attach 到 allure 报告里。replace_load 会处理 header 里的 ${get_extract_data(token)}。
def specification_yaml(self, base_info, test_case):
try:
params_type = ['data', 'json', 'params'] # 合法的参数类型
url_host = self.conf.get_section_for_data('api_envi', 'host') # 从配置取 host
api_name = base_info['api_name'] # 接口名
allure.attach(api_name, f'接口名称：{api_name}', allure.attachment_type.TEXT)
url = url_host + base_info['url'] # ★ 拼完整地址
method = base_info['method'] # 请求方法
header = self.replace_load(base_info['header']) # 请求头（含${}替换）2.eval() 把"看起来像字典的字符串"真正变成字典。base_info.get('cookies') 用 .get 取不到就返回 None（安全取法）。
cookie = None
if base_info.get('cookies') is not None: # 如果用例写了 cookies
cookie = eval(self.replace_load(base_info['cookies'])) # eval 还原成字典3.dict.pop(key) 会取出并删除这个键，这样 test_case 最后剩下的就是请求参数（data/json/params），可以整体传给请求方法。validation 从 YAML 读出来是"看起来像列表的字符串"，需要 eval 转成真正的列表。
case_name = test_case.pop('case_name') # pop：取出并删掉这个key
allure.attach(api_name, f'测试用例名称：{case_name}', ...)
val = self.replace_load(test_case.get('validation')) # 处理断言里的${}
test_case['validation'] = val
validation = eval(test_case.pop('validation')) # 还原成列表
extract = test_case.pop('extract', None) # 提取配置（单参数）
extract_list = test_case.pop('extract_list', None) # 提取配置（多参数）
for key, value in test_case.items():
if key in params_type: # 只处理 data/json/params
test_case[key] = self.replace_load(value) # 请求参数做${}替换4。处理文件上传接口。file 在 YAML 里是 file: ./data/xxx.xlsx 这种，这里用 rb 模式打开文件，包装成 requests 需要的文件字典。
file, files = test_case.pop('files', None), None
if file is not None:
for fk, fv in file.items():
allure.attach(json.dumps(file), '导入文件')
files = {fk: open(fv, mode='rb')} # 以二进制只读打开文件5.真正的请求在这里发出。**test_case 把剩余请求参数（data/json/params）展开传进去。然后拿到响应和状态码。
res = self.run.run_main(name=api_name, url=url, case_name=case_name,
header=header, method=method, file=files, cookies=cookie, **test_case)
status_code = res.status_code # 状态码
allure.attach(self.allure_attach_response(res.json()), '接口响应信息', ...)6.请求发完后做三件事：提取响应参数 → 断言。提取把下一接口要用的值（token、orderNumber 等）存进 extract.yaml，断言判断本次请求是否符合预期。
res_json = json.loads(res.text) # 响应文本转成字典
if extract is not None:
self.extract_data(extract, res.text) # ★ 提取参数写进 extract.yaml
if extract_list is not None:
self.extract_data_list(extract_list, res.text)
self.asserts.assert_result(validation, res_json, status_code) # ★ 断言字符串切片 + 反射(getattr) replace_load
def replace_load(self, data):
"""yaml数据替换解析"""
str_data = data
if not isinstance(data, str):
str_data = json.dumps(data, ensure_ascii=False)
# print('从yaml文件获取的原始数据：', str_data)
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]
# 取出yaml文件的函数名
func_name = ref_all_params[2:ref_all_params.index("(")]
# 取出函数里面的参数
func_params = ref_all_params[ref_all_params.index("(") + 1:ref_all_params.index(")")]
# 传入替换的参数获取对应的值,类的反射----getattr,setattr,del....
extract_data = getattr(DebugTalk(), func_name)(*func_params.split(',') if func_params else "")
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))
# print('通过解析后替换的数据：', str_data)
# 还原数据
if data and isinstance(data, dict):
data = json.loads(str_data)
else:
data = str_data
return data接口关联的完整闭环：
① 跑前:根 conftest 清空 extract.yaml
② 登录接口(extract: token: $.token)
→ extract_data 用 jsonpath 取出 token → write_yaml_data 写入 extract.yaml
③ 下一个接口请求参数写 token: ${get_extract_data(token)}
→ replace_load 反射 → get_extract_yaml 读 extract.yaml → 拿到真 token
函数的实现debugtalk.py：
def get_extract_data(self, node_name, randoms=None) -> str:
"""
获取extract.yaml数据，首先判断randoms是否为数字类型，如果不是就获取下一个node节点的数据
:param node_name: extract.yaml文件中的key值
:param randoms: int类型，0：随机读取；-1：读取全部，返回字符串形式；-2：读取全部，返回列表形式；其他根据列表索引取值，取第一个值为1，第二个为2，以此类推;
:return:
"""
data = self.read.get_extract_yaml(node_name)
if randoms is not None and bool(re.compile(r'^[-+]?[0-9]*\.?[0-9]+([eE][-+]?[0-9]+)?$').match(randoms)):
randoms = int(randoms)
data_value = {
randoms: self.get_extract_order_data(data, randoms),
0: random.choice(data),
-1: ','.join(data),
-2: ','.join(data).split(','),
}
data = data_value[randoms]
else:
data = self.read.get_extract_yaml(node_name, randoms)
return data
三个动作然后对应三个函数：
| 动作 | 函数 | 时机 |
|---|---|---|
| 写 | write_yaml_data() |
接口响应后,提取时 |
| 读 | get_extract_yaml() |
下个接口请求前,替换时 |
| 清 | clear_yaml_data() |
每次跑测试前 |
extract.yaml是中间的小黑板
四、项目总结
这是一个数据驱动的接口自动化测试框架——YAML 写用例，pytest 收集执行，requests 发请求，extract.yaml 传递接口依赖，allure 出报告，用例和代码分离，新增测试只写 YAML 不改代码，然后也是可以支持集成Jenkins实现自动化执行。
电商企业物流数字化转型必备！快递鸟 API 接口，72 小时快速完成物流系统集成。全流程实战1V1指导，营造开放的API技术生态圈。
更多推荐
12
0- 0
已为社区贡献1条内容
所有评论(0)