# Interface **Repository Path**: yunyoko/interface ## Basic Information - **Project Name**: Interface - **Description**: 接口自动化框架：Python+Pytest+Yaml-CSV+Allure+Log+Mysql+Mock - **Primary Language**: Python - **License**: MulanPSL-2.0 - **Default Branch**: master - **Homepage**: None - **GVP Project**: No ## Statistics - **Stars**: 0 - **Forks**: 7 - **Created**: 2026-06-23 - **Last Updated**: 2026-06-23 ## Categories & Tags **Categories**: Uncategorized **Tags**: None ## README # Interface 自动化测试框架 —— 运行原理与用例编写指南 本文档面向框架使用者和维护者，深入讲解框架的运行机制、用例编写方式以及不同模式下的执行流程。 - 阅读对象：测试工程师、自动化框架维护者 - 前置知识：Python 基础、Pytest 基础、YAML 语法、HTTP 协议 --- ## 三类读者入口 这份文档同时服务 3 类读者。为了减少“明明有内容，却不知道该先看哪”的成本，建议先按身份选入口： | 你当前的角色 | 建议先读 | 先解决什么问题 | 暂时可以先跳过 | |------|------|------|------| | 第一次接触框架的新人 | `环境准备` → `§2.0 5 分钟快速入门` → `附录：快速查阅卡` | 先把环境跑通，知道最小 YAML 和驱动文件怎么写 | 架构图、设计决策、异常体系 | | 日常主要写 YAML 用例的人 | `§2.0.3 YAML 编写格式与场景选择` → `§2.1 ~ §2.7` → `附录：快速查阅卡` | 知道当前场景该用单接口、steps、CSV、Mock、SQL、Redis 哪种写法 | 底层连接池和 Hook 细节 | | 需要改框架源码的维护者 | `项目目录结构` → `维护者源码入口` → `§一 运行机制` → `§三 核心组件交互关系` | 建立代码地图，知道改哪里、为什么这样设计 | 新手模板、基础用例示例 | ## 一页速览 如果你现在只想快速找到“我该看哪里”，先看这张表： | 当前目标 | 优先阅读 | 对应核心源码 | |------|------|------| | 5 分钟跑通一个最小用例 | [§2.0 5 分钟快速入门](#20-5分钟快速入门写出第一个用例) | `run.py` / `conftest.py` / `util_tools/execution/executor.py` | | 编写多步骤业务链路 | [§2.2 多步骤链路模式](#22-模式二多步骤链路模式steps--teardown_steps) | `util_tools/execution/executor.py` / `util_tools/execution/step_runner.py` | | 使用 Redis 共享变量 | [§2.4 跨文件 Redis 共享](#24-增强能力一跨文件-redis-共享extract--get_extract_data) | `util_tools/variables/runtime_cache.py` / `util_tools/variables/variable_resolver.py` | | 写 DB 断言或前后置 SQL | [§4.6 YAML 编写规范](#46-yaml-编写规范) + [Q3 数据库断言排查](#q3数据库断言报sql-安全拦截) | `util_tools/assertions/db_assertions.py` / `util_tools/execution/sql_executor.py` | | 使用 Mock / HAR 回放 | [§2.5 Mock 挡板](#25-增强能力二mock-挡板) + [§2.6 HAR 流量录制回放](#26-增强能力三har-流量录制回放) | `util_tools/extensions/mock_response.py` / `util_tools/generators/har_to_yaml.py` | | 定位 headers/cookies/签名/变量问题 | [附录：快速查阅卡](#附录快速查阅卡) + [§十 常见问题排查指南](#十常见问题排查指南) | `util_tools/execution/request_builder.py` / `common/runtime_functions.py` | | 维护框架核心流程 | [§一 运行机制](#一框架整体运行机制) + [§三 组件交互关系](#三核心组件交互关系) | `util_tools/execution/step_runner.py` / `util_tools/clients/http_client.py` | | 接入 Jenkins / 通知推送 | [§六 通知推送机制](#六通知推送机制) + [§九 Jenkins / CI 集成](#九jenkins--ci-集成) | `util_tools/reporting/notification_sender.py` / `util_tools/pytest_hooks/session_lifecycle_hook.py` | --- ## 目录 **开篇** - [如何阅读本文档](#如何阅读本文档) - [框架特性总览](#框架特性总览) - [项目目录结构](#项目目录结构) **正文** 1. [框架整体运行机制](#一框架整体运行机制) 2. [测试用例的几种编写方式](#二测试用例的几种编写方式) 3. [核心组件交互关系](#三核心组件交互关系) 4. [关键设计决策说明](#四关键设计决策说明) 5. [extensions 扩展层详解](#五extensions-扩展层详解) 6. [通知推送机制](#六通知推送机制) 7. [CLI 工具使用](#七cli-工具使用) 8. [自定义异常体系](#八自定义异常体系) 9. [Jenkins / CI 集成](#九jenkins--ci-集成) 10. [常见问题排查指南](#十常见问题排查指南) **附录** - [附录：快速查阅卡](#附录快速查阅卡) - [附录二：图集速查](#附录二图集速查) --- ## 如何阅读本文档 本文档篇幅较长，以下推荐三条阅读路径，你可以按照自身角色选择： ### 🌱 新手上路（第一次接触本框架） 1. 先读[框架特性总览](#框架特性总览)，30 秒了解框架卖点 2. 跟着[§2.0 5 分钟快速入门](#20-5分钟快速入门写出第一个用例)跑通最小闭环 3. 浏览[§2.1–§2.7 六种编写模式](#二测试用例的几种编写方式)，挑一种最贴近你业务的开始练手 4. 遇到具体问题去[§十、常见问题排查指南](#十常见问题排查指南) ### 🧰 老手深入（需要写复杂用例） 1. 直接读[§二 测试用例的几种编写方式](#二测试用例的几种编写方式)中的多步骤链路、CSV 数据驱动、Mock 等模式 2. 关注[§五 extensions 扩展层详解](#五extensions-扩展层详解)和[附录：快速查阅卡](#附录快速查阅卡) 3. 工程集成看[§九 Jenkins / CI 集成](#九jenkins--ci-集成) ### 🛠 维护者剖析（需要改框架源码） 1. 先看[项目目录结构](#项目目录结构)建立心智地图 2. 读[§一 框架整体运行机制](#一框架整体运行机制) → [§三 核心组件交互关系](#三核心组件交互关系) → [§四 关键设计决策说明](#四关键设计决策说明)的完整链路 3. 配合[附录二·图集速查](#附录二图集速查)一起看，文字 + 图能更快建立全局视图 ### 按任务选择最短阅读路径 如果你不是从头学框架，而是“带着具体任务来找答案”，可以直接按下面跳： | 你现在要做什么 | 最短阅读路径 | 读完后能解决什么 | |------|------|------| | 新写一个最小可运行接口用例 | `§2.0 5 分钟快速入门` → `附录：快速查阅卡` | 知道最小 YAML 结构、怎么运行、怎么校验 | | 写多步骤业务链路 | `§2.2 多步骤链路模式` → `§4.6 YAML 编写规范` | 知道 `steps`、变量共享、`teardown_steps` 怎么配 | | 接口依赖数据库准备和清理 | `§4.6 YAML 编写规范` → `Q3 数据库断言排查` | 知道 `setup_sql` / `teardown_sql` 生命周期和限制 | | 做页面接口 / Cookie 接口 | `附录：快速查阅卡` → `§2.6 HAR 流量录制回放` | 知道 `request.cookies` 怎么写、环境 Cookie 放哪 | | 排查变量没替换 / 签名不对 | `附录：快速查阅卡` → `§十 常见问题排查指南` | 知道 `${变量}`、`${get_extract_data()}`、签名函数的边界 | | 调并发、重试、熔断或 CI | `§一 运行机制` → `pytest.ini 配置详解` → `§九 Jenkins / CI 集成` | 知道默认参数、重跑和熔断的真实执行口径 | ### 新人首日建议 如果你今天第一次接手这个框架，推荐按下面顺序做，不要一上来就从头读完整份 README： 1. 先完成[环境准备](#环境准备运行前必读)，确保 Redis、Allure、环境配置能跑通。 2. 优先执行项目自带演示：[§2.0.2 框架能力拆解演示](#202-框架能力拆解演示)，因为这组用例主要依赖 Mock，成功率最高，适合先建立“这个框架怎么跑”的直觉。 3. 再看[§2.0 5 分钟快速入门](#20-5分钟快速入门写出第一个用例)，自己新建一个最小 YAML 和驱动文件。 4. 最后按你手头业务场景，进入[§2.0.3 YAML 编写格式与场景选择](#203-yaml-编写格式与场景选择)选写法。 --- ## 前言 简单讲述框架的使用 本框架主要基于：Python+Pytest+Yaml+CSV+Redis+Allure+Log+Mysql+Mock 实现接口自动化框架 Git地址：[https://gitee.com/make_a_summer/interface.git](https://gitee.com/make_a_summer/interface.git) 项目作者：唐松（挽一夏） > 沟通V：Tans-m 进沟通群 该框架已在实际接口自动化场景中验证使用。已覆盖500个接口。 Json/Yaml转换网站：esjson.com 偶然看见的一套学习视频，值得观看：[【软件测试接口自动化】 ](https://www.bilibili.com/video/BV1Um411y7mW/?share_source=copy_web&vd_source=548f70c25aefed2ddecfdc2796827683) 接口文档： > 微信公众号-标签管理： > > [https://developers.weixin.qq.com/doc/offiaccount/Basic_Information/Get_access_token.html](https://gitee.com/link?target=https%3A%2F%2Fdevelopers.weixin.qq.com%2Fdoc%2Foffiaccount%2FBasic_Information%2FGet_access_token.html) --- ## 环境准备（运行前必读） 在深入阅读本文档之前，请确保你的环境已就绪。以下五步是运行框架的最低要求： ### 1. 安装 Python 依赖 ```bash pip install -r requirements.txt ``` ### 2. 安装并启动 Redis 框架使用 Redis 作为跨进程变量共享的介质，**必须提前启动**。 #### 2.1 安装 Redis **macOS:** ```bash brew install redis brew services start redis ``` **Linux (Ubuntu/Debian):** ```bash sudo apt-get update sudo apt-get install redis-server sudo systemctl start redis-server sudo systemctl enable redis-server # 设置开机自启 ``` **Linux (CentOS/RHEL):** ```bash sudo yum install epel-release sudo yum install redis sudo systemctl start redis sudo systemctl enable redis ``` **Windows:** 1. 下载 Redis for Windows: https://github.com/tporadowski/redis/releases 2. 解压到任意目录（如 `C:\Redis`） 3. 双击运行 `redis-server.exe`，或命令行执行 `redis-server.exe redis.windows.conf` 4. 建议安装为系统服务（管理员CMD）：`redis-server.exe --service-install redis.windows.conf --loglevel verbose` #### 2.2 验证 Redis ```bash redis-cli ping # 预期输出: PONG ``` 若未安装 Redis，也可参考 [Redis 官方文档](https://redis.io/docs/getting-started/installation/) 进行安装。 ### 3. 安装 Allure Commandline（报告生成必需） `run.py` 最后会调用 `allure generate` 渲染 HTML 报告，因此需要提前安装 Allure Commandline。 **macOS:** ```bash brew install allure ``` **Linux:** ```bash # 方式1：手动下载安装 wget https://github.com/allure-framework/allure2/releases/download/2.32.0/allure-2.32.0.tgz tar -xzf allure-2.32.0.tgz -C /opt/ sudo ln -s /opt/allure-2.32.0/bin/allure /usr/local/bin/allure # 方式2：若已安装 npm npm install -g allure-commandline ``` **Windows:** 1. 下载 Allure 压缩包：https://github.com/allure-framework/allure2/releases 2. 解压到 `C:\Program Files\allure-2.x.x` 3. 将解压后的 `bin` 目录添加到系统 `PATH` 环境变量 4. 重启终端，执行 `allure --version` 验证 验证安装： ```bash allure --version # 预期输出: 2.x.x ``` ### 4. 创建环境配置文件 框架通过 `AUTO_ENV` 环境变量加载环境配置。首次运行时，建议复制一份现有模板并修改为你的测试环境： ```bash cp configs/envs/prod.yaml configs/envs/test.yaml ``` 然后编辑 `configs/envs/test.yaml`，填入你的测试环境域名： ```yaml host: "http://your-api-domain.com" base_url: base_wms_url: "http://your-wms-domain.com" redis: host: "127.0.0.1" port: 6379 db: 0 password: "" ``` ### 5. 验证安装 执行以下命令验证环境是否就绪： ```bash # Windows PowerShell $env:AUTO_ENV="test.yaml" python run.py # Linux / Mac export AUTO_ENV=test.yaml python run.py ``` 首次运行建议按以下顺序排查： 1. **Python依赖检查**：如果报 `ModuleNotFoundError`，说明 `pip install -r requirements.txt` 未执行成功，重新安装依赖 2. **Redis连接检查**：如果报 Redis 连接错误，确认 `redis-cli ping` 返回 `PONG`，且环境配置文件中的 `redis.host` 填写正确 3. **Allure命令检查**：如果报 `找不到 allure 命令`，说明 Allure Commandline 未安装或未加入 PATH，返回步骤3检查 4. **环境配置检查**：如果报 `ConfigException` 或找不到环境文件，确认 `AUTO_ENV` 指向的文件存在于 `configs/envs/` 目录下 如果终端输出 `[+] [Allure ] 报告生成成功: reports/allures/index.html`，说明 pytest 已经正常跑完一轮，且本机 Allure CLI 可用。是否“用例全部通过”仍要以 pytest 最终结果为准。 --- ## 框架特性总览 本框架是一套面向接口自动化测试的工程化解决方案，核心栈为 **Python + Pytest + YAML + CSV + Redis + Mock + Allure + MySQL**。设计目标是让测试工程师**不写代码或只写极少量代码**，就能完成复杂接口链路的自动化测试。 ### 1. 数据驱动引擎（YAML / CSV 双轨制） - **YAML 原生多组数据**：单个 YAML 文件可直接编写多组独立用例，框架通过 `read_yaml_cases` 自动解析为 Pytest 参数化用例。 - **YAML + CSV 笛卡尔积裂变**：通过 `parameters` 节点绑定 CSV 文件，实现单接口多组数据的自动化裂变。支持多 CSV 源的笛卡尔积组合，同时通过 `ast.literal_eval`（AST，Abstract Syntax Tree，抽象语法树）保证数字、布尔、字典等类型的 100% 保真。详见 [§2.3 CSV 数据驱动模式](#23-模式三csv-数据驱动模式parameters)。 - **参考样例**：`testcase/框架能力拆解演示/02_yaml_csv_driver.yaml` ### 2. 多步骤业务链路编排（steps + teardown_steps） - **steps 主链路**：支持将多个接口按顺序编排为一条完整业务链路，步骤间通过 `extract` / `extract_local` 共享变量，实现"创建 -> 查询 -> 修改 -> 删除"的闭环测试。详见 [§2.2 多步骤链路模式](#22-模式二多步骤链路模式steps--teardown_steps)。 - **teardown_steps 环境兜底**：无论主链路成功或失败，框架都会强制执行 teardown 清理，防止脏数据污染测试环境，且 teardown 报错不会掩盖主链路的真实错误。 - **参考样例**：`testcase/框架能力拆解演示/04_teardown_steps_after_case.yaml` ### 3. 双级变量池体系（Redis 全局 + 内存局部） - **Redis 全局变量池（extract / extract_list）**：支持跨 YAML 文件、跨进程的变量共享，天然兼容 `pytest-xdist` 多进程并发。底层采用 Redis Pipeline 批量写入，降低网络 RTT（Round-Trip Time，往返时延）开销。详见 [§2.4 跨文件 Redis 共享](#24-增强能力一跨文件-redis-共享extract--get_extract_data)。 - **内存局部变量池（extract_local / extract_list_local）**：变量仅存在于当前用例生命周期内，适合临时中间值，避免污染全局缓存。详见 [§2.2 多步骤链路模式](#22-模式二多步骤链路模式steps--teardown_steps)。 - **请求体逆向拦截（$.request.）**：支持从已发出的请求头、请求体、查询参数中逆向提取数据（如 TraceId、Nonce、签名参数），解决"发送了但响应没回显"的提取难题。 - **参考样例**：`testcase/框架能力拆解演示/01_memory_and_redis_variables.yaml` ### 4. 软断言与多模态断言体系 - **软断言（Soft Assert）**：原生 `assert` 是"一断就停"，本框架采用聚合报错模式，遍历完所有断言规则后一次性抛出完整错误清单，避免"修一个报一个"的低效排障。设计原理详见 [§4.3 软断言 vs 硬断言](#43-软断言-vs-硬断言)。 - **断言类型全覆盖**： - `code`：HTTP 状态码严格比对 - `eq`：字典严格全等（仅校验最外层指定字段） - `ne`：反向不等（常用于验证修改生效） - `contain`：深度包含匹配，支持 JSONPath 精确路径（`$.data.status`）和全层级模糊扫描（`status`） - `db`：数据库落库校验，支持 `eq` / `contain` / `empty` 三种模式 - **SQL 安全拦截**：数据库断言内置 DML/DDL 黑名单（insert/update/delete/drop/truncate/alter/create），仅允许 `SELECT` 查询穿透，防止测试框架误改生产数据。 ### 5. 失败重试与智能熔断 - **单步轮询重试（retry）**：针对微服务异步延迟和网络抖动，单步接口支持配置 `times` + `interval` 轮询重试。`times` 表示最大执行次数（包含首次执行），`interval` 表示每次失败后再次尝试前的等待秒数。网络异常和业务断言失败均会触发重试，耗尽后正式抛错。重试过程会完整保留在控制台日志中，Allure 报告只保留最终成功或最终失败的那一次请求现场，避免报告中出现多套重复附件。 - **用例级熔断机制（Circuit Breaker）**：基于 Redis 维护的熔断黑名单，当前置依赖用例失败时，后续关联用例自动跳过（Skip），并在 Allure 中标记"已熔断"。用例重试成功后自动"自愈"清除熔断标记。 - **瞬态故障豁免**：配合 `pytest-rerunfailures` 使用时，非末轮的失败被识别为"瞬态抖动"，不会触发熔断，保障重试管线的自我恢复能力。 ### 6. Mock 接口与流量仿真 - **请求拦截网关**：在 YAML 中配置 `mock` 节点后，框架会在真实 HTTP 请求发出前进行拦截，直接返回伪造的 Response（可自定义 status_code、json、headers 等）。详见 [§2.5 Mock 挡板](#25-增强能力二mock-挡板)。 - **HAR 流量录制回放**：通过 `cli.py record` 命令，可将浏览器抓包的 HAR 文件自动转换为 YAML 用例和 pytest 驱动文件；转换时会自动过滤静态资源、前端杂请求和非业务域名，并按 WMS/OMS 域名写入对应 Cookie 配置，实现"录制一次，回放 N 次"的回归测试。 ### 7. 高性能与工程化保障 - **orjson 极速解析**：响应体 JSON 解析、CSV 反序列化、HAR 解析、断言附件序列化等高频链路使用 `orjson`，在大型 WMS 仓储报文场景下提升解析效率。 - **多进程日志隔离**：基于 PID 隔离日志文件（`logs/日期/时间_pid{pid}.log`），配合 `propagate=False` 彻底避免 `pytest-xdist` 并发下的日志交错与重复打印。 - **大响应体截断保护**：当接口返回超大 HTML 报错页或巨型 JSON 时，控制台日志自动截断（>500 字符），Allure 附件仍保留完整数据，防止 Jenkins 控制台卡死。 - **连接池优化**：HTTP Session 全局单例 + Keep-Alive 长连接；MySQL 会按本轮用例内容自动预热连接池，并保留懒加载兜底；Redis 连接池上限扩至 50，并开启 TCP KeepAlive 与 health_check。 ### 8. Allure 深度集成与可观测性 - **步骤分层可视化**：多步骤用例在 Allure 报告中以折叠面板形式清晰展示"步骤1、步骤2"，teardown 步骤带独立图标区分。 - **请求溯源附件**：每个接口的请求 URL、Headers、Cookies、Payload、Response 均被组装为表格附件挂载到报告中。若当前 step 配置了 `retry`，中间失败尝试只保留控制台日志，Allure 只展示最终一次请求、响应、提取和断言附件。 - **环境信息自动注入**：测试结束后自动生成 `environment.properties`，包含 Python 版本、Git 分支、Jenkins 构建号、测试环境名等上下文。 ### 9. 动态函数扩展库（RuntimeFunctions） - **热调用机制**：YAML 中可直接通过 `${函数名(参数)}` 调用 Python 函数，无需编写额外代码。 - **内置函数覆盖**：毫秒时间戳、随机 Nonce、UUID 单号生成、环境配置读取、Redis 变量读取与格式化、HMAC-SHA256 签名生成等。 - **自定义扩展**：在 `common/runtime_functions.py` 中新增方法后，YAML 即可实时反射调用，框架会自动对幂等函数启用 LRU 缓存，对非幂等函数（如时间戳）强制走实时计算。 ### 10. 统计通知与 CI/CD 集成 - **运行统计**：`pytest_stats` 自动收集用例数、成功率、失败率、平均耗时、用例级重跑次数、步骤级重试次数等数据。 - **多渠道通知**：测试结束后自动触发终端图表渲染，并根据配置开关选择性推送邮件 / 企业微信机器人通知。 - **Jenkins 无缝集成**：自动嗅探 `BUILD_URL`、`JOB_NAME`、`BUILD_NUMBER` 等环境变量，动态生成 Allure 在线访问链接。 ### 功能矩阵（能力入口 + 边界） 下面这张表用于回答两个高频问题： 1. 框架现在到底支持哪些能力？ 2. 每个能力应该从哪里写、有什么边界？ | 能力 | YAML / 配置入口 | 典型场景 | 关键实现 | 当前边界 / 注意事项 | |------|------|------|------|------| | 单接口用例 | 顶层 `request` + `validation` | 验证一个接口的返回与落库 | `executor.py` / `step_runner.py` | `request.path` 与 `request.url` 二选一，若同时存在优先 `url` | | 多步骤链路 | 顶层 `steps` | 创建 → 查询 → 修改 → 删除的业务闭环 | `executor.py` / `step_runner.py` | 多步骤变量统一放 `config.variables`，不支持顶层 `variables + steps` 混用 | | 用例级兜底清理 | 顶层 `teardown_steps` | 主链路失败后仍需要清环境 | `executor.py` | 用于接口级兜底，不替代当前 step 的 `teardown_sql` | | CSV 数据驱动 | 顶层 `parameters` | 一个接口裂变出多组数据 | `data_drivers/` / `read_yaml_cases` | `parameters` 只能放单接口用例顶层，不能放进 `config` | | 请求公共配置合并 | `config.request` / `config.validation` | 多步骤共享 headers、cookies、method、公共断言 | `executor.py` | `headers/cookies` 深度合并，`validation` 是前置追加，不是覆盖 | | 页面接口 Cookie | `request.cookies` + 环境 `base_url.oms_cookie/wms_cookie` | OMS/WMS 页面接口复用登录态 | `request_builder.py` / `har_to_yaml.py` | YAML 节点名固定用 `cookies`；环境配置继续保留 `oms_cookie/wms_cookie -> cookie` 结构 | | 变量提取与复用 | `extract` / `extract_local` / `extract_list` / `extract_list_local` | 步骤间传参、跨 YAML 共享数据 | `data_extractor.py` / `runtime_cache.py` | 跨文件读取只认 `${get_extract_data(key)}`；本地变量直接用 `${变量名}` | | 前后置 SQL | `setup_sql` / `teardown_sql` | 前置造数、后置清理、状态修正 | `step_runner.py` / `sql_executor.py` | 当前只支持“字符串列表旧写法”；DML 影响行数为 0 会直接失败 | | 数据库断言 | `validation.db` | 校验接口是否真实落库 | `assertion_runner.py` / `db_assertions.py` | 只允许 `SELECT`；支持 `%s + args` 参数化；比对模式仅 `eq/contain/empty` | | 单步重试 | `retry.times` / `retry.interval` | 异步落库、短暂网络抖动 | `step_runner.py` | `times` 表示最大执行次数，包含第一次执行 | | Mock 挡板 | `mock` | 无需真实依赖服务时验证链路 | `mock_response.py` / `http_client.py` | 命中 Mock 后不会发真实请求，但提取/断言仍按正常步骤执行 | | HAR 录制回放 | `cli.py record` | 页面接口抓包转 YAML | `har_to_yaml.py` | 会优先把 WMS/OMS Cookie 归一为 `${get_base_url(wms_cookie/oms_cookie)}` | | 全局鉴权 | 环境 `auth` + `environment` | 会话启动先统一获取 token | `testcase/conftest.py` | `token_path` 按点号路径读取；多进程只登录一次，结果共享 | | 并发、熔断、统计 | `pytest.ini` + Redis 熔断状态 | xdist 并发执行、失败隔离 | `conftest.py` / `pytest_hooks/` / `pytest_stats.py` | 当前仓库默认 `-n 4 --dist=loadfile --reruns 1 --reruns-delay 1` | | 报告与通知 | Allure + `is_email_msg` / `is_wecom_msg` | 本地排查、CI 汇总、通知推送 | `allure_reporter.py` / `notification_sender.py` | Allure 生成失败不改变 pytest 退出码，通知只受开关控制 | ### 易混命名边界 下面这几个点最容易“看起来像一回事，实际上不是一个层级”： | 场景 | 正确写法 | 说明 | |------|------|------| | YAML 请求里的 Cookie | `request.cookies` | 这里是 requests 发包参数，字段名固定叫 `cookies` | | 环境文件里的 OMS/WMS Cookie | `base_url.oms_cookie.cookie` / `base_url.wms_cookie.cookie` | 这里是环境数据节点名，保留当前项目既有命名，不建议强行改成别的 | | 完整 URL vs 相对路径 | `request.url` 或 `base_url + request.path` | 同时存在时优先 `request.url` | | 当前 YAML 变量 vs 跨 YAML 变量 | `${变量名}` vs `${get_extract_data(key)}` | 本地变量不需要、也不能强行走 Redis 读取 | | 新环境函数名 vs 兼容函数名 | `${get_environment_value(key)}` 与 `${get_env_value(key)}` | 新用例优先前者，后者保留兼容旧用例 | --- ## 项目目录结构 ```text Interface/ ├── common/ # 底层基建层 │ ├── runtime_functions.py # 动态函数库（签名、造数、时间戳） │ ├── exceptions.py # 全局自定义异常体系 │ ├── logger.py # 全局日志（多进程隔离） │ ├── file_readers/ # 文件读取器（YAML、CSV） │ │ ├── csv_rows_reader.py # 纯 CSV 行数据加载工具 │ │ └── yaml_case_reader.py # 纯 YAML 用例加载工具 │ └── helpers/ # 辅助工具箱 │ └── allure_id_generator.py # Allure 报告编号生成器 ├── configs/ # ⚙️ 全局配置层：环境隔离与配置访问中心 │ ├── envs/ # 环境数据文件夹 (存放 prod.yaml, alpha1.yaml 等) │ │ ├── prod.yaml # 存放prod环境配置 │ │ └── test7.yaml # 存放test7环境配置 │ ├── config_manager.py # 唯一配置入口：提供对象化属性调用 (如 settings.mysql.host) │ └── env_reader.py # 底层工具：负责解析 YAML 配置文件 ├── data/ # 存放csv/har等文件 ├── files/ # 存放上传的文件 ├── logs/ # 📝 日志输出层：存放运行时的 .log 文件 ├── reports/ # 📊 报告输出层：存放生成的 Allure 原始数据与 HTML 报告 ├── templates/ # 邮件通知样式 ├── testcase/ # 🧪 用例数据层：存放具体的 YAML 用例与驱动脚本 ├── util_tools/ # 框架执行工具层：执行、变量、断言、客户端、扩展和报告 │ ├── clients/ # 触手：负责所有对外通信的连接器 │ │ ├── http_client.py # HTTP 客户端：封装 Session 与请求重试 │ │ ├── mysql_client.py # 数据库客户端：连接池管理与 SQL 执行 │ │ └── redis/ # Redis 能力：基础读写 + 熔断状态存储 │ │ ├── redis_client.py # Redis 基础客户端：连接池、数据类型读写、Pipeline、SCAN 清理 │ │ └── circuit_breaker_store.py # 熔断状态存储：按测试文件维护失败状态和本地缓存 │ ├── execution/ # 用例执行主流程 │ │ ├── executor.py # 执行编排，负责 steps 顺序调度和 teardown_steps 收尾清理 │ │ ├── schema_validator.py # YAML 用例结构校验 │ │ ├── request_builder.py # 请求参数组装 │ │ ├── step_runner.py # 单步骤执行辅助，负责 setup_sql、retry、请求、提取、断言、teardown_sql │ │ └── sql_executor.py # 前后置 SQL 执行处理 │ ├── variables/ # 变量处理：解析、提取、运行期缓存 │ │ ├── runtime_cache.py # 运行时变量池：管理多进程下的变量隔离与 Redis 存取 │ │ ├── data_extractor.py # 变量提取器：从响应中提取字段并存入变量池 │ │ └── variable_resolver.py # 变量解析器：将 YAML 中的 ${var} 替换为真实值 │ ├── assertions/ # 断言能力 │ │ ├── http_assertions.py # 状态码、响应体断言 │ │ ├── db_assertions.py # 数据库断言 │ │ └── assertion_runner.py # validation 列表统一调度 │ ├── data_drivers/ # 数据驱动处理 │ │ └── csv_case_expander.py # CSV 数据驱动器：处理 CSV 参数化并裂变生成用例 │ ├── reporting/ # 报告统计与通知 │ │ ├── pytest_stats.py # 运行统计收集 │ │ ├── html_summary.py # HTML 邮件摘要渲染 │ │ ├── notification_sender.py # 邮件和企微通知发送 │ │ └── channels/ # 通知渠道实现 │ │ ├── email_sender.py # 邮件通知发送逻辑 │ │ ├── terminal_reporter.py # 终端摘要展示 │ │ └── wecom_sender.py # 企业微信机器人推送逻辑 │ ├── pytest_hooks/ # pytest 生命周期 Hook 拆分 │ │ ├── circuit_breaker_hook.py # 用例熔断检查与状态更新 │ │ └── session_lifecycle_hook.py # 会话启动、xdist统计、汇总、清理与连接池关闭 │ ├── generators/ # 离线生成工具 │ │ ├── har_to_yaml.py # HAR 转 YAML：过滤请求、解析请求体、推导断言、写入 Cookie │ │ └── har_case_generator.py # HAR 用例生成器：生成 YAML + test_*.py 驱动文件 │ └── extensions/ # 扩展层：拦截并增强框架功能 │ ├── allure_reporter.py # Allure 报告附件与步骤封装 │ └── mock_response.py # Mock 响应拦截与响应伪造 ├── cli.py # HAR 流量录制工具 ├── conftest.py # Pytest 全局钩子（熔断、清理、统计） ├── pytest.ini # Pytest 配置（并发数、重试次数） ├── requirements.txt # 项目 Python 依赖清单 └── run.py # ▶️ 框架启动总入口：集成环境初始化与测试启动 ``` ### 维护者源码入口 如果你是框架维护者，不建议从 README 头到尾顺读后再找代码。更高效的方式是按“我要改什么”反查源码： | 想改的能力 | 先看这些文件 | 原因 | |------|------|------| | YAML 为什么这样执行 | `util_tools/execution/executor.py` / `util_tools/execution/step_runner.py` | 这里是主编排和单步骤生命周期入口 | | 变量为什么没替换 / 函数怎么生效 | `util_tools/variables/variable_resolver.py` / `common/runtime_functions.py` | 占位符解析和运行时函数都在这里 | | 请求为什么被改写 / cookies 怎么合并 | `util_tools/execution/request_builder.py` | headers、cookies、files、url 组装都在这里 | | DB 断言 / setup_sql / teardown_sql 为什么失败 | `util_tools/assertions/db_assertions.py` / `util_tools/execution/sql_executor.py` | SQL 校验、白名单、DML 严格模式都在这里 | | 并发、熔断、会话启动、收尾 | `conftest.py` / `testcase/conftest.py` / `util_tools/pytest_hooks/` | pytest 生命周期和全局鉴权从这里进 | | 报告、通知、终端统计 | `util_tools/extensions/allure_reporter.py` / `util_tools/reporting/` | Allure 附件、邮件、企微、终端汇总都在这里 | ### 维护者文档约定 下面这部分主要服务框架维护者。普通用例作者可以先跳过。 #### 文档事实源与同步原则 为了让这份 README 长期可维护，下面这条规则非常重要： - 本文档的目标是“帮助理解和使用框架”，不是替代源码。 - 当 README 与代码实现冲突时，**一律以代码为准**。 - 更新文档时，优先核对以下“事实源文件”： | 能力主题 | 事实源文件 | |------|------| | 环境加载 / 默认环境 | `configs/config_manager.py` / `configs/env_reader.py` | | 请求组装 / headers / cookies | `util_tools/execution/request_builder.py` | | HTTP 发包 / Retry / SSL | `util_tools/clients/http_client.py` | | 动态函数 / 签名 / nonce | `common/runtime_functions.py` | | 提取与变量解析 | `util_tools/variables/data_extractor.py` / `util_tools/variables/variable_resolver.py` | | DB 断言 / SQL 参数化 | `util_tools/assertions/assertion_runner.py` / `util_tools/assertions/db_assertions.py` | | setup_sql / teardown_sql | `util_tools/execution/step_runner.py` / `util_tools/execution/sql_executor.py` | | Redis / MySQL 客户端行为 | `util_tools/clients/redis/redis_client.py` / `util_tools/clients/mysql_client.py` | | 会话启动 / 熔断 / 预热 | `util_tools/pytest_hooks/session_lifecycle_hook.py` / `util_tools/pytest_hooks/circuit_breaker_hook.py` | | 报告与通知 | `util_tools/extensions/allure_reporter.py` / `util_tools/reporting/notification_sender.py` | #### 文档维护约定 为了让 README 不只“内容多”，还要“长期不容易过期”，建议把下面几条当成固定约定： - README 负责回答“框架支持什么、推荐怎么用、当前默认行为是什么”。 - `util_tools/README.md` 负责回答“源码从哪里读起、排查问题先看哪个目录”。 - 当代码行为发生变化时，至少同步检查下面这些章节： | 变更类型 | 至少同步更新的 README 章节 | |------|------| | YAML 字段、合并规则、默认行为变化 | `一页速览`、`§二 测试用例的几种编写方式`、`附录：快速查阅卡` | | RuntimeFunctions 新增/删除/改语义 | `变量引用速查表`、`RuntimeFunctions 内置函数完整列表` | | 环境配置字段、默认环境、默认 URL 变化 | `环境准备`、`环境配置文件完整字段说明` | | pytest 默认参数、并发策略、重试策略变化 | `pytest.ini 配置详解`、`§九 Jenkins / CI 集成` | | setup_sql / teardown_sql / DB 断言规则变化 | `§4.6 YAML 编写规范`、`Q3 数据库断言排查`、`附录：快速查阅卡` | - 如果 README 与代码实现冲突，最终必须让两者收敛到同一事实。 --- ## 一、框架整体运行机制 ### 1.1 架构分层总览 框架采用**六层架构设计**，从上到下依次是： ```mermaid graph TB L1["Layer 1: 触发与入口层"] L2["Layer 2: 数据驱动与调度层"] L3["Layer 3: 核心执行引擎层"] L4["Layer 4: 基础业务工具链"] L5["Layer 5: 底层基建与目标服务"] L6["Layer 6: 报告与通知层"] L1 --> L2 --> L3 --> L4 --> L5 --> L6 ``` 数据流向自上而下、单向依赖： 1. **Layer 1（触发与入口层）**：Git 代码合并 → Jenkins CI/CD → 本地 CLI → run.py 启动脚本，是测试的触发入口。 2. **Layer 2（数据驱动与调度层）**：Pytest 核心引擎 + conftest.py 全局钩子 + csv_case_expander 数据驱动 + ConfigManager 配置管理，负责用例收集、调度和环境装配。 3. **Layer 3（核心执行引擎层）**：Executor 场景控制总线 + context_vars 内存变量池 + ExitStack 资源托管，是框架的"大脑"，统筹单用例的全生命周期。 4. **Layer 4（基础业务工具链）**：VariableResolver（解析占位符）+ HttpClient（发包）+ DataExtractor（变量提取）+ Assertions（断言），构成单步执行的四大基础能力。 5. **Layer 5（底层基建与目标服务）**：Redis 全局缓存 + MySQL 数据库 + MockResponse 挡板 + 目标被测服务器，提供数据持久化、缓存共享、Mock 拦截和真实请求转发。 6. **Layer 6（报告与通知层）**：Allure 生成 HTML 报告 + 企微/邮件 Webhook 推送，输出测试结果。 ### 1.2 启动流程详解 #### 阶段一：环境装配（Session 启动前） ```mermaid graph TD Start(["run.py 被调用"]) --> LoadConfig["(1) 加载 pytest.ini 配置"] LoadConfig --> PytestStart["(2) Pytest 启动：pytest_sessionstart"] subgraph MasterWorker ["Master/Worker 分工"] PytestStart --> Master{仅 Master?} Master -->|是| MasterOps["Master: 环境清理 + 熔断预加载"] Master -->|否| WorkerSkip["Worker: 跳过"] end MasterOps --> Auth["(3) authorize_session fixture"] WorkerSkip --> Auth subgraph Token获取 ["全局 Token 获取"] Auth --> GenSession["生成 SessionID"] GenSession --> MasterAuth{Master?} MasterAuth -->|是| FileLock["FileLock 互斥登录 → 写入共享文件"] MasterAuth -->|否| ReadFile["读取共享文件 → 获取 Token"] FileLock --> SetScope["所有进程: 设置执行域 + Token 入 Redis"] ReadFile --> SetScope end SetScope --> End(["结束"]) ``` Session 启动前的环境装配阶段，核心目标是**让所有进程拿到一致的全局 Token 和独立的 SessionID**： 1. **加载配置**：run.py 读取 pytest.ini，获取并发数、重试次数、Allure 输出目录等。 2. **Pytest 初始化**：触发 pytest_sessionstart 钩子。**仅 Master 进程**执行清理（过期日志、Redis 脏数据、预加载熔断黑名单），**Worker 进程直接跳过**。 3. **authorize_session**：生成全局唯一的 SessionID（UUID 前 8 位）。Master 进程通过 FileLock 互斥获取登录权，调用接口获取 AccessToken 并写入共享文件；Worker 进程直接从共享文件读取 Token。所有进程最终将 Token 写入 Redis，供后续用例跨进程共享。 **为什么需要 SessionID？** 在 `pytest-xdist` 多进程并发模式下，多个 Worker 进程会同时读写 Redis。如果不做隔离，不同测试批次提取的 `order_id` 可能互相覆盖，导致用例互相污染。框架使用环境名 + SessionID 组成 Redis 前缀（如 `extract:test7:test_a1b2c3d4:order_id`），确保不同测试批次互不串数据。同一测试批次内需要跨 YAML 共享的变量请使用清晰的业务变量名，避免多个流程写入同名 Redis 变量；只在单个 YAML 链路内使用的变量优先用 `extract_local`。 #### 阶段二：用例收集与数据驱动裂变 ```mermaid graph TD Start(["Pytest 收集测试文件"]) --> Traverse["遍历 testpaths"] Traverse --> Discover["发现 test_*.py，导入 Test* 类"] Discover --> Parametrize["执行 @pytest.mark.parametrize"] subgraph 用例裂变 ["用例裂变"] Parametrize --> YamlReader["调用 read_yaml_cases"] YamlReader --> CacheCheck["缓存检查：命中则返回深拷贝"] CacheCheck --> YamlLoad["yaml.safe_load 安全解析"] YamlLoad --> Normalize["结构规范化（单字典 → 列表）"] Normalize --> csv_case_expander{含 parameters 节点?} csv_case_expander -->|是| Driver["expand_csv_cases 裂变（CSV 笛卡尔积 → 文本替换 → 反序列化）"] csv_case_expander -->|否| Register end Driver --> Register["注册为 N 个独立 TestItem"] Register --> End(["结束"]) ``` Pytest 收集阶段的目标是把 YAML 文件转化为 N 个独立的 TestItem，供后续并发执行： 1. **文件扫描**：遍历 pytest.ini 中配置的 testcase 目录，发现所有 `test_*.py` 文件。 2. **导入与参数化**：导入 Test* 类，执行 `@pytest.mark.parametrize` 装饰器，触发 read_yaml_cases 读取 YAML。 3. **read_yaml_cases 解析**：先检查 LRU 缓存（避免重复读盘）→ yaml.safe_load 安全解析 → 结构规范化（单条用例包装为列表）。 4. **expand_csv_cases 裂变**：若 YAML 中含 `parameters` 节点，读取 CSV 文件，进行**笛卡尔积组合**，然后通过文本替换 + orjson 反序列化，生成多组测试数据。 5. **注册用例**：最终通过 parametrize 注册为 N 个完全独立的 TestItem，每个 TestItem 携带一组参数。 #### 阶段三：用例执行（核心流水线） ```mermaid graph TD Start(["Pytest 调度用例"]) --> Check{熔断检查} Check -->|已熔断| Skip["pytest.skip 跳过"] Check -->|通过| Prep[准备阶段] subgraph "准备阶段" Prep --> Deepcopy[deepcopy 深拷贝隔离模板] Deepcopy --> Config[提取全局配置] Config --> Type{判断用例类型} Type -->|含 steps| Multi[多步骤链路] Type -->|无 steps| Single[单接口用例] Multi --> Init[初始化 context_vars] Single --> Init end Init --> Loop{遍历 steps} subgraph "主链路循环" Loop -->|有下一个| Merge[合并全局配置到当前步骤] Merge --> Step[单步执行] Step --> TeardownStep[teardown_sql 后置清理] TeardownStep --> Loop end subgraph "单步执行细节" Step --> Validate[validate_step_schema 语法校验] Validate --> Setup[setup_sql 前置准备] Setup --> Retry{重试循环} Retry -->|重试| Send[组装请求 → 发包 → 解析响应] Send --> Record[Allure 捕获本次尝试附件] Record --> Extract[变量提取 四大通道] Extract --> Assert[断言校验 软断言] Assert --> CheckRetry{还有重试次数?} CheckRetry -->|是| Retry CheckRetry -->|否| NextStep[本步结束 回到主链路] end Loop -->|steps 结束| Final[Teardown 链路] subgraph "收尾阶段" Final --> TdLoop{遍历 teardown_steps} TdLoop -->|有下一个| TdMerge[合并全局配置到当前步骤] TdMerge --> TdExec[执行清理步骤（异常静默隔离）] TdExec --> TdLoop TdLoop -->|结束| CheckErr{主链路有异常?} CheckErr -->|有| Raise[重新抛出异常] CheckErr -->|无| End(["结束"]) end Raise --> End Skip --> End ``` 用例执行是框架的核心流水线，分为 5 个阶段： 1. **熔断检查**：查询本地缓存，若前置用例已失败则跳过本用例（pytest.skip），避免无效执行。 2. **准备阶段**：deepcopy 隔离模板防止污染 → 提取 config 全局配置（公共 headers、断言规则等）→ 判断用例类型（单接口/多步骤）→ 初始化 context_vars 内存变量池。 3. **主链路循环**：遍历 steps 列表，每个 step 先合并全局配置，然后执行单步逻辑。`setup_sql` 成功后才算进入主流程，之后无论请求、提取或断言是否失败，都会执行 `teardown_sql`（防止脏数据）。 4. **单步执行细节**（subgraph 内）：语法校验 → 前置 SQL → 进入重试循环。循环内依次完成：组装请求数据（VariableResolver 解析 `${变量}`）→ 发包（HttpClient，含 Mock 拦截检查）→ 安全解析响应 JSON → Allure 捕获本次尝试附件 → 变量提取（extract / extract_local / extract_list / extract_list_local 四大通道）→ 软断言校验。若断言失败或网络异常且还有重试次数，则丢弃本次 Allure 附件并重新进入循环；若当前尝试成功或已经是最后一次失败，则把本次附件写入 Allure 报告。 5. **收尾阶段**：主链路所有 steps 结束后，强制执行 teardown_steps。每个 teardown_step 同样会先合并 config 全局配置，再执行单步逻辑（异常静默隔离，不掩盖主链路错误）。若主链路有异常，则重新抛出，Pytest 标红该用例。 #### 阶段四：收尾与报告 ```mermaid graph TD subgraph Report ["pytest_runtest_makereport"] ReportStart(["报告生成"]) --> CheckResult{用例结果} CheckResult -->|失败| CheckRetry{还在重试?} CheckRetry -->|耗尽| SetCircuit["写入 Redis 熔断标记，更新本地缓存"] CheckRetry -->|未耗尽| NoOp1["不标记熔断（瞬态故障豁免）"] CheckResult -->|通过| CheckPrev{之前被熔断?} CheckPrev -->|是| ClearCircuit["清除熔断标记（自愈）"] CheckPrev -->|否| NoOp2["无操作"] end subgraph Summary ["pytest_terminal_summary"] SummaryStart(["测试结束"]) --> RecordTime["记录结束时间"] RecordTime --> FakeReporter["构造 FakeReporter（兼容 xdist）"] FakeReporter --> PrintSummary["打印终端汇总图表"] PrintSummary --> GenEnv["生成 environment.properties"] end subgraph Finish ["pytest_sessionfinish"] FinishStart(["会话结束"]) --> SendNotify["发送异步通知（企微/邮件）"] SendNotify --> CloseNotify["关闭通知线程池"] CloseNotify --> CloseRedis["关闭 Redis 连接池"] CloseRedis --> CloseHttp["关闭 HTTP 全局 Session"] end subgraph Allure ["run.py 收尾"] AllureStart(["收到 pytest 退出码"]) --> AllureGen["allure generate 渲染 HTML 报告"] AllureGen --> Output["reports/allures/index.html"] end Report --> Summary Summary --> Finish Finish --> AllureStart ``` 测试收尾阶段依次执行 4 个钩子 + run.py 收尾： 1. **pytest_runtest_makereport（报告生成）**：根据用例结果更新熔断状态。失败时：若重试耗尽则写入 Redis 熔断标记并更新本地缓存；若还在重试则不标记（瞬态故障豁免）。通过时：若之前被熔断则清除标记（自愈机制）。 2. **pytest_terminal_summary（测试结束）**：记录结束时间 → 构造 FakeReporter（兼容 pytest-xdist）→ 打印终端汇总图表（通过率、耗时、失败列表）→ 生成 environment.properties（含 Python 版本、Git 分支、Jenkins 构建号等上下文）。 3. **pytest_sessionfinish（会话结束）**：发送异步通知 → 关闭通知线程池 → 关闭 Redis 连接池 → 关闭 HTTP 全局 Session。 4. **run.py 收尾**：收到 Pytest 退出码后，调用 `allure generate` 渲染 HTML 报告，输出到 `reports/allures/index.html`。 ### 1.3 环境配置与切换 框架通过 `AUTO_ENV` 环境变量决定加载哪个环境配置文件。 ```bash # Windows PowerShell $env:AUTO_ENV="test7.yaml" python run.py # Linux / Mac export AUTO_ENV=test7.yaml python run.py ``` **环境配置文件结构**（`configs/envs/*.yaml`）： ```yaml # 基础域名配置 host: "http://api.example.com" base_url: base_wms_url: "http://wms.example.com" base_wx_url: "http://wx.example.com" # MySQL 数据库配置 mysql: host: "mysql.example.com" port: 3306 username: "user" password: "123456" database: "test_db" # Redis 缓存配置 redis: host: "redis.example.com" port: 6379 db: 0 password: "" expire: 3600 # HTTP 客户端配置 http: verify_ssl: true # 默认开启 SSL 证书校验；自签名测试环境可显式改为 false # 环境级业务参数（如 AppID、Secret） environment: grant_type: "client_credential" appid: "wx1234567890" secret: "your_secret_key" # 通知开关 is_email_msg: false is_wecom_msg: false # Allure 报告地址（Jenkins 环境下自动生成） allure_url: "http://localhost:63342/interface_wms/reports/allures/index.html" ``` **配置加载流程**： ```mermaid graph TD Start(["ConfigManager 初始化"]) --> Step1["1. 确定项目根目录"] Step1 --> Step2["2. 创建必要子目录"] Step2 --> Step3["3. 读取 AUTO_ENV 环境变量"] Step3 --> Step4["4. 调用 EnvReader 解析 YAML"] Step4 --> Step5["5. 动态属性注入"] Step5 --> Check{安全性检查：key 与类属性同名?} Check -->|是| Skip["跳过注入，保留原生属性"] Check -->|否| Inject["安全注入"] Skip --> Step6 Inject --> Step6["6. 显式声明高频配置项"] Step6 --> End(["settings 就绪"]) ``` **流程说明**： | 步骤 | 说明 | |------|------| | `确定项目根目录` | configs/config_manager.py 向上回溯两级定位项目根 | | `创建子目录` | logs/、reports/、templates/、data/ | | `读取 AUTO_ENV` | 环境变量决定加载哪个环境配置文件，未设置时默认加载 `prod.yaml` | | `解析 YAML` | EnvReader().load_env_config() 安全解析 | | `动态属性注入` | 将 YAML key-value 挂载到 settings 实例，同名 key 跳过注入防覆盖 | | `显式声明` | settings.host / mysql / redis / wechat / email，IDE 代码补全友好 | ### 1.4 日志系统 框架的日志系统基于 Python `logging` + `colorlog`，针对 `pytest-xdist` 多进程并发做了专门优化。 **核心设计**： | 特性 | 说明 | |------|------| | PID 隔离 | 日志文件按进程号隔离：`logs/日期/时间_pid{pid}.log` | | propagate=False | 防止日志被 Root Logger 重复打印 | | colorlog | 控制台不同级别显示不同颜色（INFO=绿、WARNING=黄、ERROR=红） | | worker_id 注入 | 日志中自动显示 pytest-xdist 的 worker 标识（gw0/gw1/master） | | 自动清理 | 保留最近 10 个日志目录，旧目录自动删除 | **日志格式示例**： ``` 2025-01-15 14:32:08 | INFO | master | executor.py:220 | >>> [Runner ] 开始执行业务用例: 创建标签 2025-01-15 14:32:09 | INFO | master | http_client.py:169 | >>> [Request ] 请求url: [http://wx.example.com/cgi-bin/tags/create] 2025-01-15 14:32:09 | INFO | master | runtime_cache.py:119| --- [Redis ] 管道批量写入: ['tag_id'] 2025-01-15 14:32:09 | INFO | master | assertion_runner.py:180 | [+] [Assert ] 所有断言规则全部通过测试！ ``` **断言日志对齐**： 断言日志会固定“断言标题列”和“预期值列”的宽度，让 HTTP 断言、DB 断言、判空断言在控制台中更容易横向扫描。 ```text [+] [Assert ] 状态码断言 | 预期: 200 | 实际: 200 [+] [Assert ] 相等断言 | 预期: {'code': 0} | 实际: {'code': 0} [+] [Assert ] DB(contain)断言 | 预期: {'outbound_no': 'OT4H0... | 实际: {'id': 1709940521, 'outbound_no': 'OT4H00115'} [+] [Assert ] DB(eq)断言 | 预期: {'outbound_no': 'OT4H0... | 实际: {'outbound_no': 'OT4H00115'} [+] [Assert ] DB(empty)断言 | 预期: 期望为空=False | 实际: 实际为空=False ``` **日志文件位置**： ``` logs/ ├── 2025-01-14/ # 按日期归档 │ ├── 2025-01-14_09-30-15_pid12345.log # Worker 进程1 │ └── 2025-01-14_09-30-15_pid12346.log # Worker 进程2 └── 2025-01-15/ └── 2025-01-15_14-32-08_pid28421.log # Master 进程 ``` ### 1.5 性能优化专题 框架在多个核心链路做了性能优化： **orjson 极速解析**： - 响应体 JSON 解析、CSV 反序列化、HAR 解析、断言附件序列化等高频链路使用 `orjson` - 大响应体和大 HAR 文件解析效率更高，适合 WMS 大型订单/库存报文 - Redis JSON 存取当前仍使用标准库 `json`，不是全链路替换 **HTTP 连接池**： - 全局单例 Session + Keep-Alive 长连接 - pool_connections=20（最多为 20 个不同域名维护独立连接池） - pool_maxsize=50（单个域名下最多保持 50 个并发连接） - 内置 Retry 策略：幂等请求（GET/PUT/DELETE）连接失败时自动重试 1 次 **Redis Pipeline 批量写入**： - 单步提取的多个变量通过 `RuntimeCache.write_extract_variables()` 一次性 Pipeline 写入 - N 次网络往返压缩为 1 次 **大响应体截断保护**： - 控制台日志 >500 字符自动截断，防止 Jenkins 控制台卡死 - Allure 附件仍保留完整数据 - 非 JSON 响应（如 HTML 报错页）>2000 字符自动截断 **LRU 缓存加速**： - VariableResolver 的幂等函数调用启用 `lru_cache(maxsize=128)` - CSV 文件读取启用 `lru_cache(maxsize=32)` - 动态函数（时间戳、随机数）强制走实时计算，不走缓存 ### 1.6 Redis 客户端详解 框架的 Redis 能力已经拆成两个文件： - `util_tools/clients/redis/redis_client.py`：Redis 基础客户端，负责连接池、String/Hash/List/JSON、Pipeline 和 SCAN 清理。 - `util_tools/clients/redis/circuit_breaker_store.py`：测试熔断状态存储，负责把失败文件状态写入 Redis，并维护本进程本地缓存。 **连接池参数**： | 参数 | 默认值 | 说明 | |------|--------|------| | `max_connections` | 50 | 连接池最大连接数 | | `socket_connect_timeout` | 3 | TCP 连接超时（秒）| | `socket_timeout` | 10 | 读写超时（秒）| | `retry_on_timeout` | True | 超时后自动重试一次 | | `health_check_interval` | 30 | 每 30 秒检查连接存活 | | `decode_responses` | True | 返回字符串而非 bytes | **支持的数据类型**： | 类型 | 写入方法 | 读取方法 | 删除方法 | 说明 | |------|---------|---------|---------|------| | **String** | `set_redis_data(key, value, ex)` | `get_redis_data(key)` | `del_redis_data(key)` | 简单键值对，支持过期时间 | | **Hash** | `set_redis_hash(key, field, value)` | `get_redis_hash(key, field)` | `del_redis_hash(key, field)` | 嵌套字典，可单字段读写 | | **List** | `set_redis_list(key, values)` | `get_redis_list(key, start, end)` | `del_redis_list(key)` | 有序链表，支持范围读取 | | **JSON** | `set_json(key, value, ex)` | `get_json(key)` | `del_redis_data(key)` | 自动序列化/反序列化 | **Pipeline 批量操作**： ```python # 批量写入 Hash 的多个字段 client.pipeline_set_hash_batch("user:1001", {"name": "张三", "age": "25"}) # 批量读取多个字段 values = client.pipeline_get_hash_batch("user:1001", ["name", "age"]) # 返回: ["张三", "25"] # 批量写入 String 键值对 client.pipeline_set_string_batch({"key1": "val1", "key2": "val2"}, ex=3600) ``` Pipeline 将 N 次网络往返压缩为 1 次，实测 50 次操作性能提升 10~50 倍。 **SCAN + DELETE 安全批量删除**： ```python # 按模式匹配删除一批 key（不会阻塞 Redis） deleted = client.scan_and_delete("extract:test7:*", batch_size=1000) ``` 为什么不用 `KEYS` 命令？因为 `KEYS` 会一次性扫描所有 key，数据量大时会卡住 Redis 服务器。`SCAN` 是游标迭代式的，每次只返回一小批结果，不会阻塞。 **熔断状态缓存（双层架构）**： ```mermaid graph TD subgraph 查询请求 QStart(["开始查询"]) --> L1{L1 本地内存缓存} L1 -->|命中| L1Hit["纳秒级返回 O(1)"] L1 -->|未命中| L2{L2 Redis 全局缓存} L2 -->|命中| L2Hit["毫秒级返回，回填 L1"] L2 -->|未命中| ReturnNone["返回 None"] L1Hit --> QEnd(["结束"]) L2Hit --> QEnd ReturnNone --> QEnd end subgraph 写入请求 WStart(["开始写入"]) --> WriteRedis["先写 Redis（持久化）"] WriteRedis --> UpdateL1["再更新本地缓存"] UpdateL1 --> WEnd(["结束"]) end subgraph 缓存同步 Sync(["每 5 秒定时任务"]) --> Refresh["circuit_breaker_store.refresh_cache（从 Redis 同步到本地）"] Refresh --> SyncEnd(["结束"]) end ``` **架构说明**： | 层级 | 作用 | 特点 | |------|------|------| | L1 本地内存 | `CircuitBreakerStore._cache` | O(1) 纳秒级，单进程内查询 | | L2 Redis 全局 | `pytest:circuit_breaker:failed_files` | 毫秒级，多进程共享，命中后回填 L1 | | 写入流程 | 先写 Redis 再更新本地 | 保证持久化 + 后续查询本地命中 | | 同步机制 | `circuit_breaker_store.refresh_cache` 每 5 秒 | `CircuitBreakerStore._cache_lock` 保证线程安全 | 双重检查锁定（Double-Checked Locking）用于保证多线程环境下只有一个线程创建连接池或刷新缓存。 **运行期 Redis 数据边界**： | 数据类型 | Key 范围 | 写入时机 | 清理时机 | 是否跨文件可读 | |----------|----------|----------|----------|----------------| | 提取变量 | `extract:{env}:{session_id}:*` | `extract` / `extract_list` 提取成功后 | 整批测试结束时按本轮 `session_id` 精确清理 | 是 | | 本地变量 | 仅内存 `context_vars` | `extract_local` / `extract_list_local` 提取成功后 | 当前 YAML 执行结束后随内存释放 | 否 | | 熔断状态 | `pytest:circuit_breaker:failed_files` | 用例最终失败后 | 下一轮测试启动时清理框架级状态，或用例通过后自愈清理 | 是 | **清理规则说明**： 1. 测试启动时，Master 调用 `RuntimeCache.clear_framework_runtime_state()` 清理历史框架级状态，主要用于避免历史熔断影响本轮运行。 2. 测试结束时，Master 调用 `clear_session_extract_keys()` 清理本轮产生的 Redis 提取变量。 3. xdist 并发时，Worker 会把自己的 `session_id` 写入 `logs/xdist_stats/*_stats.json`，Master 通过这些统计文件知道应该清理哪些 `extract:{env}:{session_id}:*`。 4. 清理提取变量时只按本轮 `session_id` 删除，不会删除同一 Redis、同一环境下其他任务的变量。 5. 如果进程被强杀，`pytest_sessionfinish` 没有机会执行，本轮提取变量可能短暂残留；下次正常运行结束会清理本轮 session，历史异常残留需要按 `extract:{env}:*` 人工确认后再删。 **中途失败与熔断关系**： | 场景 | 是否写入熔断 | 说明 | |------|--------------|------| | 用例第一次失败，且还会被 `pytest-rerunfailures` 重试 | 否 | 中间失败会打 `_interim_failure` 标记，不作为最终失败。 | | 所有重试耗尽后仍失败 | 是 | `CircuitBreakerStore.mark_file_as_blocked()` 写入失败文件状态。 | | 同文件后续用例执行前 | 会被跳过 | `pytest_runtest_call` 阶段检查所属文件是否已熔断。 | | 触发熔断的源头用例本身 | 不跳过 | 避免源头用例重试时被自己刚写入的状态拦截。 | | 后续重跑通过 | 清理熔断 | 通过后调用 `delete_file_blocked_status()`，实现自愈。 | **`logs/xdist_stats` 目录作用**： 这个目录不是普通业务日志目录，而是 xdist 并发统计中转目录。每个 Worker 会导出一个 `*_stats.json` 文件，里面包含本 Worker 的通过数、失败数、跳过数、耗时和 `session_id`。Master 结束时先读取这些文件，再汇总统计、发送通知，并按 `session_id` 清理 Redis 提取变量。目录下的历史 `*_stats.json` 会在下一轮测试启动前自动清理。 ### 1.7 MySQL 连接池详解 框架的 MySQL 客户端（`util_tools/clients/mysql_client.py`）基于 `DBUtils.PooledDB` 实现连接池管理。 **连接池参数**： | 参数 | 默认值 | 说明 | |------|--------|------| | `maxconnections` | 5 | 连接池最大连接数（可从配置读取）| | `mincached` | 2 | 初始化时预创建连接数 | | `maxcached` | 5 | 空闲时最大保留连接数 | | `autocommit` | True | 保持查询连接无长事务状态；写操作仍由客户端显式 `commit/rollback` | | `charset` | utf8mb4 | 支持 Emoji 和生僻字 | | `cursorclass` | DictCursor | 返回字典格式（而非元组）| **懒加载 + 双重检查锁定**： ```python # 第一次调用 _get_pool() 时才创建连接池 # 使用双重检查锁定防止多线程重复创建 if _pool is None: with _instance_lock: if _pool is None: _create_pool() ``` **按需自动预热**： 框架会在 `pytest_collection_finish` 阶段扫描本轮 pytest 收集到的 `caseinfo`。只要发现以下任意配置，就说明本轮用例会访问 MySQL，框架会提前预热当前执行进程的连接池： 1. `validation` 中存在 `db` 数据库断言； 2. 用例或步骤中存在 `setup_sql`； 3. 用例或步骤中存在 `teardown_sql`。 这个机制不需要在 YAML 或环境配置中额外增加开关。用例作者只要正常编写 `db/setup_sql/teardown_sql`，框架就能自动识别。 xdist 并发运行时，MySQL 连接池是进程内资源，不能在 Master 和 Worker 之间共享。因此： - xdist Master 只负责调度和汇总，不预热 MySQL； - 每个 Worker 只扫描自己负责执行的 `items`，如果分到的用例需要 MySQL，就预热自己的连接池； - 单进程运行时，由当前进程直接预热； - 纯 HTTP 用例不会预热 MySQL，避免无意义的启动成本。 运行日志中如果在业务接口执行前看到如下信息，表示自动预热已经生效： ```text --- [MySQL ] 尝试建立连接池: mysql-dev.meiyunji.net:3306/wms_db1 [+] [MySQL ] 数据库连接池初始化成功: mysql-dev.meiyunji.net [+] [MySQL ] 数据库连接池预热完成 ``` **MySQL 功能开关**： 环境配置中可通过 `mysql.enable` 字段关闭数据库功能： ```yaml mysql: enable: false # 关闭 MySQL 功能，所有 SQL 操作将报错提示 ``` **查询接口**： ```python # 查单条（返回字典） row = db_client.select("SELECT * FROM users WHERE id = %s", (1001,)) print(row["name"]) # "张三" # 查多条（返回字典列表） rows = db_client.select("SELECT * FROM users WHERE status = %s", (1,), fetchall=True) ``` **数据变更接口**： ```python # INSERT / UPDATE / DELETE，返回受影响行数 rowcount = db_client.execute( "UPDATE users SET status = %s WHERE id = %s", (1, 1001) ) # 自动 commit，出错自动 rollback ``` SQL 参数强制使用 `%s` 占位符，由 pymysql 自动转义绑定，防止 SQL 注入。 --- ## 二、测试用例的几种编写方式 ### 2.0 5 分钟快速入门（写出你的第一个用例） 在深入 6 种模式之前，先花 5 分钟跑通一个最小闭环，建立手感。 如果你是第一次接触本框架，推荐把这节拆成两步理解： 1. 先跑项目内置演示：[§2.0.2 框架能力拆解演示](#202-框架能力拆解演示)，确认“框架本身能正常执行”。 2. 再照着下面示例自己新建 YAML 和 Python 驱动，确认“我已经会写自己的第一条用例”。 #### 第一步：写 YAML 用例文件 在 `testcase/api_auto/demo/` 下新建 `create_order.yaml`： ```yaml - name: 创建订单 base_url: ${get_base_url(base_wms_url)} request: method: post path: /api/order/create headers: Content-Type: application/json json: sku: ${get_base_data(skuA)} count: 1 validation: - code: 200 - eq: { 'code': 0 } ``` #### 第二步：写 Python 驱动文件 在同级目录下新建 `test_demo.py`： ```python import allure import pytest from common.file_readers.yaml_case_reader import read_yaml_cases from common.helpers.allure_id_generator import m_id, c_id from util_tools.execution.executor import Executor @allure.feature(next(m_id) + '演示模块') class TestDemo: @allure.story(next(c_id) + '创建订单') @pytest.mark.parametrize('caseinfo', read_yaml_cases('testcase/api_auto/demo/create_order.yaml')) def test_create_order(self, caseinfo): allure.dynamic.title(caseinfo['name']) Executor().run_case(caseinfo) ``` 说明：`m_id` 和 `c_id` 用于生成 Allure 模块编号和场景编号，推荐在测试驱动文件里统一使用 `next(m_id)`、`next(c_id)`。驱动方法中的 Pytest 参数名建议继续使用 `caseinfo`，因为熔断 Hook 会通过这个参数名读取业务用例名；框架源码内部变量统一使用 `case_info`，用于提升代码可读性。 #### 第三步：运行 ```bash python run.py ``` 完成！你刚刚完成了框架的**最小可用闭环**。接下来可以深入阅读下方 6 种编写模式，根据业务场景选择合适的方式。 --- ### 2.0.1 标准 Python 驱动文件模板 业务测试文件推荐统一使用下面这套模板。新人写用例时，优先复制这段结构，只改类名、模块名、场景名和 YAML 路径。 ```python import allure import pytest from common.file_readers.yaml_case_reader import read_yaml_cases from common.helpers.allure_id_generator import m_id, c_id from util_tools.execution.executor import Executor @allure.feature(next(m_id) + '模块名称') class TestModuleName: @allure.story(next(c_id) + '场景名称') @pytest.mark.parametrize('caseinfo', read_yaml_cases('testcase/xxx/xxx.yaml')) def test_scene_name(self, caseinfo): allure.dynamic.title(caseinfo['name']) Executor().run_case(caseinfo) ``` **固定写法说明**： | 写法 | 是否建议修改 | 说明 | |------|--------------|------| | `caseinfo` | 不建议改 | 熔断 Hook 会通过这个参数名读取业务用例名。 | | `allure.dynamic.title(caseinfo['name'])` | 不建议省略 | 保证 CSV 裂变、YAML 多用例时报告标题展示真实用例名。 | | `Executor().run_case(caseinfo)` | 不建议封装 | 驱动文件只负责把 YAML 数据交给执行器，不写业务逻辑。 | | `next(m_id)` / `next(c_id)` | 推荐保留 | Allure 编号会按本轮实际收集到的用例重新生成。 | **命名建议**： | 对象 | 建议格式 | 示例 | |------|----------|------| | 测试文件 | `test_业务含义.py` | `test_order_create.py` | | 测试类 | `Test业务模块` | `TestOrderCreate` | | 测试方法 | `test_动作_对象` | `test_create_order` | | YAML 文件 | `业务动作.yaml` 或带顺序号 | `001_create_order.yaml` | ### 2.0.2 框架能力拆解演示 项目内置了一组最小化演示用例，目录为 `testcase/框架能力拆解演示/`。这些文件不依赖真实业务接口，主要用 Mock 演示框架能力，适合新人按顺序阅读。 **新人建议先跑这一组，而不是直接写真实业务用例**。原因很简单： - 这组演示主要使用 Mock，成功率高，不依赖真实业务服务可用性。 - 你可以先确认“框架执行链路没问题”，再去区分后续失败到底是框架问题、环境问题，还是业务接口问题。 - 对第一次上手的人来说，先把“变量提取、CSV、SQL、retry、Mock”这些能力看见，比先写真实接口更容易建立信心。 | 文件 | 演示能力 | 阅读重点 | |------|----------|----------| | `test_framework_capability_demo.py` | Pytest 驱动入口 | 每个方法只负责 `read_yaml_cases()` + `Executor().run_case()`，业务逻辑全部放在 YAML。 | | `01_memory_and_redis_variables.yaml` | 内存变量 + Redis 变量 | `extract_local` 只写当前用例内存；`extract` 同时写内存和 Redis；`${get_extract_data(...)}` 从 Redis 读取变量。 | | `02_yaml_csv_driver.yaml` | YAML + CSV 数据驱动 | `parameters` 绑定 `data/csv/framework_demo_orders.csv`，框架在收集阶段把一条 YAML 裂变成多条用例。 | | `03_setup_teardown_sql.yaml` | YAML 前后置 SQL | `setup_sql` 在当前步骤请求前执行；`teardown_sql` 在当前步骤请求、提取、断言后执行。 | | `04_teardown_steps_after_case.yaml` | 用例级最终清理 | `teardown_steps` 在主流程全部结束后执行，适合做业务数据兜底清理。 | | `05_single_yaml_step_retry.yaml` | 单步骤重试 | `retry.times` 表示最大执行次数，包含首次执行；`retry.interval` 表示失败后再次尝试前等待的秒数。 | 建议阅读顺序：先看 `test_framework_capability_demo.py` 理解 Python 驱动文件的固定写法，再按 YAML 文件编号逐个学习。 **第一条最推荐跑的演示**：`06_mock.yaml` - 这是最不依赖外部环境的一组演示，用来确认 Mock、请求组装、断言、提取链路是否正常最直接。 - 如果这一组都跑不通，优先排查环境准备和框架配置，不建议立刻去写真实业务用例。 ### 2.0.3 YAML 编写格式与场景选择 框架推荐 **2 种基础用例结构 + 多种步骤增强能力**。核心原则是：**测试逻辑写在 YAML 里，Python 驱动文件只做一件事 —— 把 YAML 交给 Executor**。 一句话选择：**接口级验证用单接口；业务流程级验证用 `steps`**。 ### 日常写用例工作流 如果你平时的主要工作是“拿到一个需求后把 YAML 用例写出来”，推荐按这个顺序做： 1. 先判断是“单接口验证”还是“多步骤业务链路”。 2. 再决定是否需要增强能力：`parameters`、`extract`、`retry`、`setup_sql`、`teardown_steps`、`mock`。 3. 先写最小可运行版本，只保留 `request + validation`。 4. 跑通后再逐步补变量提取、清理逻辑、DB 断言和数据驱动。 5. 最后回到[附录：快速查阅卡](#附录快速查阅卡)做字段和写法自检。 **为什么推荐这样写**： - 新人最容易一上来就把 `steps + SQL + retry + extract + teardown_steps` 一次堆满，结果出问题后很难定位。 - 日常最稳的方式是先跑通一条最短链路，再一层层加能力。 - 这也更符合框架本身的设计：驱动文件固定，YAML 逐步增强。 **两种基础结构**： | 基础结构 | 判断标准 | 适用场景 | |----------|----------|----------| | 单接口结构 | 当前用例没有 `steps`，根节点直接写 `request`、`validation` 等字段。 | 查询接口、创建接口、单接口冒烟、单接口参数覆盖。 | | 多步骤结构 | 当前用例有 `steps`，每个 step 表示一个业务动作。 | 创建后查询、创建后修改、下单后取消、入库/出库/波次等完整业务链路。 | **能力搭配选择表**： | 目标 | 推荐组合 | 说明 | |------|----------|------| | 验证单个接口返回是否正确 | 单接口 + `validation` | 最短结构，适合接口级冒烟和回归。 | | 同一个接口跑多组入参 | 单接口 + `parameters` + CSV | `parameters` 必须放在当前单接口用例顶层，由收集阶段裂变成多条用例。 | | 一条业务链路串多个接口 | `config + steps` | 公共请求头、公共变量放在 `config`，业务动作按 `steps` 顺序写。 | | 同一个 YAML 内步骤间传值 | `steps + extract_local` | 变量只在当前用例内有效，优先用于步骤间传递临时 ID、单号、状态值。 | | 跨 YAML 或跨进程传值 | `extract + ${get_extract_data(...)}` | `extract` 会写入 Redis，适合多个测试文件之间共享登录 Token 或上游单据号。 | | 主流程失败后仍要清理业务数据 | `steps + teardown_steps` | `teardown_steps` 在主链路结束后执行，异常只记录日志，不掩盖主流程失败。 | | 当前步骤需要数据库准备或清理 | `setup_sql / teardown_sql` | 写在单接口或某个 step 内，只影响当前步骤。 | | 接口存在异步延迟或短暂失败 | `retry` | 写在单接口或某个 step 内，失败后按 `times` 和 `interval` 重试。 | | 不请求真实服务，只验证框架链路 | `mock` | 写在单接口或某个 step 内，命中后直接返回伪造响应。 | | 浏览器 HAR 回放 | `cli.py record` 自动生成 `config + steps` | 页面流量接口多、请求头重复，推荐用 CLI 生成后再人工微调。 | **常见推荐组合**： | 场景 | 推荐写法 | |------|----------| | 查询列表、查询详情、简单新增 | 单接口 + `validation` | | 同一个查询接口覆盖不同筛选条件 | 单接口 + `parameters` + CSV | | 创建单据后查询详情 | `config + steps + extract_local` | | 创建单据后无论成功失败都要作废 | `config + steps + teardown_steps` | | 下单后等待异步状态流转 | `config + steps`，状态查询 step 加 `retry` | | 测试前插入数据，测试后删除数据 | 单接口或 step 内使用 `setup_sql / teardown_sql` | | 登录 Token 给多个 YAML 使用 | 登录用例用 `extract`，其他用例用 `${get_extract_data(accessToken)}` | ### 2.1 模式一：单接口极简模式（无 steps） **适用场景**：单个接口的独立测试，不需要步骤间变量传递。 **YAML 结构**： ```yaml # 单接口用例，根节点直接是接口配置 - name: 创建标签 base_url: ${get_base_url(base_wx_url)} request: method: post path: /cgi-bin/tags/create?access_token=${get_extract_data(access_token)} json: tag: name: Tag${get_timestamp()} extract: tag_id: $.tag.id validation: - code: 200 - contain: ok ``` **进阶：config 节点示例**（在 YAML 中统一配置全局参数）： ```yaml - config: name: 标签管理测试套件 base_url: ${get_base_url(base_wx_url)} request: headers: Content-Type: application/json validation: - code: 200 # 以下用例共享 config 中的 base_url、headers 和全局断言 - name: 创建标签 request: method: post path: /cgi-bin/tags/create extract: tag_id: $.tag.id validation: - contain: ok ``` **进阶：文件上传**： ```yaml - name: 上传图片 base_url: ${get_base_url(base_wx_url)} request: method: post path: /cgi-bin/media/uploadimg files: files: 'files/mag.jpg' # key = 表单字段名，value = 文件路径（相对项目根目录） validation: - code: 200 - contain: url ``` **进阶：Cookies**： ```yaml - name: 带 Cookie 访问 base_url: ${get_base_url(base_wx_url)} request: method: get path: /api/user/profile cookies: session_id: ${get_extract_data(session_id)} user_role: admin validation: - code: 200 ``` **进阶：Form 表单（`data`）**： ```yaml - name: 表单登录 base_url: ${get_base_url(base_wx_url)} request: method: post path: /api/login headers: Content-Type: application/x-www-form-urlencoded data: username: test_user password: '123456' validation: - code: 200 ``` **进阶：直接写完整 URL**： ```yaml - name: 调用第三方接口 request: method: get url: https://httpbin.org/get # 直接写完整地址，忽略 base_url params: foo: bar validation: - code: 200 ``` **Python 驱动**： ```python import allure import pytest from common.file_readers.yaml_case_reader import read_yaml_cases from common.helpers.allure_id_generator import m_id, c_id from util_tools.execution.executor import Executor @allure.feature(next(m_id) + '标签管理') class TestTagManager: @allure.story(next(c_id) + '创建标签') @pytest.mark.parametrize('caseinfo', read_yaml_cases('testcase/tag/create_tag.yaml')) def test_create_tag(self, caseinfo): allure.dynamic.title(caseinfo['name']) Executor().run_case(caseinfo) ``` **进阶：Allure 装饰器**： 除了示例中的 `@allure.feature`、`@allure.story` 和 `allure.dynamic.title` 外，还支持以下装饰器： ```python # 为单个测试方法添加详细描述（会显示在 Allure 报告详情页） @allure.description('这里写测试场景的详细说明，支持多行文本') # 动态设置 feature/story（适合数据驱动场景，根据用例内容动态分组） allure.dynamic.feature(caseinfo['module']) allure.dynamic.story(caseinfo['sub_module']) # 为报告添加标签，方便按标签筛选 @allure.label('severity', 'critical') @allure.tag('回归测试', 'P0') ``` 完整示例： ```python import allure import pytest from common.file_readers.yaml_case_reader import read_yaml_cases from common.helpers.allure_id_generator import m_id, c_id from util_tools.execution.executor import Executor @allure.feature(next(m_id) + '订单管理') class TestOrder: @allure.story(next(c_id) + '创建订单') @allure.description('验证正常创建订单时，接口返回 code=0 且数据库正确落表') @allure.label('severity', 'blocker') @pytest.mark.parametrize('caseinfo', read_yaml_cases('create_order.yaml')) def test_create_order(self, caseinfo): allure.dynamic.title(caseinfo['name']) Executor().run_case(caseinfo) ``` **特点**： - YAML 根节点直接是单条用例（框架内部包装为 `steps = [case_info]`） - `config` 节点可统一配置 `base_url`、`request`（公共头）、`validation`（全局断言） - 变量通过 `extract` 写入 Redis，跨文件可通过 `${get_extract_data(tag_id)}` 读取 - Allure 报告中不显示步骤折叠面板（因为是单步） - 文件上传通过 `files` 字段配置，框架自动用 ExitStack 托管文件句柄 - `headers`、`cookies`、`params`、`data`、`json`、`files` 为标准 HTTP 参数，框架在发请求前会自动做变量解析 - `get_headers(data)` 和 `get_headers(json)` 可在 YAML 中快速生成 Content-Type 头，通常不需要手动调用（框架会根据请求体类型自动推断） **该模式的运行流程**： ```mermaid graph TD Start(["Pytest 调度用例"]) --> Wrap["read_yaml_cases 读取 YAML 用例列表"] Wrap --> Parametrize["@pytest.mark.parametrize 注册用例"] Parametrize --> Executor["Executor.run_case(case_info)"] subgraph 准备阶段 Executor --> Deepcopy["deepcopy 深拷贝隔离"] Deepcopy --> Config["提取 config 全局配置（如有）"] Config --> Init["初始化 context_vars"] end subgraph 单步执行 Init --> Validate["validate_step_schema 静态语法校验"] Validate --> Setup["setup_sql 前置 SQL（如有）"] Setup --> Retry{重试循环} Retry -->|重试| Parse["VariableResolver.resolve：解析 ${变量}"] Parse --> Send["HttpClient.execute_api_request：发包"] Send --> Json["安全解析响应 JSON（orjson）"] Json --> Allure["AllureReporter.record_api：捕获本次尝试附件"] Allure --> Extract["_extract_response_variables 变量提取"] Extract --> Assert["_run_response_validations 软断言遍历"] Assert --> Check{还有重试次数?} Check -->|失败| Retry Check -->|成功或耗尽| Next["本步结束"] end Next --> Teardown["teardown_sql 后置 SQL（setup_sql 成功后执行）"] Teardown --> End(["结束"]) ``` **流程说明**： | 步骤 | 说明 | |------|------| | `read_yaml_cases` | 读取 YAML、缓存结果并规范化为用例列表 | | `deepcopy` | 深拷贝隔离，防止多进程数据污染 | | `config` | 提取全局配置：`global_request`（公共 headers/cookies/method）+ `global_validation`（全局断言规则） | | `validate_step_schema` | 静态语法校验 | | `setup_sql` | 前置 SQL（如有） | | `VariableResolver.resolve` | 解析 `${变量}` 和 `${函数()}` | | `HttpClient.execute_api_request` | 发包，含 Mock 拦截检查 | | `_extract_response_variables` | 变量提取四大通道：`extract` / `extract_local` / `extract_list` / `extract_list_local` | | `_run_response_validations` | 软断言遍历：`code` / `eq` / `ne` / `contain` / `db` | | `teardown_sql` | 后置 SQL；`setup_sql` 成功后才会执行，之后请求、提取或断言失败也会执行 | --- ### 2.2 模式二：多步骤链路模式（steps + teardown_steps） **适用场景**：业务流程需要多个接口串联，如"创建 → 查询 → 修改 → 删除"，且需要步骤间共享变量。 **YAML 结构**： > 注意：多步骤流程统一使用 `config + steps`。`steps` 和 `teardown_steps` 必须与 `config` 同级，不能写到 `config` 里面。公共变量统一写在 `config.variables`，步骤内通过 `${变量名}` 读取。 ```yaml - config: name: 标签全生命周期测试 base_url: ${get_base_url(base_wx_url)} variables: tag_name: Tag${get_timestamp()} request: method: post # 主业务链路（按顺序执行） steps: - name: 步骤1-创建标签 request: path: /cgi-bin/tags/create?access_token=${get_extract_data(access_token)} json: tag: name: ${tag_name} extract_local: tag_id: $.tag.id validation: - code: 200 - contain: ok - name: 步骤2-编辑标签 request: path: /cgi-bin/tags/update json: tag: id: ${tag_id} # 引用上一步提取的内存变量 name: EditedTag${get_timestamp()} validation: - code: 200 - contain: ok - name: 步骤3-删除标签 request: path: /cgi-bin/tags/delete json: tag: id: ${tag_id} # 继续引用同一内存变量 validation: - code: 200 - contain: ok # 环境兜底清理（无论主链路成功/失败，强制执行） teardown_steps: - name: 清理-文件上传 request: method: post path: /cgi-bin/media/uploadimg files: files: 'files/mag.jpg' validation: - code: 200 ``` **Python 驱动**（与单接口模式完全相同）： ```python @pytest.mark.parametrize('caseinfo', read_yaml_cases('testcase/tag/lifecycle.yaml')) def test_tag_lifecycle(self, caseinfo): allure.dynamic.title(caseinfo['name']) Executor().run_case(caseinfo) ``` **关键机制**： | 机制 | 说明 | |------|------| | `extract_local` | 变量仅写入内存（`context_vars`），当前用例生命周期内有效，不污染 Redis | | `${tag_id}` | 内存变量直接引用，无需 `get_extract_data()` | | `teardown_steps` | 用 `try-finally` 保证执行，异常静默隔离 | | `config.variables` | 流程级初始变量池，可在 steps 前预定义公共变量 | **进阶：config 公共参数提取** 当多步骤链路中每个步骤都使用相同的 `base_url`、`method` 或 `headers` 时，可将公共参数抽到 `config` 节点中，各步骤只写差异化参数： ```yaml - config: name: yaml_内存关联_公共参数 base_url: ${get_base_url(base_wx_url)} variables: tag_name: Tag${get_timestamp()} request: method: post # 所有步骤共享 POST 方法 steps: - name: 创建标签 request: path: /cgi-bin/tags/create?access_token=${get_extract_data(access_token)} json: tag: name: ${tag_name} extract_local: tag_id: $.tag.id validation: - code: 200 - contain: ${tag_id} - name: 查询标签 request: path: /cgi-bin/tags/get params: access_token: ${get_extract_data(access_token)} extract_list_local: all_ids: "$.tags[*].id" validation: - code: 200 - contain: name - name: 编辑标签 request: path: /cgi-bin/tags/update json: tag: id: ${tag_id} name: Test${get_timestamp()} validation: - code: 200 - contain: ok - name: 删除标签 request: path: /cgi-bin/tags/delete json: tag: id: ${tag_id} validation: - code: 200 - contain: ok ``` **合并规则**：Executor 在执行每个 step（含 `teardown_steps`）前，会将 `config.request` 深度合并到 `step.request` 中： - `config.request.method = post` → 步骤未写 `method` 时自动继承 - `config.request.headers` → 与步骤级 `headers` 做字典合并（步骤级同名字段覆盖 config 级） - `config.base_url` → 步骤未写 `base_url` 时自动继承（步骤级可单独覆盖） **进阶：步骤级重试配置** 针对微服务异步延迟场景，单步骤可配置轮询重试： ```yaml steps: - name: 查询订单状态 request: method: post path: /api/order/status json: orderNo: ${order_no} retry: times: 5 # 最大执行 5 次（含首次执行 1 次 + 最多重试 4 次） interval: 3 # 每次失败后，等待 3 秒再发起下一次尝试 validation: - eq: { 'data.status': 'shipped' } ``` 说明：这里的 `retry` 是 Executor 内部的步骤级重试，只重试当前接口步骤，统计到 `step_retries`；`times` 保持历史字段名，含义是“最大执行次数”，不是“失败后额外重试次数”；`pytest-rerunfailures` 的 `--reruns` 是整条 pytest 用例重跑，统计到 `case_reruns`。报告和通知会分别展示两类数据，避免把步骤重试误读为用例重跑。控制台日志会展示每次尝试；Allure 只保留最终尝试的附件，中间失败尝试的请求、响应、提取和断言附件会被丢弃。 **进阶：变量作用域建议** 框架保留两种变量作用域： - 单个 YAML 链路内临时传递：使用 `extract_local` / `extract_list_local` 写入内存变量池，后续步骤用 `${变量名}` 读取，用例执行完即销毁。`extract_list_local` 提取的是列表，既可以整体用 `${变量名}`，也可以按 0 基下标读取单个值，例如 `${sku_ids[0]}`。 - 跨 YAML 或跨用例共享：使用 `extract` / `extract_list` 写入 Redis，其他文件用 `${get_extract_data(变量名)}` 读取。 跨 YAML 共享变量时，同一测试批次内所有用例共用同一个 SessionID Redis 前缀，所以变量名应带业务含义，避免不同流程互相覆盖。例如： ```yaml config: name: 订单链路 steps: - name: 创建订单 request: method: post path: /api/order/create extract: order_flow_order_id: $.data.orderId - name: 查询订单 request: method: get path: /api/order/${order_flow_order_id} ``` **进阶：逆向提取（extract_from_request）** 某些数据（如 TraceId、Nonce）只在请求中发送，响应不回显： ```yaml steps: - name: 创建订单 request: method: post path: /api/order/create headers: X-Request-Id: ${generate_order_number(req_)} json: sku: autoSKU extract: # 从已发出的请求头中逆向提取 TraceId trace_id: $.request.headers.X-Request-Id # 从已发出的请求体中逆向提取 sku req_sku: $.request.json.sku validation: - code: 200 - name: 查询订单 request: method: post path: /api/order/query json: # 用逆向提取的 trace_id 做断言 traceId: ${trace_id} ``` **进阶：前后置 SQL** ```yaml steps: - name: 查询用户积分 request: method: post path: /api/user/points setup_sql: # 发请求前执行 - "INSERT INTO t_user_points (user_id, points) VALUES (1001, 500)" teardown_sql: # setup_sql 成功后，步骤请求/提取/断言无论成败都会执行 - "DELETE FROM t_user_points WHERE user_id = 1001" validation: - code: 200 - db: sql: "SELECT points FROM t_user_points WHERE user_id = 1001" eq: { points: 500 } ``` **进阶：自研高阶过滤语法** 原生 jsonpath 不支持条件过滤，框架自研了过滤引擎： ```yaml # 从 rows 列表中找出 name 等于 "张三" 的所有行，提取其 order_no extract: order_no: "$.data.rows[?(@.name == '张三')].order_no" # 从 tags 列表中找出 name 包含 "测试" 的所有行，提取整行数据 extract_list: test_tags: "$.tags[?(@.name contains '测试')]" ``` | 运算符 | 说明 | 示例 | |--------|------|------| | `==` | 全等比较（字符串化后比较） | `[?(@.status == '1')]` | | `contains` | 包含判断 | `[?(@.name contains '测试')]` | **该模式的运行流程**： ```mermaid graph TD Start(["Pytest 调度用例"]) --> Yaml["read_yaml_cases 读取 YAML（含 steps → 多步骤链路）"] Yaml --> Executor["Executor.run_case(case_info)"] subgraph 初始化 Executor --> Deepcopy["deepcopy 深拷贝隔离"] Deepcopy --> Config["提取 config 全局配置（如有）"] Config --> ExtractSteps["提取 steps 列表"] ExtractSteps --> ExtractTeardown["提取 teardown_steps 列表"] ExtractTeardown --> InitVars["初始化 context_vars + config.variables"] end subgraph 主链路 try 块 InitVars --> ExitStack["with ExitStack 资源托管"] ExitStack --> Loop{遍历 steps} Loop -->|有下一个| Merge["合并全局配置到当前 step"] Merge --> AllureStep["allure_step_context 折叠面板"] AllureStep --> ExecStep["StepRunner.execute_step"] ExecStep --> CheckStep{步骤结果?} CheckStep -->|成功| Loop CheckStep -->|失败| SaveEx["保存异常 + break 中断"] SaveEx --> Finally Loop -->|结束| Finally["finally 块"] end subgraph Teardown 兜底 Finally --> TdLoop{遍历 teardown_steps} TdLoop -->|有下一个| SilentExec["StepRunner.execute_step（异常静默隔离）"] SilentExec --> TdLoop TdLoop -->|结束| CheckEx{main_exception 存在?} CheckEx -->|是| Raise["重新抛出异常（Pytest 标红）"] CheckEx -->|否| End(["结束"]) end Raise --> End ``` **流程说明**： | 步骤 | 说明 | |------|------| | `deepcopy` | 深拷贝隔离，防止多进程数据污染 | | `context_vars + config.variables` | 将 `config.variables` 初始化到内存变量池，供步骤间共享 | | `ExitStack` | 资源托管，确保文件句柄等自动释放 | | `合并全局配置` | 将 config.request 深度合并到当前 step（含 teardown_steps） | | `allure_step_context` | 生成 Allure 折叠面板，步骤可视化 | | `VariableResolver.resolve` | 解析 `${变量}`，含上一步 `extract_local` 的变量 | | `_extract_response_variables` | `extract_local` → 写入 `context_vars`，供后续步骤读取 | | `teardown_sql` | 单步后置 SQL；`setup_sql` 成功后，请求、提取或断言失败也会执行 | | `异常静默隔离` | teardown 异常只打印警告，不掩盖主链路错误 | | `main_exception` | 主链路异常在 teardown 后重新抛出 | **关键执行顺序**： 1. `config.variables` 初始值 → 放入 `context_vars` 2. 步骤1 `extract_local` → 写入 `context_vars` → 步骤2 通过 `${变量名}` 读取 3. 某步骤失败 → 保存异常 → break 中断剩余 steps 4. `finally` 强制执行 `teardown_steps` 5. teardown 异常静默隔离 → 主链路异常重新抛出 --- ### 2.3 模式三：CSV 数据驱动模式（parameters） **适用场景**：同一个接口需要跑多组不同参数，如正例/反例、边界值测试。 > 注意：`parameters` 必须放在当前单接口用例顶层，不能写成 `config.parameters`。CSV 数据驱动用于展开当前用例模板，放进 `config` 后框架不会识别。 **CSV 文件**（`data/wx_tag_create.csv`）： ```csv name,method,validation 正常创建_创建标签1,post,${get_extract_data(003_tag_id)} 正常创建_创建标签2,post,${get_extract_data(003_tag_name)} 创建异常_请求方式错误,get,44002 ``` **YAML 结构**： ```yaml - name: $csv{name} # 用例名称从 CSV 读取 base_url: ${get_base_url(base_wx_url)} parameters: name-method-validation: data/wx_tag_create.csv # Key = CSV 列名组合（-连接） request: method: $csv{method} # HTTP 方法从 CSV 读取 path: /cgi-bin/tags/create?access_token=${get_extract_data(access_token)} json: tag: name: Tag${get_timestamp()} extract: 003_tag_id: $.tag.id 003_tag_name: $.tag.name validation: - contain: $csv{validation} # 断言预期值从 CSV 读取 ``` **裂变过程**： ```mermaid graph TD Start(["YAML 模板（1 条）"]) --> ReadCsv["expand_csv_cases() 读取 CSV"] subgraph CSV 数据 ReadCsv --> Header["表头: [name, method, validation]"] Header --> Row1["数据行1: 正常创建_创建标签1, post, ..."] Header --> Row2["数据行2: 正常创建_创建标签2, post, ..."] Header --> Row3["数据行3: 创建异常_请求方式错误, get, 44002"] end Row1 --> Replace["文本替换 + orjson 反序列化"] Row2 --> Replace Row3 --> Replace subgraph 裂变结果 Replace --> Case1["裂变用例1（name=正常创建_创建标签1）"] Replace --> Case2["裂变用例2（name=正常创建_创建标签2）"] Replace --> Case3["裂变用例3（name=创建异常_请求方式错误）"] end Case1 --> End(["N 个独立 TestItem"]) Case2 --> End Case3 --> End ``` **裂变说明**： | 阶段 | 说明 | |------|------| | `expand_csv_cases()` | 读取 CSV 文件，解析表头和数据行 | | `文本替换` | 将 `$csv{列名}` 替换为实际值 | | `orjson 反序列化` | 保证数字、布尔、字典等类型保真 | | `裂变结果` | 1 个 YAML + 1 个 CSV → N 条独立用例，各自独立重试/报告/熔断 | **特点**： - 1 个 YAML + 1 个 CSV → 裂变出 N 条独立用例 - 多 CSV 源支持笛卡尔积（3 行 × 2 行 = 6 条用例） - CSV 单元格支持 AST 类型推导（`{` 或 `[` 开头自动解析为 dict/list） **数据驱动裂变的本质**： 不是"一个用例跑多次"，而是"一个 YAML 模板在收集阶段就裂变成 N 个独立的 TestItem"。这 N 个 TestItem 在 Pytest 看来是完全独立的用例，可以独立重试、独立报告、独立熔断。 **该模式的运行流程**： ```mermaid graph TD Start(["Pytest 收集阶段（pytest_generate_tests）"]) --> Yaml["read_yaml_cases 读取 YAML"] Yaml --> Check{发现 parameters 节点?} Check -->|是| csv_case_expander["expand_csv_cases()"] subgraph 用例裂变 csv_case_expander --> ReadCsv["读取 CSV 文件（带 LRU 缓存）"] ReadCsv --> ParseCsv["解析表头 + 数据行"] ParseCsv --> BuildRule["构建替换规则（$csv{列名}）"] BuildRule --> Product["itertools.product 笛卡尔积组合"] Product --> Replace["str.replace 全局文本替换"] Replace --> Deserialize["orjson.loads 反序列化"] Deserialize --> AddList["加入 new_case_list"] end Check -->|否| Parametrize AddList --> Parametrize["@pytest.mark.parametrize 注册用例"] Parametrize --> End1(["收集阶段结束"]) Start2(["Pytest 执行阶段"]) --> Execute["每个 TestItem 独立调度到 Worker"] Execute --> Single["执行流程同【单接口极简模式】"] Single --> End2(["结束"]) ``` **流程说明**： | 步骤 | 说明 | |------|------| | `parameters` 节点 | 触发 expand_csv_cases 进行用例裂变 | | `LRU 缓存` | CSV 文件读取结果缓存，避免重复读盘 | | `itertools.product` | 多 CSV 源时做笛卡尔积组合 | | `str.replace` | 将 `$csv{列名}` 替换为实际值 | | `orjson.loads` | 反序列化保证数字、布尔等类型保真 | | 裂变时机 | 发生在**收集阶段**，不是运行阶段 | | 独立性 | 裂变后的每个用例独立：独立重试、独立报告、独立熔断 | **关键要点**： - 裂变发生在**收集阶段**，不是运行阶段 - 裂变后的每个用例都是**独立**的：独立重试、独立报告、独立熔断 - `$csv{xxx}` 占位符在裂变时就被替换为实际值，运行时不再处理 CSV --- ### 2.4 增强能力一：跨文件 Redis 共享（extract + get_extract_data） **适用场景**：多个 YAML 文件之间需要传递变量，如文件 A 创建订单 → 文件 B 查询订单 → 文件 C 取消订单。 **文件 A**（`001_create_order.yaml`）： ```yaml - name: 创建订单 base_url: ${get_base_url(base_wx_url)} request: method: post path: /api/order/create json: sku: autoSKU count: 1 extract: # 写入 Redis（跨文件共享） order_id: $.data.orderId validation: - code: 200 - eq: { code: 0 } ``` **文件 B**（`002_query_order.yaml`）： ```yaml - name: 查询订单 base_url: ${get_base_url(base_wx_url)} request: method: post path: /api/order/query json: orderId: ${get_extract_data(order_id)} # 从 Redis 读取文件 A 写入的变量 validation: - code: 200 - contain: autoSKU ``` **Python 驱动**： ```python import allure import pytest from common.file_readers.yaml_case_reader import read_yaml_cases from common.helpers.allure_id_generator import m_id, c_id from util_tools.execution.executor import Executor @allure.feature(next(m_id) + "订单全链路") class TestOrderFlow: @allure.story(next(c_id) + '创建订单') @pytest.mark.parametrize('caseinfo', read_yaml_cases('001_create_order.yaml')) def test_create_order(self, caseinfo): allure.dynamic.title(caseinfo['name']) Executor().run_case(caseinfo) @allure.story(next(c_id) + '查询订单') @pytest.mark.parametrize('caseinfo', read_yaml_cases('002_query_order.yaml')) def test_query_order(self, caseinfo): allure.dynamic.title(caseinfo['name']) Executor().run_case(caseinfo) ``` `m_id` 和 `c_id` 是 `allure_id_generator.py` 提供的 Allure 编号迭代器，用于在报告中自动生成有序的 feature/story 编号。测试文件推荐使用 `next(m_id)`、`next(c_id)` 这种短写法，阅读成本更低；底层会在 pytest 收集完成后，按本轮实际收集到的用例重新编号，保证全量运行或只运行部分目录时都从 `M001_`、`C001_` 开始。 **关键区别**： | 特性 | `extract_local` | `extract` | |------|-----------------|-----------| | 存储位置 | 内存（`context_vars`） | 内存（`context_vars`）+ Redis | | 作用范围 | 当前 YAML 链路生命周期 | 当前 YAML 链路 + 跨文件、跨进程 | | 读取方式 | 当前 YAML 内 `${变量名}` | 当前 YAML 内 `${变量名}`；跨 YAML 用 `${get_extract_data(变量名)}` | | 并发安全 | 当前 YAML 链路内安全 | 多进程安全（Key 含环境名 + SessionID 前缀） | | 适用场景 | 单文件内步骤间传递 | 多文件间传递、pytest-xdist 并发 | **跨文件传递的关键**： 1. **写入方**（文件 A）：用 `extract`（不是 `extract_local`）→ 数据写入 Redis 2. **读取方**（文件 B）：用 `${get_extract_data(order_id)}` → RuntimeFunctions 直接读取 Redis 3. **批次隔离**：Redis Key 包含环境名 + SessionID，不同测试批次互不串数据 4. **同一批次内跨文件/跨进程**：因为 SessionID 相同，文件 A 和文件 B 的 Redis Key 前缀一致，可以互相读取 **该模式的运行流程**： ```mermaid graph TD subgraph 文件A 写入方 A1["read_yaml_cases('001_create.yaml')"] --> A2["Executor.run_case"] A2 --> A3["StepRunner.execute_step"] A3 --> A4["HttpClient.execute_api_request（真实网络请求）"] A4 --> A5["响应成功"] A5 --> A6["_extract_response_variables"] A6 --> A7["extract: order_id → context_vars"] A7 --> A8["RuntimeCache.write_extract_variables → Redis"] A8 --> A9["Key: extract:{env}:{session}:order_id，Value: ORD123"] A9 --> A10["pytest_runtest_makereport（无失败 → 不触发熔断）"] end subgraph 文件B 读取方 B1["read_yaml_cases('002_query.yaml')"] --> B2["Executor.run_case"] B2 --> B3["StepRunner.execute_step"] B3 --> B4["build_request_data"] B4 --> B5["VariableResolver.resolve（${get_extract_data(order_id)}）"] B5 --> B6["RuntimeFunctions.get_extract_data"] B6 --> B7["RuntimeCache.read_extract_variable → 查 Redis"] B7 --> B8["Key 命中 → 返回 ORD123"] B8 --> B9["组装 json: {orderId: ORD123}"] B9 --> B10["HttpClient.execute_api_request（真实请求）"] B10 --> B11["断言通过"] B11 --> B12["pytest_runtest_makereport → 通过"] end A9 -.->|跨文件变量传递| B7 ``` **流程说明**： | 步骤 | 说明 | |------|------| | `extract` | 写入当前 YAML 的 `context_vars` + Redis，Redis Key 含环境名 + SessionID 前缀 | | `get_extract_data` | RuntimeFunctions 直接读取 Redis | | `SessionID 隔离` | 同一测试批次：SessionID 相同 → Key 前缀相同 → 可互相读取 | | `跨进程共享` | 同一测试批次内多个 Worker 共享 SessionID，可读取同一批次的 Redis 变量 | | `熔断影响` | 文件 A 失败被熔断 → 文件 B 在 `pytest_runtest_call` 阶段被 skip | **关键要点**： - 文件 A 的 `extract` 规则执行成功后，数据会同时写入当前 YAML 的 `context_vars` 和 Redis - 如果文件 A 失败被熔断，文件 B 会在 `pytest_runtest_call` 阶段被 `pytest.skip()` 跳过 - 同一测试批次内跨文件/跨进程：SessionID 相同 → Redis Key 前缀相同 → 可以互相读取 - 不同测试批次：SessionID 不同 → Redis Key 前缀不同，避免串数据 --- ### 2.5 增强能力二：Mock 挡板 **适用场景**：下游接口不可用、需要模拟异常响应（如超时、404）、或测试尚未对接的接口。 **YAML 结构**： ```yaml # 模式 A：简单 Mock（固定返回） - name: 模拟商品详情 base_url: ${get_base_url(base_wx_url)} mock: status_code: 200 headers: Content-Type: application/json json: code: 0 data: sku_name: "iPhone 15" price: 5999 request: method: get path: /api/sku/detail params: id: 1001 validation: - eq: { code: 0 } # 模式 B：顺序 Mock（第1次返回A，第2次返回B） - name: 模拟异步轮询 mock: chain: - status_code: 202 json: { status: "PENDING", progress: 50 } - status_code: 200 json: { status: "SUCCESS", result: "http://download/report.pdf" } request: method: get path: /api/task/status validation: - eq: { status: "PENDING" } # 模式 C：条件 Mock（根据请求参数匹配） - name: 模拟订单状态 mock: rules: - when: { order_id: "A001" } then: status_code: 200 json: { id: "A001", status: "SHIPPED" } - when: { order_id: "B999" } then: status_code: 404 json: { code: 404001, msg: "Not Found" } request: method: POST path: /api/order/info json: order_id: "A001" validation: - eq: { status: "SHIPPED" } ``` **进阶：数据库断言（db）在单接口中的使用** ```yaml - name: 创建订单并校验落库 base_url: ${get_base_url(base_wx_url)} request: method: post path: /api/order/create json: sku: autoSKU count: 1 extract: order_no: $.data.orderNo validation: - code: 200 - eq: { code: 0 } # 校验数据库是否正确落表 # args 是可选的；当 SQL 里使用 %s 占位符时才需要传 - db: sql: "SELECT order_no, status FROM t_order WHERE sku = %s ORDER BY id DESC LIMIT 1" args: ["autoSKU"] eq: { order_no: '${order_no}', status: 0 } # 校验关联表数据 - db: sql: "SELECT count(*) as cnt FROM t_order_item WHERE order_no = %s" args: ["${order_no}"] contain: { cnt: 1 } # 校验数据是否存在：empty=false 表示期望 SQL 能查到数据 - db: sql: "SELECT id FROM t_order WHERE order_no = %s" args: ["${order_no}"] empty: false # 校验数据是否不存在：empty=true 表示期望 SQL 查不到数据 - db: sql: "SELECT id FROM t_order_error WHERE order_no = %s" args: ["${order_no}"] empty: true ``` **DB 三种断言模式**： | 模式 | 适用场景 | 比对规则 | |------|----------|----------| | `eq` | 查询字段和预期字段需要完全一致 | 实际第一行必须和预期字典完全相等，字段多或字段少都会失败 | | `contain` | 只关心部分字段是否正确 | 预期字典中的字段和值都能在实际第一行中匹配即可 | | `empty` | 校验数据是否存在或不存在 | `empty: true` 表示期望查不到数据；`empty: false` 表示期望查到数据 | **SQL 安全拦截**：数据库断言内置 DML/DDL 黑名单（insert/update/delete/drop/truncate/alter/create），仅允许 `SELECT` 查询穿透，防止测试框架误改生产数据。 **Mock 拦截原理**： ```mermaid graph TD Start(["HttpClient.execute_api_request"]) --> Check{case_info 含 mock 节点?} Check -->|命中| Mock["MockResponse.build_fake_response"] Mock --> FakeResp["_FakeResp（不发起真实网络请求）"] Check -->|未命中| Real["GLOBAL_SESSION.request"] Real --> Response["真实 Response"] FakeResp --> Return["统一返回 Response/FakeResp"] Response --> Return Return --> End(["上层代码无感知"]) ``` **拦截说明**： | 特性 | 说明 | |------|------| | 拦截位置 | HttpClient.execute_api_request 内部，真实网络请求发出之前 | | 上层无感知 | Executor、DataExtractor、Assertions 统一按正常 Response 处理 | | _FakeResp | 模拟 requests.Response 的常用接口（status_code、text、json()、headers 等） | **_FakeResp 的接口模拟**： | 属性/方法 | 说明 | |-----------|------| | `status_code` | HTTP 状态码 | | `text` | 响应文本（JSON 序列化后的字符串）| | `content` | UTF-8 字节流 | | `json()` | 返回 Python 对象（省去反序列化开销）| | `ok` | 200~299 返回 True | | `headers` | 响应头字典 | | `elapsed` | 模拟耗时（固定 0 秒）| **该模式的运行流程**： ```mermaid graph TD Start(["StepRunner.execute_step"]) --> Validate["validate_step_schema 静态语法校验"] Validate --> Setup["setup_sql 前置 SQL（如有）"] Setup --> Retry{重试循环} Retry -->|重试| Prepare["build_request_data：组装请求"] Prepare --> Http["HttpClient.execute_api_request"] subgraph Mock 拦截 Http --> Log["打印审计日志"] Log --> CheckMock{case_info 含 mock 节点?} CheckMock -->|否| RealReq["GLOBAL_SESSION.request（真实网络）"] CheckMock -->|是| MockBuild["MockResponse.build_fake_response"] MockBuild --> MockType{Mock 类型判断} MockType -->|简单 Mock| Simple["直接返回固定数据"] MockType -->|chain 顺序| Chain["按 _mock_idx 轮询返回"] Chain --> Idx["_mock_idx += 1 写回 case_info"] MockType -->|rules 条件| CheckRule{"when 是 request_json 子集?"} CheckRule -->|是| ReturnThen["返回 then 对应数据"] CheckRule -->|否| RealReq end RealReq --> Response["返回 Response"] Simple --> Response Idx --> Response ReturnThen --> Response Response --> Json["安全解析响应 JSON（orjson）"] Json --> Allure["AllureReporter.record_api：捕获本次尝试附件"] Allure --> Extract["_extract_response_variables 变量提取"] Extract --> Assert["_run_response_validations 软断言"] Assert --> CheckRetry{还有重试次数?} CheckRetry -->|失败| Retry CheckRetry -->|成功或耗尽| Next["本步结束"] Next --> Teardown["teardown_sql 后置 SQL（setup_sql 成功后执行）"] Teardown --> End(["结束"]) ``` **流程说明**： | 步骤 | 说明 | |------|------| | `Mock 拦截位置` | HttpClient.execute_api_request 内部，真实网络请求发出之前 | | `上层无感知` | Executor、DataExtractor、Assertions 统一按正常 Response 处理 | | `简单 Mock` | 无条件返回固定 status_code、json、headers | | `chain 顺序` | `_mock_idx` 写回 `case_info`，保证同用例多次调用顺序轮询 | | `rules 条件` | `when` 的所有键值对都存在于 `request_json` 中即命中，否则走真实网络 | | `teardown_sql` | 无论 Mock 或真实请求，后置 SQL 均执行 | **关键要点**： - Mock 拦截发生在 **HttpClient.execute_api_request** 内部，在真实网络请求发出之前 - 上层代码（Executor、DataExtractor、Assertions）**完全无感知**，统一按正常 Response 处理 - chain 模式的 `_mock_idx` 写入 `case_info` 字典，保证同一用例多次调用时按顺序轮询 - rules 条件匹配使用子集判断：`when` 的所有键值对都存在于 `request_json` 中即命中 --- ### 2.6 增强能力三：HAR 流量录制回放 **适用场景**：已有线上流量或浏览器抓包数据，需要快速转换为可回归执行的自动化用例。 **使用方式**： ```bash # 将浏览器抓包的 HAR 文件转换为 YAML + pytest 驱动文件 python cli.py record data/har/flow.har ``` 默认输出目录： ```text testcase/har_case/flow/ ├── flow.yaml └── test_flow.py ``` **CLI 参数说明**： | 参数 | 说明 | 默认值 | |------|------|--------| | `har_path` | HAR 文件路径（位置参数，必填） | - | | `-o, --output` | 输出根目录，最终会在下面按 HAR 文件名创建子目录 | `testcase/har_case` | | `-b, --base-url` | 基础 URL 占位符 | `${get_base_url(base_wms_url)}` | | `-d, --domain` | 在业务域名白名单基础上进一步收窄，只录制指定域名 | "" | | `--name` | 指定生成目录名、YAML 文件名和 pytest 文件名 | HAR 文件名 | | `--feature` | 生成 pytest 文件中的 Allure feature 文案 | `HAR自动生成用例` | | `--story` | 生成 pytest 文件中的 Allure story 文案 | 生成后的用例名 | **生成限制与前置条件**： | 限制项 | 当前规则 | 处理建议 | |--------|----------|----------| | 业务域名白名单 | 默认只保留 `configs/envs/*.yaml` 的 `base_url` 中出现过的域名。 | 录制前确认目标域名已经写入环境配置；如果 HAR 里混有多个业务域名，可用 `-d` 进一步收窄。 | | API 路径白名单 | 只保留 `/api/`、`/wms/`、`/oms/`、`/proxy/`、`/pda/` 开头的请求。 | 即使域名已配置到 `base_url`，路径不命中这些前缀也会被过滤。例如微信 `https://api.weixin.qq.com/cgi-bin/token` 的域名能命中 `base_wx_url`，但 `/cgi-bin/` 不在白名单内，默认不会生成。 | | 静态资源过滤 | `.js`、`.css`、图片、字体、`.map`、`/speed`、`/version.json` 等不会生成到 YAML。 | 这是正常行为，HAR 里有这些请求不代表生成失败。 | | OPTIONS 预检 | 浏览器 CORS 预检请求会跳过。 | 自动化用例只保留真实业务请求，预检请求不需要手写。 | | 请求体类型 | 只自动解析 `application/json` 和 `application/x-www-form-urlencoded`。 | 文件上传、二进制、GraphQL、自定义 MIME 类型需要生成后人工补充。 | | 请求头 | 会过滤 `host`、`user-agent`、`accept`、`origin`、`referer`、`sec-*`、`cookie` 等浏览器噪音头。 | 如业务确实依赖某个被过滤的头，需要在生成后手动补回 YAML。 | | Cookie | WMS/OMS 域名优先写成 `${get_base_url(wms_cookie)}` / `${get_base_url(oms_cookie)}`。 | 环境配置里的 Cookie 必须可用；HAR 中录到的一次性 Cookie 不建议长期固化。 | | 动态参数 | 只自动替换常见字段：`userId`、`clientId`、`accessToken`、`warehouseId`、`timestamp`、`nonce`。 | 订单号、SKU、批次号、业务 ID 等仍需人工改成 `${generate_order_number()}` 或变量提取。 | | 变量提取 | 默认生成 `extract:` 空节点，不会自动推断步骤间依赖。 | 需要人工补 `extract` / `extract_local`，再在后续步骤用 `${变量名}` 引用。 | | 自动断言 | 只生成基础断言：HTTP 状态码、`success`、有效 `code`、成功文案、常见 `data` 主键。 | 业务强断言必须人工补充，不能只依赖录制时返回值。 | | 生成结果 | 如果所有请求都被过滤，不会生成 YAML/PY 文件。 | 查看终端的 HAR 过滤统计，确认是域名、路径、静态资源还是请求体问题。 | **生成的 YAML 结构**： ```yaml - config: name: 基于流量回放自动生成的测试链路 (包含 4 个接口) base_url: ${get_base_url(base_wms_url)} variables: {} request: method: POST headers: Content-Type: application/json;charset=UTF-8 cookies: ${get_base_url(oms_cookie)} steps: - name: '步骤 1: create - /oms/order/create' request: path: /oms/order/create json: warehouseId: 7694 trackingNo: HAHAHA9999 extract: validation: - code: 200 - eq: {success: true} - name: '步骤 2: list - /oms/order/list' request: path: /oms/order/list extract: validation: - code: 200 - eq: {success: true} ``` **生成的 pytest 驱动文件结构**： ```python @allure.feature(next(m_id) + "HAR自动生成用例") class TestFlowAutoGen: @allure.story(next(c_id) + "flow") @pytest.mark.parametrize("caseinfo", read_yaml_cases("testcase/har_case/flow/flow.yaml")) def test_flow(self, caseinfo): allure.dynamic.title(caseinfo.get("name", "flow")) Executor().run_case(caseinfo) ``` **特点**： - 自动从 HAR 文件中提取 URL、Method、Headers、Body、Cookies - 默认只保留 `configs/envs/*.yaml` 中配置过的业务域名，请求路径还必须命中 `/api/`、`/wms/`、`/oms/`、`/proxy/`、`/pda/` 等业务前缀 - 自动过滤静态资源和前端杂请求，例如 `.js`、`.css`、`.png`、`/speed`、`/version.json` - 页面接口 Cookie 优先按域名映射为 `${get_base_url(wms_cookie)}` 或 `${get_base_url(oms_cookie)}`；如果域名不在配置中，才从 HAR 的 `request.cookies` 或 `Cookie` 请求头解析 - 智能推导基础断言规则：永远生成 `code` 状态码断言；JSON 响应中优先生成 `success` 断言；`code` 为空字符串时不会生成无意义断言 - 多步骤自动编排为 `steps` 链路结构 - 默认同时生成 YAML 和 `test_*.py`，pytest 可以直接发现并执行 - 生成后仍建议人工检查业务强断言、变量提取规则和动态参数 **该模式的运行流程**： ```mermaid graph TD Start(["CLI 命令执行：python cli.py record"]) --> Cmd["handle_record_command()"] Cmd --> Clean["清洗参数（去掉拖拽带入的引号/空格）"] Clean --> Generator["HarCaseGenerator.generate()"] Generator --> Engine["HarToYaml.convert()"] subgraph 转换引擎 Engine --> ParseHar["解析 HAR JSON（提取 log.entries）"] ParseHar --> Filter["业务域名 + 业务路径 + 静态资源过滤"] Filter --> Loop{遍历每个 entry} Loop -->|有下一个| Extract1["提取 request.method"] Extract1 --> Extract2["提取 request.url → 解析为 path"] Extract2 --> Extract3["提取 request.headers（保留 Content-Type 等关键头）"] Extract3 --> ExtractCookie["按域名写入 wms_cookie / oms_cookie"] ExtractCookie --> Extract4["提取 request.postData.text → 解析为 json/data"] Extract4 --> Extract5["提取 response.content.text（智能推导断言）"] Extract5 --> Derive{"响应体是 JSON?"} Derive -->|是| EqAssert["优先推导 success/code/msg/data 主键断言"] Derive -->|否| Assemble["组装为 step 字典"] EqAssert --> Assemble Assemble --> Loop Loop -->|结束| Combine["提取公共 method/headers/cookies → 组装 YAML"] Combine --> WriteYaml["写入 .yaml"] end WriteYaml --> WritePy["生成 test_.py"] WritePy --> End1(["离线转换结束"]) Start2(["测试执行阶段（生成文件后）"]) --> Manual["人工检查业务断言、extract、动态参数"] Manual --> Exec["执行流程同【多步骤链路模式】（steps 结构）"] Exec --> End2(["结束"]) ``` **流程说明**： | 步骤 | 说明 | |------|------| | `离线转换` | HAR 转换是一次性离线操作，输出 YAML + pytest 驱动文件 | | `请求过滤` | 默认按环境配置中的业务域名过滤，同时跳过静态资源、埋点、版本文件等杂请求 | | `Cookie 处理` | WMS 域名写入 `${get_base_url(wms_cookie)}`，OMS 域名写入 `${get_base_url(oms_cookie)}` | | `智能推导断言` | 若响应体是 JSON，优先推导 `success`、有效 `code`、成功文案和常见业务主键断言 | | `人工补充` | 需补充变量提取规则、动态参数替换、业务强断言、CSV 绑定等 | | `断言局限` | 自动断言只适合快速起步，复杂业务规则仍需人工补充 | | `执行阶段` | 生成的是多步骤 `steps` 结构，执行流程与手写多步骤链路完全一致 | **关键要点**： - HAR 转换是**一次性的离线操作**，不参与测试运行时 - HAR 生成结果是“可运行模板”，不是最终业务用例。生成后必须检查路径、Cookie、动态参数、变量提取和业务断言 - 生成的 YAML 是多步骤 `steps` 结构，执行流程与手写多步骤链路完全一致 - 默认会同时生成 `test_*.py`，文件名满足 pytest 自动发现规则 - 页面接口 Cookie 会优先引用环境配置，不直接固化 HAR 中容易过期的真实 Cookie - 需要人工补充的内容：变量提取规则、动态参数（如订单号/时间戳）、CSV 绑定、业务强断言等 - 自动断言是基础模板，复杂断言需人工补充 --- ### 2.7 用例结构与增强能力对比总结 | 结构/能力 | YAML 结构特征 | Python 驱动特征 | 适用场景 | 变量传递方式 | |------|--------------|----------------|---------|-------------| | **单接口极简** | 无 `steps`，根节点直接用例配置 | 标准 parametrize + Executor | 独立接口测试 | `extract` → 内存 + Redis | | **多步骤链路** | 有 `steps` + `teardown_steps` | 同上 | 业务流程闭环测试 | `${变量名}` 读内存；需要跨文件时用 `extract` 写 Redis | | **CSV 数据驱动** | 含 `parameters` 节点 | 同上 | 多参数组合测试 | `extract` → 内存 + Redis | | **跨文件 Redis** | 普通单接口结构 | 多个 test_ 方法分别读取不同 YAML | 多文件业务链路 | `extract` → Redis → `${get_extract_data()}` | | **Mock 挡板** | 含 `mock` 节点 | 同上 | 下游不可用/异常模拟 | 不涉及变量传递 | | **HAR 流量回放** | `steps` 链路（由 HAR 自动生成） | 自动生成 `test_*.py` + Executor | 页面接口录制回放 / 线上流量回归 | 生成后按业务需要补 `extract` / `extract_local` | ## 三、核心组件交互关系 ### 3.1 组件职责矩阵 | 组件 | 文件路径 | 核心职责 | 输入 | 输出 | |------|---------|---------|------|------| | **run.py** | `/run.py` | 启动入口，调用 pytest + 生成 Allure 报告 | CLI 调用 | exit_code | | **ConfigManager** | `configs/config_manager.py` | 单例配置管理，动态加载环境 YAML | `AUTO_ENV` | settings 对象 | | **conftest** (根) | `/conftest.py` | 全局生命周期钩子（熔断、统计、报告） | Pytest 事件 | 副作用（Redis、日志） | | **conftest** (testcase) | `testcase/conftest.py` | 全局 Fixture（登录、SessionID、缓存清理） | Pytest Fixture | access_token, session_id | | **read_yaml_cases** | `common/file_readers/yaml_case_reader.py` | 读取 + 缓存 + 规范化 YAML | 文件路径 | 用例字典列表 | | **expand_csv_cases** | `util_tools/data_drivers/csv_case_expander.py` | CSV 数据驱动裂变 | 含 parameters 的用例 | N 条独立用例 | | **Executor** | `util_tools/execution/executor.py` | 用例执行总调度 | case_info | 异常/通过 | | **VariableResolver** | `util_tools/variables/variable_resolver.py` | 变量解析替换引擎 | 含 ${} 的数据结构 | 替换后的数据结构 | | **HttpClient** | `util_tools/clients/http_client.py` | HTTP 发包 + Mock 拦截 | 请求参数 | Response/FakeResp | | **DataExtractor** | `util_tools/variables/data_extractor.py` | 响应数据提取（JSONPath/正则/逆向） | 响应文本 + 规则 | 提取字典 | | **AssertionRunner** | `util_tools/assertions/assertion_runner.py` | validation 调度，分发 HTTP/DB 断言 | 实际结果 + 预期规则 | 通过/ValidationException | | **RuntimeCache** | `util_tools/variables/runtime_cache.py` | Redis 读写 + 前缀管理 | Key-Value | 带前缀的 Redis 操作 | | **RuntimeFunctions** | `common/runtime_functions.py` | 动态函数库（签名、造数、配置读取） | 函数名 + 参数 | 函数返回值 | | **LoggerUtil** | `common/logger.py` | 多进程隔离日志系统 | 日志消息 | 日志文件 + 控制台输出 | | **AllureReporter** | `util_tools/extensions/allure_reporter.py` | Allure 步骤上下文 + 附件挂载 | 请求/响应数据 | Allure JSON 原始数据 | | **sql_executor** | `util_tools/execution/sql_executor.py` | 前后置 SQL 执行，核心函数为 `execute_sql_statements` | SQL 语句列表 | 执行结果/异常 | | **MockResponse** | `util_tools/extensions/mock_response.py` | Mock 响应拦截 | case_info | FakeResp/None | | **PytestStatsManager** | `util_tools/reporting/pytest_stats.py` | 运行统计 + 通知触发 | Pytest 报告 | 邮件/企微通知 | ### 3.2 Allure 报告深度集成 框架在每个接口步骤中自动挂载以下附件到 Allure 报告： 如果当前步骤配置了 `retry`，框架会先把每次尝试的附件暂存在内存中： 中间失败且还会继续重试时，丢弃本次附件； 当前尝试成功，或已经是最后一次失败时，才把本次附件写入 Allure。 因此控制台日志能看到完整重试过程，Allure 报告只展示最终有效现场。 | 附件名称 | 内容 | 挂载时机 | |---------|------|---------| | 请求 URL | 完整的请求地址 | 请求发送后 | | 请求头 | Headers 字典 | 请求发送后 | | 请求 Cookie | Cookies 字典 | 请求发送后 | | 请求体 | JSON / Form / Params 数据 | 请求发送后 | | 响应体 | 接口返回的完整响应 | 响应接收后 | | 响应状态码 | HTTP status_code | 响应接收后 | | 接口响应提取数据 | Extract 提取结果（JSON 格式） | 提取完成后 | | XX断言结果 | 每条断言的预期 vs 实际（JSON 格式） | 断言执行后 | | 环境信息 | Python 版本、Git 分支、Jenkins 构建号等 | 会话结束时 | **多步骤链路在 Allure 中的展示**： ```mermaid graph TD Root["标签全生命周期测试"] Root --> Step1["▶ 步骤1-创建标签"] Root --> Step2["▶ 步骤2-编辑标签"] Root --> Step3["▶ 步骤3-删除标签"] Root --> Teardown["🗑 清理-文件上传"] Step1 --> ReqURL["请求 URL"] Step1 --> ReqHeader["请求头"] Step1 --> ReqBody["请求体"] Step1 --> RespBody["响应体"] Step1 --> Extract["接口响应提取数据"] Step1 --> Assert["XX断言结果"] Step2 --> Dots1["..."] Step3 --> Dots2["..."] Teardown --> Dots3["..."] ``` **挂载时机说明**： | 附件 | 挂载时机 | |------|---------| | 请求 URL / 请求头 / 请求 Cookie / 请求体 | 请求发送后 | | 响应体 / 响应状态码 | 响应接收后 | | 接口响应提取数据 | 提取完成后 | | XX断言结果 | 断言执行后 | | 环境信息 | 会话结束时 | ### 3.3 数据流转路径 ```mermaid graph LR subgraph 输入层 YAML[YAML 文件] --> Reader[read_yaml_cases] Reader --> Driver[expand_csv_cases] Driver --> Executor[Executor] end subgraph 工具链路 RuntimeFunctions[RuntimeFunctions] --> VariableResolver[VariableResolver] CacheMgr[RuntimeCache] <--> Redis[(Redis)] CacheMgr --> VariableResolver end subgraph 数据链路 VariableResolver --> HttpClient[HttpClient] HttpClient --> MockResponse[MockResponse] VariableResolver --> Executor end subgraph 输出层 Executor --> DataExtractor[DataExtractor] DataExtractor --> Assertions[Assertions] Assertions --> Allure[Allure 报告] end ``` **数据流转说明**： 1. **输入层**：YAML 文件 → read_yaml_cases（含缓存和深拷贝）→ expand_csv_cases（CSV 裂变）→ Executor（场景控制总线）。 2. **工具链路**：VariableResolver 调用 RuntimeFunctions 动态函数，RuntimeCache 读写 Redis 全局缓存。 3. **数据链路**：VariableResolver 解析变量 → HttpClient 发包（含 MockResponse 拦截）。 4. **输出层**：Executor → DataExtractor 变量提取 → Assertions 断言校验 → Allure 报告。 ### 3.4 核心工具链内部机制 #### 3.4.1 read_yaml_cases：为什么不用 lru_cache？ `read_yaml_cases` 使用手动维护的 `_YAML_CACHE` 字典，而不是 `@lru_cache` 装饰器。原因是： `lru_cache` 缓存的是对象引用。YAML 解析出来的是**可变字典**，如果 Worker A 从缓存取到字典并修改了里面的变量，Worker B 再取缓存时就会拿到**脏数据**。 手动缓存 + `copy.deepcopy()` 返回，确保每个调用方拿到的是独立副本。 ```python if cache_key in _YAML_CACHE: return copy.deepcopy(_YAML_CACHE[cache_key]) # 深拷贝隔离 ``` #### 3.4.2 csv_case_expander / expand_csv_cases：AST 类型推导规则 CSV 单元格内容在替换到 YAML 模板时，框架会尝试推导其真实类型： | 条件 | 推导行为 | 示例 | |------|---------|------| | 以 `{` 或 `[` 开头 | 用 `ast.literal_eval` 解析为 dict/list | `"{\"a\": 1}"` → `{"a": 1}` | | 以 `True` 或 `False` 开头 | 解析为布尔值 | `"True"` → `True` | | 去掉小数点后是纯数字，且不以 `0` 开头 | 解析为 int/float | `"123"` → `123` | | 以上都不满足 | 保持字符串 | `"hello"` → `"hello"` | 为什么用 `ast.literal_eval` 而不是 `eval`？因为 `eval` 会执行任意代码，有安全风险。`ast.literal_eval` 只允许解析字面量（数字、字符串、列表、字典、布尔值等），不会执行函数调用。 #### 3.4.3 DataExtractor：正则提取与自研过滤语法 **正则提取**： DataExtractor 通过检查表达式中是否包含正则特征来判断是否走正则分支： ```python _REGEX_PATTERNS = [r'(.*?)', r'(.+?)', r'(\d+)', r'(\d*)'] ``` - 如果表达式包含上述特征 → 走 `re.search` / `re.findall` - 如果正则里写了 `(\d+)` → 结果自动转为 `int` **自研高阶过滤语法**： 原生 jsonpath 不支持条件过滤，框架自研了过滤引擎，支持 `==` 和 `contains` 两种运算符： ``` $.data.rows[?(@.user.id == '123')].order_no └─────┘ └──────┘ └─┘ └──┘ └──────┘ 基础路径 过滤字段 运算符 期望值 目标字段 ``` 匹配过程： 1. 用正则拆解表达式为 5 个分组（base_path, filter_field, operator, filter_val, target_field） 2. 先用原生 jsonpath 获取基础路径下的所有数据 3. 遍历每条数据，检查过滤字段是否满足条件 4. 满足条件的，提取目标字段的值 #### 3.4.4 HttpClient：Retry 策略详解 ```python retry_strategy = Retry( total=1, # 总共最多重试 1 次 connect=1, # 连接失败时重试 1 次 read=0, # 读取超时不再重试（防止重复提交） backoff_factor=0.1,# 重试间隔退避因子 allowed_methods=["HEAD", "GET", "OPTIONS", "PUT", "DELETE"] # POST 不重试 ) ``` | 参数 | 值 | 说明 | |------|-----|------| | `total` | 1 | 所有原因合计最多重试 1 次 | | `connect` | 1 | TCP 连接失败时重试 1 次（网络抖动）| | `read` | 0 | 读取超时不重试。因为请求可能已经到达服务器并被执行，重试可能造成重复操作 | | `backoff_factor` | 0.1 | 重试间隔 = 0.1 * (2^retry_count) 秒 | | `allowed_methods` | 幂等方法 | POST 不在白名单内，防止重复提交造成脏数据 | SSL 证书校验默认开启。推荐在环境配置中使用 `http.verify_ssl` 控制：生产和公共网络环境保持 `true`；自签名测试环境如确需关闭，再显式配置为 `false`，此时框架才会屏蔽 `InsecureRequestWarning`。 #### 3.4.5 ConfigManager：动态属性注入的安全机制 ```python for key, value in self._env_data.items(): if not hasattr(self, key): setattr(self, key, value) # 安全注入 else: pass # 与类已有属性同名，跳过注入，防止覆盖核心属性 ``` 如果 YAML 中的 key 与类已有属性（如 `BASE_DIR`、`LOG_PATH`）同名，会静默跳过，保留原生属性。业务方仍可通过 `_env_data` 字典读取被跳过的配置。 #### 3.4.6 allure_id_generator：Allure 报告编号生成器 ```python from common.helpers.allure_id_generator import m_id, c_id @allure.feature(next(m_id) + "订单管理") class TestOrder: @allure.story(next(c_id) + "创建订单") def test_create_order(self): ... ``` **设计要点**： - `m_id` 用于生成 Feature 编号，例如 `M001_` - `c_id` 用于生成 Story 编号，例如 `C001_` - 测试文件只需要使用 `next(m_id)`、`next(c_id)`，不需要手动维护序号 - `next(m_id)`、`next(c_id)` 在装饰器阶段只生成内部占位符，不立即消费真实编号 - `conftest.py` 会在 pytest 收集完成后，按本轮实际 `items` 顺序把占位符替换成真实编号 - 全量运行时按全量用例从 `M001_`、`C001_` 开始；只运行部分目录时按这部分目录重新从 `M001_`、`C001_` 开始 - pytest-xdist 多 Worker 下不会因为每个 Worker 重复收集用例而放大编号 #### 3.4.7 Jenkins 环境变量嗅探 ConfigManager 自动嗅探以下 Jenkins 环境变量，动态生成 Allure 报告在线访问链接： | 环境变量 | 用途 | |---------|------| | `BUILD_URL` | Jenkins 构建页面 URL | | `JOB_NAME` | Jenkins 任务名称 | | `BUILD_NUMBER` | Jenkins 构建编号 | | `JENKINS_URL` | 判断是否在 Jenkins 环境运行 | | `GIT_BRANCH` | 当前 Git 分支 | Allure 报告地址拼接规则： ``` {jenkins_root}/job/{job_name}/{build_number}/allure ``` #### 3.4.8 HMAC-SHA256 签名机制 框架内置了 `hmac_sha256_sign` 函数，用于生成接口请求的防重放签名。该函数在 `common/runtime_functions.py` 中实现，核心设计要点如下： **签名参数来源**： | 参数 | 来源 | 说明 | |------|------|------| | `timestamp` | TLS 线程缓存 | 毫秒级时间戳，由 `generate_and_store_timestamp_nonce()` 生成 | | `nonce` | TLS 线程缓存 | 2 位随机数，与时间戳一同生成 | | `method` | TLS 线程缓存 | 当前步骤的真实 HTTP 方法（GET/POST/PUT/DELETE），由 `set_method()` 写入 | | `accessToken` | Redis | 通过 `get_extract_data('accessToken')` 读取 | | `clientId` | 环境配置 | 通过 `get_env_value('clientId')` 读取 | | `userId` | 环境配置 | 通过 `get_env_value('userId')` 读取 | | `url` | 请求路径 | 函数传入的参数，即接口的相对路径 | **签名计算流程**： 1. `StepRunner.execute_step()` 在发请求前自动调用 `RuntimeFunctions.generate_and_store_timestamp_nonce()`，生成时间戳和随机数并存入 TLS（Thread Local Storage）线程缓存 2. 同时调用 `RuntimeFunctions.set_method(method)` 将当前 HTTP 方法（如 GET/POST）写入 TLS，确保签名使用真实方法而非硬编码 POST 3. 当 YAML 中遇到 `${hmac_sha256_sign(/api/order/create)}` 时，`VariableResolver` 调用 `RuntimeFunctions.hmac_sha256_sign()` 4. 函数从 TLS 读取 timestamp、nonce、method，从 Redis 读取 accessToken，从环境配置读取 clientId/userId 5. 按 `accessToken`、`clientId`、`method`、`nonce`、`timestamp`、`url`、`userId` 的顺序拼接签名字符串 6. 使用环境配置中的 `clientSecret` 作为密钥，执行 HMAC-SHA256 运算，输出 64 位十六进制字符串 **线程安全设计**： 使用 `threading.local()` 实现 TLS，确保 `pytest-xdist` 多进程并发时，每个线程的签名参数互不干扰。Worker A 的时间戳不会覆盖 Worker B 的时间戳。 **YAML 中的典型用法**： ```yaml request: headers: timestamp: ${get_timestamp()} nonce: ${get_nonce()} sign: ${hmac_sha256_sign(/api/order/create)} ``` **注意事项**： - `get_timestamp()` 和 `get_nonce()` 优先读取 TLS 缓存值，若未命中则自动调用 `generate_and_store_timestamp_nonce()` 生成新值 - 如果手动调用了 `hmac_sha256_sign` 但没有先发送请求，可能拿到的是上一次请求的过期时间戳，导致服务端验签失败 - 框架已在 `StepRunner.execute_step()` 中保证每次请求前自动刷新时间戳和 nonce，YAML 中通常无需手动处理 --- ## 四、关键设计决策说明 ### 4.1 为什么要 deepcopy？ 数据驱动（parametrize）会把同一份 YAML 数据发给多个测试函数。如果不 deepcopy，测试函数 A 修改了 case_info 里的某个字段，测试函数 B 会拿到被污染的数据。 同理，多步骤链路中的 `global_request` 合并也用了 deepcopy，防止步骤1 修改了全局 headers 影响到步骤2。 ### 4.2 为什么要用 ExitStack？ 文件上传场景：`files: {'file': 'files/mag.jpg'}` 会打开文件句柄。如果在发请求过程中抛出异常，文件可能没有被关闭，导致内存泄漏。 ExitStack 保证：无论成功还是异常，`with` 块结束时所有注册的资源都会被自动释放。 ### 4.3 软断言 vs 硬断言 **硬断言**（原生 assert）：第一个失败就停止，一次只能发现一个错误。修完再跑，再发现下一个错误，效率低。 **软断言**（框架实现）：遍历所有断言规则，收集全部错误，最后一次性抛出。一次运行暴露所有问题，排障效率高。 ### 4.4 熔断机制的设计意图 假设 5 个接口组成一条业务链路：登录 → 创建订单 → 查询订单 → 修改订单 → 删除订单。 如果第2步"创建订单"失败了，第3、4、5步肯定也会失败（因为订单不存在）。继续跑下去只是浪费时间和资源。 熔断机制：第2步失败后标记文件为"已熔断"，后续同文件用例直接跳过。等修复后重跑，成功后自动"自愈"清除熔断标记。 ### 4.5 为什么 VariableResolver 要分三种匹配模式？ 1. **纯变量全量匹配** `"${order_id}"`：类型保真，替换后返回原始类型（int/dict/list），不强制转字符串。 2. **纯函数全量匹配** `"${get_timestamp()}"`：同上，函数返回值是什么类型就是什么类型。 3. **字符串混编** `"id=${order_id}"`：无法保留原始类型（因为要和字符串拼接），强制降级为字符串。 这个设计确保了 JSON 请求体中的数字字段不会被误转成字符串，避免接口因类型不匹配而报错。 ### 4.6 YAML 编写规范 YAML 是本框架的业务用例主体。推荐把“测试数据、请求参数、提取规则、断言规则”都写在 YAML 中，Python 驱动文件只做参数化和调用 `Executor`。 **最小必填字段**： | 字段 | 位置 | 是否必填 | 说明 | |------|------|----------|------| | `name` | 用例顶层或 step | 是 | 用例名或步骤名，会进入日志和 Allure 标题。 | | `request` | 用例顶层或 step | 是 | HTTP 请求配置。 | | `request.method` | request 内 | 是 | HTTP 方法，支持 GET/POST/PUT/DELETE/PATCH/HEAD/OPTIONS。 | | `request.path` / `request.url` | request 内 | 二选一 | `url` 是完整地址；`path` 会与 `base_url` 拼接。 | | `validation` | 用例顶层或 step | 是 | 断言列表，可以为空列表，但建议至少校验状态码。 | **`url` 与 `path` 的选择**： | 写法 | 适用场景 | 示例 | |------|----------|------| | `base_url + path` | 项目内大部分接口 | `base_url: ${get_base_url(base_wms_url)}` + `path: /api/order/create` | | `url` | 第三方完整地址或临时调试地址 | `url: https://example.com/open/api` | 如果同时写了 `request.url` 和 `request.path`，框架优先使用 `url`。普通项目接口建议优先使用 `base_url + path`，环境切换更清晰。 **`config` 合并规则**： | 配置 | 合并方式 | 说明 | |------|----------|------| | `config.base_url` | 作为默认基础域名 | step 没写 `base_url` 时使用它。 | | `config.request.method` | step 未写时继承 | 适合同一 YAML 全部是 POST 的场景。 | | `config.request.headers` | 字典深度合并 | 全局为底，step 同名字段覆盖全局。 | | `config.request.cookies` | 字典深度合并 | 规则同 headers。 | | `config.validation` | 追加到 step validation 前面 | 适合统一校验 HTTP 状态码。 | **变量提取选择**： | 字段 | 存储位置 | 读取方式 | 使用场景 | |------|----------|----------|----------| | `extract` | 内存 + Redis | 当前链路 `${变量名}`，跨文件 `${get_extract_data(变量名)}` | 需要跨 YAML、跨文件、跨进程共享。 | | `extract_local` | 仅内存 | 当前链路 `${变量名}` | 只给当前 YAML 后续步骤用。 | | `extract_list` | 内存 + Redis | 同 `extract` | 提取列表结果并跨文件共享。 | | `extract_list_local` | 仅内存 | 同 `extract_local` | 提取列表结果，仅当前 YAML 使用。 | **SQL 与清理边界**： | 字段 | 执行时机 | 失败影响 | 建议用途 | |------|----------|----------|----------| | `setup_sql` | 当前 step 发请求前 | 失败会阻断当前 step | 准备测试数据。 | | `teardown_sql` | 当前 step 主流程开始后 | `setup_sql` 失败时不执行；请求、提取或断言失败时仍会执行；自身异常会按当前 step 逻辑暴露 | 清理当前 step 产生的数据。 | | `teardown_steps` | 主链路结束后统一执行 | 异常静默隔离，不掩盖主链路错误 | 删除业务数据、释放状态、兜底回滚。 | | `validation.db` | 断言阶段 | 失败会作为断言失败 | 只允许 SELECT，用于校验落库结果。 | 补充说明：`setup_sql` / `teardown_sql` 中的 `INSERT` / `UPDATE` / `DELETE` 采用严格模式，影响行数为 0 会直接失败。写 SQL 时要确保 `WHERE` 条件能命中目标数据，避免前置数据没有准备成功或后置数据没有清理成功却被误认为通过。 **推荐写法**： ```yaml config: name: 订单创建链路 base_url: ${get_base_url(base_wms_url)} request: method: post headers: Content-Type: application/json validation: - code: 200 steps: - name: 创建订单 request: path: /api/order/create json: orderNo: ${generate_order_number(ORD)} extract: order_id: $.data.id validation: - contain: code: 0 teardown_steps: - name: 删除订单 request: path: /api/order/delete json: id: ${order_id} validation: - code: 200 ``` ### 4.7 开发者扩展规范 维护框架时优先遵循现有目录职责，不要把新能力都塞回 `executor.py` 或 `conftest.py`。 | 需求 | 推荐位置 | 说明 | |------|----------|------| | 新增 `${函数()}` | `common/runtime_functions.py` | 写成 `RuntimeFunctions` 方法，供 `VariableResolver` 动态调用。 | | 新增 HTTP 断言 | `util_tools/assertions/http_assertions.py` | 只处理响应体、状态码等 HTTP 结果。 | | 新增数据库断言 | `util_tools/assertions/db_assertions.py` | 保持数据库查询和断言聚合职责。 | | 新增断言调度规则 | `util_tools/assertions/assertion_runner.py` | 负责把 validation 规则分发到具体断言类。 | | 新增请求组装规则 | `util_tools/execution/request_builder.py` 或 `step_runner.py` | 纯参数组装放 request_builder，执行流程放 step_runner。 | | 新增外部资源客户端 | `util_tools/clients/` | 例如 MQ、对象存储、第三方 SDK。 | | 新增 pytest 生命周期逻辑 | `util_tools/pytest_hooks/` | 根 `conftest.py` 只保留 hook 入口。 | | 新增通知渠道 | `util_tools/reporting/channels/` | 例如钉钉、飞书等渠道。 | | 新增报告汇总逻辑 | `util_tools/reporting/` | 统计、HTML 摘要、通知调度放这里。 | | 新增 HAR/YAML 生成工具 | `util_tools/generators/` | 离线生成类工具放这里。 | **代码注释要求**： 1. 新增公共方法必须写 docstring，说明用途、参数、返回值和失败行为。 2. 新增复杂分支必须写行内注释，解释“为什么这样做”，不要只重复代码含义。 3. 新增调试入口只放在适合单独运行的工具文件底部，格式统一使用 `if __name__ == '__main__':`。 4. `executor.py`、`conftest.py`、pytest hooks 不建议写随意调试入口，这类文件依赖完整 pytest 生命周期。 --- ## 五、extensions 扩展层详解 框架通过 extensions 扩展机制在核心执行链路的关键节点插入增强能力，Allure、Mock 等扩展模块位于 `util_tools/extensions/` 目录下；前后置 SQL 属于接口执行链路，位于 `util_tools/execution/sql_executor.py`；pytest 统计和通知位于 `util_tools/reporting/` 目录下。 ### 5.1 Allure 报告扩展（allure_reporter.py） **职责**：Allure 步骤分层、附件挂载、JSON 序列化缓存。 **核心组件**： | 组件 | 说明 | |------|------| | `allure_step_context(step_name)` | 上下文管理器，包裹代码块形成 Allure 可折叠步骤 | | `AllureBuffer` | 基于线程本地存储（TLS，Thread Local Storage）的缓冲池，防止多进程数据串扰 | | `AllureReporter.record_api()` | 聚合 HTTP 请求全生命周期信息并写入 AllureBuffer；普通步骤直接落报告，retry 尝试由 StepRunner 决定提交或丢弃 | | `format_allure_attachment_data()` | 带 LRU 缓存的安全 JSON 序列化 | | `begin_attempt_capture()` / `commit_attempt_capture()` / `discard_attempt_capture()` | retry 尝试级附件控制；中间失败丢弃，最终尝试写入 | **AllureBuffer 的工作机制**： ```python # 数据先压入当前线程的缓冲队列（不立即写磁盘） AllureBuffer.attach(json_body, "请求参数") AllureBuffer.attach(json_body, "响应结果") # 统一 flush 落盘（由 allure_step_context 的 finally 自动调用） AllureBuffer.flush() # 一次性写入 Allure 报告文件 ``` **retry 场景下的附件策略**： ```python # 每次 retry 尝试开始时，先开启临时捕获池 AllureBuffer.begin_attempt_capture() # 当前尝试成功，或已经是最后一次失败：提交附件 AllureBuffer.commit_attempt_capture() # 当前尝试失败且后面还会继续重试：丢弃附件 AllureBuffer.discard_attempt_capture() ``` 这样做的原因是：控制台日志用于看完整过程，Allure 报告用于看最终现场。 如果中间失败尝试也全部写入 Allure，同一个 step 下会出现多套相似的请求、响应和断言附件，新人排查时容易误判。 **为什么用 TLS（线程本地存储）？** pytest-xdist 多进程并发时，如果全局列表存报告数据，会出现 A 用例的报告挂载了 B 用例的数据。TLS 让每个线程有独立的存储栈，互不干扰。 ### 5.2 SQL 执行辅助（sql_executor.py） **职责**：执行前后置 SQL，并把执行记录写入 Allure 报告。 **核心设计**： | 设计点 | 说明 | |--------|------| | SQL 类型白名单 | 仅允许 `SELECT` / `INSERT` / `UPDATE` / `DELETE`，以及 `WITH` CTE 后接这些主语句；`DROP` / `TRUNCATE` / `ALTER` / `CREATE` 等危险语句会被拦截 | | 多语句拦截 | 单条 SQL 配置中不允许用分号拼接多条语句，避免 `SELECT ...; DROP ...` 这类误操作 | | DML 有效性检查 | INSERT/UPDATE/DELETE 影响行数为 0 时直接报错，防止后续断言基于错误数据状态 | | 类型校验 | `sql` 字段必须是列表（`["SELECT * FROM user"]`），字符串会报格式错误 | | 异常现场保留 | SQL 执行出错时，把出错前的执行记录和错误信息一起写入 Allure 报告 | **执行流程**： ```python def execute_sql_statements(sql_list, hook_name="执行 SQL"): for sql in sql_list: # 清洗注释、拒绝多语句、判断 SQL 主类型 sql_type = _detect_sql_type(sql) if sql_type == "select": res = db_client.select(sql, fetchall=True) else: # insert/update/delete 走执行通道，并检查影响行数 rowcount = db_client.execute(sql) if rowcount == 0: raise ValueError("影响行数为 0，可能是 WHERE 条件没命中") # 记录执行结果 # 统一写入 Allure 报告 ``` ### 5.3 Mock 响应扩展（mock_response.py） **职责**：在发真实网络请求前，根据配置返回假响应。 **三种 Mock 模式**： | 模式 | 配置方式 | 适用场景 | |------|---------|---------| | **简单 Mock** | `mock: {status_code, json, headers}` | 无条件返回固定响应 | | **链式 Mock** | `mock: {chain: [{...}, {...}]}` | 模拟异步状态流转（处理中→成功）| | **规则 Mock** | `mock: {rules: [{when, then}]}` | 不同请求参数返回不同结果 | **_FakeResp 对象**： 模拟 `requests.Response` 的常用接口，让上层代码无感知： ```python class _FakeResp: status_code = 200 text = '{"code": 0}' content = b'{"code": 0}' headers = {"Content-Type": "application/json"} elapsed = timedelta(seconds=0) def json(self): return {"code": 0} # 直接返回 Python 对象，省去反序列化 ``` ### 5.4 统计报告扩展（pytest_stats.py） **职责**：收集测试执行统计信息，生成报告并触发通知推送。 **统计维度**： | 指标 | 说明 | |------|------| | passed | 通过的用例数 | | failed | 断言失败的用例数 | | error | 环境异常（setup/teardown 失败）的用例数 | | skipped | 跳过的用例数（含熔断跳过）| | case_reruns | 用例级重跑次数，来源于 `pytest-rerunfailures` 的 `--reruns` | | step_retries | 步骤级重试次数，来源于 YAML step.retry | | reruns | 重试合计字段，值为 `case_reruns + step_retries`，供报告模板和通知统一展示 | | failed_test_names | 失败用例的方法名列表（用于告警）| | total_files | 涉及的测试文件总数 | | actual_duration | 实际耗时（秒）| **Fake Reporter 机制**： pytest-xdist 分布式执行时，官方的 TerminalReporter 对象可能丢失或数据不全。PytestStatsManager 自己构造一个属性结构相同的"假 Reporter"，让下游的邮件模板和企微通知能正常渲染。 **通知触发流程**： ```mermaid graph TD Start(["pytest_sessionfinish"]) --> Send["send_test_summary_notifications()"] Send --> Fake["构造 FakeReporter（兼容 pytest-xdist）"] Fake --> Render["渲染 HTML 邮件正文"] Render --> CheckEmail{is_email_msg?} CheckEmail -->|是| Email["后台线程发送邮件（不阻塞主进程）"] CheckEmail -->|否| CheckWecom Email --> CheckWecom{is_wecom_msg?} CheckWecom -->|是| Wecom["后台线程发送企微 Markdown"] CheckWecom -->|否| End(["结束"]) Wecom --> End ``` **流程说明**： | 步骤 | 说明 | |------|------| | `FakeReporter` | pytest-xdist 分布式执行时官方 TerminalReporter 可能丢失，PytestStatsManager 自行构造兼容结构 | | `后台线程发送` | ThreadPoolExecutor(max_workers=1)，不阻塞主进程退出 | | `发送失败` | 仅记录日志，不影响测试状态（通知是锦上添花，不是必需路径） | --- ## 六、通知推送机制 框架支持两种通知渠道：**邮件**和**企业微信机器人**，通过环境配置文件中的开关控制。 ### 6.1 邮件通知 **环境配置**（`configs/envs/*.yaml`）： ```yaml is_email_msg: true email: email_name: "your_email@qq.com" # 发件邮箱 email_password: "your_smtp_password" # 邮箱授权码 email_port: 465 email_server: "smtp.qq.com" email_receiver: "receiver1@example.com,receiver2@example.com" ``` **邮件内容**：基于 `templates/email_summary.html` 模板渲染，包含： - 测试执行时间戳 - 用例总数、通过数、失败数、错误数、跳过数 - 通过率、耗时 - 涉及文件数 - 重试次数 ### 6.2 企业微信机器人通知 **环境配置**： ```yaml is_wecom_msg: true wechat: # 两种写法二选一： webhook_key: "your_key" # webhook_url: "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your_key" ``` **消息格式**：Markdown 格式，包含： - 测试汇总统计（通过率饼图文本版） - 失败用例列表 - Allure 报告在线链接 ### 6.3 通知触发时机 通知在 `pytest_sessionfinish` 钩子中触发，通过 `ThreadPoolExecutor(max_workers=1)` 后台线程发送，不阻塞主进程退出。 **注意**：通知发送失败不会导致测试失败，仅记录日志。这是"通知是锦上添花，不是必需路径"的设计。 --- ## 七、CLI 工具使用 框架提供命令行工具 `cli.py`，用于辅助测试开发。 ### 7.1 HAR 生成 YAML + pytest（record 命令） 将浏览器抓包的 HAR 文件转换为标准 YAML 测试用例，并同步生成 pytest 驱动文件。 ```bash # 基础用法 python cli.py record data/har/flow.har # 指定输出目录 python cli.py record data/har/flow.har -o testcase/har_case # 指定基础 URL 占位符 python cli.py record data/har/flow.har -b '${get_base_url(base_wms_url)}' # 只录制指定域名的请求 python cli.py record data/har/flow.har -d jfwms.net # 指定生成目录名和文件名 python cli.py record data/har/song.jfwms.net.har --name song_jfwms_net ``` 默认生成结果： ```text testcase/har_case/flow/ ├── flow.yaml └── test_flow.py ``` **参数说明**： | 参数 | 必填 | 说明 | 默认值 | |------|------|------|--------| | `har_path` | 是 | HAR 文件路径 | - | | `-o, --output` | 否 | 输出根目录，命令会在该目录下创建 `/` 子目录 | `testcase/har_case` | | `-b, --base-url` | 否 | 基础 URL 占位符 | `${get_base_url(base_wms_url)}` | | `-d, --domain` | 否 | 域名过滤，在配置业务域名基础上进一步收窄 | "" | | `--name` | 否 | 指定生成目录名和 YAML/PY 文件名 | HAR 文件名 | | `--feature` | 否 | 生成 pytest 文件中的 Allure feature 文案 | `HAR自动生成用例` | | `--story` | 否 | 生成 pytest 文件中的 Allure story 文案 | 生成后的用例名 | **转换逻辑**： 1. 解析 HAR 文件的 `log.entries` 列表 2. 默认读取 `configs/envs/*.yaml` 中的业务域名，只保留 WMS/OMS 等配置内域名 3. 过滤静态资源、前端埋点和版本文件，例如 `.js`、`.css`、`/speed`、`/version.json` 4. 提取每条请求的 Method、URL、Headers、Cookies、Body 5. Cookie 优先按域名写成 `${get_base_url(wms_cookie)}` 或 `${get_base_url(oms_cookie)}` 6. 智能推导基础断言规则（HTTP 状态码 + success/code/msg/data 常见字段） 7. 输出标准 YAML（多步骤 `steps` 结构）和 pytest 驱动文件 **生成限制**： | 限制项 | 说明 | |--------|------| | 域名限制 | 默认只录 `configs/envs/*.yaml` 的 `base_url` 中配置过的业务域名。 | | 路径限制 | 只录 `/api/`、`/wms/`、`/oms/`、`/proxy/`、`/pda/` 开头的业务 API。其他前缀会被过滤，例如微信接口 `/cgi-bin/token` 默认不会生成。 | | 请求体限制 | 只自动解析 JSON 和 Form 表单；文件上传、二进制、自定义 MIME 类型需要人工补。 | | 断言限制 | 自动断言只是基础模板，不能替代业务强断言。 | | 变量限制 | 自动替换少量通用字段，订单号、SKU、业务 ID、步骤依赖变量需要人工补 `extract`。 | | Cookie 限制 | 页面接口优先使用环境配置中的 `wms_cookie` / `oms_cookie`，所以环境 Cookie 必须可用。 | 生成后建议先检查 YAML 中的 `steps` 数量是否符合预期。如果明显少于 HAR 中的业务请求，优先查看终端输出的过滤统计，通常是域名未配置、路径前缀不在白名单、或请求被识别为静态资源。 --- ### 7.2 工具文件开发调试入口 为了方便新人理解工具方法的真实调用方式，部分工具文件底部保留了 `if __name__ == '__main__'` 开发调试入口。 运行方式统一使用模块方式，避免直接执行文件时找不到项目包： ```bash # Windows PowerShell 示例 python -m common.file_readers.yaml_case_reader python -m util_tools.clients.http_client python -m util_tools.variables.runtime_cache ``` **适合直接运行的入口**： | 模块 | 调试内容 | 是否有外部副作用 | |------|----------|------------------| | `common.file_readers.yaml_case_reader` | 读取真实 YAML 用例 | 无 | | `common.file_readers.csv_rows_reader` | 读取真实 CSV 数据 | 无 | | `common.runtime_functions` | 查看时间戳、nonce、配置读取等运行时函数 | 无，不调用签名函数 | | `util_tools.clients.http_client` | 通过 Mock 分支验证请求入口 | 不发真实 HTTP | | `util_tools.generators.har_to_yaml` | HAR 转临时 YAML 并自动清理 | 只生成临时文件 | | `util_tools.variables.runtime_cache` | 写入/读取/清理 Redis extract 变量 | 只使用 debug session 并精确清理 | | `util_tools.clients.redis.redis_client` | Redis String/Hash/List/Pipeline/JSON/SCAN 调试 | 只使用 `debug:redis_client:*` 前缀并清理 | | `util_tools.clients.redis.circuit_breaker_store` | 熔断状态读写调试 | 只写 debug 文件路径字段并清理 | | `util_tools.reporting.channels.email_sender` | 使用假 SMTP 验证邮件组装 | 不发真实邮件 | | `util_tools.reporting.channels.wecom_sender` | 验证企微配置读取和 nodeid 简化 | 不发真实企微 | **不建议单独运行的文件**： `util_tools.execution.executor`、`step_runner` 依赖完整 YAML 执行上下文； `util_tools.pytest_hooks/*` 依赖 pytest 生命周期； `util_tools.generators.har_case_generator` 是 `cli.py record` 调用的内部生成器，不单独提供命令行入口； 通知发送总入口默认由 `pytest_sessionfinish` 调用。 这些文件单独加调试入口容易制造误导，应通过 `python run.py` 或指定 pytest 用例验证。 --- ## 八、自定义异常体系 框架定义了完整的分层异常体系（`common/exceptions.py`），所有异常继承自 `FrameworkException`，统一包含 **错误码**、**错误消息**、**上下文数据**。 ### 8.1 异常分层结构 ```mermaid graph TD Base["FrameworkException / 基类 code=1000"] subgraph L1 配置层异常 1000-1999 Base --> C[ConfigException 1001] end subgraph L2 网络层异常 2000-2999 Base --> N[NetworkException 2001] end subgraph L3 数据层异常 3000-3999 Base --> D[DataException 3001] end subgraph L4 存储层异常 4000-4999 Base --> S[StorageException 4001] end subgraph L5 用例层异常 5000-5999 Base --> E[CaseException 5001] end subgraph L6 熔断/重试层异常 6000-6999 Base --> CB[CircuitBreakerException 6001] end subgraph L7 安全/权限层异常 7000-7999 Base --> SE[SecurityException 7001] end ``` **各层子类明细**： | 层级 | 父类 | 子类 | |------|------|------| | L1 配置层 | ConfigException 1001 | EnvFileNotFoundException 1002、EnvVarMissingException 1003、YamlParseException 1004、SecretMissingException 1005、ConfigTypeErrorException 1006 | | L2 网络层 | NetworkException 2001 | DNSResolveException 2002、ConnectionTimeoutException 2003、RequestTimeoutException 2004、ConnectionRefusedException 2005、SSLErrorException 2006、ProxyErrorException 2007、StatusCodeException 2008、ResponseParseException 2009 | | L3 数据层 | DataException 3001 | JsonDecodeException 3002、ExtractException 3003、ExtractPathNotFoundException 3004、ValidationException 3005、SchemaValidationException 3006、DataTransformException 3007 | | L4 存储层 | StorageException 4001 | RedisConnectionException 4002、RedisTimeoutException 4003、RedisKeyNotFoundException 4004、MySQLConnectionException 4005、MySQLQueryException 4006 | | L5 用例层 | CaseException 5001 | CaseNotFoundException 5002、CaseFormatException 5003、CaseSyntaxException 5004、DependencyException 5005、PreconditionFailedException 5006、HarParseException 5007、StepExecutionException 5008 | | L6 熔断/重试 | CircuitBreakerException 6001 | RetryExhaustedException 6002、RateLimitException 6003 | | L7 安全/权限 | SecurityException 7001 | AuthenticationException 7002、AuthorizationException 7003 | ### 8.2 异常使用场景 | 异常类 | 触发场景 | 日志示例 | |--------|---------|---------| | `CaseFormatException` | YAML 缺少必填字段 | `[创建订单流程] 用例文件 '第二步_查询订单' 格式错误: 缺少字段 'method'` | | `DependencyException` | 变量依赖缺失 | `[创建订单流程] 依赖变量 'order_id (request.json.orderId)' 未找到 (步骤: 第二步_查询订单)` | | `ValidationException` | 断言失败 | `字段 'code' eq 校验失败: 期望=0, 实际=500` | | `ExtractException` | JSONPath 提取失败 | `使用 'jsonpath' 从路径 '$.data.id' 提取变量失败` | | `CircuitBreakerException` | 熔断跳过 | `前置用例 [test_create] 失败，当前用例被熔断跳过` | | `StatusCodeException` | HTTP 非 2xx | `GET /api/order 返回 404 (接口不存在)` | | `RedisConnectionException` | Redis 连不上 | `无法连接到 Redis redis.example.com:6379/0` | | `MySQLQueryException` | SQL 执行失败 | `SQL执行失败: Table 'test.user' doesn't exist` | --- ## 九、Jenkins / CI 集成 本框架已内置 Jenkins 无缝集成能力，自动嗅探 `BUILD_URL`、`JOB_NAME`、`BUILD_NUMBER` 等环境变量，动态生成 Allure 在线访问链接。以下介绍完整的 Jenkins 配置方案。 ### 9.1 指定环境运行 在 Jenkins 的 **General → 参数化构建过程** 中，添加 **Choice Parameter**： - **名称**：`AUTO_ENV` - **选项**：`configs/envs/` 目录下所有环境文件名（如 `test7.yaml`、`prod.yaml`） - **描述**：选择本次构建要使用的测试环境 手动构建时，Jenkins 弹出的下拉框中选择对应环境文件即可。 ### 9.2 提交代码自动推送 在 Jenkins **源码管理** 中配置 Git 仓库，然后在 Git 仓库的 **Webhooks** 中添加 Jenkins 触发地址。提交代码至 Git 后，即可自动触发构建执行。 如果需要再发布环境自动运行，需要找运维配置。 ### 9.3 配置 Shell 构建脚本 用来安装环境依赖与清理历史数据。脚本会检查容器是否安装 Redis、MySQL，安装依赖包，并在构建结束后回收资源。 ```bash #!/bin/bash # ========================================== # Interface 自动化测试框架 Jenkins 执行脚本 # ========================================== # 设计目标： # 1. 在 Jenkins 工作空间内启动测试所需环境。 # 2. 复用 Python 虚拟环境和依赖缓存，减少构建耗时。 # 3. 执行 run.py，并把真实退出码透传给 Jenkins。 # 4. 测试失败时仍生成 Allure 报告，方便排查。 # ========================================== set -e exec 2>&1 echo "==========================================" echo "🚀 开始构建自动化测试任务" echo "📂 工作空间: ${WORKSPACE}" echo "🔢 构建编号: ${BUILD_NUMBER}" echo "🌍 目标测试环境: ${AUTO_ENV}" echo "⏰ 开始时间: $(date)" echo "==========================================" # AUTO_ENV 是框架选择环境配置的关键变量。 # 如果没有显式传入，框架可能会回退到默认环境，导致误跑 prod.yaml。 if [ -z "${AUTO_ENV}" ]; then echo "❌ 致命错误: AUTO_ENV 未设置，请在 Jenkins 参数中选择 test7.yaml/prod.yaml" exit 1 fi # 所有相对路径都基于 Jenkins 当前工作空间。 cd "${WORKSPACE}" # ------------------------------------------ # [1/5] 启动并检查 Redis 缓存中间件 # ------------------------------------------ echo -e "\n▶ [1/5] 启动中间件 (Redis)..." REDIS_PORT=6379 REDIS_PID_FILE="${WORKSPACE}/redis_${BUILD_NUMBER}.pid" if command -v redis-server &> /dev/null; then # 如果 Redis 已经运行，直接复用，避免误杀其他任务正在使用的 Redis。 if redis-cli -p ${REDIS_PORT} ping 2>/dev/null | grep -q PONG; then echo "✅ Redis 已在 ${REDIS_PORT} 端口运行，直接复用" else # 只记录本次构建启动的 Redis pid，收尾时只关闭这个进程。 redis-server --daemonize yes --port ${REDIS_PORT} --pidfile "${REDIS_PID_FILE}" sleep 2 if redis-cli -p ${REDIS_PORT} ping 2>/dev/null | grep -q PONG; then echo "✅ Redis 启动成功" else echo "❌ Redis 启动失败" exit 1 fi fi else echo "❌ 致命错误: 当前 Jenkins 服务器未安装 Redis" exit 1 fi # ------------------------------------------ # [2/5] 装载 Python 独立虚拟环境 # ------------------------------------------ echo -e "\n▶ [2/5] 装载 Python 运行环境..." if [ -d "venv" ] && [ -f "venv/bin/activate" ]; then source venv/bin/activate echo "✅ 成功激活已有虚拟环境 (venv)" else echo "⏳ 首次运行，正在创建全新虚拟环境..." python3 -m venv venv source venv/bin/activate fi # 将虚拟环境 bin 放到 PATH 最前面，确保 python/pytest/allure 走当前项目环境。 export PATH="${WORKSPACE}/venv/bin:${PATH}" # requirements.txt 未变化时跳过依赖安装，提升 Jenkins 构建速度。 if [ -f "requirements.txt" ]; then if [ ! -f "venv/.deps_installed" ] || [ "requirements.txt" -nt "venv/.deps_installed" ]; then echo "📦 检测到依赖清单有更新，正在同步安装第三方库..." pip install -r requirements.txt --upgrade --quiet touch venv/.deps_installed echo "✅ 依赖同步完成" else echo "⚡ 依赖清单无变化，触发缓存极速跳过" fi fi # ------------------------------------------ # [3/5] 初始化测试与报告目录 # ------------------------------------------ echo -e "\n▶ [3/5] 清理历史报告残骸..." mkdir -p reports/temps # 只清理 Allure 原始结果，避免上一次构建数据混入本次报告。 find reports/temps -maxdepth 1 -type f \( -name "*.json" -o -name "*.txt" -o -name "*.xml" \) -delete 2>/dev/null || true echo "🧹 历史临时数据清理完毕" # ------------------------------------------ # [4/5] 执行自动化测试核心引擎 # ------------------------------------------ echo -e "\n▶ [4/5] 开始执行测试流水线..." EXIT_CODE=0 # 注入 Jenkins 上下文，框架通知模块会用这些变量拼接报告地址和构建信息。 export JENKINS_BUILD_NUMBER="${BUILD_NUMBER}" export JENKINS_BUILD_URL="${BUILD_URL}" export JENKINS_JOB_NAME="${JOB_NAME}" if [ -f "run.py" ]; then echo "🎯 触发入口: python run.py" # timeout 防止测试进程死锁卡住 Jenkins 节点。 # 捕获退出码而不是直接中断，是为了让 run.py/后续逻辑仍有机会生成报告。 timeout 1800 python run.py || EXIT_CODE=$? else echo "⚠️ 未找到 run.py，使用原生 pytest 兜底执行..." pytest testcase/ -v --alluredir=reports/temps --disable-warnings || EXIT_CODE=$? fi echo "📊 测试执行完毕，底层退出状态码: ${EXIT_CODE}" # ------------------------------------------ # [5/5] 收尾打扫与环境重置 # ------------------------------------------ echo -e "\n▶ [5/5] 执行退出回收协议..." # 只关闭本次构建启动的 Redis。 # 如果 Redis 是 Jenkins 机器上已有服务，则不处理，避免影响其他任务。 if [ -f "${REDIS_PID_FILE}" ]; then REDIS_PID=$(cat "${REDIS_PID_FILE}") if kill -0 "${REDIS_PID}" 2>/dev/null; then kill "${REDIS_PID}" 2>/dev/null || true echo "✅ 已关闭本次构建启动的 Redis 进程: ${REDIS_PID}" fi rm -f "${REDIS_PID_FILE}" else echo "ℹ️ Redis 非本次构建启动，跳过关闭" fi # 输出 Allure 原始结果数量，方便判断报告是否成功落地。 if [ -d "reports/temps" ]; then JSON_COUNT=$(find reports/temps -name "*.json" 2>/dev/null | wc -l) echo "📁 本轮共生成 Allure JSON 数据切片: ${JSON_COUNT} 个" fi if [ ${EXIT_CODE} -eq 0 ]; then echo "🎉 完美！所有接口测试全部通过！" else echo "🔴 警告：存在失败的用例或断言，请前往 Allure 报告查看详情！" fi echo "==========================================" echo "🏁 构建流水线运行结束" # 透传测试退出码，让 Jenkins 正确标记构建状态。 exit ${EXIT_CODE} ``` ### 9.4 构建后查看 Allure 报告 在 Jenkins 的 **构建后操作** 中添加 **Allure Report** 插件。执行完毕后，点击构建记录的 "Allure Report" 链接即可查看 HTML 报告。确保 Jenkins 已安装 Allure 插件。 --- ## 十、常见问题排查指南 ### Q1：为什么我的 `${变量}` 没有被替换？ 检查以下几点： 1. 变量名是否拼写正确，区分大小写 2. 变量是否在 `extract` / `extract_local` 中正确提取 3. 如果是跨文件变量，检查是否使用了 `extract`（而非 `extract_local`）写入 Redis 4. 变量是否含有未闭合的引号或特殊字符，导致 YAML 解析异常 5. 如果是 `${get_extract_data(xxx)}`，检查 Redis 中是否存在该 key（可用 `redis-cli keys '*xxx*'` 排查） ### Q2：Allure 报告中用例名显示为 `$csv{name}`，没有替换？ CSV 数据驱动的名称替换发生在 `pytest.mark.parametrize` 之后。建议在 `test_xxx.py` 的驱动方法中使用 `allure.dynamic.title(caseinfo['name'])`，这样 Pytest 会正确显示裂变后的用例名。 ### Q3：数据库断言报"SQL 安全拦截"？ 框架仅允许 `SELECT` 语句。如果 SQL 里包含了 `insert`、`update`、`delete`、`drop`、`truncate`、`alter`、`create` 等关键字，会被直接熔断。测试数据预埋请使用 `setup_sql`，测试清理请使用 `teardown_sql` 或 `teardown_steps`。 注意：这里说的是 `validation.db` 数据库断言，只允许查询。`setup_sql` / `teardown_sql` 走 `sql_executor.execute_sql_statements`，允许 `SELECT` / `INSERT` / `UPDATE` / `DELETE`，但会拒绝多语句和 DDL 危险操作。 ### Q4：用例失败后为什么后续用例都被跳过了？ 这是框架的**熔断机制**在生效。默认情况下，若某个用例文件失败，框架会将该文件标记为"已熔断"，避免无效执行。你可以在 `pytest.ini` 中调整重试次数，或在本地调试时单文件运行来规避。 ### Q5：如何只运行某一个 YAML 文件？ ```bash pytest testcase/api_auto/demo/test_demo.py -vs ``` ### Q6：如何调试单个步骤？ 打开 `logs/日期/` 目录下对应 PID 的日志文件，搜索用例名称。框架会完整打印：请求 URL、请求头、请求体、实际响应体、断言结果。 ### Q7：pytest-xdist 并发下日志错乱？ 框架已通过 PID 隔离日志文件，每个 Worker 进程有独立的日志文件。如果控制台输出仍然交错，这是正常现象（多个进程同时向 stdout 写入）。查看 `logs/日期/` 下的各 PID 日志文件即可看到隔离后的完整日志。 ### Q8：Redis 连接失败怎么办？ 1. 检查环境配置文件中的 Redis 配置是否正确 2. 确认 Redis 服务已启动：`redis-cli ping` 应返回 `PONG` 3. 检查防火墙是否放行了 Redis 端口 4. 如果 Redis 设置了密码，确认配置文件中 `password` 字段已填写 ### Q9：如何查看当前加载的是哪个环境配置？ 在日志中搜索 `[Config ]` 关键字，第一行会打印加载的环境文件名。或者在 Python 中直接打印： ```python from configs.config_manager import settings print(settings.env_name) # 如 'test7.yaml' ``` ### Q10：Mock 没有生效，还是发了真实请求？ 1. 确认 `mock` 节点在 YAML 中的位置正确（与 `request` 同级） 2. 确认 `mock` 节点格式正确：简单 Mock 直接写 `{status_code, json}`，链式 Mock 写 `{chain: [...]}` 3. 检查 HttpClient 的日志输出，如果看到 `>>> [Mock ] 网关拦截生效` 说明 Mock 命中了；如果看到 `>>> [Request ]` 则是真实请求 ### Q11：为什么运行结束后 Redis 里还有数据？ 先区分数据类型： 1. `extract:{env}:{session_id}:*` 是本轮提取变量，正常情况下会在 `pytest_sessionfinish` 阶段清理。 2. `pytest:circuit_breaker:failed_files` 是熔断状态，下一轮启动时会清理框架级状态，也可能在用例通过后自愈删除。 3. 如果进程被强杀、机器断电、pytest 没有走到 `sessionfinish`，本轮提取变量可能残留。 排查方式： ```bash redis-cli keys "extract:*" redis-cli keys "pytest:circuit_breaker:*" ``` 确认是历史异常残留后，再按环境和 session 前缀精确删除，不建议直接 `FLUSHDB`。 ### Q12：为什么只运行部分目录时 Allure 编号还是从 1 开始？ 这是预期行为。`next(m_id)` / `next(c_id)` 在装饰器阶段只生成占位符，真实编号会在 pytest 收集完成后，按本轮实际收集到的 items 重新生成。因此全量运行从 `M001_`、`C001_` 开始，只运行某几个目录也会按这几个目录重新从 `M001_`、`C001_` 开始。 ### Q13：CSV 数据驱动裂变失败怎么办？ 重点检查以下几项： 1. YAML 中 `parameters` 指向的 CSV 路径是否存在。 2. CSV 表头是否和 YAML 中 `$csv{列名}` 完全一致。 3. CSV 单元格如果是字典、列表、布尔值、数字，格式是否能被 `ast.literal_eval` 正确识别。 4. 替换后的 YAML 是否仍是合法 JSON/YAML 结构，尤其注意引号和冒号。 5. 如果用例名里有 `$csv{name}`，驱动方法中必须写 `allure.dynamic.title(caseinfo['name'])`。 ### Q14：用例被熔断后如何快速定位源头？ 1. 查看 Allure 中被跳过用例的“熔断详情”附件。 2. 日志中搜索 `依赖链路熔断`，可以看到触发熔断的 `nodeid`。 3. 熔断粒度是测试文件，同一个文件里前置用例最终失败后，后续用例会被跳过。 4. 如果启用了重试，中间失败不会触发熔断；只有重试耗尽后的最终失败才会写入熔断。 ### Q15：新增动态函数后 YAML 调不到怎么办？ 1. 方法必须写在 `common/runtime_functions.py` 的 `RuntimeFunctions` 类中。 2. YAML 调用格式必须是 `${函数名()}` 或 `${函数名(参数)}`。 3. 函数名区分大小写。 4. 非幂等函数，例如时间戳、随机数、nonce，不要强行加缓存。 5. 如果函数需要读取 Redis 跨文件变量，优先复用 `RuntimeCache.read_extract_variable()`。 ### Q16：什么时候用 `teardown_sql`，什么时候用 `teardown_steps`？ | 场景 | 推荐写法 | 原因 | |------|----------|------| | 清理当前接口直接产生的数据 | `teardown_sql` | 和当前 step 生命周期绑定，位置更近。 | | 需要调用接口释放业务状态 | `teardown_steps` | 例如取消订单、删除标签、释放库存。 | | 主链路失败后也必须执行的兜底清理 | `teardown_steps` | 主链路异常后仍会进入 finally 执行。 | | 只做落库校验 | `validation.db` | 数据库断言只允许 SELECT，更安全。 | --- ## 附录：快速查阅卡 ### YAML 字段速查表 | 字段 | 层级 | 必填 | 说明 | |------|------|------|------| | `name` | 顶层/step | 是 | 用例/步骤名称 | | `base_url` | 顶层/step | 否 | 基础域名（step 级可覆盖顶层） | | `config` | 顶层 | 否 | 全局配置节点（name/base_url/variables/request/validation） | | `config.variables` | config | 否 | 多步骤流程级内存变量初始值 | | `config.request` | config | 否 | 多步骤流程公共请求配置，会合并到每个 step | | `config.validation` | config | 否 | 多步骤流程公共断言，会合并到每个 step | | `steps` | 顶层 | 否 | 多步骤主链路列表 | | `teardown_steps` | 顶层 | 否 | 清理链路列表 | | `request` | step | 是 | 请求配置 | | `request.method` | step | 是 | HTTP 方法 | | `request.path` | step | 条件 | URL 路径（与 url 二选一） | | `request.url` | step | 条件 | 完整 URL（与 path 二选一） | | `request.headers` | step | 否 | 请求头字典 | | `request.cookies` | step | 否 | Cookie 配置；页面接口常写 `${get_base_url(wms_cookie)}` / `${get_base_url(oms_cookie)}` | | `request.json` | step | 否 | JSON 请求体 | | `request.params` | step | 否 | URL 查询参数 | | `request.data` | step | 否 | Form 表单数据 | | `request.files` | step | 否 | 文件上传 | | `request.timeout` | step | 否 | 超时时间（秒） | | `extract` | step | 否 | 单值提取 → 本地 + Redis | | `extract_local` | step | 否 | 单值提取 → 仅本地 | | `extract_list` | step | 否 | 列表提取 → 本地 + Redis | | `extract_list_local` | step | 否 | 列表提取 → 仅本地 | | `validation` | step | 是 | 断言规则列表 | | `retry` | step | 否 | 重试配置 `{times, interval}`；`times` 表示最大执行次数，包含首次执行 | | `mock` | step | 否 | Mock 配置 | | `setup_sql` | step | 否 | 前置 SQL 列表 | | `teardown_sql` | step | 否 | 后置 SQL 列表 | | `parameters` | 顶层 | 否 | CSV 数据驱动配置 | **不支持的位置**： | 写法 | 原因 | |------|------| | `config.steps` | `steps` 是主业务链路，必须和 `config` 同级。 | | `config.teardown_steps` | `teardown_steps` 是用例级收尾流程，必须和 `config` 同级。 | | `config.parameters` | `parameters` 是 CSV 裂变入口，必须放在单接口用例顶层。 | | 顶层 `variables` + `steps` | 多步骤流程统一使用 `config.variables`，避免同一能力出现两种入口。 | ### 变量引用速查表 | 语法 | 示例 | 适用场景 | |------|------|---------| | `${变量名}` | `${order_id}` | 引用当前 YAML 内存变量（`variables` / `extract` / `extract_local`） | | `${函数名()}` | `${get_timestamp()}` | 调用 RuntimeFunctions 无参函数 | | `${函数名(参数)}` | `${get_extract_data(order_id)}` | 调用带1个参数的函数 | | `${函数名(参数1, 参数2)}` | `${get_extract_data(sku_list, -1)}` | 调用带多个参数的函数 | | `$csv{列名}` | `$csv{name}` | 引用 CSV 数据（YAML 模板中） | | `$.data.xxx` | `$.data.orderId` | JSONPath 提取表达式 | | `$.request.headers.xxx` | `$.request.headers.Trace-Id` | 逆向提取（从请求参数中抠数据） | | `${get_base_url(key)}` | `${get_base_url(base_wms_url)}` | 读取环境配置中的基础域名 | | `${get_environment_value(key)}` | `${get_environment_value(clientId)}` | 读取 environment 节点配置；新用例推荐使用 | | `${get_env_value(key)}` | `${get_env_value(appid)}` | 读取环境配置中的鉴权参数 | | `${get_base_data(key)}` | `${get_base_data(skuA)}` | 读取 base_web_data 中的枚举值 | | `${get_base_data_int(key)}` | `${get_base_data_int(customerId)}` | 读取 base_web_data 中的数字枚举值 | | `${hmac_sha256_sign(path)}` | `${hmac_sha256_sign(/api/order/create)}` | 生成 HMAC-SHA256 签名 | ### 断言类型速查表 | 类型 | 示例 | 说明 | |------|------|------| | `code` | `- code: 200` | HTTP 状态码严格比对 | | `eq` | `- eq: {code: 0, msg: success}` | 字典严格全等（仅校验指定字段） | | `ne` | `- ne: {code: 1000}` | 反向不等 | | `contain` | `- contain: success` | 深度包含（支持 JSONPath 精确路径） | | `contain` | `- contain: {status: ok}` | 字典深度包含（全层级扫描 Key） | | `contain` | `- contain: {"$.data.list[0].status": 1}` | 精确 JSONPath 包含 | | `db` | `- db: {sql: "SELECT ... WHERE id = %s", args: [1], eq: {id: 1}}` | 数据库断言（仅允许 SELECT，`args` 可选；使用 `%s` 占位符时再传） | | `db` | `- db: {sql: "SELECT count(*) AS cnt FROM t_order WHERE status = %s", args: [1], contain: {cnt: 5}}` | 数据库包含断言 | | `db` | `- db: {sql: "SELECT id FROM t_order WHERE order_no = %s", args: ["${order_no}"], empty: true}` | 数据库判空；`true` 表示期望查不到，`false` 表示期望查到 | ### `get_extract_data` 格式化参数速查表 | out_format | 作用 | 示例说明 | |-----------|------|---------| | 不传 / None | 原始返回 | 列表稳定返回第一个元素，字典/字符串/数字直接返回 | | `0` | 首元素快捷写法 | 从列表中返回第一个元素，行为与默认值一致 | | `-1` | 序列降维 | 将列表拼接为逗号分隔的字符串，常用于 SQL 的 `IN` 语法 | | `-2` | 类型纯化 | 剔除列表中的非数字项，全部转为 `int` 列表 | | 正整数 `N` | 绝对索引 | 取列表的第 `N` 个元素（注意是 1-based 索引） | **示例**： ```yaml # 假设 Redis 中存了 sku_id_list = [101, 102, 103] json: # 取第一个 SKU skuId: ${get_extract_data(sku_id_list, 1)} # 取第一个 SKU（和默认行为一致） randomSku: ${get_extract_data(sku_id_list, 0)} # 拼成 SQL 的 IN 参数字符串："101,102,103" skuInParam: ${get_extract_data(sku_id_list, -1)} # 全部转为 int 列表 skuIdList: ${get_extract_data(sku_id_list, -2)} ``` **重要**：`get_extract_data` 只能读取通过 `extract` / `extract_list` 落盘到 Redis 的变量。`extract_local` / `extract_list_local` 写入的内存变量，在当前 YAML 文件内直接通过 `${变量名}` 引用即可，**无需也不能**用 `get_extract_data` 读取。 ### RuntimeFunctions 内置函数完整列表 | 函数 | 参数 | 返回值 | 说明 | 是否缓存 | |------|------|--------|------|---------| | `get_extract_data(key, format)` | key: 变量名; format: 格式化指令 | Any | 从 Redis 读取变量，支持多态格式化 | 否 | | `get_base_url(node_name)` | node_name: 配置节点名 | str | 读取环境配置中的基础域名 | 是 | | `get_mysql_config(node_name)` | node_name: 配置节点名 | dict | 读取数据库实例拓扑配置 | 是 | | `get_environment_value(key)` | key: 环境变量名 | str | 读取 environment 节点配置；新用例推荐使用 | 否 | | `get_env_value(key)` | key: 环境变量名 | str | 读取 environment 节点配置 | 是 | | `get_base_data(key)` | key: 数据节点名 | str | 读取 base_web_data 节点枚举值，统一按字符串返回 | 是 | | `get_base_data_int(key)` | key: 数据节点名 | int | 读取 base_web_data 节点数字枚举值，适合断言接口返回的数字 ID | 是 | | `get_headers(params_type)` | params_type: 'data'/'json' | dict | 返回对应 Content-Type 的请求头 | 是 | | `get_timestamp()` | 无 | str | 返回当前毫秒时间戳（从 TLS 线程缓存读取，若未命中则自动生成） | 否 | | `get_nonce()` | 无 | str | 返回当前线程缓存中的 2 位随机数；未命中时自动生成 | 否 | | `timestamp()` | 无 | str | 每次实时生成新的毫秒时间戳，不复用线程缓存 | 否 | | `generate_order_number(prefix)` | prefix: 前缀字符串 | str | 生成带前缀的随机单号 | 否 | | `hmac_sha256_sign(path)` | path: 接口路径 | str | 生成 HMAC-SHA256 签名 | 否 | **缓存说明**： - **是**：走 `lru_cache`，相同参数多次调用直接返回缓存结果，适合读取配置类函数 - **否**：强制实时计算，适合时间戳、随机数等动态函数 ### pytest.ini 配置详解 以下示例为**当前仓库默认值**，不是演示值： ```ini [pytest] # -v: 详细输出，显示每个用例的执行结果 # -s: 输出 print/logging 到控制台 # --alluredir: Allure 原始数据输出目录 # --clean-alluredir: 运行前清理历史 Allure 数据 # --reruns: 用例失败后的重试次数 # --reruns-delay: 每次重试的间隔秒数 # -n: 并发进程数（auto=自动按 CPU 核心数，具体数字如 3=3个进程） # --dist: 用例分发策略（loadscope=按类分发，loadfile=按文件分发） addopts = -vs --alluredir=reports/temps --clean-alluredir --reruns 1 --reruns-delay 1 -n 4 --dist=loadfile # 测试用例搜索路径 testpaths = testcase # 文件/类/函数的命名规则 python_files = test_*.py python_classes = Test* python_functions = test_* # 自定义标记 markers = smoke: 冒烟用例 ``` **常用配置调整**： | 场景 | 修改建议 | |------|---------| | 本地调试单文件 | `-n 1`（关闭并发）| | CI/CD 加速 | `-n auto --dist=loadfile` | | 不稳定接口重试 | `--reruns 2 --reruns-delay 3` | | 只看失败详情 | 去掉 `-s`，只保留 `-v` | **默认值解读**： - `--dist=loadfile`：同一个测试文件尽量分配到同一个 Worker，和当前熔断、缓存、日志阅读习惯更匹配。 - `--reruns 1 --reruns-delay 1`：默认给一次瞬态失败自愈机会，但不会把重跑拉得过重。 - `-n 4`：当前仓库默认按 4 个进程跑；如果本机资源紧张或本地排查，建议临时降到 `-n 1`。 ### 环境配置文件完整字段说明 ```yaml # ========================================== # 基础配置 # ========================================== # 默认基础域名（优先级低于 base_url 配置） host: "http://api.example.com" # 命名基础域名（通过 ${get_base_url(key)} 读取） base_url: base_wms_url: "http://wms.example.com" base_wx_url: "http://wx.example.com" base_oms_url: "http://oms.example.com" # 页面接口登录态常通过 request.cookies 引用这两个配置 oms_cookie: cookie: "JSESSIONID=xxx; oms_u=xxx" wms_cookie: cookie: "JSESSIONID=xxx; wms_u=xxx" # ========================================== # MySQL 数据库配置 # ========================================== mysql: enable: true host: "mysql.example.com" port: 3306 username: "user" password: "123456" database: "test_db" charset: "utf8" # 可选：连接池大小 maxconnections: 10 # ========================================== # Redis 缓存配置 # ========================================== redis: host: "redis.example.com" port: 6379 db: 0 password: "" # 缓存过期时间（秒） expire: 3600 # ========================================== # HTTP 客户端配置 # ========================================== http: # 是否校验 HTTPS 证书。未配置时默认 true。 verify_ssl: true # ========================================== # 环境级业务参数 # ========================================== environment: # 微信公众号相关 grant_type: "client_credential" appid: "wx1234567890" secret: "your_secret_key" # 默认登录鉴权接口参数。auth.param_keys 会按这些 key 从 environment 中取值 clientId: "your_client_id" userId: "your_user_id" refreshToken: "your_refresh_token" # 其他业务参数... # ========================================== # 全局登录鉴权配置 # ========================================== # pytest 会话启动时，authorize_session 会根据 auth 配置调用登录/刷新 token 接口 # 获取到的 token 会写入共享文件和 Redis，后续 YAML 可通过 ${get_extract_data(accessToken)} 读取 auth: base_url_key: "base_wms_url" # 从 base_url 哪个 key 取登录域名 endpoint: "/api/oauth/refreshToken" # 登录/刷新 token 接口路径 method: "GET" # GET 使用 params 传参，POST 使用 json 传参 param_keys: # 参数名列表，值从 environment 节点读取 - "clientId" - "userId" - "refreshToken" token_path: "data.accessToken" # 响应 JSON 中 accessToken 的点号路径 # ========================================== # 通知配置 # ========================================== # 邮件通知开关 is_email_msg: false # 邮件服务器配置 email: email_name: "your_email@qq.com" email_password: "your_smtp_password" email_port: 465 email_server: "smtp.qq.com" email_receiver: "receiver1@example.com" # 企微通知开关 is_wecom_msg: false # 企微机器人 Webhook wechat: webhook_key: "xxx" # 如果已经有完整 Webhook 地址，也可以写： # webhook_url: "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx" # 兼容旧配置写法：如果历史环境文件使用 wechat_webhook，框架同样可读取 # wechat_webhook: # webhook_key: "xxx" # ========================================== # Allure 报告配置 # ========================================== # 本地 Allure 报告地址（Jenkins 环境下自动生成） allure_url: "http://localhost:63342/interface_wms/reports/allures/index.html" ``` --- ## 附录二：图集速查 本附录集中收录文档中所有架构图和运行流程图，方便快速查阅和复习。 ### 图 1：六层架构总览 ```mermaid graph TB L1["Layer 1: 触发与入口层"] L2["Layer 2: 数据驱动与调度层"] L3["Layer 3: 核心执行引擎层"] L4["Layer 4: 基础业务工具链"] L5["Layer 5: 底层基建与目标服务"] L6["Layer 6: 报告与通知层"] L1 --> L2 --> L3 --> L4 --> L5 --> L6 ``` 框架采用六层架构设计，从上到下依次是：触发入口 → 数据驱动调度 → 核心执行引擎 → 基础业务工具链 → 底层基建 → 报告通知。每一层只依赖下层，上层通过 Executor 统一调度。 ### 图 2：全局生命周期钩子交互 ```mermaid graph TD Start(["用户/Jenkins 执行 python run.py"]) --> RunPy["run.py 调用 pytest.main()"] subgraph 启动阶段 ["启动阶段"] RunPy --> SessionStart["pytest_sessionstart（仅 Master 执行）"] SessionStart --> S1["清理 Redis + 预加载熔断缓存"] S1 --> Auth["authorize_session（所有进程参与）"] Auth --> Auth1["FileLock 互斥 → 登录 → 写入共享文件"] end subgraph 收集阶段 ["收集阶段"] Auth1 --> Collect["收集用例 + parametrize 裂变"] Collect --> CollectionFinish["pytest_collection_finish（批量检查熔断 + 按需预热 MySQL）"] end subgraph 执行循环 ["执行循环"] CollectionFinish --> Loop{循环：每个 TestItem} Loop -->|有下一个| Call["pytest_runtest_call"] Call --> Check{熔断检查} Check -->|通过| Execute["执行用例"] Check -->|已熔断| Skip["标记 Skip"] Execute --> Report["pytest_runtest_makereport"] Skip --> Report Report --> Update["更新熔断状态"] Update --> Loop end subgraph 收尾阶段 ["收尾阶段"] Loop -->|结束| Summary["pytest_terminal_summary"] Summary --> S2["打印汇总 + environment.properties"] S2 --> Finish["pytest_sessionfinish"] Finish --> F1["发送通知 + 关闭连接池"] F1 --> Allure["allure generate 渲染 HTML"] Allure --> End(["结束"]) end ``` 全局生命周期钩子交互全景。框架通过 Pytest 的钩子机制在 5 个关键时间点介入： 1. **pytest_sessionstart**：仅 Master 进程执行环境清理和熔断缓存预加载，所有进程参与 authorize_session 获取全局 Token。 2. **pytest_collection_finish**：用例收集完成后，批量检查各文件的熔断状态；如果本轮 `caseinfo` 中存在 `db/setup_sql/teardown_sql`，则在执行进程内按需预热 MySQL 连接池。 3. **pytest_runtest_call / makereport**：每个 TestItem 执行前后进行熔断检查和状态更新。已熔断的用例直接 Skip，避免无效执行。 4. **pytest_terminal_summary / sessionfinish**：测试结束后打印汇总、生成环境信息、发送通知、关闭所有连接池。 5. **allure generate**：最后由 run.py 调用 Allure CLI 渲染 HTML 报告。 全局生命周期钩子时间线。pytest_sessionstart 仅 Master 执行清理，authorize_session 所有进程参与获取 Token，pytest_collection_finish 批量检查熔断，最后用例循环中逐条执行并更新熔断状态。 ### 图 3：单接口极简模式运行流程 ```mermaid graph TD Start(["Pytest 调度用例"]) --> Wrap["read_yaml_cases 读取 YAML 用例列表"] Wrap --> Parametrize["@pytest.mark.parametrize 注册用例"] Parametrize --> Executor["Executor.run_case"] subgraph 准备阶段 ["准备阶段"] Executor --> Deepcopy["deepcopy 深拷贝隔离模板"] Deepcopy --> Config["提取 config 全局配置"] Config --> Init["初始化 context_vars"] end subgraph 单步执行 ["单步执行"] Init --> Validate["validate_step_schema 语法校验"] Validate --> Setup["setup_sql 前置 SQL"] Setup --> Retry{重试循环} Retry -->|重试| Send["VariableResolver 解析变量 → HttpClient 发包（含 Mock 拦截检查）"] Send --> ParseJson["安全解析响应 JSON"] ParseJson --> Record["Allure 捕获本次尝试附件"] Record --> Extract["变量提取（四大通道）"] Extract --> Assert["软断言校验"] Assert --> CheckRetry{还有重试次数?} CheckRetry -->|是| Retry CheckRetry -->|否| Next[本步结束] end Next --> Teardown["teardown_sql 后置 SQL（setup_sql 成功后执行）"] Teardown --> End(["结束"]) ``` 单接口极简模式的执行流程（无 steps 节点）。read_yaml_cases 负责读取并返回用例列表；Executor 在运行时将无 steps 的单接口用例统一按单步骤执行，并调度 StepRunner 依次执行 validate_step_schema → setup_sql → 重试循环（解析/发包/提取/断言）→ teardown_sql。注意：如果 setup_sql 阶段失败，teardown_sql 不会执行；setup_sql 成功后，请求、提取或断言失败都会继续触发 teardown_sql。 ### 图 4：多步骤链路模式运行流程 ```mermaid graph TD Start(["Executor.run_case"]) --> Deepcopy["deepcopy 深拷贝隔离模板"] Deepcopy --> Config["提取 config 全局配置"] Config --> Steps["提取 steps 列表"] Steps --> TeardownSteps["提取 teardown_steps 列表"] TeardownSteps --> Init["初始化 context_vars + config.variables"] subgraph 主链路 ["主链路"] Init --> Try["try: 主链路"] Try --> ExitStack["with ExitStack 资源托管"] ExitStack --> Loop{遍历 steps} Loop -->|有下一个| Merge["合并全局配置到当前步骤"] Merge --> Step["单步执行"] Step --> CheckSuccess{步骤结果?} CheckSuccess -->|成功| Loop CheckSuccess -->|失败| SaveException["保存异常 + break 中断"] SaveException --> Finally Loop -->|结束| Finally end subgraph Teardown兜底 ["Teardown 兜底"] Finally --> TeardownLoop{遍历 teardown_steps} TeardownLoop -->|有下一个| TdMerge["合并全局配置到当前步骤"] TdMerge --> Silent["异常静默隔离（不掩盖主链路错误）"] Silent --> TeardownLoop TeardownLoop -->|结束| CheckException{主链路有异常?} CheckException -->|有| Raise["重新抛出异常（Pytest 标红）"] CheckException -->|无| End(["结束"]) end Raise --> End ``` 多步骤链路用例的执行流程。主链路 steps 按顺序执行，任一步骤失败即 break；无论主链路成败，finally 中强制执行 teardown_steps，teardown 异常静默隔离不掩盖主链路错误。 ### 图 5：CSV 数据驱动裂变运行流程 ```mermaid graph TD subgraph 收集阶段 ["Pytest 收集阶段"] Start(["开始"]) --> YamlReader["read_yaml_cases 读取 YAML"] YamlReader --> CheckParams{发现 parameters?} CheckParams -->|是| csv_case_expander["expand_csv_cases()"] csv_case_expander --> ReadCsv["读取 CSV（带 LRU 缓存）"] ReadCsv --> Parse["解析表头 + 数据行"] Parse --> Product["itertools.product 笛卡尔积"] Product --> Replace["str.replace 文本替换"] Replace --> Deserialize["orjson.loads 反序列化"] Deserialize --> Register1["@pytest.mark.parametrize 注册用例"] CheckParams -->|否| Register1 end subgraph 执行阶段 ["Pytest 执行阶段"] Register1 --> Execute["每个 TestItem 独立执行"] Execute --> End(["结束"]) end ``` CSV 驱动裂变过程。read_yaml_cases 发现 parameters 后调用 expand_csv_cases，读取 CSV → 笛卡尔积组合 → 文本替换 → orjson 反序列化，最终注册为 N 个独立 TestItem。 ### 图 6：跨文件 Redis 变量传递运行流程 ```mermaid graph TD subgraph 文件A ["文件 A 写入方"] A1["文件 A 执行用例"] --> A2["单步执行完成"] A2 --> A3["extract 提取变量"] A3 --> A4["RuntimeCache.write_extract_variables 写入 Redis"] A4 --> A5["Redis Key: extract:{env}:{session}:order_id"] end subgraph 文件B ["文件 B 读取方"] B1["文件 B 执行用例"] --> B2["VariableResolver 解析变量"] B2 --> B3["${get_extract_data(order_id)}"] B3 --> B4["RuntimeFunctions 读取 Redis"] B4 --> B5["命中 → 返回 ORD123"] B5 --> B6["组装请求 → 发包"] end A5 -.->|跨文件变量传递| B4 ``` 跨文件变量传递原理。文件 A 通过 extract 将变量写入 Redis（Key 含环境名 + SessionID），文件 B 通过 `${get_extract_data()}` 从 Redis 读取，实现同一测试批次内跨 YAML 文件的数据共享。同一测试批次内要避免不同流程写入同名 Redis 变量，建议使用带业务前缀的变量名。 ### 图 7：Mock 挡板模式运行流程 ```mermaid graph TD Start(["HttpClient.execute_api_request"]) --> Log["打印审计日志"] Log --> CheckMock{case_info 含 mock?} CheckMock -->|否| RealRequest["GLOBAL_SESSION.request（真实网络）"] CheckMock -->|是| MockBuild["MockResponse.build_fake_response"] MockBuild --> MockType{Mock 类型} MockType -->|简单| SimpleMock["直接返回固定数据"] MockType -->|chain 顺序| ChainMock["按 _mock_idx 轮询返回"] MockType -->|rules 条件| CheckRule{when 是 request_json 子集?} CheckRule -->|是| ReturnThen["返回 then 数据"] CheckRule -->|否| RealRequest RealRequest --> Response["返回 Response"] SimpleMock --> Response ChainMock --> Response ReturnThen --> Response Response --> End(["上层代码完全无感知"]) ``` Mock 拦截网关。HttpClient 发包前检查 mock 节点，命中则直接返回 FakeResponse，未命中则走真实网络。支持简单 Mock、chain 顺序轮询、rules 条件匹配三种模式。 ### 图 8：数据流转路径总图 ```mermaid graph LR YAML[YAML 文件] --> Reader[read_yaml_cases] Reader --> Driver[expand_csv_cases] Driver --> Executor[Executor] Executor --> DataExtractor[DataExtractor] DataExtractor --> Assertions[Assertions] Assertions --> Allure[Allure 报告] Executor --> VariableResolver[VariableResolver] VariableResolver --> HttpClient[HttpClient] HttpClient --> MockResponse[MockResponse] VariableResolver --> RuntimeFunctions[RuntimeFunctions] VariableResolver --> CacheMgr[RuntimeCache] CacheMgr <--> Redis[(Redis)] ``` 框架核心数据流转全景。YAML → read_yaml_cases → expand_csv_cases → Executor → 依次经过 VariableResolver（解析）→ HttpClient（发包）→ DataExtractor（提取）→ Assertions（断言）→ Allure 报告。 ### 图 9：Redis 熔断状态缓存双层架构 ```mermaid graph TD subgraph 查询请求 ["查询请求"] QStart(["开始"]) --> L1Check{L1 本地内存缓存} L1Check -->|命中| L1Hit["O(1) 纳秒级返回"] L1Check -->|未命中| L2Check{L2 Redis 全局缓存} L2Check -->|命中| L2Hit["毫秒级返回（回填 L1）"] L2Check -->|未命中| ReturnNone["返回 None"] end subgraph 写入请求 ["写入请求"] WStart(["开始"]) --> WriteRedis["先写 Redis（持久化）"] WriteRedis --> UpdateL1["再更新本地缓存"] UpdateL1 --> WEnd(["结束"]) end L1Hit --> QEnd(["结束"]) L2Hit --> QEnd ReturnNone --> QEnd ``` 熔断状态采用 L1 本地内存 + L2 Redis 双层缓存。查询先走本地（纳秒级），未命中再走 Redis（毫秒级）；写入先写 Redis 再更新本地。每 5 秒通过 `circuit_breaker_store.refresh_cache()` 同步，`CircuitBreakerStore._cache_lock` 保证线程安全。 --- ### 图 10：完整执行时序图 ```mermaid sequenceDiagram participant USER as 用户/Jenkins participant RUN as run.py participant PT as Pytest participant HOOK as conftest / pytest_hooks participant READER as YAML/CSV读取器 participant EXE as Executor participant SQL as sql_executor / MySQL participant RESOLVE as VariableResolver / RuntimeFunctions participant HTTP as HttpClient / Mock / 目标服务 participant EXT as DataExtractor participant ASSERT as Assertions participant CACHE as RuntimeCache participant REDIS as Redis participant ALLURE as Allure participant NOTIFY as 企微/邮件 %% 阶段一：启动、清理、收集 USER->>RUN: ① 执行 python run.py RUN->>RUN: ② 加载环境配置和 pytest 参数 RUN->>PT: ③ pytest.main() 启动测试会话 PT->>HOOK: ④ pytest_sessionstart HOOK->>REDIS: ⑤ 清理框架级状态 + 预加载熔断缓存 PT->>HOOK: ⑥ authorize_session 初始化登录 HOOK->>HOOK: ⑦ FileLock 互斥，避免多进程重复登录 HOOK->>CACHE: ⑧ 写入全局 accessToken CACHE->>REDIS: ⑨ 保存跨进程 Token 变量 PT->>READER: ⑩ pytest_generate_tests 读取 YAML READER->>READER: ⑪ safe_load + 结构规范化 alt YAML 包含 parameters READER->>READER: ⑫ CSV 笛卡尔积裂变 else 普通 YAML READER->>READER: ⑫ 保持 YAML 原始用例列表 end READER-->>PT: ⑬ 注册 N 个 TestItem PT->>HOOK: ⑭ pytest_collection_finish HOOK->>REDIS: ⑮ 批量检查熔断状态 opt 本轮存在 db / setup_sql / teardown_sql HOOK->>SQL: ⑯ 按需预热 MySQL 连接池 end %% 阶段二：每个用例执行 loop 每个 TestItem PT->>HOOK: ⑰ pytest_runtest_call 熔断检查 alt 当前文件已熔断 HOOK-->>PT: ⑱ pytest.skip 跳过当前用例 else 允许执行 PT->>EXE: ⑱ 下发 case_info EXE->>EXE: ⑲ deepcopy 隔离模板 + 提取 config EXE->>EXE: ⑳ 读取外层 steps / teardown_steps loop 主链路每个 step EXE->>EXE: ㉑ 合并 config.request / config.validation EXE->>EXE: ㉒ validate_step_schema 静态校验 opt 存在 setup_sql EXE->>RESOLVE: ㉓ 解析 setup_sql 中的变量 RESOLVE-->>EXE: ㉔ 返回已替换 SQL EXE->>SQL: ㉕ 执行前置 SQL SQL-->>EXE: ㉖ 返回查询结果或影响行数 end alt setup_sql 执行失败 EXE->>EXE: ㉗ 记录主流程异常，跳过本 step 的 teardown_sql else setup_sql 成功或未配置 loop 当前 step 的 retry.times 最大执行次数 EXE->>RESOLVE: ㉘ 解析 request / headers / body / path opt 读取当前 YAML 内存变量 RESOLVE->>RESOLVE: ㉙ 读取 context_vars end opt 调用 RuntimeFunctions RESOLVE->>RESOLVE: ㉙ 调用运行时函数 end opt 调用 get_extract_data RESOLVE->>CACHE: ㉙ 读取 Redis 提取变量 CACHE->>REDIS: ㉚ GET extract:{env}:{session}:key REDIS-->>CACHE: ㉛ 返回变量值 CACHE-->>RESOLVE: ㉜ 返回变量值 end RESOLVE-->>EXE: ㉝ 返回完整请求参数 EXE->>HTTP: ㉞ 发送请求 alt 命中 mock HTTP-->>EXE: ㉟ 返回 FakeResponse else 未命中 mock HTTP-->>EXE: ㉟ 返回真实 Response end EXE->>ALLURE: ㊱ 捕获本次尝试的请求、响应、耗时附件 EXE->>EXT: ㊲ 执行 extract / extract_local EXT->>EXE: ㊳ 写入 context_vars opt extract / extract_list 需要跨文件共享 EXT->>CACHE: ㊴ 批量写入提取变量 CACHE->>REDIS: ㊵ Pipeline SET end EXE->>ASSERT: ㊶ 执行 code / eq / ne / contain / db 断言 opt validation.db ASSERT->>SQL: ㊷ 执行 SELECT 落库校验 SQL-->>ASSERT: ㊸ 返回查询结果 end ASSERT-->>EXE: ㊹ 返回断言结果 alt 当前 step 通过 EXE->>EXE: ㊺ 退出 retry 循环 else 断言失败或网络异常，且还有重试次数 EXE->>ALLURE: ㊺ 丢弃本次尝试附件 EXE->>EXE: ㊺ 等待 retry.interval 秒后重试当前 step end opt 重试耗尽仍失败 EXE->>EXE: ㊺ 保存主流程异常 end end opt 存在 teardown_sql EXE->>RESOLVE: ㊻ 解析 teardown_sql 中的变量 RESOLVE-->>EXE: ㊼ 返回已替换 SQL EXE->>SQL: ㊽ 执行后置 SQL SQL-->>EXE: ㊾ 返回查询结果或影响行数 end end alt 当前 step 存在主流程异常 EXE->>EXE: ㊿ break 主链路，进入 teardown_steps end end opt 存在 teardown_steps loop 每个 teardown_step EXE->>EXE: 51. 合并全局配置并执行清理步骤 EXE->>EXE: 52. teardown_steps 异常只记录日志，不掩盖主流程异常 end end end end %% 阶段三：收尾与通知 PT->>HOOK: 53. pytest_runtest_makereport alt 用例最终失败 HOOK->>REDIS: 54. 写入熔断状态 else 用例通过且之前被熔断 HOOK->>REDIS: 54. 清除熔断状态 end HOOK->>ALLURE: 55. 写入失败现场 / 崩溃日志附件 PT->>HOOK: 56. pytest_terminal_summary HOOK->>HOOK: 57. 打印终端汇总 + 生成 environment.properties PT->>HOOK: 58. pytest_sessionfinish HOOK->>NOTIFY: 59. 根据开关发送邮件 / 企微通知 HOOK->>CACHE: 60. 清理本轮 Redis 提取变量 CACHE->>REDIS: 61. 按 session_id 精确删除 extract Key HOOK->>SQL: 62. 关闭 MySQL 连接池 HOOK->>REDIS: 63. 关闭 Redis 连接池 PT-->>RUN: 64. pytest 退出并交还状态码 RUN->>ALLURE: 65. allure generate 渲染 HTML 报告 ALLURE-->>RUN: 报告生成完毕 ``` 完整执行时序图覆盖框架从触发到报告生成的完整生命周期，共 65 个关键动作，分为三个阶段： **阶段一（①~⑯）：启动、清理与用例收集** | 步骤 | 说明 | |------|------| | ①~③ | 用户或 Jenkins 执行 `python run.py`，run.py 加载环境配置并启动 pytest。 | | ④~⑤ | `pytest_sessionstart` 负责测试会话级清理，包括框架级 Redis 状态和熔断缓存预加载。 | | ⑥~⑨ | `authorize_session` 使用 FileLock 控制多进程只登录一次，并把全局 Token 写入 Redis，供 Worker 共享。 | | ⑩~⑬ | YAML 读取器解析用例；如果发现 `parameters`，会触发 CSV 笛卡尔积裂变，最终注册为多个 TestItem。 | | ⑭~⑯ | 用例收集完成后批量检查熔断状态；如果本轮存在 DB 断言、`setup_sql` 或 `teardown_sql`，提前预热 MySQL 连接池。 | **阶段二（⑰~52）：单个 TestItem 的业务执行链路** | 步骤 | 说明 | |------|------| | ⑰~⑱ | 执行前先检查熔断状态；已熔断的测试文件会直接跳过，避免无效请求。 | | ⑲~㉒ | Executor 深拷贝用例，提取 `config`，读取外层 `steps / teardown_steps`，并对当前 step 合并全局配置。 | | ㉓~㉖ | 如果配置了 `setup_sql`，先解析变量再执行 SQL；当前严格模式下 DML 影响行数为 0 会失败。 | | ㉗ | `setup_sql` 失败时，本 step 的接口请求不会开始，`teardown_sql` 也不会执行，避免提前读取尚未提取的变量。 | | ㉘~㉝ | 进入请求前变量解析阶段：`${变量}` 查 `context_vars`，`${函数()}` 调用 RuntimeFunctions，`${get_extract_data()}` 从 Redis 读取跨文件变量。 | | ㉞~㉟ | HttpClient 发起请求；如果配置了 `mock` 且命中规则，则直接返回 FakeResponse，不访问真实服务器。 | | ㊱~㊵ | 捕获本次尝试的 Allure 请求响应附件；`extract_local` 只写内存，`extract / extract_list` 同时写内存和 Redis。retry 中间失败会丢弃本次附件，最终尝试才写入报告。 | | ㊶~㊹ | 执行软断言；普通响应断言和 DB 断言统一汇总，DB 断言只允许 SELECT 查询。 | | ㊺ | 若网络异常或断言失败且仍有 retry 次数，先丢弃本次 Allure 附件，再等待 `retry.interval` 后重试当前 step；`retry.times` 是最大执行次数，包含首次执行。 | | ㊻~㊾ | `setup_sql` 成功后才会执行 `teardown_sql`；请求、提取或断言失败时仍会执行，用于清理当前 step 产生的数据。 | | ㊿~52 | 主链路任一步骤失败后停止后续业务步骤，进入 `teardown_steps`；teardown_steps 异常只记录日志，不掩盖主流程原始失败。 | **阶段三（53~65）：结果回写、资源清理与报告生成** | 步骤 | 说明 | |------|------| | 53~55 | pytest 生成测试报告时回写熔断状态：最终失败写入熔断，通过则清除历史熔断，并把失败现场写入 Allure 附件。 | | 56~57 | `pytest_terminal_summary` 打印终端汇总，并生成 Allure 环境信息。 | | 58~63 | `pytest_sessionfinish` 发送通知、按 session_id 精确清理本轮 Redis 提取变量，并关闭 MySQL / Redis 等连接池。 | | 64~65 | pytest 退出后，run.py 调用 Allure CLI 把 `reports/temps` 渲染成 `reports/allures` HTML 报告。 |