风铃选品接口自动化项目架构设计

一.项目介绍该项目是一个在线购物的商城网站包括用户注册登录下单上架/下架商品下单支付等相关功能,主要针对该网站搭建了接口自动化框架,封装了相关请求,实现接口关联,结果断言,数据库断言等脚本技术栈PythonpytestpymysqlrequestsallurejsonpathyamlJenkinsMySQL二.项目结构说明2.1 目录结构request_auto_project ├── base/ # 基础层请求核心逻辑与工具 │ ├── apiutil.py # 请求基类 RequestBase数据解析、占位符替换、提取、断言调度 │ ├── generateId.py # Allure 报告目录编号生成器M01_/C01_ │ └── removefile.py # 文件清理工具清理 allure 临时文件 ├── common/ # 公共层通用能力封装 │ ├── assertions.py # 断言模块5 种断言方式 分发入口 │ ├── sendrequest.py # 请求发送模块session 管理、cookie 持久化、异常处理 │ ├── yaml_handler.py # YAML 读写测试用例解析 extract.yaml 关联数据读写 │ ├── debugtalk.py # 动态数据工厂加密、时间戳、关联变量取值 │ ├── connection.py # MySQL 数据库连接封装 │ ├── recordlog.py # 日志模块文件 控制台双输出 │ ├── dingRobot.py # 钉钉机器人通知 │ ├── feishuRobot.py # 飞书机器人通知 │ └── jenkins_handler.py # Jenkins API 封装构建查询、报告统计 ├── config/ # 配置层分层配置管理 │ ├── setting.py # 基础参数路径、日志级别、报告类型、通知开关不随环境变化 │ ├── config_test.yaml # 环境配置接口地址、数据库、Jenkins、机器人密钥随环境变化 │ └── operationConfig.py # YAML 配置读取工具类 ├── testcases/ # 用例层测试用例仅编排不含业务逻辑 │ ├── conftest.py # 会话级 fixture登录、Allure 环境信息、日志打印 │ ├── test_productList.py # 商品管理单接口用例 │ ├── test_user_manager.py # 用户管理单接口用例 │ └── test_business_scenario.py # 业务场景下单支付全流程用例 ├── testdata/ # 数据层YAML 测试数据按业务模块划分 │ ├── LoginManager/ # 登录接口数据 │ ├── ProductManager/ # 商品相关接口数据 │ ├── UserManager/ # 用户相关接口数据 │ └── BusinessManager/ # 业务场景数据 ├── report/ # 报告输出allure 临时数据 results.xml ├── logs/ # 运行日志按日期滚动 ├── extract.yaml # 接口关联数据中转文件运行时动态生成 ├── conftest.py # 根级 conftest清理关联数据、测试结束推送钉钉/飞书通知 ├── pytest.ini # pytest 配置用例发现规则、警告策略 ├── requirements.txt # 依赖清单 └── run.py # 程序入口2.2 分层架构与执行流程框架遵循「配置层 → 数据层 → 用例层 → 执行层 → 报告/通知层」的单向依赖run.py入口 │ ▼ pytest 收集 testcases/ 下的用例 │ ├─ 根 conftest清空 extract.yaml、清理报告临时文件 ├─ testcases/conftest会话级自动登录、写入 Allure 环境信息 │ ▼ 用例方法pytest.mark.parametrize 参数化注入 YAML 数据 │ ▼ RequestBase.specification_yaml() ← 执行核心 ├─ replace_load()解析 ${func()} 占位符反射调用 DebugTalk ├─ SendRequestsession 发送请求自动持久化 cookie ├─ extract_data()jsonpath / 正则提取 → 写入 extract.yaml └─ Assertions.assert_result()按断言类型分发执行 │ ▼ Allure 报告 results.xml 钉钉/飞书通知2.3 配置分层设计配置被刻意拆分为两份以区分「稳定参数」与「环境参数」配置文件内容特点config/setting.py路径、日志级别、超时时间、报告类型、通知开关与运行环境无关代码级常量config/config_test.yaml接口 host、MySQL 账号、Jenkins、机器人密钥随环境变化切换环境只改此文件三、核心代码介绍3.1 程序入口run.pyimport pytest import os import webbrowser from config.setting import REPORT_TYPE, FILE_PATH if __name__ __main__: report_temp FILE_PATH[TEMP] report_tm FILE_PATH[TMR] if REPORT_TYPE allure: pytest.main([ -s, -v, f--alluredir{report_temp}, ./testcases, --clean-alluredir, f--junitxml{FILE_PATH[RESULTXML]}/results.xml #供Jenkins工具解析 ]) os.system(fchcp 65001 allure serve {report_temp}) elif REPORT_TYPE tm: pytest.main([ -vs, --pytest-tmreport-nametestReport.html, f--pytest-tmreport-path{report_tm} ]) webbrowser.open_new_tab(f{report_tm}/testReport.html)代码说明准备报告输出目录report_temp FILE_PATH[TEMP] # report/temp —— Allure 原始数据目录 report_tm FILE_PATH[TMR] # report/tmreport —— tm 风格 HTML 报告目录Allure 报告分支if REPORT_TYPE allure: pytest.main([ -s, -v, f--alluredir{report_temp}, ./testcases, --clean-alluredir, f--junitxml{FILE_PATH[RESULTXML]}/results.xml #供Jenkins工具解析 ]) os.system(fchcp 65001 allure serve {report_temp})参数说明参数作用-s -v不捕获控制台输出 详细模式--alluredir{report_temp}Allure 原始结果数据JSON输出目录./testcases只执行 testcases 目录下的测试用例--clean-alluredir运行前清空旧数据避免历史结果污染报告--junitxmlreport/results.xml额外生成 JUnit 格式 XML供 Jenkins 的测试报告插件解析统计elif REPORT_TYPE tm: pytest.main([ -vs, --pytest-tmreport-nametestReport.html, f--pytest-tmreport-path{report_tm} ]) webbrowser.open_new_tab(f{report_tm}/testReport.html)通过 --pytest-tmreport-name / --pytest-tmreport-path 参数让 pytest-tmreport 插件生成一个名为 testReport.html 的静态离线报告随后用 webbrowser.open_new_tab 自动在浏览器中打开该报告无需本地起服务可直接分发3.2 测试用例层testcases/用例层是框架的一层——不含任何请求与断言逻辑只负责三件事 读YAML、参数化、交给RequestBaseallure.feature(next(m_id) 商品管理单接口) class TestProductManager: allure.story(next(c_id) 获取商品列表) pytest.mark.run(order1) pytest.mark.parametrize(base_info,testcase, get_testcase_yaml(./testdata/ProductManager/getProductList.yaml)) def test_get_product_list(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(./testdata/ProductManager/getProductList.yaml))实现参数化get_testcase_yaml()将 YAML 中的baseInfo与每条testCase组合成[[baseInfo, testCase], ...]一个 YAML 中的多条用例自动展开为多条独立测试 且单条失败不影响其他用例参数说明base_info接口基本信息接口名、URL、请求方法、请求头、cookiestestcase用例内容用例名、请求参数、断言规则、提取规则其他代码设计执行顺序控制pytest.mark.run(order1)保证业务场景按列表 → 详情 → 下单 → 支付顺序执行Allure 编号生成器generateId.py用生成器按序产出M01_、C01_前缀 使报告目录顺序与用例执行顺序一致会话前置testcases/conftest.pysystem_login会话开始先调用登录接口登录失败直接exit()终止整个会话allure_environment动态生成environment.properties报告首页展示 Python 版本、 操作系统、被测环境等信息3.3 测试数据规范YAML 数据驱动示例数据YAML 文件内容- baseInfo: # 接口基本信息同一接口的多条用例共享 api_name: 提交订单 url: /coupApply/cms/placeAnOrder # 相对路径自动拼接配置中的 host method: post header: Content-Type: application/json;charsetUTF-8 testCase: # 用例列表一个接口可挂多条用例 - case_name: 提交订单 json: goods_id: ${get_extract_data(goodsId,0)} # 占位符引用上游提取的数据 number: 2 consignee_info: {name: 张三,phone: 13800000000} validation: # 断言配置可叠加多种断言 - eq: { message: 提交订单成功 } - eq: { error_code: 0000 } extract: # 响应数据提取写入 extract.yaml orderNumber: $.orderNumber userId: $.userId字段约定字段说明params/data/json三种请求参数形态分别对应 GET 查询、表单、JSON bodyvalidation断言列表支持contains、eq、ne、rv、db五种extract提取单值支持 jsonpath$.xxx与正则(.*?)extract_list提取多值列表如$.goodsList[*].goodsId${func(params)}占位符运行时反射调用DebugTalk中的同名方法3.4 请求核心specification_yaml()方法specification_yaml()是整个框架的执行接口请求的核心该方法用于处理 YAML 文件中定义的接口测试用例包括请求参数构造、动态变量替换、接口调用、响应断言以及数据提取等核心功能def specification_yaml(self, base_info, test_case): 接口请求处理主流程解析参数 → 发送请求 → 提取数据 → 执行断言 :param base_info: yaml 文件中的 baseInfo 段 :param test_case: yaml 文件中的 testCase 段 params_type [data, json, params] # 解析接口基本信息 url_host self.conf.get_section_for_data(api_envi, host) api_name base_info[api_name] url url_host base_info[url] method base_info[method] # 记录请求信息到 allure 报告 allure.attach(f接口名称{api_name}, 接口名称, allure.attachment_type.TEXT) allure.attach(f接口地址{url}, 接口地址, allure.attachment_type.TEXT) allure.attach(f请求方法{method}, 请求方法, allure.attachment_type.TEXT) header self.replace_load(base_info[header]) allure.attach(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(f测试用例名称{case_name}, 测试用例名称, allure.attachment_type.TEXT) # 处理断言和参数提取配置 validation eval(self.replace_load(test_case.pop(validation, None))) 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) # 处理文件上传 files None file test_case.pop(files, None) if file is not None: for fk, fv in file.items(): allure.attach(json.dumps(file), 导入文件) files {fk: open(fv, moderb)} # 发送请求 res self.run.run_main(nameapi_name, urlurl, case_namecase_name, headerheader, methodmethod, filefiles, cookiescookie, **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) 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: logs.error(系统异常或接口未请求) raise except Exception as e: logs.error(f断言或提取异常: {e}) raise代码说明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] url url_host base_info[url] method base_info[method] # 记录请求信息到 allure 报告 allure.attach(f接口名称{api_name}, 接口名称, allure.attachment_type.TEXT) allure.attach(f接口地址{url}, 接口地址, allure.attachment_type.TEXT) allure.attach(f请求方法{method}, 请求方法, allure.attachment_type.TEXT)从base_info中提取接口名称、URL 和请求方法使用allure.attach将这些信息附加到 Allure 报告中便于测试结果查看3. 处理请求头和 Cookieheader self.replace_load(base_info[header]) allure.attach(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)调用热加载方法replace_load对 header 进行变量动态替换如果存在 cookies则同样进行替换并使用eval转换为字典类型4. 提取测试用例名称并添加报告附件case_name test_case.pop(case_name) allure.attach(f测试用例名称{case_name}, 测试用例名称, allure.attachment_type.TEXT)从test_case中提取用例名称并删除原始字段添加到 Allure 报告中5. 处理断言逻辑validation eval(self.replace_load(test_case.pop(validation, None)))替换断言表达式中的变量并使用eval执行断言表达式生成实际断言规则6. 提取数据字段extract / extract_listextract test_case.pop(extract, None) extract_list test_case.pop(extract_list, None)从test_case中提取需要提取的字段名供后续从响应中提取数据7. 处理请求参数data /json/paramsfor key, value in test_case.items(): if key in params_type: test_case[key] self.replace_load(value)遍历所有测试用例参数若为data、json或params类型则对其值进行变量替换8. 处理文件上传files None file test_case.pop(files, None) if file is not None: for fk, fv in file.items(): allure.attach(json.dumps(file), 导入文件) files {fk: open(fv, moderb)}如果存在files字段表示是文件上传接口使用open(..., moderb)读取文件内容并附加到 Allure 报告中9. 发送接口请求res self.run.run_main(nameapi_name, urlurl, case_namecase_name, headerheader, methodmethod, filefiles, cookiescookie, **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) 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: logs.error(系统异常或接口未请求) raise except Exception as e: logs.error(f断言或提取异常: {e}) raise将响应内容转为 JSON 字典若有extract或extract_list则调用对应方法提取数据使用断言方法对响应结果进行验证关键机制热加载replace_load()将 YAML 中的${func_name(params)}解析为函数名与参数通过getattr(DebugTalk(), func_name)(*args)反射调用完成动态替换。DebugTalk提供了三类能力加密类md5_encryption、sha1_encryption、base64_encryption时间类各精度时间戳、当天/本月起止时间、指定偏移日期时间等关联类get_extract_data(node, randoms)从extract.yaml取值randoms支持按序取值1第一个、随机取值0、全部拼接-1、返回列表-2四种模式接口关联机制extract.yaml运行时通过extract_data()将提取结果通过YamlHandler.write_yaml_data()写入extract.yaml每次会话开始时由根conftest.py的clear_extract清空保证数据不跨轮次污染登录返回的 cookie 也由SendRequest自动捕获并持久化到该文件3.5 断言模块Assertionsassertions.py实现了五种断言方式 字典分发的设计import allure import jsonpath import operator from common.recordlog import logs from common.connection import ConnectMysql class Assertions: 接口断言模块支持以下断言模式 1) 响应文本字符串包含断言 2) 响应结果相等 / 不相等断言 3) 响应结果任意值断言 4) 响应时间断言 5) 数据库断言 def contains_assert(self, value, response, status_code): 字符串包含断言断言预期结果的字符串是否包含在接口响应中 :param value: 预期结果yaml 文件中的断言配置 :param response: 接口实际响应 :param status_code: 响应状态码 :return: 0 表示通过非 0 表示失败 flag 0 for assert_key, assert_value in value.items(): if assert_key status_code: # 状态码断言 if assert_value ! status_code: flag 1 allure.attach(f预期结果{assert_value}\n实际结果{status_code}, 响应代码断言结果:失败, attachment_typeallure.attachment_type.TEXT) logs.error(fcontains 断言失败接口返回码【{status_code}】不等于【{assert_value}】) else: # 通过 jsonpath 提取响应中的字段值注意参数顺序为表达式在前数据在后 resp_list jsonpath.search(f$..{assert_key}, response) if isinstance(resp_list[0], str): resp_list .join(resp_list) if resp_list: # 预期值为 NONE 时视为 None assert_value None if assert_value.upper() NONE else assert_value if assert_value in resp_list: logs.info(f字符串包含断言成功预期结果【{assert_value}】,实际结果【{resp_list}】) else: flag 1 allure.attach(f预期结果{assert_value}\n实际结果{resp_list}, 响应文本断言结果失败, attachment_typeallure.attachment_type.TEXT) logs.error(f响应文本断言失败预期结果为【{assert_value}】,实际结果为【{resp_list}】) return flag def _compare_dict_assert(self, expected, actual, op_func, op_name): 字典比较断言内部方法提取相等/不相等断言的公共逻辑 :param expected: 预期结果字典 :param actual: 实际结果字典 :param op_func: 比较函数operator.eq 或 operator.ne :param op_name: 操作名称相等 或 不相等 :return: 0 表示通过非 0 表示失败 if not (isinstance(expected, dict) and isinstance(actual, dict)): raise TypeError(f{op_name}断言 -- 预期结果和实际响应结果必须为字典类型) # 取第一个共同 key构造实际结果子集进行比较 common_key list(expected.keys() actual.keys())[0] new_actual {common_key: actual[common_key]} result op_func(new_actual, expected) if result: logs.info(f{op_name}断言成功实际结果{new_actual}预期结果{expected}) allure.attach(f预期结果{expected}\n实际结果{new_actual}, f{op_name}断言结果成功, attachment_typeallure.attachment_type.TEXT) else: allure.attach(f预期结果{expected}\n实际结果{new_actual}, f{op_name}断言结果失败, attachment_typeallure.attachment_type.TEXT) logs.error(f{op_name}断言失败实际结果{new_actual}预期结果{expected}) return 1 return 0 def equal_assert(self, expected_results, actual_results): 相等断言 return self._compare_dict_assert(expected_results, actual_results, operator.eq, 相等) def not_equal_assert(self, expected_results, actual_results): 不相等断言 return self._compare_dict_assert(actual_results, expected_results, operator.ne, 不相等) def assert_response_any(self, actual_results, expected_results): 断言响应 body 中任意字段值是否匹配 :param actual_results: 接口实际响应 :param expected_results: 预期结果单键值对 :return: 0 表示通过非 0 表示失败 try: exp_key list(expected_results.keys())[0] exp_value list(expected_results.values())[0] if exp_key in actual_results and actual_results[exp_key] exp_value: logs.info(响应结果任意值断言成功) return 0 else: logs.error(f响应结果任意值断言失败预期 {exp_key}{exp_value}) return 1 except Exception as e: logs.error(f响应结果任意值断言异常: {e}) raise def assert_response_time(self, res_time, exp_time): 断言接口响应时间是否小于预期时间 :param res_time: 实际响应时间秒 :param exp_time: 预期响应时间秒 try: assert res_time exp_time return True except AssertionError: logs.error(f接口响应时间[{res_time}s]大于预期时间[{exp_time}s]) raise def assert_mysql_data(self, expected_results): 数据库断言执行 SQL 查询有数据则通过 :param expected_results: SQL 语句 :return: 0 表示通过非 0 表示失败 conn ConnectMysql() db_value conn.query_all(expected_results) if db_value is not None: logs.info(数据库断言成功) return 0 else: logs.error(数据库断言失败请检查数据库是否存在该数据) return 1 # 断言类型 → 处理方法的映射 ASSERT_DISPATCH { contains: contains_assert, eq: equal_assert, ne: not_equal_assert, rv: assert_response_any, db: assert_mysql_data, } def assert_result(self, expected, response, status_code): 断言总入口根据 yaml 中的断言类型分发到对应的断言方法 :param expected: 预期结果包含断言类型和断言值的列表 :param response: 实际响应结果 :param status_code: 响应状态码 all_flag 0 try: logs.info(fyaml 文件预期结果{expected}) for yq in expected: for key, value in yq.items(): method_name self.ASSERT_DISPATCH.get(key) if method_name is None: logs.error(f不支持的断言方式: {key}) continue method getattr(self, method_name) # contains 需要额外传 response 和 status_code if key contains: all_flag method(value, response, status_code) elif key in (eq, ne): all_flag method(value, response) else: all_flag method(response, value) except Exception: logs.error(接口断言异常请检查 yaml 预期结果值是否正确填写!) raise assert all_flag 0, 测试失败断言键说明断言键方法说明containscontains_assert通过 jsonpath 定位字段断言响应文本包含预期值支持状态码断言eqequal_assert响应字段与预期字典相等nenot_equal_assert响应字段与预期字典不相等rvassert_response_any响应 body 中任意字段值匹配dbassert_mysql_data执行 SQL查询到数据即通过数据库层验证总入口assert_result()通过ASSERT_DISPATCH映射表将断言类型分发到对应方法 累计失败计数all_flag后统一assert all_flag 0即一条用例内所有断言都会执行完 不会因第一个断言失败而跳过后续断言相等/不相等断言通过_compare_dict_assert()提取公共逻辑仅传入operator.eq / operator.ne区分行为避免重复代码3.6 通知机制结果通知根conftest.py的pytest_terminal_summary钩子测试结束→收集 总数/通过/失败/错误/跳过/耗时→JENKINS_ENHANCE 开启时追加构建编号、构建地址、Allure 报告链接查询失败自动降级为普通通知不影响流程→按 setting.py 开关推送钉钉/飞书通知与邮件的职责边界清晰钉钉/飞书消息由框架代码推送附带 Jenkins 构建链接 点击直达报告邮件通知完全交给 Jenkins 邮件插件框架本身不含任何邮件逻辑3.7 Jenkins 集成jenkins_handler.py 基于 python-jenkins 封装查询最新构建号、构建状态、测试报告统计通过/失败/跳过/时长从控制台日志正则提取 Allure 报告链接配合 Jenkins 侧的 Allure 插件与 results.xml即可实现定时构建 → Allure 报告 → 邮件/群通知的完整 CI 闭环四、测试结果4.1 Allure 报告4.2 邮箱通知4.3 飞书/钉钉通知五、项目结语本项目基于 pytest Allure 构建了一套完整、稳定、可扩展的接口自动化测试框架适用于电商系统的业务流程测试。通过数据驱动、变量替换、断言机制、报告可视化等功能有效提高了测试效率与质量。未来将持续优化提升其通用性与智能化水平