提交 f5003c87 authored 作者: 陈泽健's avatar 陈泽健

feat(executor): 新增 SelectorMapper 选择器增强与步骤级实时推送

核心变更:
- SelectorMapper 集成:支持 page_key + element_key 多选择器回退解析
- 步骤级实时推送:通过 WebSocket 广播每一步执行状态
- 单步重试机制:step_retry_count 配置支持
- 录制器统一使用 StepDefinition 格式,透传新字段
- 前端实时步骤时间线组件

新增文档:
- README.md / CLAUDE.md / HANDOFF.md 项目文档
- Handoff skill 自动生成交接文档
- 执行器准确率优化相关 PRD/计划文档
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 3c4cff22
@echo off @echo off
cd /d E:\GithubData\ubains-module-test\platform-auto-test cd /d C:\PycharmData\ubains-module-test\platform-auto-test
E:\nodejs\claude.cmd --permission-mode bypassPermissions C:\nvm4w\nodejs\claude.cmd --permission-mode bypassPermissions
\ No newline at end of file
---
name: Handoff
description: 生成交接文档 HANDOFF.md,供下一次会话快速恢复上下文
---
## Usage
/Handoff
## Description
在当前会话结束时执行,生成一份 **HANDOFF.md** 交接文档,内容包括:
- **项目概览**:项目是什么、技术栈、目录结构
- **当前任务**:本次会话在做什么
- **已完成事项**:已完成的里程碑和关键成果
- **卡点和问题**:当前卡在什么地方、未解决的问题
- **下一步计划**:优先级最高的待办任务
- **踩坑记录**:绝对不要重复踩的坑(含 Windows 平台特有坑)
- **启动指南**:如何快速启动项目
- **关键人员/文档索引**:重要文档和 PRD 位置
## 执行步骤
1. 读取项目关键状态:
- `git log --oneline -5`(最近提交)
- `git status --short`(未提交变更)
- `git branch --show-current`(当前分支)
- `ls Docs/PRD/自动化测试平台/`(检查文档目录)
2. 读取已有进度文档:`Docs/PRD/自动化测试平台/当前进度记录.md`
3. 根据上述信息合成 HANDOFF.md,写入项目根目录 `HANDOFF.md`
4. 告知用户 HANDOFF.md 已生成
## 输出格式
- 写入 `C:\PycharmData\ubains-module-test\platform-auto-test\HANDOFF.md`
- 使用 Markdown 格式,中文字符
- 每个章节用 `---` 分隔,便于快速扫读
\ No newline at end of file
# CLAUDE.md
> 本文件为 Claude Code 在此项目中工作时的指导文档。Claude Code 在每次会话开始时会自动读取本文件。
> 维护者:czj · 最后更新:2026-07-13
---
## 项目概述
**平台自动化测试可视化系统 (platform-auto-test)** — 一个 Web 可视化自动化测试平台,核心创新是用例录制器:用户在目标网站上操作,系统自动捕获行为并转换为测试步骤,无需手动编写测试代码。
- **被测系统**:统一管理平台 (https://192.168.5.44)
- **登录凭据**:admin@xty / Ubains@13579 · 验证码固定 `csba`
- **远程仓库**:http://git.ubainsyun.com/bing/ubains-module-test.git
- **当前分支**`platform-auto-test`(主分支:`main`
---
## 常用命令
### 启动服务
```bash
# 后端(端口 8001,工作目录 backend/)
cd backend
pip install -r requirements.txt
playwright install chromium
uvicorn app.main:app --reload --port 8001
# 前端(端口 3000,代理指向 8001,工作目录 frontend/)
cd frontend
npm install
npm run dev
```
### 构建 & 测试
```bash
# 前端构建(类型检查 + Vite 打包)
cd frontend && npm run build
# 前端 Lint
cd frontend && npm run lint
# 后端测试
cd backend && pytest tests/ -v --cov=app --cov-report=html
# WebSocket 流程集成测试
cd backend && python scripts/test_websocket_flow.py
```
### 访问地址
- 前端:http://localhost:3000
- API 文档:http://localhost:8001/docs
- 健康检查:http://localhost:8001/health
---
## 代码架构
### 后端分层(`backend/app/`)
- **`main.py`** — FastAPI 入口,注册 10 个路由模块(modules / cases / executions / recorder / stats / reports / cleanup / batch / dependencies)。**已配置 Windows ProactorEventLoop,勿删**
- **`config.py`**`Settings` 类,支持环境变量覆盖(DATABASE_URL / PLAYWRIGHT_HEADLESS / SCREENSHOT_DIR 等)。
- **`database.py`** — SQLite 异步引擎 + 表初始化。已配置 `check_same_thread=False` + `pool_pre_ping=True`
- **`models/`** — SQLAlchemy ORM:module / test_case / execution / case_result / case_dependency。
- **`schemas/`** — Pydantic 校验层。
- **`routers/`** — API 路由(薄层,仅参数校验 + 调用 service)。
- **`services/`** — 业务逻辑层(核心业务在此)。
- **`executors/playwright_executor.py`** — 执行引擎核心,**同步 API**,13 种动作 + 8 种断言。
- **`websocket/manager.py`** — WebSocket 连接管理(分组广播 + 断线重连)。
- **`utils/selector_mapper.py`** — 选择器映射,多选器回退策略。
- **`scripts/`** — 15 个辅助脚本(create_cases / test_websocket_flow / 数据清理等)。
### 前端分层(`frontend/src/`)
- **`App.vue`** — 根布局(侧边栏 220px + 顶部栏 + 内容区)。
- **`router/index.ts`** — 7 个路由,懒加载。
- **`api/`** — 各模块的 axios 封装调用。
- **`utils/request.ts`** — Axios 全局封装,**baseURL 为空字符串**,超时 5 分钟。
- **`utils/websocket.ts`** — WS 客户端,事件驱动 + 自动重连 3 次。
- **`views/`** — 7 个页面:Dashboard / Modules / Cases / Recorder / Execution / Reports / Settings。
- **`types/`** — TypeScript 类型定义。
### API 前缀约定
所有 API 前缀为 `/api/{模块}`(如 `/api/cases``/api/executions`)。前端 axios `baseURL` **必须为空字符串**,路由路径写完整 `/api/...`,否则会出现 `/api/api/cases` 双重前缀 404。
---
## 关键约束(务必遵守)
### 1. Playwright 必须用同步 API
Windows 上 Playwright 异步 API 在 FastAPI asyncio 循环中无法启动。**不要**尝试改回 `async_playwright`,也不要用 `asyncio.create_task` / `threading` / `multiprocessing` 在 Windows 上实现异步执行——都已验证失败。
正确模式(已固化):
1. `main.py` 设置 `asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())`
2. `playwright_executor.py` 使用 `sync_playwright` 同步 API
3. `execution_service.py` 通过 `loop.run_in_executor()` 在线程池中执行
4. **每个测试用例创建独立的 `PlaywrightExecutor` 实例**(共享实例在 `stop()``_is_running=False`,下次 `start()` 会失败)
### 2. 前端 useRoute() 必须在 setup 顶层
`useRoute()` 不能在 `computed()` 回调内调用,否则报 `inject() can only be used inside setup()`,导致页面主内容区域为空。统一在 `<script setup>` 顶层调用。
### 3. 前端 axios 超时设 5 分钟
默认 30s 太短,Playwright 执行可能需要几分钟。`request.ts` 已配置 `timeout: 300000`
### 4. SQLite 并发写锁
已配置 `check_same_thread=False` + `pool_pre_ping=True`。同步执行期间的并发写入仍需注意;如遇 `database is locked`,检查是否有未关闭的会话。`error_message` 字段必须 `nullable=True`(成功执行后为 None)。
### 5. 数据目录(gitignored)
- `data/test_platform.db` — SQLite 数据库
- `data/screenshots/` — 执行截图
- `data/reports/` — 生成的报告
这些目录在应用启动时由 `lifespan` 自动创建,无需手动 mkdir。
---
## 踩坑记录(绝对不要重复踩)
| # | 现象 | 根因 | 正确做法 |
|---|------|------|---------|
| 1 | `NotImplementedError` / Sync API inside asyncio | Windows 默认 SelectorEventLoop 不支持子进程 | 同步 API + ProactorEventLoop + 线程池 |
| 2 | 第二次执行 `start()` 失败 | 共享执行器 `_is_running` 状态 | 每用例新建 PlaywrightExecutor 实例 |
| 3 | 请求 `/api/api/cases` 404 | `baseURL:'/api'` 与路径叠加 | baseURL 留空,路径写完整 `/api/...` |
| 4 | `SQLite database is locked` | 同步执行并发写入 | check_same_thread=False + pool_pre_ping |
| 5 | bash 中 `start` 启动 CMD 失败 | MSYS 封装损坏 `&&` 引号 | `MSYS_NO_PATHCONV=1 cmd.exe /c start ...` |
| 6 | `claude` 命令不在 PATH | Windows 未配全局 PATH | `which claude` 找路径,补 `.cmd` 后缀 |
| 7 | 前端页面空白 | `useRoute()` 在 computed 内调用 | 移到 setup 顶层 |
| 8 | `NOT NULL constraint failed: error_message` | 字段 nullable=False 但成功时为 None | 改为 nullable=True |
详见 `HANDOFF.md` 第五节。
---
## 文档体系
PRD 与进度文档全部在 `Docs/PRD/自动化测试平台/`,遵循工作流:
```
PRD 需求文档 → (prd-plan skill) → 计划执行文档 → (prd-code skill) → 代码
```
| 文档 | 用途 |
|------|------|
| `_PRD_平台自动化测试可视化系统需求文档.md` | 主 PRD |
| `_PRD_..._计划执行.md` | 分阶段执行计划 |
| `_开发进度报告_20260709.md` | 进度 + 完整 API 清单 + Phase 5 待办 |
| `当前进度记录.md` | 26 个用例测试结果 + 关键决策 |
| `测试结果汇总报告.md` | 执行结果详细数据 |
| `HANDOFF.md` | 会话交接(由 /Handoff skill 生成) |
**修改 PRD 或新增功能时**:先写 PRD → 走 prd-plan → 走 prd-code,保持文档与代码同步。
---
## Git 工作流
- **当前分支**`platform-auto-test`,主分支:`main`
- **提交规范**:Conventional Commits(`feat:` / `fix:` / `docs:` / `test:` / `refactor:`
- **提交工具**:使用 `/GitCommit` skill(三重门控 + 分支检查 + 推送确认)
- **提交后缀**:所有 commit message 结尾加 `Co-Authored-By: Claude <noreply@anthropic.com>`
- **未在用户明确要求时不要自动 commit/push**
---
## 可用 Skills(`.claude/skills/`)
| Skill | 用途 |
|-------|------|
| `CreateCMD` | 在当前目录打开 N 个 CMD 窗口运行 Claude |
| `GitCommit` | 代码提交辅助(审查 + 生成 commit + add/commit/push) |
| `Handoff` | 会话结束生成 HANDOFF.md 交接文档 |
| `prd-plan` | PRD 需求文档 → 执行计划文档 |
| `prd-code` | 执行计划文档 → 代码 |
---
## 当前进度与待办
- **Phase 1-4** 全部完成(基础框架、核心功能、执行与可视化、分析与报告)
- **Phase 5** 进行中(约 30%)— 已完成 Excel 导入、执行引擎调通、数据库锁修复、用例依赖管理、选择器优化、CI/CD
- **Phase 5 剩余 P0**:Settings.vue 功能实现、Pinia 状态管理、单元测试
- **远期规划**:迁移 PostgreSQL、用户权限系统、分布式执行、多浏览器录制
### 已知风险
- **SQLite 并发瓶颈** — 多用户并发场景可能 `database is locked`,远期迁移 PostgreSQL
- **Windows 子进程限制** — 当前用同步 + 线程池规避,非完美方案
- **无用户权限系统** — 当前无认证/授权,仅限内网使用
---
## 代码风格约定
- 后端 Python 文件头部注释模板:模块名称、模块描述、作者、创建日期、最后修改日期
- 前端 Vue/TS 文件头部注释:组件/文件名称、描述、@author、@date
- 命名:后端用 snake_case,前端用 camelCase,类型用 PascalCase
- API 路径统一 `/api/{模块}` 前缀
- 数据库 ID 使用 UUID(`utils/id_generator.py`
---
*本文件由 Claude Code 维护,修改项目结构或新增约束时请同步更新。*
\ No newline at end of file
# 执行器准确率优化需求文档
> **文档版本**: v1.0
> **创建日期**: 2026-07-13
> **文档状态**: 初稿待评审
> **负责人**: czj
---
## 一、背景
### 1.1 问题描述
当前 `platform-auto-test` 平台的测试用例执行准确率低于预期。对比同仓库的 `pc自动化测试`(Playwright 原生 E2E 测试,51 个工作流,通过率 100%),平台在相同目标系统上的执行成功率明显偏低。
### 1.2 根因分析
通过深度对比两套方案的实现源码,发现主要差距在以下维度:
| 维度 | pc自动化测试 | platform-auto-test | 影响程度 |
|------|-------------|-------------------|----------|
| 选择器质量 | 手写多级 CSS 链,唯一性高 | 录制器生成单层 class,易匹配多个 | ★★★★★ |
| 等待策略 | 显式 waitFor(visible) + 500-800ms 延时 | 等待超时后静默继续 | ★★★★☆ |
| 交互健壮性 | force click + 多选器回退 + triple-click 清空 | 简单 click/fill,无回退 | ★★★★☆ |
| 重试粒度 | 单步重试 + 动作特定 recovery | 整用例从第 1 步重来 | ★★★☆☆ |
| 日志反馈 | 每步控制台输出带图标 | logger.debug 级别,默认不可见 | ★★☆☆☆ |
### 1.3 关键发现
代码审查中发现两个关键问题:
1. **SelectorMapper 悬空**`backend/app/utils/selector_mapper.py` 已定义丰富的多选器回退链,但执行器从未导入使用
2. **WebSocket 回调未接通**`execution_service.py``_create_step_callback` 已实现但从未传给 `executor.execute_case()`,导致前端无法实时看到步骤执行
---
## 二、需求内容
### 2.1 目标
提升平台执行器的执行准确率,使其在同等条件下达到与 `pc自动化测试` 接近的通过率(>90%),同时保持平台的"可视化/录制器"定位。
### 2.2 功能需求
#### 2.2.1 多选择器回退机制
**现状**:每个步骤依赖单一 CSS 选择器,无备选方案。
**需求**
- 执行器支持单个步骤传入多个选择器(`selectors: List[str]`
- 主选择器失败时自动尝试下一个,全部失败才标记为 failed
- 集成 SelectorMapper 中预定义的丰富选择器链
- 兼容旧格式(仅 `params.selector` 单值)
#### 2.2.2 显式等待策略
**现状**:动作前等待过于简单,缺少延时和状态检查。
**需求**
- 每个动作前执行 `wait_for_selector(state="visible")` 显式等待
- 每个动作后根据类型增加 300-2000ms 延时(应对页面动画)
- 导航后等待 `networkidle`
- 加载遮罩两阶段等待(先等 visible 再等 hidden)
#### 2.2.3 交互健壮性增强
**现状**`_do_click` 无 force 参数,`_do_fill` 不清空输入,`_do_assert` 无重试。
**需求**
- Click 支持 `force=True` 参数,绕过覆盖层
- Fill 执行 triple-click → Ctrl+A → Delete → fill 清空模式
- 输入框填充前等待元素可见
- 断言验证支持重试(如列表数量断言等待 3 秒后重试)
#### 2.2.4 单步重试机制
**现状**:整用例从头重试,浪费时间和资源。
**需求**
- 单步执行失败后自动重试(默认 2 次)
- 重试间隔 1 秒
- 配置项 `step_retry_count` 可调
- 整用例重试保留为可选配置(默认禁用)
#### 2.2.5 实时执行反馈
**现状**:WebSocket 步骤回调已实现但未接通。
**需求**
- 接通 `_create_step_callback``executor.execute_case()`
- 每步执行完成通过 WebSocket 推送 `step_update` 事件
- 前端 Execution.vue 实时显示步骤执行情况
#### 2.2.6 统一的步骤数据结构
**现状**:Schema / Recorder / Executor 三处步骤表示不一致。
**需求**
- `StepDefinition` 增加可选字段(向后兼容):
- `selectors: List[str]` — 多选择器回退链
- `page_key: str` — SelectorMapper 页面标识
- `element_key: str` — 元素标识
- `force: bool` — 强制点击
- `wait_after: int` — 自定义操作后延时
- Recorder 统一使用 `StepDefinition` 格式
- 旧数据(仅 `params.selector`)继续有效
### 2.3 非功能需求
- 向后兼容:现有 26 个用例无需修改即可继续执行
- 执行效率:单步重试不显著增加整体执行时间
- 稳定性:Windows 11 + Playwright 同步 API + 线程池仍正常
---
## 三、Why
执行准确率是自动化测试平台的**核心指标**。当前平台虽然有完整的 Web 管理界面、录制器、报告系统,但如果执行准确率不可靠,这些功能的价值将大打折扣。`pc自动化测试` 项目已验证了这些最佳实践在同一个目标系统上的有效性,本需求旨在将这些已证明有效的策略引入平台。
## 四、How to apply
当后续修改执行器逻辑时:
1. 确保新增字段全部 Optional,不影响旧数据
2. 多选择器回退按优先级排序,首个为最优选择器
3. 等待时间以实际页面动画时间为准,不过度延长
4. 日志统一使用图标前缀(✓/⚠/✗/ℹ/⏳)
5. 修改后运行现有 26 个用例回归验证
---
## 五、验收标准
| 验收项 | 标准 |
|-------|------|
| 多选择器回退 | 主选择器失效时,自动切换到备选选择器执行成功 |
| 等待策略 | 页面动画/弹窗后,操作正常执行不报元素不可见 |
| 交互健壮性 | 覆盖层上的元素可通过 force click 点击;输入框先清空再填充 |
| 单步重试 | 某步失败后自动重试 2 次,不重新执行前置步骤 |
| 实时反馈 | 前端 WebSocket 收到每步的 step_update 事件 |
| 向后兼容 | 旧格式 26 个用例无需修改,正常执行 |
| 日志 | 后端控制台输出带图标前缀的步骤执行日志 |
\ No newline at end of file
# 执行器准确率优化 执行计划文档
> **文档版本**: v1.0
> **创建日期**: 2026-07-13
> **关联 PRD**: _PRD_需求文档_执行器准确率优化.md v1.0
> **负责人**: czj
---
## 一、执行概述
### 1.1 项目背景
`platform-auto-test` 平台的测试用例执行准确率低于预期。经对比同仓库 `pc自动化测试`(Playwright 原生 E2E 测试,51 个工作流,通过率 100%),发现差距主要在:选择器质量、等待策略、交互健壮性、重试粒度、日志反馈五个维度。本计划针对这些差距,从 `pc自动化测试` 提取已验证的最佳实践,融入平台执行器。
### 1.2 执行目标
| 目标 | 描述 | 验收标准 |
|------|------|---------|
| 多选择器回退 | 单步骤支持多个选择器,主选择器失败自动切换 | 录制时可指定 selectors 列表,执行时逐一尝试 |
| 显式等待 | 操作前等待元素可见,操作后等待页面动画完成 | 弹窗/列表加载后操作正常执行 |
| 交互健壮性 | force click 绕过覆盖层、fill 先清空再输入 | 覆盖层上元素可点击,输入框填充正确 |
| 单步重试 | 失败步骤自动重试,不重新执行前置步骤 | 某步失败后重试 2 次,前置步骤不再执行 |
| 实时反馈 | WebSocket 步骤回调接通,前端实时显示 | 执行时浏览器 console 可见 step_update 事件 |
| 向后兼容 | 现有 26 个用例无需修改 | 旧格式数据执行正常 |
### 1.3 技术选型
| 项目 | 选择 | 说明 |
|------|------|------|
| 执行器 | Playwright Sync API | 保持现有同步方案,避免 Windows 兼容问题 |
| 选择器回退 | SelectorMapper + 步骤级 selectors 列表 | 已有 SelectorMapper 直接集成 |
| 实时通信 | WebSocket(FastAPI 内置) | 已有 manager.py,接通回调即可 |
| 数据库 | SQLite | 步骤数据存 JSON 列,新字段 Optional 无 schema 变更 |
### 1.4 预计工期
| 阶段 | 内容 | 预计时间 |
|------|------|---------|
| Phase 1 | 执行器核心改进(P0) | 2-3 天 |
| Phase 2 | 数据结构统一(P0) | 1 天 |
| Phase 3 | 执行服务改进(P1) | 1 天 |
| **总计** | | **4-5 天** |
---
## 二、任务分解与实施计划
### 2.1 Phase 1: 执行器核心改进(预计 2-3 天)
#### 2.1.1 任务清单
| 任务ID | 任务名称 | 说明 | 优先级 | 预计工时 |
|--------|---------|------|--------|---------|
| P1-001 | 集成 SelectorMapper + 多选择器回退 | 导入 SelectorMapper,新增 `_resolve_selectors` 方法,改造所有 `_do_*` 动作方法 | P0 | 1 天 |
| P1-002 | 显式等待策略 | 新增 `_wait_for_element``_safe_is_visible``_post_action_wait` 方法,改造 execute_step | P0 | 0.5 天 |
| P1-003 | 强制点击 + 输入清空 | `_do_click` 支持 force 参数,`_do_fill` 执行 triple-click 清空模式 | P0 | 0.3 天 |
| P1-004 | 图标日志 | 全部 logger 改为图标前缀,execute_step 入口/出口加日志 | P0 | 0.2 天 |
| P1-005 | 加载遮罩两阶段等待 | 新增 `_wait_for_loading_overlay`,导航后和点击前调用 | P0 | 0.3 天 |
#### 2.1.2 代码变更
**文件:`backend/app/executors/playwright_executor.py`**
```python
# 新增导入
from app.utils.selector_mapper import SelectorMapper
# __init__ 新增
self._selector_mapper = SelectorMapper()
# 新增方法
def _resolve_selectors(self, params: dict) -> List[str]:
"""解析多选择器回退链"""
# 1. 优先 params.selectors(列表)
# 2. 回退 params.selector(单值)
# 3. 若有 page_key+element_key,调 mapper 取完整链
...
def _wait_for_element(self, selector, timeout=None, state="visible") -> bool:
"""安全等待元素,不抛异常"""
...
def _safe_is_visible(self, selector: str) -> bool:
"""安全检查可见性,对应 .isVisible().catch(() => false)"""
...
def _post_action_wait(self, action_type: str):
"""按动作类型等待"""
...
def _wait_for_loading_overlay(self, timeout=10000):
"""两阶段加载遮罩等待"""
...
# 改造 _do_click — 多选择器 + force 参数
def _do_click(self, params: dict) -> None:
selectors = self._resolve_selectors(params)
force = params.get("force", False)
for sel in selectors:
try:
self._wait_for_element(sel, timeout=5000)
self._page.click(sel, timeout=5000, force=force)
return
except Exception as e:
logger.debug(f" ⚠ 点击失败 {sel}: {e}")
continue
# iframe fallback 保留
...
# 改造 _do_fill — triple-click 清空模式
def _do_fill(self, params: dict) -> None:
selectors = self._resolve_selectors(params)
value = params.get("value", "")
for sel in selectors:
try:
self._wait_for_element(sel, timeout=5000)
self._page.click(sel, click_count=3, force=True, timeout=5000)
self._page.keyboard.press("Control+A")
self._page.keyboard.press("Delete")
self._page.wait_for_timeout(100)
self._page.fill(sel, str(value))
logger.info(f" ✓ 填充: {sel} = {value}")
return
except Exception as e:
logger.debug(f" ⚠ 填充失败 {sel}: {e}")
continue
raise Exception(f"无法填充元素: {selectors}")
# 图标日志改造示例
# _do_click 成功: logger.info(f" ✓ 点击: {selector}")
# _do_navigate 成功: logger.info(f" ✓ 导航到: {url}")
# _do_assert 成功: logger.info(f" ✓ 断言通过: {assert_type}")
# execute_step 入口: logger.info(f" ⏳ 步骤 {order}/{total}: {name}")
# execute_step 成功: logger.info(f" ✓ 步骤 {order} 完成 ({duration:.2f}s)")
# execute_step 失败: logger.warning(f" ✗ 步骤 {order} 失败: {error}")
```
#### 2.1.3 验证方法
1. 执行一个旧格式用例(仅 `params.selector`),确认走通
2. 手动构造一个带 `selectors` 列表的用例,主选择器故意写错,检查备选是否生效
3. 检查控制台输出是否带 `✓`/`⏳`/`✗` 图标
4. 检查 force click 参数是否透传成功
---
### 2.2 Phase 2: 数据结构统一(预计 1 天)
#### 2.2.1 任务清单
| 任务ID | 任务名称 | 说明 | 优先级 | 预计工时 |
|--------|---------|------|--------|---------|
| P2-001 | 扩展 StepDefinition Schema | 增加 selectors/page_key/element_key/force/wait_after 可选字段 | P0 | 0.3 天 |
| P2-002 | Recorder 统一格式 | RecordedStep 复用 StepDefinition,add_step/add_assertion 存 selectors 列表 | P0 | 0.3 天 |
| P2-003 | Executor 读取新字段 | execute_step 读取 selectors/page_key/element_key 等新字段 | P0 | 0.2 天 |
| P2-004 | 前端类型同步 | StepDefinition 接口添加对应字段 | P0 | 0.2 天 |
#### 2.2.2 代码变更
**文件:`backend/app/schemas/test_case.py`**
```python
class StepDefinition(BaseModel):
order: int = Field(..., ge=1)
name: str = Field(..., min_length=1, max_length=200)
action: str = Field(...)
params: Dict[str, Any] = Field(default_factory=dict)
expected: str = Field(default="")
# 新增字段(全部 Optional,向后兼容)
selectors: Optional[List[str]] = Field(None, description="选择器列表(带回退)")
page_key: Optional[str] = Field(None, description="SelectorMapper 页面标识")
element_key: Optional[str] = Field(None, description="元素标识")
force: Optional[bool] = Field(None, description="是否强制点击")
wait_after: Optional[int] = Field(None, description="操作后等待时间(ms)")
```
**文件:`backend/app/routers/recorder.py`**
- `RecordedStep` 改为指向 `StepDefinition`(或直接替换)
- `add_step` 接收 `selectors` 列表替代顶层 `selector`/`value`
- `add_assertion` 同步使用 `selectors` 列表
- `save_as_case` 删除手动转换逻辑
**文件:`frontend/src/types/case.ts`**
```typescript
export interface StepDefinition {
order: number
name: string
action: string
params: Record<string, any>
expected?: string
selectors?: string[]
pageKey?: string
elementKey?: string
force?: boolean
waitAfter?: number
}
```
#### 2.2.3 验证方法
1. 录制器创建用例 → 查 DB 的 steps JSON 含 `selectors` 列表
2. 旧格式用例(只有 `params.selector`)执行正常
3. 新用例(含 `selectors` 列表)执行时多选器回退生效
---
### 2.3 Phase 3: 执行服务改进(预计 1 天)
#### 2.3.1 任务清单
| 任务ID | 任务名称 | 说明 | 优先级 | 预计工时 |
|--------|---------|------|--------|---------|
| P3-001 | 单步重试 | execute_step 内层循环失败后重试,默认 2 次 | P1 | 0.3 天 |
| P3-002 | 接通 WebSocket 回调 | 将 _create_step_callback 传给 executor.execute_case | P1 | 0.5 天 |
| P3-003 | 前端实时时间线 | Execution.vue 增加 liveSteps 时间线展示 | P1 | 0.5 天 |
#### 2.3.2 代码变更
**文件:`backend/app/executors/playwright_executor.py`**
```python
# __init__ 新增
self.step_retry_count: int = self.config.get("step_retry_count", 2)
# execute_step 改造 — 单步重试
def execute_step(self, step, callback=None):
...
for attempt in range(self.step_retry_count + 1):
try:
# 执行动作
...
step_result.status = "passed"
break
except Exception as e:
if attempt < self.step_retry_count:
logger.warning(f" ⚠ 步骤 {step_result.order} 第{attempt+1}次重试: {e}")
step_result.log += f"\n⚠ 第{attempt+1}次重试..."
self._page.wait_for_timeout(1000)
else:
raise
...
```
**文件:`backend/app/services/execution_service.py`**
```python
# run_all_cases_sync 改造 — 接收 loop + broadcast_fn,接通回调
def run_all_cases_sync(cases_list, exec_config, execution_id, loop, broadcast_fn):
executor = PlaywrightExecutor(config=exec_config)
executor.start()
results = []
try:
for i, case_dict in enumerate(cases_list):
case_id = case_dict.get("id")
def step_callback(step_result, cid=case_id):
try:
asyncio.run_coroutine_threadsafe(
broadcast_fn(execution_id, {
"type": "step_update",
"data": {
"execution_id": execution_id,
"case_id": cid,
"step_order": step_result.order,
"step_name": step_result.name,
"action": step_result.action,
"status": step_result.status,
"duration": step_result.duration,
"screenshot": step_result.screenshot,
"log": step_result.log,
"error": step_result.error,
},
}),
loop
)
except Exception as e:
logger.error(f"步骤回调失败: {e}")
result = executor.execute_case(case=case_dict, callback=step_callback)
results.append(result)
finally:
executor.stop()
return results
# 调用处
all_results = await loop.run_in_executor(
None, run_all_cases_sync,
[c["case_dict"] for c in cases_to_run],
config, execution_id, loop, manager.broadcast,
)
```
#### 2.3.3 验证方法
1. 配置 `step_retry_count=2`,让步骤 3 故意失败,检查步骤 3 重试 2 次后标记 failed,步骤 1-2 不重新执行
2. 浏览器 WebSocket 连接后执行,console 检查 `step_update` 事件
3. 前端 Execution.vue 执行时时间线逐步更新
---
## 三、代码文件规划
### 3.1 修改文件清单
```
backend/
├── app/
│ ├── executors/
│ │ └── playwright_executor.py # 核心:SelectorMapper 集成、多选器回退、等待策略、force、清空、图标日志、单步重试、加载遮罩
│ ├── utils/
│ │ └── selector_mapper.py # 新增 resolve_selectors 方法(现有选择器定义不动)
│ ├── schemas/
│ │ └── test_case.py # StepDefinition 扩展(全部 Optional)
│ ├── routers/
│ │ └── recorder.py # RecordedStep 统一为 StepDefinition
│ └── services/
│ └── execution_service.py # WebSocket 回调接通、step_retry_count 透传
frontend/
├── src/
│ ├── types/
│ │ └── case.ts # 类型同步
│ └── views/
│ └── Execution.vue # 实时步骤时间线
```
### 3.2 依赖关系
```
Phase 1 (执行器) ← 独立,无外部依赖
Phase 2 (数据结构) ← 独立,但需与 Phase 1 配合可获得完整效果
Phase 3 (执行服务) ← 依赖 Phase 1(执行器改造完毕)/ Phase 2(新字段读取)
└→ 前端 Execution.vue ← 依赖 Phase 3 WebSocket 接通
```
建议按 Phase 1 → Phase 2 → Phase 3 顺序实施。
---
## 四、验证方案
### 4.1 单元测试
| 测试项 | 方法 | 预期 |
|-------|------|------|
| 多选择器回退 | 构造带 3 个选择器的 params,第 1 个无效 | 第 2 个或第 3 个生效,步骤 passed |
| 向后兼容 | 构造仅 params.selector 的 params | 执行正常,不抛 KeyError |
| 单步重试 | 设 step_retry_count=2,步骤抛异常 | 重试 2 次后 failed |
| force click | 传 force=True | click 调用的 force 参数为 True |
### 4.2 集成测试
| 测试项 | 方法 | 预期 |
|-------|------|------|
| 录制→执行闭环 | 录制器创建含 selectors 的用例 → 执行 | 步骤正确执行,DB 存储完整 |
| 旧数据回归 | 执行现有 26 个用例 | 通过率不低于改动前 |
| 实时反馈 | 浏览器 WebSocket 连接 → 执行 | step_update 事件实时推送 |
| Windows 兼容 | Windows 11 全流程执行 | Playwright 启动正常,执行完成 |
### 4.3 回滚方案
如果改动后出现异常:
1. 数据结构改动(Optional 字段)不影响旧数据,无需回滚
2. 执行器逻辑可恢复为 git checkout 旧版本
3. 数据库无迁移操作,不需回滚数据
---
## 五、参考文档
| 文档 | 路径 | 说明 |
|------|------|------|
| PRD 需求文档 | `Docs/PRD/自动化测试平台/_PRD_需求文档_执行器准确率优化.md` | 本计划对应需求 |
| 方案对比分析 | `Docs/自动化测试方案对比.md` | 两套方案多维度对比 |
| 执行准确率差异分析 | `Docs/执行准确率差异分析.md` | 根因分析和改进建议 |
| pc自动化测试设计文档 | `pc自动化测试/pc自动化测试/BookV3_E2E_设计文档.md` | 参考项目设计 |
| pc自动化测试 Page Object | `pc自动化测试/pc自动化测试/e2e-bookv3/pages/*.js` | 选择器策略参考 |
| SelectorMapper | `backend/app/utils/selector_mapper.py` | 待集成组件 |
\ No newline at end of file
# 执行准确率差异分析:`pc自动化测试` vs `platform-auto-test`
## 核心结论
**`pc自动化测试` 准确率远高于 `platform-auto-test`,根本原因不是框架本身的差异,而是"选择器质量"和"等待策略"两个维度的差距。**
---
## 差异一:选择器的精确度(最关键差异)
### `pc自动化测试` — 精确到具体组件的 CSS class
Page Object 中使用**从 Vue 组件源码直接确定的、多级嵌套的具体 class**
```javascript
// Step1Page.js — 选择器非常具体
get meetingNameInput() {
return this.page.locator('.bookv3-slim-form .field-name input[placeholder="请输入"]').first();
}
get roomInput() {
return this.page.locator('.bookv3-slim-form .field-room .room-input');
}
get nextButton() {
return this.page.locator('.footer-next-btn-v3');
}
```
**特点**
- 选择器由**人写代码时直接从组件 DOM 结构确定**
- 多层嵌套(form > field > input),唯一性极高
- 使用 `placeholder``class` 等稳定属性组合定位
### `platform-auto-test` — 通用选择器,录制器自动生成
```python
# playwright_executor.py — 自动登录时的选择器
inputs = self._page.query_selector_all('input:visible')
for inp in inputs:
ph = inp.get_attribute('placeholder') or ''
if '手机' in ph or '用户' in ph or '邮箱' in ph or 'account' in ph.lower():
inp.fill(self.LOGIN_CONFIG["username"])
```
**特点**
- 选择器存在数据库的步骤 params 中,由**录制器自动生成**
- 录制器生成的选择器通常是单层 class(如 `.el-input``button`
- 自动生成的选择器缺乏上下文,容易匹配到多个元素
- 录制器捕获时的 DOM 结构与执行时的 DOM 结构可能有差异
### 对比
| 维度 | pc自动化测试 | platform-auto-test |
|------|-------------|-------------------|
| 选择器来源 | 开发人员手写 | 录制器自动生成 |
| 选择器粒度 | 多级嵌套,精确到具体字段 | 单层/通用 |
| 唯一性 | 高 — 一个选择器对应一个元素 | 低 — 易匹配到多个 |
| 稳定性 | 高 — 针对组件结构设计 | 中 — 依赖录制时 DOM |
| 抗页面变化 | 中 — class 变化会断 | 低 — 通用选择器更容易误点 |
---
## 差异二:等待策略
### `pc自动化测试` — 显式等待 + 固定延时 + 容错
```javascript
// 1. 明确等待元素可见
await this.meetingNameInput.waitFor({ state: 'visible' });
// 2. 固定延时确保弹窗动画完成
await this.page.waitForTimeout(800);
// 3. 多重容错:多种方式尝试
if (await emptyBtn.first().isVisible().catch(() => false)) {
// 方式A
} else if (await addSquare.first().isVisible().catch(() => false)) {
// 方式B
}
```
### `platform-auto-test` — 简单等待 + 静默失败
```python
def _do_click(self, params: dict) -> None:
selector = params.get("selector", "")
# 等待10秒,失败则直接忽略
try:
self._page.wait_for_selector(selector, state="visible", timeout=10000)
except Exception:
pass # ← 静默忽略,继续尝试点击
# 尝试点击,失败则静默忽略
try:
self._page.click(selector, timeout=5000)
return
except Exception:
pass # ← 静默忽略
# iframe fallback
for frame in self._page.frames:
...
```
**问题**:静默太多,步骤报告显示 "passed" 但实际上可能根本没点到正确元素。
### 对比
| 等待策略 | pc自动化测试 | platform-auto-test |
|---------|-------------|-------------------|
| 元素可见等待 | ✅ 明确 waitFor({state:'visible'}) | ⚠️ 等待超时后静默继续 |
| 弹窗/动画等待 | ✅ 固定延时 500-800ms | ❌ 无 |
| 操作间间隔 | ✅ 每步间有 waitForTimeout | ❌ 连续执行 |
| 失败反馈 | ❌ throw error(测试直接 fail) | ⚠️ 静默 pass,但操作无效 |
---
## 差异三:操作交互的健壮性
### `pc自动化测试` 的做法
```javascript
// 清空输入框:先 triple-click 全选 → Ctrl+A → Delete → fill
await this.meetingNameInput.click({ clickCount: 3, force: true });
await this.page.keyboard.press('Control+A');
await this.page.keyboard.press('Delete');
await this.meetingNameInput.fill(name);
// 点击被覆盖的元素:force: true 绕过覆盖层
await moreBtn.click({ force: true });
// 多选器回退:一个选择器不行就换一个
var confirmBtn = dialogContainer.locator('button').filter({ hasText: /^确定$/ }).first();
if (!btnVisible) {
confirmBtn = dialogContainer.locator('.el-dialog__footer button, .el-message-box__btns button')
.filter({ hasText: '确定' }).first();
}
```
### `platform-auto-test` 的做法
```python
# 简单的 fill
self._page.fill(selector, str(value))
# 简单的 click
self._page.click(selector, timeout=5000)
```
### 关键差别
| 操作 | pc自动化测试 | platform-auto-test |
|------|-------------|-------------------|
| 输入框清空 | triple-click + Ctrl+A + Delete + fill | 直接 fill(不清空已有内容) |
| 覆盖层点击 | `force: true` 绕过 | 无 force,被覆盖时失败 |
| 多选器回退 | 2-3 种选择器依次尝试 | 无回退 |
| 按钮匹配 | `filter({hasText})` 文本匹配 | 无 |
---
## 差异四:浏览器配置
```javascript
// pc自动化测试 — 使用系统已安装的 Chrome
channel: 'chrome',
launchOptions: {
args: [
'--start-maximized',
'--window-size=1920,1080',
'--disable-infobars',
],
},
```
```python
# platform-auto-test — 使用 Playwright 内置 Chromium
self._browser = self._playwright.chromium.launch(
headless=self.headless,
args=["--no-sandbox", "--disable-setuid-sandbox"],
)
```
**差异**:系统 Chrome 更接近用户真实环境,内置 Chromium 的行为可能与真实环境有偏差。`--no-sandbox` 模式在某些场景下会导致 CSS 渲染差异。
---
## 差异五:日志和调试能力
```javascript
// pc自动化测试 — 每步都有详细日志
console.log(` ✓ 已填写会议名称: ${name}`);
console.log(` ✓ 已打开会议室选择弹窗`);
console.log(` ✓ 已点击"下一步"`);
```
```python
# platform-auto-test — 缺乏细粒度执行日志
logger.debug(f"填充输入框: {params.get('selector', '')} = {params.get('value', '')}")
```
**问题**`platform-auto-test` 的日志级别是 `debug`,在默认配置下**看不到**。即使步骤失败了,也只能看到 "failed" 状态,不知道具体卡在哪一步的哪个选择器。
---
## 差异六:测试用例的设计粒度
| 维度 | pc自动化测试 | platform-auto-test |
|------|-------------|-------------------|
| 用例粒度 | **业务场景级**(51 个完整工作流) | **页面访问级**(26 个基本验证) |
| 单用例步骤数 | 10-30+ 步(全流程) | 3 步(导航 + 等待 + 截图) |
| 复杂交互 | 多步表单填写、弹窗交互、下拉菜单 | 基本页面访问 |
| 数据清理 | 自动取消会议,不留脏数据 | 无 |
`pc自动化测试` 的用例更"深"——每个用例经过多次"创建→验证→编辑→验证"的循环,每一步都验证了上一步确实生效。而 `platform-auto-test` 的用例更"浅"——大多是"访问页面 → 等待 → 截图",即使有操作也是单步的,无法形成验证闭环。
---
## 根因总结
| 根因 | 影响程度 | 说明 |
|------|---------|------|
| **选择器质量** | ★★★★★ | pc 的手写选择器 vs 平台的录制自动生成,这是最大差距 |
| **等待策略** | ★★★★☆ | pc 有显式等待 + 固定延时 + 容错,平台过于简陋 |
| **交互健壮性** | ★★★★☆ | pc 有 force click、多选器回退、文本匹配,平台没有 |
| **用例深度** | ★★★☆☆ | pc 的用例步骤多、形成验证闭环,平台的用例太浅 |
| **浏览器配置** | ★★☆☆☆ | 系统 Chrome vs 内置 Chromium,影响较小 |
| **日志调试** | ★★☆☆☆ | pc 能看到每步详细输出,平台日志不直观 |
---
## 改进建议
### 短期(可快速落地)
1. **选择器优化**:改进录制器,生成带上下文的多级选择器(如 `.bookv3-slim-form .field-name input`),而非单层 class
2. **增加等待延时**:每个关键操作后加 300-800ms 等待,给页面动画留时间
3. **添加 force click**:对 click 操作增加 `force=True` 参数,绕过覆盖层
4. **优化日志**:将关键操作日志从 debug 改为 info,让用户在界面上能看到每步执行情况
### 中期
5. **引入多选器回退机制**:每个步骤存 2-3 个备选选择器,主选择器失败时自动回退
6. **改进录制器**:录制时不仅捕获 class,还捕获 `placeholder``aria-label``text` 等稳定属性
7. **丰富用例深度**:将现有的 26 个浅用例扩展为包含"操作+验证"的完整业务流程
### 长期
8. **引入视觉定位**:使用 AI/Screenshot 比对来定位元素,而非完全依赖 DOM 选择器
9. **自动愈合 (Self-Healing)**:当选择器失效时,自动分析 DOM 差异并修复选择器
---
## 一句话总结
**不是平台本身不准,而是"手写测试代码"天然比"录制器自动生成"更精准**——人写代码时知道要点击哪个具体的按钮,而录制器只能记录"点击了第 3 个 `.el-button`",后者的容错空间小得多。改进录制器的选择器生成策略,是拉平差距的最有效手段。
\ No newline at end of file
# 自动化测试方案对比文档
## 概述
当前项目中有**两套自动化测试方案**,它们的设计理念、技术栈和适用场景完全不同。本文档从多个维度进行对比分析,帮助理解各自的定位和优劣势。
---
## 方案一:`pc自动化测试` — Playwright 原生 E2E 测试
### 技术栈
| 项目 | 选择 |
|------|------|
| 测试框架 | Playwright Test Runner (v1.61) |
| 语言 | JavaScript (Node.js) |
| 设计模式 | Page Object Model (POM) |
| 运行方式 | CLI + HTML 测试面板 |
| 报告 | Playwright HTML Reporter |
| 被测系统 | 会议管理系统 BookV3 (真实业务系统) |
### 架构
```
用户操作
├─ CLI 方式: node run-tests.js → 交互菜单 → npx playwright test
└─ 面板方式: node e2e-panel/server.js → 浏览器面板 → 勾选用例 → 执行
spawn playwright test
Chrome 浏览器 (有头模式)
真实前端 (localhost:8080)
真实后端 (139.159.185.67:8999)
```
### 核心设计
1. **Page Object 模式**:将页面元素和操作封装为独立类
- `Step1Page.js``Step2Page.js``MeetingListPage.js` 等 10+ 个 Page Object
- 每个 Page Object 封装 getter(元素定位)和 method(操作步骤)
2. **工作流驱动**:每个测试用例是一个完整业务场景
- 工作流 1:新建会议 → 修改会议名称 → 取消会议
- 工作流 8:新建(内容+会议室+参会人+抄送人+领导) → 编辑 → 取消
- 共 51 个工作流,覆盖 18+ 模块
3. **测试数据策略**:时间戳后缀避免重复
```javascript
meetingName: '恭的自动测试' + (Date.now() % 100000)
```
4. **测试面板**:轻量 HTTP 服务器 + SSE 实时推送
- 零依赖(纯 Node.js http 模块)
- 动态从 `workflows.spec.js` 提取用例列表
- SSE 流式推送测试执行输出
### 覆盖范围
| 模块 | 用例数 |
|------|--------|
| 预订流程 | 15 |
| 首页 | 2 |
| 我的会议 | 2 |
| 会议列表 | 2 |
| 会议模板 | 3 |
| 群组管理 | 3 |
| 会议看板 | 2 |
| 关注中心 | 2 |
| 会议计划 | 3 |
| 个人设置 | 1 |
| 系统设置 | 1 |
| 任务 | 5 |
| 会议材料 | 5 |
| 统计报表 | 5 |
| 审批 | 3 |
| 参会确认 | 1 |
| **合计** | **51** |
---
## 方案二:`platform-auto-test` — 可视化自动化测试平台
### 技术栈
| 项目 | 选择 |
|------|------|
| 后端框架 | FastAPI (Python) |
| 前端框架 | Vue 3 + Element Plus + Vite 5 + TypeScript |
| 数据库 | SQLite (SQLAlchemy ORM) |
| 自动化引擎 | Playwright (同步 API,线程池执行) |
| 实时通信 | WebSocket |
| 图表 | ECharts |
| 报告 | HTML 报告生成 |
### 架构
```
用户浏览器
├─ 前端 (Vue 3, localhost:3000)
│ ├─ Dashboard.vue — 统计看板 (ECharts)
│ ├─ Cases.vue — 用例管理
│ ├─ Modules.vue — 模块管理
│ ├─ Execution.vue — 执行管理
│ ├─ Recorder.vue — 用例录制器 ⭐
│ ├─ Reports.vue — 报告查看
│ └─ Settings.vue — 设置 (占位)
└─ 后端 API (FastAPI, localhost:8001)
├─ routers/ — RESTful API 路由
├─ services/ — 业务逻辑层
├─ models/ — 数据模型
├─ executors/ — Playwright 执行引擎
├─ websocket/ — 实时推送
└─ scripts/ — 辅助脚本
Playwright (Chromium, 线程池)
被测系统 (192.168.5.44)
```
### 核心设计
1. **用例录制器 (Recorder)** — 核心创新点
- 用户在目标网站上操作,系统自动捕获行为并转换为测试步骤
- 无需手动编写测试代码
- 录制器会话管理:`RecorderSession` 类管理录制状态
2. **Web 可视化用例管理**
- 用例 CRUD:增删改查
- 模块管理:按模块分组组织用例
- 依赖管理:用例间依赖关系,拓扑排序执行
- 批量操作:批量导入/导出/执行
3. **后台执行引擎**
- Playwright 同步 API + 线程池 (`run_in_executor`)
- 每个用例独立执行器实例
- WebSocket 实时推送执行状态
4. **报告系统**
- HTML 报告生成
- 预览/导出/下载
- ECharts 统计图表
### 覆盖范围
| 功能 | 状态 |
|------|------|
| 用例管理 (CRUD) | ✅ |
| 模块管理 | ✅ |
| 用例录制器 | ✅ |
| 手动执行 | ✅ |
| 批量执行 | ✅ |
| 依赖管理 | ✅ |
| 实时推送 (WebSocket) | ✅ |
| 报告生成 | ✅ |
| 统计看板 (ECharts) | ✅ |
| 数据清理 | ✅ |
| 配置页面 | ⬜ 占位 |
| Pinia 状态管理 | ⬜ 未实现 |
| 单元测试 | ⬜ 未实现 |
---
## 多维度对比
### 1. 设计理念
| 维度 | pc自动化测试 | platform-auto-test |
|------|-------------|-------------------|
| **核心理念** | 用代码写测试,自动化执行 | 可视化操作,降低测试编写门槛 |
| **目标用户** | 开发人员/测试开发 | 测试人员/业务人员 |
| **测试用例来源** | 手写 JavaScript 代码 | 录制器自动生成 + 手动录入 |
| **组织方式** | 工作流编号 + 模块分组 | 模块树 + 用例列表 + 依赖关系 |
### 2. 技术架构
| 维度 | pc自动化测试 | platform-auto-test |
|------|-------------|-------------------|
| **代码量** | 约 15 个源文件 | 40+ 个 Python/TS 源文件 |
| **系统复杂度** | 低 (纯测试框架) | 高 (Web 平台) |
| **部署方式** | `npm install` + `npx playwright test` | 后端 + 前端双服务启动 |
| **依赖** | Node.js + Playwright | Python + Node.js + SQLite |
| **可扩展性** | 低 (加测试用例) | 高 (加功能模块) |
### 3. 测试能力
| 维度 | pc自动化测试 | platform-auto-test |
|------|-------------|-------------------|
| **用例数量** | 51 个 | 26 个 |
| **覆盖模块** | 18+ | 6+ |
| **业务场景深度** | 深 (完整业务流程) | 中 (页面访问验证为主) |
| **执行方式** | 串行 (1 worker) | 线程池并行 |
| **执行环境** | 系统 Chrome | Playwright 内置 Chromium |
| **超时配置** | 600s/用例 | 300s/步骤 |
### 4. 用户体验
| 维度 | pc自动化测试 | platform-auto-test |
|------|-------------|-------------------|
| **上手难度** | 需要会写 Playwright | 浏览器操作,零代码 |
| **调试体验** | VS Code 插件 + 有头浏览器 | Web 界面 + 日志 |
| **实时反馈** | 终端输出 / SSE 面板 | WebSocket 实时推送 |
| **报告展示** | Playwright HTML 报告 | 自定义 HTML + ECharts |
| **IDE 集成** | VS Code Playwright 插件 | 无 (Web 平台) |
### 5. 维护性
| 维度 | pc自动化测试 | platform-auto-test |
|------|-------------|-------------------|
| **用例维护** | 改代码 + 改 Page Object | Web 界面直接编辑 |
| **元素定位变更** | 需更新 Page Object 文件 | 重新录制或编辑步骤 |
| **测试数据** | 硬编码在 fixtures 中 | 数据库存储 |
| **版本控制** | 代码文件直接 Git 管理 | 数据在 SQLite 中 |
| **CI/CD 集成** | GitHub Actions (配置好) | GitHub Actions (已配置) |
### 6. 局限性
| 维度 | pc自动化测试 | platform-auto-test |
|------|-------------|-------------------|
| **主要瓶颈** | 用例量大时维护成本高 | 录制器生成步骤的健壮性 |
| **Windows 兼容** | 良好 (Playwright 原生) | 有坑 (需同步 API + 线程池) |
| **并发能力** | 单线程串行 | 线程池,但 SQLite 有写锁 |
| **认证授权** | 无 | 无 |
| **数据库** | 无 | SQLite (不适用多用户) |
---
## 场景推荐
### 适用 `pc自动化测试` 的场景
1. **开发阶段快速验证**:开发新功能后,快速运行 E2E 测试确认不破坏现有功能
2. **回归测试**:51 个工作流覆盖 18+ 模块,适合全量回归
3. **CI/CD 流水线**:轻量级,`npm install && npx playwright test` 即可集成
4. **深度业务场景**:需要测试完整的业务流程(新建 → 编辑 → 取消)
5. **调试定位**:有头模式 + 截图/视频,方便定位问题
### 适用 `platform-auto-test` 的场景
1. **非技术人员编写测试**:业务人员/测试人员通过录制器创建测试用例
2. **测试用例管理**:Web 界面管理用例、模块、依赖关系
3. **测试报告可视化**:ECharts 统计看板、HTML 报告导出
4. **批量执行与调度**:通过 Web 界面批量选择用例执行
5. **团队协作**:多人共享测试用例数据库
---
## 互补性分析
两套方案并非替代关系,而是**互补关系**
| 场景 | 推荐方案 | 原因 |
|------|---------|------|
| 开发期快速回归 | pc自动化测试 | 轻量、快速、直接 |
| 业务人员编写测试 | platform-auto-test | 录制器零代码 |
| 深度 E2E 场景 | pc自动化测试 | 51 个用例覆盖更全 |
| 测试报告展示 | platform-auto-test | ECharts 看板更直观 |
| CI/CD 集成 | pc自动化测试 | 更轻量、启动快 |
| 团队协作管理 | platform-auto-test | Web 平台天然适合 |
建议的策略是:
- **保留两套方案**,各自发挥优势
- `pc自动化测试` 作为开发期回归测试工具
- `platform-auto-test` 作为测试管理平台 + 面向非技术用户的测试入口
- 未来可考虑将 `pc自动化测试` 中的 51 个工作流逐步导入到 `platform-auto-test` 平台中统一管理
---
## 建设成本对比
| 项目 | pc自动化测试 | platform-auto-test |
|------|-------------|-------------------|
| 初始建设 | 低 (Playwright 原生) | 高 (全栈 Web 平台) |
| 新增用例成本 | 中 (写代码) | 低 (录制/表单) |
| 维护成本 | 中高 (元素变更需改代码) | 中 (需维护平台本身) |
| 学习成本 | 高 (需 Playwright 技能) | 低 (浏览器操作) |
| 运行环境要求 | 低 (仅 Node.js) | 高 (Python + Node.js 双环境) |
| 部署复杂度 | 低 | 中 |
---
*本文档由 Claude Code 自动生成,基于对 `pc自动化测试/` 和 `platform-auto-test/` 两个项目的源码分析。*
\ No newline at end of file
# HANDOFF — 会话交接文档
> **生成时间**: 2026-07-13
> **当前分支**: `platform-auto-test`
> **最近提交**: `3c4cff2` — feat(test): 新增测试用例依赖管理和批量操作功能
---
## 一、项目概览
**项目名称**: 平台自动化测试可视化系统 (platform-auto-test)
**一句话描述**: 一个 Web 可视化自动化测试平台,核心创新点在于**用例录制器**功能——用户在目标网站上操作,系统自动捕获行为并转换为测试步骤,无需手动编写测试代码。
### 技术栈
| 层级 | 技术 | 版本 |
|------|------|------|
| 后端 | FastAPI + SQLAlchemy + SQLite | Python 3.x |
| 前端 | Vue 3 + Element Plus + Vite 5 + TypeScript | Node 18+ |
| 浏览器自动化 | Playwright (Chromium) | 1.40.x |
| 实时通信 | WebSocket (FastAPI 内置) | — |
| 图表库 | ECharts | 5.5.x |
### 目录结构(关键路径)
```
platform-auto-test/
├── backend/app/
│ ├── main.py # FastAPI 入口
│ ├── executors/playwright_executor.py # 执行引擎核心
│ ├── services/ # 业务逻辑层
│ ├── routers/ # API 路由层
│ ├── models/ # 数据模型(含新增的依赖模型)
│ └── scripts/ # 辅助脚本(15个)
├── frontend/src/
│ ├── views/ # 7 个页面组件
│ └── api/ # API 调用层
├── Docs/PRD/自动化测试平台/ # 所有需求/计划/进度文档
│ ├── _PRD_平台自动化测试可视化系统需求文档.md
│ ├── _PRD_平台自动化测试可视化系统需求文档_计划执行.md
│ ├── 当前进度记录.md
│ ├── 测试结果汇总报告.md
│ └── ... (其他 PRD 文档)
└── .claude/skills/ # Claude Code 技能
├── CreateCMD/ # 打开 CMD 窗口
├── GitCommit/ # 代码提交辅助
├── Handoff/ # 生成交接文档(本技能)
├── prd-plan/ # PRD → 执行计划
└── prd-code/ # 执行计划 → 代码
```
---
## 二、当前任务
本次会话主要做了以下工作:
1. **修复了 `CreateCMD` skill 的路径配置**:将 bat 文件中的 `E:\nodejs\claude.cmd` 更新为 `C:\nvm4w\nodejs\claude.cmd``cd` 目录从 `E:\GithubData\ubains-module-test\platform-auto-test` 更新为 `C:\PycharmData\ubains-module-test\platform-auto-test`
2. **新增了 `Handoff` skill**:用于在会话结束时自动生成 HANDOFF.md 交接文档。
3. **生成了本 HANDOFF.md**
---
## 三、已完成事项
| 里程碑 | 状态 | 说明 |
|--------|------|------|
| Phase 1-4 全部完成 | ✅ | 基础框架、核心功能、执行与可视化、分析与报告 |
| Phase 5 约 30% | 🔄 | 已完成为 Excel 导入、执行引擎调通、数据库锁修复 |
| 26 个测试用例全部通过 | ✅ | 含 7 高优先级 + 12 中优先级 + 7 低优先级 |
| 深层交互测试(5 个) | ✅ | 表单填写、按钮点击、断言验证 |
| WebSocket 实时推送 | ✅ | 毫秒级延迟,6 种消息类型 |
| 报告生成和导出 | ✅ | HTML 报告,支持预览/导出/下载 |
| 数据清理功能 | ✅ | 截图清理、孤儿清理、数据库优化 |
| 选择器优化 | ✅ | 多选器回退策略,19 个用例更新 |
| CI/CD 集成 | ✅ | GitHub Actions 工作流 |
| 测试用例依赖管理 | ✅ | 新增依赖模型、批量操作、拓扑排序 |
| GitCommit skill 完善 | ✅ | 三重门控、分支检查、推送确认 |
---
## 四、卡点和问题
### 当前无严重阻塞项
Phase 5 剩余任务尚未启动,但无技术阻断:
| 待办 | 优先级 | 说明 |
|------|--------|------|
| 系统配置页面 (Settings.vue) | P0 | 占位页面,需实现功能 |
| Pinia 状态管理 | P0 | 需创建 executionStore / settingsStore |
| 单元测试 | P0 | 后端 pytest + 前端 Vitest |
| 数据库迁移 (Alembic) | P1 | 当前手动管理表结构 |
| 后台执行优化 | P1 | Windows 子进程/队列方式异步执行 |
| 前端布局抽取 | P1 | 独立 Layout 组件 + 侧边栏折叠 |
| 定时任务 | P2 | Celery/APScheduler |
| Docker 部署 | P2 | Docker Compose 配置 |
### 已知风险
- **SQLite 并发瓶颈**:当前使用 SQLite,多用户并发场景下可能出现 `database is locked`,远期需迁移 PostgreSQL
- **Windows 子进程限制**:Playwright 在 Windows asyncio 下无法正常启动,当前用同步 + 线程池规避,非完美方案
- **无用户权限系统**:当前无认证/授权机制,仅限内网使用
---
## 五、踩坑记录(绝对不要重复踩)
### 坑 1:Playwright 在 Windows 上必须用同步 API
- **现象**`NotImplementedError` (SelectorEventLoop 不支持子进程),`It looks like you are using Playwright Sync API inside the asyncio loop`
- **根因**:Windows 默认 `SelectorEventLoop` 不支持子进程管道;`ProactorEventLoop` 虽支持但 Playwright 仍会检测 asyncio 环境
- **正确做法**
1. `main.py` 设置 `asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())`
2. 使用 `sync_playwright`(同步 API)而非 `async_playwright`
3. 通过 `loop.run_in_executor()` 在线程池中执行
4. 每个测试用例创建独立的 `PlaywrightExecutor` 实例
- **不要**:尝试用 `asyncio.create_task``threading``multiprocessing` 在 Windows 上实现异步执行——都已试过,全部失败
### 坑 2:每个用例必须用独立的执行器实例
- **现象**:第二次执行时 `execute_case()` 调用 `start()` 失败
- **根因**:共享执行器在 `stop()` 后将 `_is_running` 设为 `False`,下次 `start()` 在 asyncio 环境中失败
- **正确做法**:每次执行都 `PlaywrightExecutor(...)` 新实例
### 坑 3:前端 baseURL 不能带 `/api`
- **现象**:请求路径变成 `/api/api/cases` 导致 404
- **根因**`baseURL: '/api'` 与路由路径 `/api/cases` 叠加
- **正确做法**`baseURL` 设为空字符串,路由路径写完整 `/api/...`
### 坑 4:SQLite 并发写锁
- **现象**`SQLite database is locked`
- **正确做法**`check_same_thread=False` + `pool_pre_ping=True`,仍需注意同步执行期间的并发写入
### 坑 5:bash 中 `start` 不是 Windows 原生 cmd
- **现象**:CMD 窗口打开、cd 成功但 claude 不执行
- **根因**:bash 里的 `start` 解析为 `/usr/bin/start`(MSYS 封装),会损坏 `&&` 复合命令的引号
- **正确做法**`MSYS_NO_PATHCONV=1 cmd.exe /c start "Title" cmd /k "<bat 路径>"`
### 坑 6:`claude` 命令不在 Windows PATH
- **现象**:新开的 cmd 窗口敲 `claude` 报"不是内部或外部命令"
- **正确做法**:用 `which claude` 找出完整路径,补 `.cmd` 后缀,用完整路径调用
### 坑 7:前端 axios 超时
- 默认 30s 超时,但 Playwright 执行可能需要几分钟
- **正确做法**:超时设为 5 分钟(300000ms)
---
## 六、重要文档索引
| 文档 | 路径 | 说明 |
|------|------|------|
| PRD 主文档 | `Docs/PRD/自动化测试平台/_PRD_平台自动化测试可视化系统需求文档.md` | 项目需求定义 |
| 执行计划 | `Docs/PRD/自动化测试平台/_PRD_平台自动化测试可视化系统需求文档_计划执行.md` | 分阶段执行计划 |
| 当前进度记录 | `Docs/PRD/自动化测试平台/当前进度记录.md` | 含 26 个测试用例结果、关键决策 |
| 进度报告 | `Docs/PRD/自动化测试平台/_开发进度报告_20260709.md` | 含完整 API 清单、Phase 5 待办 |
| 测试结果汇总 | `Docs/PRD/自动化测试平台/测试结果汇总报告.md` | 执行结果详细数据 |
| 登录模块优化 | `Docs/PRD/自动化测试平台/_PRD_需求文档_登录模块测试优化.md` | 最新 PRD 需求 |
| 登录模块执行计划 | `Docs/PRD/自动化测试平台/_执行计划_登录模块测试优化.md` | 对应的执行计划 |
| 用例管理完成报告 | `Docs/PRD/自动化测试平台/测试用例管理优化-第一阶段完成报告.md` | 依赖管理功能完成报告 |
---
## 七、启动指南
```bash
# 后端(端口 8001)
cd backend
pip install -r requirements.txt
playwright install chromium
uvicorn app.main:app --reload --port 8001
# 前端(端口 3000,代理指向 8001)
cd frontend
npm install
npm run dev
```
**访问地址**
- 前端页面: http://localhost:3000
- API 文档: http://localhost:8001/docs
- 健康检查: http://localhost:8001/health
**被测系统**:https://192.168.5.44 (统一管理平台)
- 登录用户: admin@xty / Ubains@13579
- 验证码: csba(固定值)
---
## 八、本次会话详细信息
### 会话时间
2026-07-13 ~ 2026-07-13
### 执行的操作
1. 使用 `/CreateCMD 3` 打开 3 个 CMD 窗口运行 Claude
2. 修复 `CreateCMD` skill 的路径配置(从 `E:\` 更新为 `C:\`
3. 创建 `Handoff` skill
4. 生成本 HANDOFF.md
### 未提交的变更
- `.claude/skills/CreateCMD/_run_claude.bat` — 路径配置修改
- `pc自动化测试/` 目录 — 新出现的 untracked 目录
### 远程仓库
- `origin`: http://git.ubainsyun.com/bing/ubains-module-test.git
---
*本文档由 `/Handoff` skill 自动生成,供下一次会话快速恢复上下文。*
\ No newline at end of file
# 平台自动化测试可视化系统
> **Web 可视化自动化测试平台** — 用户通过在目标网站上实际操作,系统自动捕获行为并转换为测试步骤,无需手动编写测试代码。
![Python](https://img.shields.io/badge/Python-3.10+-blue)
![FastAPI](https://img.shields.io/badge/FastAPI-0.110-green)
![Vue 3](https://img.shields.io/badge/Vue_3-3.4-brightgreen)
![Playwright](https://img.shields.io/badge/Playwright-1.40-orange)
![License](https://img.shields.io/badge/License-MIT-yellow)
![Status](https://img.shields.io/badge/Status-开发中-lightgrey)
---
## 📋 目录
- [项目概述](#-项目概述)
- [技术栈](#-技术栈)
- [项目结构](#-项目结构)
- [功能特性](#-功能特性)
- [快速开始](#-快速开始)
- [使用指南](#-使用指南)
- [API 接口](#-api-接口)
- [执行引擎能力](#-执行引擎能力)
- [开发进度](#-开发进度)
- [CI/CD](#-cicd)
- [常见问题](#-常见问题)
- [开发团队](#-开发团队)
---
## 🎯 项目概述
**平台自动化测试可视化系统** 是一个针对企业内部统一管理平台的 Web 可视化自动化测试工具。它解决了传统自动化测试需要编写代码的门槛问题,通过**用例录制器**功能,让测试人员通过实际操作即可生成测试用例。
### 核心创新
| 特性 | 说明 |
|------|------|
| 🎥 **可视化录制** | 在目标网站上操作,系统自动捕获点击、输入、导航等行为 |
| 🚀 **一键执行** | 支持单个或批量执行测试用例,实时查看执行进度 |
| 📊 **实时推送** | 基于 WebSocket 的毫秒级实时状态推送 |
| 📝 **报告生成** | 自动生成 HTML 格式的测试报告,支持导出和下载 |
| 🔄 **用例依赖** | 支持测试用例间的依赖管理,自动拓扑排序 |
| 🧹 **数据清理** | 截图、报告等资源的自动清理和数据库优化 |
---
## 🛠 技术栈
| 层级 | 技术 | 版本 |
|------|------|------|
| **后端框架** | FastAPI | ^0.110.x |
| **ORM** | SQLAlchemy | ^2.0.x |
| **数据库** | SQLite (aiosqlite) | 3.x |
| **前端框架** | Vue 3 + TypeScript | ^3.4.x |
| **UI 组件库** | Element Plus | ^2.5.x |
| **构建工具** | Vite | ^5.1.x |
| **浏览器自动化** | Playwright (Chromium) | ^1.40.x |
| **实时通信** | WebSocket | FastAPI 内置 |
| **图表库** | ECharts | ^5.5.x |
| **状态管理** | Pinia | ^2.1.x |
---
## 📁 项目结构
```
platform-auto-test/
├── backend/ # 后端服务 (Python FastAPI)
│ ├── requirements.txt # Python 依赖清单
│ └── app/
│ ├── main.py # FastAPI 应用入口
│ ├── config.py # 应用配置管理
│ ├── database.py # SQLite 异步引擎 & 会话管理
│ ├── models/ # 数据模型层
│ │ ├── module.py # 模块表模型
│ │ ├── test_case.py # 测试用例模型
│ │ ├── execution.py # 执行记录模型
│ │ ├── case_result.py # 用例结果模型
│ │ └── case_dependency.py # 用例依赖模型
│ ├── schemas/ # Pydantic 数据校验层
│ │ ├── module.py # 模块 CRUD Schema
│ │ └── test_case.py # 用例 CRUD + 导入导出 Schema
│ ├── routers/ # API 路由层 (10 个路由模块)
│ │ ├── modules.py # 模块管理 API
│ │ ├── cases.py # 用例管理 API
│ │ ├── executions.py # 执行管理 API + WebSocket
│ │ ├── recorder.py # 录制器 API
│ │ ├── stats.py # 统计分析 API
│ │ ├── reports.py # 报告管理 API
│ │ ├── cleanup.py # 数据清理 API
│ │ ├── batch.py # 批量操作 API
│ │ └── dependencies.py # 依赖管理 API
│ ├── services/ # 业务逻辑层
│ │ ├── module_service.py # 模块服务
│ │ ├── case_service.py # 用例服务
│ │ ├── execution_service.py # 执行调度服务
│ │ ├── report_service.py # 报告生成服务
│ │ ├── cleanup_service.py # 数据清理服务
│ │ ├── batch_service.py # 批量操作服务
│ │ └── dependency_service.py # 依赖管理服务
│ ├── executors/ # 测试执行引擎
│ │ └── playwright_executor.py # Playwright 执行器
│ ├── websocket/ # 实时通信
│ │ └── manager.py # WebSocket 连接管理
│ ├── utils/ # 工具函数
│ │ ├── id_generator.py # UUID 生成器
│ │ ├── selector_mapper.py # 选择器映射器
│ │ └── session_manager.py # 会话管理器
│ └── scripts/ # 辅助脚本 (15 个)
├── frontend/ # 前端应用 (Vue 3 + TS)
│ ├── package.json # 依赖配置
│ ├── vite.config.ts # Vite 构建配置
│ ├── index.html # HTML 入口
│ └── src/
│ ├── main.ts # 应用入口
│ ├── App.vue # 根组件 (布局 + 导航)
│ ├── router/index.ts # 路由配置 (7 个路由)
│ ├── types/ # TypeScript 类型定义
│ │ ├── module.ts
│ │ ├── case.ts
│ │ └── execution.ts
│ ├── api/ # API 调用层
│ │ ├── modules.ts
│ │ ├── cases.ts
│ │ ├── executions.ts
│ │ ├── stats.ts
│ │ ├── reports.ts
│ │ └── recorder.ts
│ ├── utils/ # 工具函数
│ │ ├── request.ts # Axios 封装
│ │ └── websocket.ts # WebSocket 客户端
│ ├── components/ # 公共组件
│ │ └── StepDetailPanel.vue # 步骤详情面板
│ └── views/ # 页面组件 (7 个)
│ ├── Dashboard.vue # 测试总览
│ ├── Modules.vue # 模块管理
│ ├── Cases.vue # 用例管理
│ ├── Recorder.vue # 用例录制
│ ├── Execution.vue # 执行中心
│ ├── Reports.vue # 报告中心
│ └── Settings.vue # 系统配置
├── Docs/ # 项目文档
│ └── PRD/自动化测试平台/ # PRD 需求 & 计划文档
├── data/ # 运行数据 (gitignored)
│ ├── test_platform.db # SQLite 数据库
│ ├── screenshots/ # 执行截图
│ └── reports/ # 生成的报告
├── .github/workflows/ # CI/CD 配置
│ └── ci-cd.yml # GitHub Actions 工作流
└── .claude/ # Claude Code 配置
└── skills/ # 自定义技能
```
---
## ✨ 功能特性
### 🎥 用例录制器
在目标网站上通过真实操作自动生成测试用例,无需编写代码:
1. **启动录制** — 配置目标 URL,打开录制浏览器
2. **操作捕获** — 自动记录点击、输入、导航、选择等操作
3. **添加断言** — 支持可见性判断、文本验证、URL 断言等
4. **保存用例** — 录制完成自动保存为标准用例格式
### 🚀 执行引擎
支持 13 种动作类型和 8 种断言类型:
**动作类型**`navigate` · `click` · `fill` · `select` · `hover` · `wait` · `scroll` · `press` · `check` · `uncheck` · `assert` · `screenshot` · `upload`
**断言类型**`visible` · `hidden` · `text` · `contains` · `value` · `count` · `url` · `title`
### 📊 实时监控
- 基于 WebSocket 的毫秒级实时状态推送
- 6 种消息类型:`execution_start` · `case_start` · `step_update` · `case_complete` · `execution_complete` · `execution_cancelled`
- 自动断线重连(最多 3 次)
### 📝 报告系统
- HTML 格式测试报告,包含统计卡片、进度条、用例详情
- 支持报告生成、导出、下载
- 截图自动关联展示
### 🔄 用例依赖管理
- 支持测试用例间的依赖关系定义
- 自动拓扑排序,确保执行顺序正确
- 批量操作支持
---
## 🚀 快速开始
### 环境要求
| 环境 | 版本要求 |
|------|---------|
| Python | ^3.10 |
| Node.js | ^18.0 |
| npm | ^9.0 |
| SQLite | ^3.0 |
### 安装与启动
```bash
# 1️⃣ 克隆项目
git clone http://git.ubainsyun.com/bing/ubains-module-test.git
cd platform-auto-test
# 2️⃣ 安装后端依赖
cd backend
pip install -r requirements.txt
# 3️⃣ 安装 Playwright 浏览器
playwright install chromium
# 4️⃣ 启动后端服务 (端口 8001)
uvicorn app.main:app --reload --port 8001
# 5️⃣ 安装前端依赖 (新开终端)
cd frontend
npm install
# 6️⃣ 启动前端服务 (端口 3000)
npm run dev
```
### 访问地址
| 服务 | 地址 |
|------|------|
| 🌐 前端页面 | http://localhost:3000 |
| 📚 API 文档 (Swagger) | http://localhost:8001/docs |
| 📖 ReDoc 文档 | http://localhost:8001/redoc |
| ❤️ 健康检查 | http://localhost:8001/health |
---
## 📖 使用指南
### 1. 模块管理
- 创建测试模块分类(如"会议管理"、"登录模块"等)
- 支持卡片式展示,直观查看模块统计信息
- 模块级联删除(自动删除模块下所有用例)
### 2. 用例管理
- 在模块下创建测试用例
- 每个用例包含多个测试步骤(JSON 格式)
- 支持用例复制、导入/导出
- 支持用例依赖关系配置
### 3. 用例录制
1. 点击「启动录制」配置录制参数
2. 在弹出的浏览器窗口中操作目标系统
3. 系统自动捕获每一步操作
4. 可手动添加断言验证
5. 录制完成后保存为测试用例
### 4. 执行中心
- 选择需要执行的测试用例
- 支持批量选择和顺序执行
- WebSocket 实时推送执行进度
- 查看每一步的截图和日志
- 支持取消正在执行的测试
### 5. 报告中心
- 查看历史执行报告
- 生成 HTML 格式详细报告
- 支持报告导出和下载
---
## 🔌 API 接口
### 模块管理
| 方法 | 路径 | 说明 |
|------|------|------|
| GET | `/api/modules` | 获取模块列表(分页) |
| POST | `/api/modules` | 创建模块 |
| GET | `/api/modules/{id}` | 获取模块详情 |
| PUT | `/api/modules/{id}` | 更新模块 |
| DELETE | `/api/modules/{id}` | 删除模块 |
| GET | `/api/modules/{id}/stats` | 获取模块统计 |
### 用例管理
| 方法 | 路径 | 说明 |
|------|------|------|
| GET | `/api/cases` | 获取用例列表 |
| POST | `/api/cases` | 创建用例 |
| GET | `/api/cases/{id}` | 获取用例详情 |
| PUT | `/api/cases/{id}` | 更新用例 |
| DELETE | `/api/cases/{id}` | 删除用例 |
| POST | `/api/cases/{id}/copy` | 复制用例 |
| POST | `/api/cases/import` | 批量导入用例 |
| POST | `/api/cases/export` | 导出用例 |
### 执行管理
| 方法 | 路径 | 说明 |
|------|------|------|
| GET | `/api/executions` | 获取执行记录列表 |
| POST | `/api/executions` | 创建执行任务 |
| GET | `/api/executions/{id}` | 获取执行详情 |
| POST | `/api/executions/{id}/run` | 启动执行 |
| POST | `/api/executions/{id}/cancel` | 取消执行 |
| GET | `/api/executions/{id}/results` | 获取执行结果 |
| WS | `/api/executions/ws/{id}` | WebSocket 实时推送 |
### 统计分析
| 方法 | 路径 | 说明 |
|------|------|------|
| GET | `/api/stats/overview` | 总览统计 |
| GET | `/api/stats/trend` | 趋势统计 |
| GET | `/api/stats/modules` | 模块统计 |
| GET | `/api/stats/module/{id}` | 模块详情统计 |
| GET | `/api/stats/priority` | 优先级分布 |
| GET | `/api/stats/status` | 状态分布 |
### 报告管理 & 数据清理 & 录制器
| 方法 | 路径 | 说明 |
|------|------|------|
| GET | `/api/reports/generate/{id}` | 生成 HTML 报告 |
| POST | `/api/reports/export/{id}` | 导出报告 |
| GET | `/api/reports/download/{id}` | 下载报告 |
| POST | `/api/recorder/start` | 启动录制 |
| POST | `/api/recorder/stop` | 停止录制 |
| POST | `/api/recorder/save` | 保存录制用例 |
| GET | `/api/cleanup/stats` | 存储统计 |
| POST | `/api/cleanup/cleanup` | 执行清理 |
---
## 🧪 执行引擎能力
### 选择器优先级策略
| 优先级 | 策略 | 示例 |
|--------|------|------|
| 1 | `test-id` 属性 | `[data-testid="login-btn"]` |
| 2 | `id` 属性 | `#login-btn` |
| 3 | 语义化属性 | `[aria-label="登录"]` |
| 4 | `role` 属性 | `button[role="button"]` |
| 5 | 文本内容 | `text=登录` |
| 6 | CSS 类名 | `.login-button` |
| 7 | 层级组合 | `div > button.submit` |
### 多选器回退策略
每个步骤可配置多个选择器,执行时按优先级依次尝试,前一个失败则自动回退到下一个选择器,大幅提高测试稳定性。
---
## 📈 开发进度
| 阶段 | 内容 | 状态 | 完成度 |
|------|------|------|--------|
| **Phase 1** | 基础框架搭建 | ✅ 已完成 | 100% |
| **Phase 2** | 核心功能开发 | ✅ 已完成 | 100% |
| **Phase 3** | 执行与可视化 | ✅ 已完成 | 100% |
| **Phase 4** | 分析与报告 | ✅ 已完成 | 100% |
| **Phase 5** | 扩展与优化 | 🔄 进行中 | ~30% |
### 最新测试结果
| 指标 | 数值 |
|------|------|
| 测试用例总数 | 26 |
| 通过率 | 100% (26/26) |
| 总执行次数 | 89 |
| 模块覆盖率 | 96.3% (26/27) |
| 深层交互测试 | 5 个 (含表单、搜索、断言) |
### Phase 5 待办
| 优先级 | 任务 | 状态 |
|--------|------|------|
| 🔴 P0 | 系统配置页面 (Settings.vue) | ⏳ 待开始 |
| 🔴 P0 | Pinia 状态管理 | ⏳ 待开始 |
| 🔴 P0 | 单元测试 (pytest + Vitest) | ⏳ 待开始 |
| 🟡 P1 | 数据库迁移 (Alembic) | ⏳ 待开始 |
| 🟡 P1 | 后台执行优化 | ⏳ 待开始 |
| 🟡 P1 | 前端布局抽取 | ⏳ 待开始 |
| 🔵 P2 | 定时任务 | ⏳ 待开始 |
| 🔵 P2 | Docker 部署 | ⏳ 待开始 |
---
## 🔄 CI/CD
项目配置了 GitHub Actions 自动流水线:
- **Backend Tests** — pytest + coverage
- **Frontend Tests** — lint + build
- **Integration Tests** — WebSocket 流程验证
- **Build** — 打包构建产物
- **Deploy** — 仅 master 分支触发
---
## ❗ 常见问题
### Playwright 在 Windows 上无法运行
**问题**`NotImplementedError` (SelectorEventLoop 不支持子进程)
**解决**:项目已适配 Windows 平台,使用同步 API + 线程池方案:
```python
# main.py 已配置
asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
```
每个测试用例会创建独立的 `PlaywrightExecutor` 实例,在线程池中隔离执行。
### 前端页面空白
**问题**`inject() can only be used inside setup()`
**解决**:确保 `useRoute()``<script setup>` 顶层调用,不要在 `computed()` 回调内调用。
### 请求报 404
**问题**:API 请求路径变成 `/api/api/cases`
**解决**:前端 `baseURL` 设为空字符串,路由路径写完整 `/api/...`
### 数据库锁定
**问题**`SQLite database is locked`
**解决**:项目已配置 `check_same_thread=False` + `pool_pre_ping=True`
---
## 👥 开发团队
| 角色 | 姓名 |
|------|------|
| 开发者 | czj |
---
## 📄 文档索引
| 文档 | 路径 | 说明 |
|------|------|------|
| PRD 主文档 | `Docs/PRD/自动化测试平台/_PRD_平台自动化测试可视化系统需求文档.md` | 完整需求定义 |
| 执行计划 | `Docs/PRD/自动化测试平台/_PRD_平台自动化测试可视化系统需求文档_计划执行.md` | 分阶段执行计划 |
| 进度报告 | `Docs/PRD/自动化测试平台/_开发进度报告_20260709.md` | 含 API 清单、Phase 5 待办 |
| 测试结果 | `Docs/PRD/自动化测试平台/测试结果汇总报告.md` | 执行结果详细数据 |
| 当前进度 | `Docs/PRD/自动化测试平台/当前进度记录.md` | 含 26 个用例结果 |
---
*📅 最后更新: 2026-07-13*
\ No newline at end of file
...@@ -25,6 +25,7 @@ from playwright.sync_api import ( ...@@ -25,6 +25,7 @@ from playwright.sync_api import (
from playwright._impl._errors import TimeoutError as PlaywrightTimeoutError from playwright._impl._errors import TimeoutError as PlaywrightTimeoutError
from app.config import settings from app.config import settings
from app.utils.selector_mapper import SelectorMapper
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
...@@ -150,6 +151,10 @@ class PlaywrightExecutor: ...@@ -150,6 +151,10 @@ class PlaywrightExecutor:
self.retry_count: int = self.config.get("retry", 0) self.retry_count: int = self.config.get("retry", 0)
self.ignore_https_errors: bool = self.config.get("ignore_https_errors", True) self.ignore_https_errors: bool = self.config.get("ignore_https_errors", True)
self.auto_login: bool = self.config.get("auto_login", False) self.auto_login: bool = self.config.get("auto_login", False)
self.step_retry_count: int = self.config.get("step_retry_count", 2)
# SelectorMapper 实例(用于 page_key + element_key 解析)
self._selector_mapper: SelectorMapper = SelectorMapper()
# 内部状态 # 内部状态
self._playwright: Optional[Playwright] = None self._playwright: Optional[Playwright] = None
...@@ -271,6 +276,144 @@ class PlaywrightExecutor: ...@@ -271,6 +276,144 @@ class PlaywrightExecutor:
logger.error(f"自动登录异常: {str(e)}") logger.error(f"自动登录异常: {str(e)}")
return False return False
# ==================== 选择器与等待辅助 ====================
def _resolve_selectors(self, params: Dict[str, Any]) -> List[str]:
"""
解析参数中的选择器链
优先级:
1. params.selectors(多选择器回退链)
2. params.selector(单选择器,向后兼容)
3. params.page_key + params.element_key(调 SelectorMapper 取完整链)
Args:
params (dict): 步骤参数
Returns:
List[str]: 选择器列表(可能为空)
"""
selectors = params.get("selectors")
if isinstance(selectors, list) and selectors:
return [str(s) for s in selectors if s]
# 单选择器向后兼容
single = params.get("selector")
if single:
return [str(single)]
# SelectorMapper 解析
page_key = params.get("page_key")
element_key = params.get("element_key")
if page_key and element_key:
return self._selector_mapper.resolve_selectors(page_key, element_key)
return []
def _wait_for_element(
self,
selector: str,
timeout: Optional[int] = None,
state: str = "visible",
) -> bool:
"""
显式等待元素到达指定状态
失败不抛异常,返回 False,由调用方决定后续行为。
Args:
selector (str): 选择器
timeout (Optional[int]): 超时毫秒,None 用默认
state (str): 状态 visible/hidden/attached/detached
Returns:
bool: 是否在超时前达到状态
"""
try:
self._page.wait_for_selector(
selector,
state=state,
timeout=timeout or self.timeout,
)
return True
except Exception as e:
logger.debug(f"⏳ 等待元素未就绪: {selector} state={state} ({e})")
return False
def _safe_is_visible(self, selector: str) -> bool:
"""安全判断元素是否可见,不抛异常"""
try:
return self._page.is_visible(selector)
except Exception:
return False
def _wait_for_loading_overlay(self, timeout: int = 10000) -> None:
"""
等待加载遮罩两阶段:先短暂等出现,再等消失
用于导航、点击触发列表加载等场景,避免在遮罩期间操作。
"""
overlay_selectors = [
".el-loading-mask",
".loading",
"[class*='loading']",
]
# 第一阶段:短暂等待出现(很短,避免无遮罩场景空等)
appeared = False
for sel in overlay_selectors:
try:
self._page.wait_for_selector(sel, state="visible", timeout=800)
appeared = True
break
except Exception:
continue
if not appeared:
return
# 第二阶段:等待所有遮罩隐藏
try:
for sel in overlay_selectors:
try:
self._page.wait_for_selector(sel, state="hidden", timeout=timeout)
except Exception:
continue
except Exception:
pass
def _post_action_wait(self, action_type: str, override: Optional[int] = None) -> None:
"""
动作完成后短等待,保证 UI 响应
Args:
action_type (str): 动作类型
override (Optional[int]): 用户自定义延时(毫秒),非空则覆盖默认
"""
if override is not None:
try:
self._page.wait_for_timeout(int(override))
except Exception:
pass
return
defaults = {
"navigate": 2000,
"click": 500,
"fill": 300,
"select": 500,
"hover": 300,
"check": 300,
"uncheck": 300,
"press": 200,
"dialog": 800,
"scroll": 200,
}
delay = defaults.get(action_type, 300)
try:
self._page.wait_for_timeout(delay)
except Exception:
pass
def execute_case( def execute_case(
self, self,
case: Dict[str, Any], case: Dict[str, Any],
...@@ -331,28 +474,19 @@ class PlaywrightExecutor: ...@@ -331,28 +474,19 @@ class PlaywrightExecutor:
result.steps_result.append(step_result) result.steps_result.append(step_result)
# 如果步骤失败,后续步骤标记为跳过 # 如果步骤失败,后续步骤标记为跳过
# (单步重试已在 execute_step 内完成,这里不再整用例重试)
if step_result.status == "failed": if step_result.status == "failed":
result.status = "failed" result.status = "failed"
result.error_message = step_result.error result.error_message = step_result.error
# 尝试重试 # 后续步骤跳过
if result.retry_count < self.retry_count:
result.retry_count += 1
logger.warning(f"重试用例 {case_name},第{result.retry_count}次重试")
# 重置步骤结果重新执行
result.steps_result = []
result.error_message = None
result.status = "pending"
continue
# 不重试,后续步骤跳过
for remaining in steps[len(result.steps_result):]: for remaining in steps[len(result.steps_result):]:
skipped = StepResult( skipped = StepResult(
order=remaining.get("order", 0), order=remaining.get("order", 0),
name=remaining.get("name", ""), name=remaining.get("name", ""),
action=remaining.get("action", ""), action=remaining.get("action", ""),
status="skipped", status="skipped",
log="前置步骤失败,跳过执行", log="前置步骤失败,跳过执行",
) )
result.steps_result.append(skipped) result.steps_result.append(skipped)
break break
...@@ -378,7 +512,7 @@ class PlaywrightExecutor: ...@@ -378,7 +512,7 @@ class PlaywrightExecutor:
callback: Optional[Callable[[StepResult], None]] = None, callback: Optional[Callable[[StepResult], None]] = None,
) -> StepResult: ) -> StepResult:
""" """
执行单个测试步骤 执行单个测试步骤(支持多选择器回退 + 单步重试)
Args: Args:
step (Dict): 步骤定义 step (Dict): 步骤定义
...@@ -387,6 +521,11 @@ class PlaywrightExecutor: ...@@ -387,6 +521,11 @@ class PlaywrightExecutor:
- action (str): 动作类型 - action (str): 动作类型
- params (dict): 动作参数 - params (dict): 动作参数
- expected (str): 预期结果 - expected (str): 预期结果
- selectors (list): 多选择器回退链(可选)
- page_key (str): 页面标识(可选,与 element_key 配合)
- element_key (str): 元素标识(可选)
- force (bool): 强制点击(可选)
- wait_after (int): 操作后延时(可选)
callback (Optional[Callable]): 步骤完成回调 callback (Optional[Callable]): 步骤完成回调
Returns: Returns:
...@@ -401,79 +540,114 @@ class PlaywrightExecutor: ...@@ -401,79 +540,114 @@ class PlaywrightExecutor:
) )
start_time = datetime.now() start_time = datetime.now()
try:
action = step.get("action", "") action = step.get("action", "")
params = step.get("params", {}) params = step.get("params", {})
expected = step.get("expected", "") expected = step.get("expected", "")
force = step.get("force", params.get("force", False))
wait_after = step.get("wait_after", params.get("wait_after", None))
logger.debug(f"执行步骤 {step_result.order}: {step_result.name} (action={action})") logger.info(f"⏳ 步骤 {step_result.order}/{params.get('_total_steps', '?')}: {step_result.name}")
# 单步重试循环
attempt = 0
max_attempts = self.step_retry_count + 1 # 至少执行一次
last_error: Optional[Exception] = None
while attempt < max_attempts:
attempt += 1
step_result.status = "running"
step_result.error = None
try:
logger.debug(f"⏳ 执行步骤 {step_result.order}: {step_result.name} (action={action}, attempt={attempt}/{max_attempts})")
# 智能处理:如果已登录且步骤是导航到登录页或首页,跳过 # 智能处理:如果已登录且步骤是导航到登录页或首页,跳过
if action == "navigate" and self._is_logged_in: if action == "navigate" and self._is_logged_in:
url = params.get("url", "") url = params.get("url", "")
# 跳过登录页、首页导航
if (self.LOGIN_CONFIG["base_url"] in url and if (self.LOGIN_CONFIG["base_url"] in url and
("login" in url.lower() or url == self.LOGIN_CONFIG["base_url"] or url.endswith("/"))): ("login" in url.lower() or url == self.LOGIN_CONFIG["base_url"] or url.endswith("/"))):
step_result.status = "passed" step_result.status = "passed"
step_result.log = f"已登录,跳过导航: {url}" step_result.log = f"✓ 已登录,跳过导航: {url}"
logger.info(f"已登录,跳过步骤 {step_result.order}: {step_result.name}") logger.info(f"✓ 已登录,跳过步骤 {step_result.order}: {step_result.name}")
return step_result break
else:
logger.debug(f"导航URL不匹配跳过条件: {url}")
# 根据动作类型分发执行 # 根据动作类型分发执行
if action == "navigate": if action == "navigate":
self._do_navigate(params) self._do_navigate(params)
step_result.log = f"导航到: {params.get('url', '')}" step_result.log = f"✓ 导航到: {params.get('url', '')}"
elif action == "click": elif action == "click":
self._do_click(params) self._do_click(params, force=force)
step_result.log = f"点击元素: {params.get('selector', '')}" step_result.log = f"✓ 点击元素: {params.get('selector', '') or '多选择器'}"
elif action == "fill": elif action == "fill":
self._do_fill(params) self._do_fill(params, force=force)
step_result.log = f"填充输入框: {params.get('selector', '')} = {params.get('value', '')}" step_result.log = f"✓ 填充输入框: {params.get('selector', '') or '多选择器'} = {params.get('value', '')}"
elif action == "select": elif action == "select":
self._do_select(params) self._do_select(params, force=force)
step_result.log = f"选择下拉框: {params.get('selector', '')} = {params.get('value', '')}" step_result.log = f"✓ 选择下拉框: {params.get('selector', '') or '多选择器'} = {params.get('value', '')}"
elif action == "hover": elif action == "hover":
self._do_hover(params) self._do_hover(params)
step_result.log = f"悬停元素: {params.get('selector', '')}" step_result.log = f"✓ 悬停元素: {params.get('selector', '') or '多选择器'}"
elif action == "wait": elif action == "wait":
self._do_wait(params) self._do_wait(params)
step_result.log = f"等待元素: {params.get('selector', '')}" step_result.log = f"✓ 等待元素: {params.get('selector', '') or '多选择器'}"
elif action == "scroll": elif action == "scroll":
self._do_scroll(params) self._do_scroll(params)
step_result.log = f"滚动页面: {params.get('direction', 'down')}" step_result.log = f"✓ 滚动页面: {params.get('direction', 'down')}"
elif action == "press": elif action == "press":
self._do_press(params) self._do_press(params)
step_result.log = f"按键: {params.get('key', '')}" step_result.log = f"✓ 按键: {params.get('key', '')}"
elif action == "check": elif action == "check":
self._do_check(params, check=True) self._do_check(params, check=True, force=force)
step_result.log = f"勾选: {params.get('selector', '')}" step_result.log = f"✓ 勾选: {params.get('selector', '') or '多选择器'}"
elif action == "uncheck": elif action == "uncheck":
self._do_check(params, check=False) self._do_check(params, check=False, force=force)
step_result.log = f"取消勾选: {params.get('selector', '')}" step_result.log = f"✓ 取消勾选: {params.get('selector', '') or '多选择器'}"
elif action == "assert": elif action == "assert":
self._do_assert(params, expected) self._do_assert(params, expected)
step_result.log = f"断言验证: {params.get('type', 'visible')} - 通过" step_result.log = f"✓ 断言验证: {params.get('type', 'visible')} - 通过"
elif action == "screenshot": elif action == "screenshot":
self._do_screenshot(params) self._do_screenshot(params)
step_result.log = f"截图保存" step_result.log = "✓ 截图保存"
else: else:
raise ValueError(f"不支持的动作类型: {action}") raise ValueError(f"不支持的动作类型: {action}")
# 成功:退出重试循环
step_result.status = "passed" step_result.status = "passed"
last_error = None
break
except Exception as e: except Exception as e:
last_error = e
step_result.status = "failed" step_result.status = "failed"
step_result.error = str(e) step_result.error = str(e)
step_result.log = f"执行失败: {str(e)}" step_result.log = f"✗ 执行失败: {str(e)}"
logger.warning(f"步骤 {step_result.order} 失败: {str(e)}") logger.warning(f"⚠ 步骤 {step_result.order} 尝试 {attempt}/{max_attempts} 失败: {str(e)}")
finally: # 如果还有重试机会,短暂等待后重试
if attempt < max_attempts:
try:
self._page.wait_for_timeout(1000)
except Exception:
pass
else:
# 达到最大重试次数,最终失败
step_result.log = f"✗ 步骤失败(已重试 {self.step_retry_count} 次): {str(e)}"
logger.error(f"✗ 步骤 {step_result.order} 最终失败: {str(e)}")
# 最终状态设置
if last_error:
step_result.status = "failed"
elif step_result.status != "passed":
step_result.status = "failed"
# 结束计时
step_result.end_time = datetime.now() step_result.end_time = datetime.now()
step_result.duration = (step_result.end_time - start_time).total_seconds() step_result.duration = (step_result.end_time - start_time).total_seconds()
# 成功后执行后置等待
if step_result.status == "passed":
self._post_action_wait(action, wait_after)
# 截图(无论成功与否,如果开启了截图) # 截图(无论成功与否,如果开启了截图)
if self.screenshot_enabled: if self.screenshot_enabled:
try: try:
...@@ -482,14 +656,20 @@ class PlaywrightExecutor: ...@@ -482,14 +656,20 @@ class PlaywrightExecutor:
self._page.screenshot(path=screenshot_path) self._page.screenshot(path=screenshot_path)
step_result.screenshot = screenshot_path step_result.screenshot = screenshot_path
except Exception as e: except Exception as e:
logger.error(f"截图失败: {str(e)}") logger.error(f"✗ 截图失败: {str(e)}")
# 回调通知 # 回调通知
if callback: if callback:
try: try:
callback(step_result) callback(step_result)
except Exception as e: except Exception as e:
logger.error(f"回调函数执行失败: {str(e)}") logger.error(f"✗ 回调函数执行失败: {str(e)}")
# 图标日志
if step_result.status == "passed":
logger.info(f"✓ 步骤 {step_result.order} 完成 (耗时 {step_result.duration:.2f}s)")
else:
logger.error(f"✗ 步骤 {step_result.order} 失败: {step_result.error}")
return step_result return step_result
...@@ -506,93 +686,169 @@ class PlaywrightExecutor: ...@@ -506,93 +686,169 @@ class PlaywrightExecutor:
if not url: if not url:
raise ValueError("导航URL不能为空") raise ValueError("导航URL不能为空")
self._page.goto(url, wait_until="networkidle") self._page.goto(url, wait_until="networkidle")
# 等待加载遮罩消失
self._wait_for_loading_overlay()
def _do_click(self, params: dict) -> None: def _do_click(self, params: dict, force: bool = False) -> None:
""" """
点击元素 点击元素(多选择器回退 + 强制点击)
Args: Args:
params (dict): {"selector": str} params (dict): 支持 {"selector": str} / {"selectors": [str]} / {"page_key","element_key"}
force (bool): 是否强制点击(忽略覆盖层)
""" """
selector = params.get("selector", "") selectors = self._resolve_selectors(params)
if not selector: if not selectors:
raise ValueError("选择器不能为空") raise ValueError("选择器不能为空")
# 尝试等待元素可见后再点击 last_error: Optional[Exception] = None
try:
self._page.wait_for_selector(selector, state="visible", timeout=10000) for selector in selectors:
except Exception: # 等待元素可见
# 如果等待失败,直接尝试点击 self._wait_for_element(selector, timeout=10000, state="visible")
pass
# 尝试在主页面点击 # 主页面点击
try: try:
self._page.click(selector, timeout=5000) self._page.click(selector, timeout=5000, force=force)
logger.debug(f"✓ 主页面点击成功: {selector}")
# 点击触发列表加载时等待遮罩
self._wait_for_loading_overlay()
return return
except Exception: except Exception as e:
pass last_error = e
logger.debug(f"⚠ 主页面点击失败 {selector}: {e}")
# 如果主页面失败,尝试在iframe中点击 # iframe 中点击
for frame in self._page.frames: for frame in self._page.frames:
if frame != self._page.main_frame: if frame == self._page.main_frame:
continue
try: try:
# 等待iframe中的元素
frame.wait_for_selector(selector, state="visible", timeout=5000) frame.wait_for_selector(selector, state="visible", timeout=5000)
frame.click(selector, timeout=5000) frame.click(selector, timeout=5000, force=force)
logger.debug(f"在iframe中点击成功: {selector}") logger.debug(f"✓ iframe 中点击成功: {selector}")
self._wait_for_loading_overlay()
return return
except Exception: except Exception as e2:
last_error = e2
continue continue
# 如果都失败,抛出异常 raise Exception(f"✗ 无法点击元素(已尝试 {len(selectors)} 个选择器): {selectors} 最后错误: {last_error}")
raise Exception(f"无法点击元素: {selector}")
def _do_fill(self, params: dict) -> None: def _do_fill(self, params: dict, force: bool = False) -> None:
""" """
填充输入框 填充输入框(多选择器回退 + 强制清空 + 强制点击)
清空再输入。 click(count=3, force=True) 全选 → Ctrl+A → Delete → 等待 100ms → fill(value)
Args: Args:
params (dict): {"selector": str, "value": str} params (dict): {"selector"|"selectors"|"page_key","element_key": str, "value": str}
force (bool): 是否强制点击
""" """
selector = params.get("selector", "") selectors = self._resolve_selectors(params)
value = params.get("value", "") if not selectors:
if not selector:
raise ValueError("选择器不能为空") raise ValueError("选择器不能为空")
self._page.fill(selector, str(value)) value = params.get("value", "")
last_error: Optional[Exception] = None
for selector in selectors:
try:
# 等待元素可见
self._wait_for_element(selector, timeout=10000, state="visible")
# 强制清空:三击全选 → Ctrl+A → Delete → 等待 100ms
self._page.click(selector, click_count=3, force=True, timeout=5000)
self._page.keyboard.press("Control+A")
self._page.keyboard.press("Delete")
self._page.wait_for_timeout(100)
# 填充
self._page.fill(selector, str(value), timeout=5000)
logger.debug(f"✓ 填充成功: {selector} = {value}")
return
except Exception as e:
last_error = e
logger.debug(f"⚠ 填充失败 {selector}: {e}")
continue
raise Exception(f"✗ 无法填充元素(已尝试 {len(selectors)} 个选择器): {selectors} 最后错误: {last_error}")
def _do_select(self, params: dict) -> None: def _do_select(self, params: dict, force: bool = False) -> None:
""" """
选择下拉框 选择下拉框(多选择器回退)
Args: Args:
params (dict): {"selector": str, "value": str} params (dict): {"selector"|"selectors": str|[str], "value": str}
force (bool): 保留参数以保持接口一致
""" """
selector = params.get("selector", "") selectors = self._resolve_selectors(params)
if not selectors:
raise ValueError("选择器不能为空")
value = params.get("value", "") value = params.get("value", "")
self._page.select_option(selector, str(value))
last_error: Optional[Exception] = None
for selector in selectors:
try:
self._wait_for_element(selector, timeout=10000, state="visible")
self._page.select_option(selector, str(value), timeout=5000)
logger.debug(f"✓ 下拉选择成功: {selector} = {value}")
return
except Exception as e:
last_error = e
logger.debug(f"⚠ 下拉选择失败 {selector}: {e}")
continue
raise Exception(f"✗ 无法选择下拉框(已尝试 {len(selectors)} 个选择器): {selectors} 最后错误: {last_error}")
def _do_hover(self, params: dict) -> None: def _do_hover(self, params: dict) -> None:
""" """
鼠标悬停 鼠标悬停(多选择器回退)
Args: Args:
params (dict): {"selector": str} params (dict): {"selector"|"selectors": str|[str]}
""" """
selector = params.get("selector", "") selectors = self._resolve_selectors(params)
self._page.hover(selector) if not selectors:
raise ValueError("选择器不能为空")
last_error: Optional[Exception] = None
for selector in selectors:
try:
self._wait_for_element(selector, timeout=10000, state="visible")
self._page.hover(selector, timeout=5000)
logger.debug(f"✓ 悬停成功: {selector}")
return
except Exception as e:
last_error = e
logger.debug(f"⚠ 悬停失败 {selector}: {e}")
continue
raise Exception(f"✗ 无法悬停元素(已尝试 {len(selectors)} 个选择器): {selectors} 最后错误: {last_error}")
def _do_wait(self, params: dict) -> None: def _do_wait(self, params: dict) -> None:
""" """
等待元素出现 等待元素出现(多选择器回退)
Args: Args:
params (dict): {"selector": str, "timeout": int} params (dict): {"selector"|"selectors": str|[str], "timeout": int}
""" """
selector = params.get("selector", "") selectors = self._resolve_selectors(params)
if not selectors:
raise ValueError("选择器不能为空")
timeout = params.get("timeout", self.timeout) timeout = params.get("timeout", self.timeout)
last_error: Optional[Exception] = None
for selector in selectors:
try:
self._page.wait_for_selector(selector, timeout=timeout) self._page.wait_for_selector(selector, timeout=timeout)
logger.debug(f"✓ 等待元素成功: {selector}")
return
except Exception as e:
last_error = e
logger.debug(f"⚠ 等待元素失败 {selector}: {e}")
continue
raise Exception(f"✗ 等待元素超时(已尝试 {len(selectors)} 个选择器): {selectors} 最后错误: {last_error}")
def _do_scroll(self, params: dict) -> None: def _do_scroll(self, params: dict) -> None:
""" """
...@@ -626,19 +882,35 @@ class PlaywrightExecutor: ...@@ -626,19 +882,35 @@ class PlaywrightExecutor:
else: else:
self._page.keyboard.press(key) self._page.keyboard.press(key)
def _do_check(self, params: dict, check: bool = True) -> None: def _do_check(self, params: dict, check: bool = True, force: bool = False) -> None:
""" """
勾选/取消勾选 勾选/取消勾选(多选择器回退 + force)
Args: Args:
params (dict): {"selector": str} params (dict): {"selector"|"selectors": str|[str]}
check (bool): True=勾选, False=取消勾选 check (bool): True=勾选, False=取消勾选
force (bool): 是否强制点击
""" """
selector = params.get("selector", "") selectors = self._resolve_selectors(params)
if not selectors:
raise ValueError("选择器不能为空")
last_error: Optional[Exception] = None
for selector in selectors:
try:
self._wait_for_element(selector, timeout=10000, state="visible")
if check: if check:
self._page.check(selector) self._page.check(selector, timeout=5000, force=force)
else: else:
self._page.uncheck(selector) self._page.uncheck(selector, timeout=5000, force=force)
logger.debug(f"✓ {'勾选' if check else '取消勾选'}成功: {selector}")
return
except Exception as e:
last_error = e
logger.debug(f"⚠ {'勾选' if check else '取消勾选'}失败 {selector}: {e}")
continue
raise Exception(f"✗ 无法勾选/取消勾选(已尝试 {len(selectors)} 个选择器): {selectors} 最后错误: {last_error}")
def _do_screenshot(self, params: dict) -> None: def _do_screenshot(self, params: dict) -> None:
""" """
...@@ -653,68 +925,114 @@ class PlaywrightExecutor: ...@@ -653,68 +925,114 @@ class PlaywrightExecutor:
def _do_assert(self, params: dict, expected: str = "") -> None: def _do_assert(self, params: dict, expected: str = "") -> None:
""" """
执行断言验证 执行断言验证(多选择器回退用于 visible/text/contains/value/count)
支持多种断言类型:visible, hidden, text, contains, value, count, url, title 支持多种断言类型:visible, hidden, text, contains, value, count, url, title
Args: Args:
params (dict): 断言参数 params (dict): 断言参数
- selector (str): 元素选择器
- type (str): 断言类型
- expected (str): 期望值
expected (str): 期望描述 expected (str): 期望描述
Raises: Raises:
AssertionError: 断言失败时抛出 AssertionError: 断言失败时抛出
""" """
assert_type = params.get("type", "visible") assert_type = params.get("type", "visible")
selector = params.get("selector", "") selectors = self._resolve_selectors(params)
expected_value = params.get("expected", "") expected_value = params.get("expected", "")
if assert_type == "visible": if assert_type == "visible":
element = self._page.wait_for_selector(selector, state="visible") if not selectors:
if not element: raise AssertionError("断言 visible 缺少选择器")
raise AssertionError(f"断言失败: 元素 {selector} 不可见") for selector in selectors:
if self._wait_for_element(selector, timeout=self.timeout, state="visible"):
return
raise AssertionError(f"✗ 断言失败: 所有选择器均不可见: {selectors}")
elif assert_type == "hidden": elif assert_type == "hidden":
element = self._page.wait_for_selector(selector, state="hidden") if not selectors:
if not element: raise AssertionError("断言 hidden 缺少选择器")
raise AssertionError(f"断言失败: 元素 {selector} 仍然可见") for selector in selectors:
try:
self._page.wait_for_selector(selector, state="hidden", timeout=self.timeout)
return
except Exception:
continue
raise AssertionError(f"✗ 断言失败: 元素仍然可见: {selectors}")
elif assert_type == "text": elif assert_type == "text":
element = self._page.wait_for_selector(selector) if not selectors:
text = element.text_content() raise AssertionError("断言 text 缺少选择器")
if text.strip() != expected_value.strip(): last_err: Optional[Exception] = None
raise AssertionError(f"断言失败: 期望文本 '{expected_value}',实际 '{text.strip()}'") for selector in selectors:
try:
element = self._page.wait_for_selector(selector, timeout=self.timeout)
if element:
text = element.text_content() or ""
if text.strip() == expected_value.strip():
return
last_err = AssertionError(f"断言失败: 期望文本 '{expected_value}',实际 '{text.strip()}'")
except Exception as e:
last_err = e
continue
raise AssertionError(f"✗ 断言 text 失败: {selectors} 错误: {last_err}")
elif assert_type == "contains": elif assert_type == "contains":
element = self._page.wait_for_selector(selector) if not selectors:
text = element.text_content() raise AssertionError("断言 contains 缺少选择器")
if expected_value not in text: last_err = None
raise AssertionError(f"断言失败: 文本不包含 '{expected_value}',实际 '{text}'") for selector in selectors:
try:
element = self._page.wait_for_selector(selector, timeout=self.timeout)
if element:
text = element.text_content() or ""
if expected_value in text:
return
last_err = AssertionError(f"文本不包含 '{expected_value}',实际 '{text}'")
except Exception as e:
last_err = e
continue
raise AssertionError(f"✗ 断言 contains 失败: {selectors} 错误: {last_err}")
elif assert_type == "value": elif assert_type == "value":
element = self._page.wait_for_selector(selector) if not selectors:
raise AssertionError("断言 value 缺少选择器")
last_err = None
for selector in selectors:
try:
element = self._page.wait_for_selector(selector, timeout=self.timeout)
if element:
actual_value = element.input_value() actual_value = element.input_value()
if actual_value != expected_value: if actual_value == expected_value:
raise AssertionError(f"断言失败: 期望值 '{expected_value}',实际 '{actual_value}'") return
last_err = AssertionError(f"期望值 '{expected_value}',实际 '{actual_value}'")
except Exception as e:
last_err = e
continue
raise AssertionError(f"✗ 断言 value 失败: {selectors} 错误: {last_err}")
elif assert_type == "count": elif assert_type == "count":
elements = self._page.query_selector_all(selector) if not selectors:
actual_count = len(elements) raise AssertionError("断言 count 缺少选择器")
expected_count = int(expected_value) expected_count = int(expected_value)
if actual_count != expected_count: total = 0
raise AssertionError(f"断言失败: 期望数量 {expected_count},实际 {actual_count}") for selector in selectors:
try:
elements = self._page.query_selector_all(selector)
total += len(elements)
except Exception:
continue
if total != expected_count:
raise AssertionError(f"✗ 断言失败: 期望数量 {expected_count},实际 {total}")
elif assert_type == "url": elif assert_type == "url":
current_url = self._page.url current_url = self._page.url
if expected_value not in current_url: if expected_value not in current_url:
raise AssertionError(f"断言失败: 当前URL '{current_url}' 不包含 '{expected_value}'") raise AssertionError(f"断言失败: 当前URL '{current_url}' 不包含 '{expected_value}'")
elif assert_type == "title": elif assert_type == "title":
current_title = self._page.title() current_title = self._page.title()
if expected_value not in current_title: if expected_value not in current_title:
raise AssertionError(f"断言失败: 当前标题 '{current_title}' 不包含 '{expected_value}'") raise AssertionError(f"断言失败: 当前标题 '{current_title}' 不包含 '{expected_value}'")
else: else:
raise ValueError(f"不支持的断言类型: {assert_type}") raise ValueError(f"不支持的断言类型: {assert_type}")
......
...@@ -19,6 +19,7 @@ from sqlalchemy.ext.asyncio import AsyncSession ...@@ -19,6 +19,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db from app.database import get_db
from app.services.case_service import CaseService from app.services.case_service import CaseService
from app.schemas.test_case import StepDefinition
from app.utils.id_generator import generate_id from app.utils.id_generator import generate_id
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
...@@ -60,21 +61,10 @@ class StartRecordingResponse(BaseModel): ...@@ -60,21 +61,10 @@ class StartRecordingResponse(BaseModel):
message: str message: str
class RecordedStep(BaseModel):
"""录制的步骤"""
order: int
name: str
action: str
selector: Optional[str] = None
value: Optional[str] = None
params: Optional[dict] = None
expected: Optional[str] = None
class AddStepRequest(BaseModel): class AddStepRequest(BaseModel):
"""添加步骤请求""" """添加步骤请求"""
session_id: str session_id: str
step: RecordedStep step: StepDefinition
class AddAssertionRequest(BaseModel): class AddAssertionRequest(BaseModel):
...@@ -176,13 +166,14 @@ async def add_step( ...@@ -176,13 +166,14 @@ async def add_step(
if not session.is_recording: if not session.is_recording:
raise HTTPException(status_code=400, detail="录制会话已停止") raise HTTPException(status_code=400, detail="录制会话已停止")
step_data = request.step.dict() # 直接以 StepDefinition 格式存储(含 selectors/page_key/element_key/force/wait_after)
step_data = request.step.model_dump(by_alias=False, exclude_none=True)
session.steps.append(step_data) session.steps.append(step_data)
logger.info(f"步骤已添加: session_id={request.session_id}, step_order={step_data['order']}") logger.info(f"步骤已添加: session_id={request.session_id}, step_order={step_data.get('order')}")
return { return {
"message": "步骤已添加", "message": "步骤已添加",
"step_order": step_data["order"], "step_order": step_data.get("order"),
"total_steps": len(session.steps), "total_steps": len(session.steps),
} }
...@@ -208,13 +199,15 @@ async def add_assertion( ...@@ -208,13 +199,15 @@ async def add_assertion(
raise HTTPException(status_code=400, detail="录制会话已停止") raise HTTPException(status_code=400, detail="录制会话已停止")
order = len(session.steps) + 1 order = len(session.steps) + 1
# 统一存储为 StepDefinition 格式:selectors 列表 + params
step = { step = {
"order": order, "order": order,
"name": f"断言验证: {request.selector}", "name": f"断言验证: {request.selector}",
"action": "assert", "action": "assert",
"selector": request.selector, "selectors": [request.selector] if request.selector else [],
"params": { "params": {
"selector": request.selector, "selector": request.selector,
"selectors": [request.selector] if request.selector else [],
"type": request.type, "type": request.type,
"expected": request.expected or "", "expected": request.expected or "",
}, },
...@@ -284,23 +277,22 @@ async def save_as_case( ...@@ -284,23 +277,22 @@ async def save_as_case(
try: try:
case_service = CaseService(db) case_service = CaseService(db)
# 构建步骤数据 # 录制器已统一为 StepDefinition 格式(含 selectors/page_key/element_key/force/wait_after)
# 直接透传给 CaseService,无需手动字段转换
steps = [] steps = []
for s in session.steps: for s in session.steps:
# 保留所有已知字段,未知字段原样放在 params 中
step = { step = {
"order": s.get("order"), "order": s.get("order"),
"name": s.get("name", ""), "name": s.get("name", ""),
"action": s.get("action", ""), "action": s.get("action", ""),
"params": s.get("params", {}),
"expected": s.get("expected", ""),
} }
if s.get("params"): # 透传可选字段(执行器识别)
step["params"] = s["params"] for opt_field in ("selectors", "page_key", "element_key", "force", "wait_after"):
else: if opt_field in s and s[opt_field] is not None:
step["params"] = { step[opt_field] = s[opt_field]
"selector": s.get("selector", ""),
"value": s.get("value", ""),
}
if s.get("expected"):
step["expected"] = s["expected"]
steps.append(step) steps.append(step)
# 创建用例 # 创建用例
...@@ -314,6 +306,7 @@ async def save_as_case( ...@@ -314,6 +306,7 @@ async def save_as_case(
"config": { "config": {
"timeout": 30000, "timeout": 30000,
"screenshot": True, "screenshot": True,
"step_retry_count": 2,
}, },
} }
......
...@@ -32,6 +32,11 @@ class StepDefinition(BaseModel): ...@@ -32,6 +32,11 @@ class StepDefinition(BaseModel):
action (str): 动作类型 (navigate/click/fill/assert等) action (str): 动作类型 (navigate/click/fill/assert等)
params (dict): 动作参数 params (dict): 动作参数
expected (str): 预期结果 expected (str): 预期结果
selectors (Optional[List[str]]): 多选择器回退链
page_key (Optional[str]): SelectorMapper 页面标识
element_key (Optional[str]): 元素标识
force (Optional[bool]): 强制点击
wait_after (Optional[int]): 自定义操作后延时(ms)
""" """
order: int = Field(..., ge=1, description="步骤顺序") order: int = Field(..., ge=1, description="步骤顺序")
...@@ -39,6 +44,11 @@ class StepDefinition(BaseModel): ...@@ -39,6 +44,11 @@ class StepDefinition(BaseModel):
action: str = Field(..., description="动作类型") action: str = Field(..., description="动作类型")
params: Dict[str, Any] = Field(default_factory=dict, description="动作参数") params: Dict[str, Any] = Field(default_factory=dict, description="动作参数")
expected: str = Field(default="", description="预期结果") expected: str = Field(default="", description="预期结果")
selectors: Optional[List[str]] = Field(None, description="多选择器回退链")
page_key: Optional[str] = Field(None, description="SelectorMapper 页面标识")
element_key: Optional[str] = Field(None, description="元素标识")
force: Optional[bool] = Field(None, description="强制点击")
wait_after: Optional[int] = Field(None, description="自定义操作后延时(ms)")
class TestCaseBase(BaseModel): class TestCaseBase(BaseModel):
......
...@@ -229,10 +229,20 @@ class ExecutionService: ...@@ -229,10 +229,20 @@ class ExecutionService:
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()
# 定义一个共享的执行函数,复用同一个执行器 # 定义一个共享的执行函数,复用同一个执行器
def run_all_cases_sync(cases_list, exec_config): def run_all_cases_sync(cases_list, exec_config, ev_loop, broadcast_fn):
"""在一个执行器中执行所有用例,保持登录态""" """
在一个执行器中执行所有用例,保持登录态
Args:
cases_list: 用例字典列表
exec_config: 执行配置
ev_loop: 主事件循环(用于线程安全广播)
broadcast_fn: 异步广播函数(async broadcast)
"""
executor_config = exec_config or {} executor_config = exec_config or {}
executor_config["auto_login"] = True executor_config["auto_login"] = True
# 透传单步重试次数(默认 2)
executor_config.setdefault("step_retry_count", exec_config.get("step_retry_count", 2) if exec_config else 2)
executor = PlaywrightExecutor(config=executor_config) executor = PlaywrightExecutor(config=executor_config)
results = [] results = []
...@@ -242,8 +252,36 @@ class ExecutionService: ...@@ -242,8 +252,36 @@ class ExecutionService:
# 第一个用例会自动登录 # 第一个用例会自动登录
for i, case_dict in enumerate(cases_list): for i, case_dict in enumerate(cases_list):
logger.info(f"执行用例 {i+1}/{len(cases_list)}: {case_dict.get('name', '')}") case_id = case_dict.get("id", "")
result = executor.execute_case(case=case_dict) case_name = case_dict.get("name", "")
logger.info(f"⏳ 执行用例 {i+1}/{len(cases_list)}: {case_name}")
# 构造线程安全的步骤回调(跨线程提交到事件循环)
def step_callback(step_result, _cid=case_id, _cname=case_name):
try:
asyncio.run_coroutine_threadsafe(
broadcast_fn(execution_id, {
"type": "step_update",
"data": {
"execution_id": execution_id,
"case_id": _cid,
"case_name": _cname,
"step_order": step_result.order,
"step_name": step_result.name,
"action": step_result.action,
"status": step_result.status,
"duration": step_result.duration,
"screenshot": step_result.screenshot,
"log": step_result.log,
"error": step_result.error,
},
}),
ev_loop,
)
except Exception as e:
logger.error(f"✗ 步骤回调提交失败: {str(e)}")
result = executor.execute_case(case=case_dict, callback=step_callback)
results.append(result) results.append(result)
return results return results
...@@ -261,12 +299,14 @@ class ExecutionService: ...@@ -261,12 +299,14 @@ class ExecutionService:
"case_dict": case.to_dict(), "case_dict": case.to_dict(),
}) })
# 在线程池中批量执行 # 在线程池中批量执行(传入事件循环和广播函数以接通 WebSocket 步骤回调)
all_results = await loop.run_in_executor( all_results = await loop.run_in_executor(
None, None,
run_all_cases_sync, run_all_cases_sync,
[c["case_dict"] for c in cases_to_run], [c["case_dict"] for c in cases_to_run],
config config,
loop,
manager.broadcast,
) )
# 更新结果 # 更新结果
......
...@@ -389,6 +389,57 @@ class SelectorMapper: ...@@ -389,6 +389,57 @@ class SelectorMapper:
"""获取通用选择器""" """获取通用选择器"""
return self.selectors.get("common", {}) return self.selectors.get("common", {})
def resolve_selectors(
self,
page_key: Optional[str] = None,
element_key: Optional[str] = None,
) -> List[str]:
"""
通过 page_key + element_key 解析完整选择器链
会在 SELECTORS 中查找以 page_key 为根的所有 category,
收集 element_key 对应的选择器列表。
Args:
page_key (Optional[str]): 页面标识(如 "login")
element_key (Optional[str]): 元素标识(如 "username")
Returns:
List[str]: 选择器列表(可能为空)
"""
if not page_key or not element_key:
return []
page_def = self.selectors.get(page_key)
if not isinstance(page_def, dict):
return []
collected: List[str] = []
# 情况1:element_key 直接映射到顶层列表(如 common.loading)
top_candidate = page_def.get(element_key)
if isinstance(top_candidate, list) and top_candidate:
collected.extend(top_candidate)
# 情况2:element_key 嵌套在某个 category 下
for category_name, category in page_def.items():
# 跳过非字典字段(如 url)
if not isinstance(category, dict):
continue
# 直接以 element_key 为键的列表
candidate = category.get(element_key)
if isinstance(candidate, list) and candidate:
collected.extend(candidate)
# 去重但保留顺序
seen = set()
deduped: List[str] = []
for s in collected:
if s not in seen:
seen.add(s)
deduped.append(s)
return deduped
# 全局选择器映射器实例 # 全局选择器映射器实例
mapper = SelectorMapper() mapper = SelectorMapper()
......
...@@ -18,6 +18,16 @@ export interface StepDefinition { ...@@ -18,6 +18,16 @@ export interface StepDefinition {
params: Record<string, any> params: Record<string, any>
/** 预期结果 */ /** 预期结果 */
expected?: string expected?: string
/** 多选择器回退链 */
selectors?: string[]
/** SelectorMapper 页面标识 */
pageKey?: string
/** 元素标识 */
elementKey?: string
/** 强制点击 */
force?: boolean
/** 自定义操作后延时(ms) */
waitAfter?: number
} }
/** 测试用例接口 */ /** 测试用例接口 */
......
...@@ -62,6 +62,27 @@ ...@@ -62,6 +62,27 @@
</span> </span>
</el-progress> </el-progress>
</div> </div>
<!-- 实时步骤时间线 -->
<div class="live-timeline" v-if="hasLiveSteps">
<div class="timeline-title">实时步骤时间线</div>
<el-scrollbar max-height="280px">
<div
v-for="(step, idx) in allLiveSteps"
:key="`${step.case_id}-${step.step_order}-${idx}`"
class="timeline-item"
>
<span class="timeline-icon">{{ getStepIcon(step.status) }}</span>
<span class="timeline-step-name">{{ step.step_name }}</span>
<el-tag :type="getResultType(step.status)" size="small" effect="plain">
{{ getResultText(step.status) }}
</el-tag>
<span class="timeline-duration" v-if="step.duration">
{{ step.duration.toFixed(2) }}s
</span>
</div>
</el-scrollbar>
</div>
</el-card> </el-card>
<!-- 执行操作栏 --> <!-- 执行操作栏 -->
...@@ -298,6 +319,9 @@ const resultsLoading = ref<Record<string, boolean>>({}) ...@@ -298,6 +319,9 @@ const resultsLoading = ref<Record<string, boolean>>({})
// 用例列表缓存(用于选择用例时显示名称) // 用例列表缓存(用于选择用例时显示名称)
const casesCache = ref<Record<string, any>>({}) const casesCache = ref<Record<string, any>>({})
// 实时步骤时间线(按 case_id 聚合)
const liveSteps = ref<Record<string, any[]>>({})
const runConfig = ref({ const runConfig = ref({
name: '', name: '',
environment: 'default', environment: 'default',
...@@ -339,6 +363,32 @@ const passRateClass = computed(() => { ...@@ -339,6 +363,32 @@ const passRateClass = computed(() => {
return 'rate-danger' return 'rate-danger'
}) })
// 实时步骤时间线:聚合所有用例的步骤
const allLiveSteps = computed(() => {
const all: any[] = []
for (const cid in liveSteps.value) {
for (const s of liveSteps.value[cid]) {
all.push({ ...s, case_id: cid })
}
}
// 按 step_order 排序
all.sort((a, b) => (a.step_order || 0) - (b.step_order || 0))
return all
})
const hasLiveSteps = computed(() => allLiveSteps.value.length > 0)
const getStepIcon = (status: string) => {
const map: Record<string, string> = {
passed: '✓',
failed: '✗',
running: '⏳',
skipped: '⏭',
pending: '⋯',
}
return map[status] || '•'
}
// ==================== 方法定义 ==================== // ==================== 方法定义 ====================
const getStatusType = (status: string) => { const getStatusType = (status: string) => {
...@@ -558,6 +608,8 @@ const cancelExecution = async (id: string) => { ...@@ -558,6 +608,8 @@ const cancelExecution = async (id: string) => {
const connectWebSocket = (executionId: string) => { const connectWebSocket = (executionId: string) => {
// 先断开旧连接 // 先断开旧连接
disconnectWebSocket() disconnectWebSocket()
// 清空实时步骤时间线
liveSteps.value = {}
wsClient = new WebSocketClient(executionId) wsClient = new WebSocketClient(executionId)
...@@ -575,8 +627,19 @@ const connectWebSocket = (executionId: string) => { ...@@ -575,8 +627,19 @@ const connectWebSocket = (executionId: string) => {
}) })
wsClient.on('step_update', (data: any) => { wsClient.on('step_update', (data: any) => {
// 实时更新步骤状态 // 实时更新步骤状态 + 聚合到时间线
console.log('步骤更新:', data.step_name, data.status) console.log('步骤更新:', data.step_name, data.status)
const cid = data.case_id
if (!liveSteps.value[cid]) {
liveSteps.value[cid] = []
}
const arr = liveSteps.value[cid]
const idx = arr.findIndex(s => s.step_order === data.step_order)
if (idx >= 0) {
arr[idx] = data
} else {
arr.push(data)
}
}) })
wsClient.on('case_complete', (data: any) => { wsClient.on('case_complete', (data: any) => {
...@@ -590,6 +653,28 @@ const connectWebSocket = (executionId: string) => { ...@@ -590,6 +653,28 @@ const connectWebSocket = (executionId: string) => {
results.push(data) results.push(data)
} }
} }
// 用例完成时合并时间线到 rowResults 的 steps_result
const steps = liveSteps.value[data.case_id]
if (steps && steps.length) {
// 将时间线步骤映射回 steps_result 格式
const mapped = steps.map(s => ({
order: s.step_order,
name: s.step_name,
action: s.action,
status: s.status,
duration: s.duration,
screenshot: s.screenshot,
log: s.log,
error: s.error,
}))
if (rowResults.value[activeExecution.value?.id]) {
const results = rowResults.value[activeExecution.value?.id]
const ridx = results.findIndex(r => r.case_id === data.case_id)
if (ridx >= 0 && (!results[ridx].steps_result || results[ridx].steps_result.length === 0)) {
results[ridx].steps_result = mapped
}
}
}
}) })
wsClient.on('execution_complete', (data: any) => { wsClient.on('execution_complete', (data: any) => {
...@@ -670,6 +755,44 @@ onUnmounted(() => { ...@@ -670,6 +755,44 @@ onUnmounted(() => {
font-size: 14px; font-size: 14px;
} }
} }
.live-timeline {
margin-top: 16px;
padding-top: 12px;
border-top: 1px dashed #ebeef5;
.timeline-title {
font-size: 14px;
font-weight: 600;
color: #303133;
margin-bottom: 8px;
}
.timeline-item {
display: flex;
align-items: center;
gap: 8px;
padding: 6px 0;
font-size: 13px;
border-bottom: 1px dashed #f0f0f0;
.timeline-icon {
width: 18px;
text-align: center;
font-weight: 600;
}
.timeline-step-name {
flex: 1;
color: #303133;
}
.timeline-duration {
color: #909399;
font-size: 12px;
}
}
}
} }
.page-header { .page-header {
......
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论