前言
在书驿书城项目测试过程中,随着业务迭代,接口数量持续增加。每次版本上线前,都需要对用户注册登录、图书上下架、下单、支付等核心接口进行手工回归测试,重复工作量大,人工测试容易出现漏测,回归测试效率较低。
为解决接口重复回归的痛点,本项目基于 Python+pytest 搭建了一套接口自动化测试框架。框架实现用例数据与脚本代码分离,支持接口间参数关联、接口响应断言以及数据库数据校验;集成 Allure 生成可视化测试报告,并接入 Jenkins 完成持续集成自动化回归。
一、项目介绍
基于 Python+pytest+sqlalchemy+requests+allure+jsonpath+yaml+Jenkins+Linux
被测系统为书驿线上图书商城网站,包含用户注册、登录、下单、图书上架 / 下架、下单支付等相关业务功能;本项目完成该书城接口自动化测试框架搭建,实现请求封装、接口关联、响应与数据库断言。框架支持后续接入 Jenkins+Linux 做持续集成,本文侧重框架搭建设计。
二、项目结构说明
pythonproject ├── base # 底层基础封装层,基础接口能力与辅助工具 │ ├── __init__.py │ ├── apiutil.py # 基础http接口请求封装 │ ├── apiutil_business.py # 业务层接口二次封装 │ ├── generateId.py # 生成各类测试所需ID │ ├── new_testcase_tools.py # 测试用例生成工具 │ ├── new_tools.ui # 简易UI界面文件 │ └── removefile.py # 文件清理工具 ├── common # 公共工具模块,通用能力封装 │ ├── __init__.py │ ├── assertions.py # 自定义断言封装 │ ├── connection.py # 数据库连接封装 │ ├── debugtalk.py # 接口前置后置、自定义函数 │ ├── dingRobot.py # 钉钉机器人消息通知 │ ├── handleExcel.py # Excel文件读写工具 │ ├── operationcsv.py # csv文件操作工具 │ ├── operxml.py # xml文件解析工具 │ ├── Pjenkins.py # Jenkins相关操作封装(预留扩展) │ ├── readyaml.py # yaml文件读取解析工具 │ ├── recordlog.py # 日志记录封装 │ └── semail.py # 邮件发送工具 ├── conf # 全局配置目录 │ ├── __init__.py │ ├── config.ini # 配置参数文件 │ ├── operationConfig.py # 读取配置文件的工具 │ └── setting.py # 全局环境、路径设置 ├── data # 测试数据存放目录 │ ├── __init__.py │ ├── sql # 测试sql脚本 │ ├── login_data.csv │ ├── loginName.yaml # yaml格式测试数据 │ ├── vehicleNo.csv │ └── 测试数据.xls ├── logs # 运行日志目录,存放自动化执行日志 ├── report # 测试报告输出目录 │ ├── allureReport # allure报告生成文件 │ ├── html │ ├── temp │ ├── tmreport # 简易html报告 │ └── results.xml ├── testcase # pytest测试用例目录 │ ├── __init__.py │ ├── conftest.py # pytest全局夹具、全局钩子、参数提取 │ ├── Business interface # 业务场景接口用例 │ ├── ProductManager # 商品管理模块接口用例 │ └── Single interface # 单个独立接口用例 ├── environment.xml # Allure报告环境信息配置 ├── extract.yaml # 接口返回值提取配置文件 ├── pytest.ini # pytest框架主配置文件 ├── requirements.txt # python第三方依赖清单 └── run.py # 项目执行入口,一键执行自动化三、核心代码
1.程序的入口run.py
代码如下:
import shutil import pytest import os import subprocess import webbrowser import sys #新增 from conf.setting import REPORT_TYPE if __name__ == '__main__': if REPORT_TYPE == 'allure': # 改用 python -m pytest 方式启动,解决Windows找不到pytest命令 cmd = [ sys.executable, "-m", "pytest", "-s", "-v", "--alluredir", "./report/temp", "--clean-alluredir", "--junitxml", "./report/results.xml", "./testcase" ] subprocess.run(cmd) env_file = './environment.xml' target_dir = './report/temp' if os.path.exists(env_file): shutil.copy(env_file, target_dir) os.system(f'allure serve {target_dir}') elif REPORT_TYPE == 'tm': pytest.main(['-v', '-s', '--pytest-tmreport-name=testReport.html', '--pytest-tmreport-path=./report/tmreport']) webbrowser.open_new_tab(os.getcwd() + '/report/tmreport/testReport.html')代码说明:
if __name__ == '__main__':- Python 标准写法,表示当脚本作为主程序运行时才执行以下代码块。
- 防止模块被其他脚本导入时意外执行测试逻辑。
基于 subprocess 启动 pytest,生成 Allure 测试原始数据:
if REPORT_TYPE == 'allure': # 改用 python -m pytest 方式启动,解决Windows找不到pytest命令 cmd = [ sys.executable, "-m", "pytest", "-s", "-v", "--alluredir", "./report/temp", "--clean-alluredir", "--junitxml", "./report/results.xml", "./testcase" ] subprocess.run(cmd)参数说明:
| 参数 | 含义 |
|---|---|
| sys.executable | 获取当前项目正在使用的 Python 解释器路径,保证使用虚拟环境里的 Python,而不是系统全局 Python。 |
| -m pytest | 以模块方式运行 pytest,就是python -m pytest,规避环境变量问题。 |
| -s | 输出控制台打印内容。 |
| -v | verbose,详细模式,展示每个用例的执行结果。 |
| --alluredir ./report/temp | 指定 Allure 原始结果数据,输出到report/temp文件夹,Allure 需要这些临时文件来生成可视化报告。 |
| --clean-alluredir | 执行前清空 temp 目录旧数据,防止上次测试残留的报告数据干扰本次报告。 |
| --junitxml ./report/results.xml | 同时生成 junit 格式 xml 测试结果文件,兼容其他 CI 平台读取测试结果。 |
| ./testcase | 指定运行testcase目录下所有测试用例。 |
| subprocess.run(cmd) | 执行上面组装好的命令,等待测试全部运行完成后,再继续执行后面的代码。 |
env_file = './environment.xml' target_dir = './report/temp' if os.path.exists(env_file): shutil.copy(env_file, target_dir)- 判断环境描述文件是否存在,如果存在,就把
environment.xml复制到 Allure 的临时结果文件夹内。 - Allure 生成报告时,会读取这个 xml,在报告页面展示测试环境信息(Python 版本、系统、被测服务版本等)。
os.system(f'allure serve {target_dir}')- 调用系统命令,启动 Allure 的本地服务,自动打开浏览器,渲染并展示 Allure 测试报告。
target_dir就是存放 allure 原始结果的文件夹./report/temp。
2.测试用例testcase
商品管理代码示例:
import allure import pytest from base.generateId import m_id, c_id from base.apiutil import RequestBase from common.readyaml import get_testcase_yaml @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) @allure.story(next(c_id) + "获取商品详情信息") @pytest.mark.run(order=2) @pytest.mark.parametrize('base_info,testcase', get_testcase_yaml('./testcase/ProductManager/productDetail.yaml')) def test_get_product_detail(self, base_info, testcase): allure.dynamic.title(testcase['case_name']) RequestBase().specification_yaml(base_info, testcase) @allure.story(next(c_id) + "提交订单") @pytest.mark.run(order=3) @pytest.mark.parametrize('base_info,testcase', get_testcase_yaml('./testcase/ProductManager/commitOrder.yaml')) def test_commit_order(self, base_info, testcase): allure.dynamic.title(testcase['case_name']) RequestBase().specification_yaml(base_info, testcase) @allure.story(next(c_id) + "订单支付") @pytest.mark.run(order=4) @pytest.mark.parametrize('base_info,testcase', get_testcase_yaml('./testcase/ProductManager/orderPay.yaml')) def test_order_pay(self, base_info, testcase): allure.dynamic.title(testcase['case_name']) RequestBase().specification_yaml(base_info, testcase)代码说明:
@pytest.mark.parametrize参数化装饰器
@pytest.mark.parametrize('base_info,testcase', get_testcase_yaml('./testcase/ProductManager/getProductList.yaml'))作用:
- 实现参数化测试,即一个测试方法可以运行多组不同的输入数据。
每一组 b
ase_info数据都会触发一次完整的测试执行。
参数说明:
- 'base_info,testcase':一般存放公共基础信息和单条测试用例数据
- get_testcase('...'):调用函数读取指定路径下的 YAML 文件内容,返回一个包含多个测试用例的列表。
示例(yaml文件内容):
- 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(goodsId)} page: 1 size: 20 validation: - contains: { 'error_code': '4000' }测试方法定义
def test_get_product_detail(self, base_info, testcase):- 这是一个pytest 测试方法,每个参数化的
case_info都会触发一次该方法的执行。 self表示这是类中的一个实例方法(属于TestEBusinessScenario类)。base_info:第一个参数名,一般存放公共基础信息(比如环境地址、token、公共请求头、数据库基础配置等)testcase:第二个参数名,存放单条测试用例数据(接口 url、请求方法、请求参数、预期结果、断言信息)
动态设置 Allure 报告标题
allure.dynamic.title(testcase['case_name'])作用:
- 在生成的 Allure 报告中,为当前测试用例设置一个可读性更强的标题。
- 标题来自于yaml文件里的api_name字段
解释说明:
如果yaml里面case_name的值是"获取商品列表",那么在报告里显示的就是这个值
api_name: 提交订单接口请求和校验
RequestBase().specification_yaml(base_info, testcase)作用:
- 创建
RequestBase实例,并调用其specification_yaml()方法。 - 读取
base_info获取接口地址、请求方式、请求头 - 读取
testcase获取接口入参、预期校验规则 - 同时会进行响应断言、日志记录等操作,完成整个接口测试流程。
3.接口测试specification_yaml()方法
代码示例
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 especification_yaml方法解析
该方法用于处理 YAML 文件中定义的接口测试用例,包括请求参数构造、动态变量替换、接口调用、响应断言以及数据提取等核心功能。
方法定义
def specification_yaml(self, base_info, test_case): """ 接口请求处理基本方法 :param base_info: yaml文件里面的baseInfo :param test_case: yaml文件里面的testCase :return: """步骤详解
1. 定义常量与基础配置
params_type = ['data', 'json', 'params'] url_host = self.conf.get_section_for_data('api_envi', 'host')params_type:表示请求参数可能包含的类型字段。url_host:从配置文件中获取当前环境的主机地址。
2. 提取接口基本信息并添加 Allure 报告附件
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)- 从
base_info中提取接口名称、URL 和请求方法。 - 使用
allure.attach将这些信息附加到 Allure 报告中,便于测试结果查看。
3. 处理请求头和 Cookie
header = self.replace_load(base_info['header']) allure.attach(api_name, f'请求头:{header}', allure.attachment_type.TEXT) cookie = None if base_info.get('cookies') is not None: cookie = eval(self.replace_load(base_info['cookies']))- 调用
replace_load对 header 进行变量替换。 - 如果存在 cookies,则同样进行替换并使用
eval转换为字典格式。
4. 提取测试用例名称并添加报告附件
case_name = test_case.pop('case_name') allure.attach(api_name, f'测试用例名称:{case_name}', allure.attachment_type.TEXT)- 从
test_case中提取用例名称并删除原始字段。 - 添加到 Allure 报告中。
5. 处理断言逻辑
val = self.replace_load(test_case.get('validation')) test_case['validation'] = val validation = eval(test_case.pop('validation'))- 替换断言表达式中的变量。
- 使用
eval执行断言表达式,生成实际断言规则。
6. 提取数据字段(extract / extract_list)
extract = test_case.pop('extract', None) extract_list = test_case.pop('extract_list', None)- 把 yaml 用例里【需要提取保存的参数规则】拿出来,单独存到变量里,同时把这两个配置项从用例字典移除,避免当成接口请求参数提交。
7. 处理请求参数(data /json/params)
for key, value in test_case.items(): if key in params_type: test_case[key] = self.replace_load(value)- 遍历所有测试用例参数,若为
data、json或params类型,则对其值进行变量替换。
8. 处理文件上传
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')}- 如果存在
files字段,表示是文件上传接口。 - 使用
open(..., mode='rb')读取文件内容,并附加到 Allure 报告中。
9. 发送接口请求
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)- 调用
run_main方法发送请求。 - 获取状态码和响应内容。
- 将响应内容附加到 Allure 报告中。
10. 处理响应与断言
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- 将响应内容转为 JSON 字典。
- 若有
extract或extract_list,则调用对应方法提取数据。 - 使用断言方法对响应结果进行验证。
11. 异常捕获
except Exception as e: raise e- 捕获所有异常并重新抛出,确保测试框架可以正确识别失败情况。
示例说明:
- 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(goodsId)} page: 1 size: 20 validation: - contains: { 'error_code': '4000' }经过specification_yaml处理后:
${get_exact_data(goodsId)}会被替换成实际商品id。- 请求会携带正确的 JSON 数据。。
- 断言会检查是否返回预期结果。
4.断言判断
接口断言代码示例
def assert_response_time(self, res_time, exp_time): """ 通过断言接口的响应时间与期望时间对比,接口响应时间小于预期时间则为通过 :param res_time: 接口的响应时间 :param exp_time: 预期的响应时间 :return: """ try: assert res_time < exp_time return True except Exception as e: logs.error('接口响应时间[%ss]大于预期时间[%ss]' % (res_time, exp_time)) raise方法说明
方法名称
def assert_response_time(self, res_time, exp_time):功能描述
该方法用于校验接口响应性能:对比接口实际响应时间与预期最大时间,若实际耗时小于预期则校验通过;超时则打印错误日志并抛出异常,标记测试失败。
示例场景
测试用户登录接口,要求接口响应不能超过 0.5 秒(exp_time=0.5)
1.调用接口后拿到实际耗时res_time=0.3
- 执行 assert 0.3 < 0.5,条件成立,函数返回 True,测试通过
2.再次执行,网络波动,实际耗时res_time=0.6
assert 0.6 < 0.5断言失败,进入 except:打印错误日志接口响应时间[0.6s]大于预期时间[0.5s],抛出异常,pytest 标记该用例失败
参数说明
| 参数名 | 类型 | 描述 |
|---|---|---|
res_time | float | 接口实际响应时间 |
exp_time | float | 设置的最大允许预期时间 |
返回值说明
返回一个布尔类型
- True表示断言成功
- False表示断言失败
执行逻辑详解
1. 定义函数与文档说明
def assert_response_time(self, res_time, exp_time): """ 通过断言接口的响应时间与期望时间对比,接口响应时间小于预期时间则为通过 :param res_time: 接口的响应时间 :param exp_time: 预期的响应时间 :return: """- 定义断言响应时间的方法,接收 2 个参数:实际响应时间
res_time、预期最大响应时间exp_time。
2.try 代码块执行断言
try: assert res_time < exp_time return Trueassert res_time < exp_time:判断实际响应时间是否小于预期时间
1. 条件成立:无异常,执行return True,函数直接返回,校验通过,不会进入 except 分支
2. 条件不成立:抛出AssertionError断言异常,跳出 try,进入 except 捕获分支
3.except 捕获异常分支
except Exception as e: logs.error('接口响应时间[%ss]大于预期时间[%ss]' % (res_time, exp_time)) raise- 捕获 try 中抛出的异常,打印 error 级日志,输出实际耗时和预期阈值,方便定位问题
raise:重新抛出原异常,将异常向上传递给 pytest 测试框架,标记测试用例失败
四、结语
本书城接口自动化框架依托 pytest 完成全流程接口回归,通过 yaml 管理参数、分层设计提升脚本可维护性,结合自定义断言、数据库校验保障接口质量,接入 Jenkins 实现持续测试。自动化的核心在于解耦与复用,后续会继续补充异常场景用例,进一步提升框架覆盖度。