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

feat(test): 新增测试用例依赖管理和批量操作功能

- 新增 TestCase 模型字段:depends_on、parameters、condition、order、version
- 新增 CaseDependency 依赖关系模型,支持强依赖/弱依赖配置
- 新增 batch 路由、schema、service,支持批量创建、更新、删除、执行测试用例
- 新增 dependencies 路由、schema、service,支持依赖关系拓扑排序和条件执行
- 新增 backend/scripts 辅助脚本目录(15个脚本):用例创建、迁移、清理等
- 新增 PRD 文档:登录模块测试优化需求文档和执行计划
- 新增测试用例 Excel 和 backend/tests/ 测试目录
- 完善 GitCommit skill:增加分支检查、推送确认、三重门控机制
- 修改 Playwright 执行器默认 headless 为 False,便于调试
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 726e4713
--- ---
name: git-commit name: git-commit
description: 代码提交辅助 - 审查待提交文件、自动生成 Conventional Commits 信息、执行 git add/commit description: 代码提交辅助 - 审查待提交文件、自动生成 Conventional Commits 信息、执行 git add/commit/push(仅限当前分支)
--- ---
代码提交辅助工具。审查当前工作区所有改动文件,过滤无关/临时文件,与用户确认提交范围后自动生成符合 Conventional Commits 规范的提交信息,执行提交。 代码提交辅助工具。审查当前工作区所有改动文件,过滤无关/临时文件,与用户确认提交范围后自动生成符合 Conventional Commits 规范的提交信息,执行提交。
**🚨 重要限制:**
- **仅限当前分支**:只能提交到当前分支,禁止切换分支或提交到其他分支
- **推送需确认**:每次 push 到远程仓库前必须与用户确认
## Usage ## Usage
``` ```
...@@ -14,6 +18,32 @@ description: 代码提交辅助 - 审查待提交文件、自动生成 Conventio ...@@ -14,6 +18,32 @@ description: 代码提交辅助 - 审查待提交文件、自动生成 Conventio
## 执行流程 ## 执行流程
### 阶段 0:分支检查(前置步骤)
**在执行任何 git 操作前,必须先检查当前分支:**
1. **获取当前分支信息**
```bash
git branch --show-current # 获取当前分支名
git rev-parse --abbrev-ref HEAD # 备用方法
```
2. **获取远程分支信息**
```bash
git remote -v # 查看远程仓库
git branch -vv # 查看分支跟踪关系
```
3. **明确告知用户**
- 当前分支名
- 远程仓库名(通常是 origin)
- 远程跟踪分支
4. **锁定分支**
- ✅ 只能在当前分支执行 `git add``git commit`
- ❌ 禁止执行 `git checkout``git switch``git push origin <other-branch>`
- ❌ 禁止提交到非当前分支
### 阶段 1:扫描与分类 ### 阶段 1:扫描与分类
1. **收集改动**:执行 `git status --porcelain``git diff --stat`,获取所有 staged 和 unstaged 文件 1. **收集改动**:执行 `git status --porcelain``git diff --stat`,获取所有 staged 和 unstaged 文件
...@@ -132,18 +162,38 @@ Co-Authored-By: Claude <noreply@anthropic.com> ...@@ -132,18 +162,38 @@ Co-Authored-By: Claude <noreply@anthropic.com>
1. `git add <确认的文件列表>`**逐文件 add,不使用 `git add -A` 或 `git add .`** 1. `git add <确认的文件列表>`**逐文件 add,不使用 `git add -A` 或 `git add .`**
2. `git commit -m "<生成的 message>"` 2. `git commit -m "<生成的 message>"`
3. **不要自动 push**:提交成功后告知用户,并询问是否需要 `git push`(push 也需用户明确同意才执行) 3. **🚨 推送确认(三重门控)**
- 提交成功后,**必须**询问用户是否需要 `git push`
- 使用 `AskUserQuestion` 让用户选择:
- ✅ 推送到远程仓库(当前分支)
- ❌ 不推送,仅本地提交
- 如果用户选择推送,**必须再次确认**
- 显示即将推送的分支:`git push origin <当前分支名>`
- 用户确认后才执行推送
- **禁止行为**
- ❌ 自动推送(必须用户明确同意)
- ❌ 推送到其他分支(只能推送到当前分支的远程跟踪分支)
- ❌ 执行 `git push --all``git push --tags`(除非用户明确要求)
**禁止行为:** **禁止行为:**
- ❌ 跳过任一确认门直接 add/commit - ❌ 跳过任一确认门直接 add/commit/push
- ❌ 把"未明确拒绝"当作"同意"——必须用户主动确认才执行 - ❌ 把"未明确拒绝"当作"同意"——必须用户主动确认才执行
- ❌ 一次性把"提交 + push"打包执行,两者必须分别确认 - ❌ 一次性把"提交 + push"打包执行,两者必须分别确认
- ❌ 提交到非当前分支
- ❌ 推送到非当前分支的远程跟踪分支
## 注意事项 ## 注意事项
- **🚨 双重确认门控(最高优先级)**:范围确认(门1)+ message 确认(门2)都必须用户主动点头,未明确同意绝不执行 add/commit/push - **🚨 三重确认门控(最高优先级)**
1. 范围确认(门1)- 用户确认提交哪些文件
2. Message 确认(门2)- 用户确认 commit message
3. 推送确认(门3)- 用户确认是否推送及推送目标
- **🚨 分支限制(严格执行)**
- 只能在当前分支操作,禁止切换分支
- 只能推送到当前分支的远程跟踪分支
- 禁止使用 `git push origin <其他分支名>`
- **绝不使用 `git add .` 或 `git add -A`**:必须逐文件 add,只添加用户确认的文件 - **绝不使用 `git add .` 或 `git add -A`**:必须逐文件 add,只添加用户确认的文件
- **不自动 push**:push 需用户明确指令 - **推送必须确认**:每次 push 前必须与用户确认,显示完整的推送命令和目标分支
- **不修改 .gitignore**:如需添加忽略规则,提醒用户自行添加 - **不修改 .gitignore**:如需添加忽略规则,提醒用户自行添加
- **commit message 末尾必须有 `Co-Authored-By: Claude <noreply@anthropic.com>`** - **commit message 末尾必须有 `Co-Authored-By: Claude <noreply@anthropic.com>`**
- **中文描述优先**:commit 摘要使用中文,与仓库已有提交风格保持一致 - **中文描述优先**:commit 摘要使用中文,与仓库已有提交风格保持一致
......
# PRD需求文档 - 登录模块测试用例优化
## 一、需求背景
### 1.1 问题描述
登录模块的测试用例在执行过程中出现超时失败问题,即使实际登录成功,测试结果仍显示失败。
### 1.2 问题影响
- 测试结果不准确,误报失败
- 无法真实反映登录功能的实际状态
- 影响测试平台的可信度和使用体验
- 浪费调试时间和资源
### 1.3 现状分析
**失败的测试步骤:**
- 步骤8:验证登录成功(断言URL包含"platform")- 超时失败
**根本原因:**
1. **断言时机不当**:URL断言是立即检查,不等待页面跳转完成
2. **超时时间不足**:页面跳转需要时间,默认30秒不够
3. **等待策略问题**:等待`networkidle`过于严格,页面未完全加载就超时
4. **元素选择器不精确**:部分选择器在实际页面上匹配不到元素
---
## 二、需求目标
### 2.1 总体目标
优化登录模块测试用例,确保测试结果准确反映实际功能状态,提高测试稳定性。
### 2.2 具体目标
| 目标项 | 当前状态 | 目标状态 | 优先级 |
|--------|---------|----------|--------|
| 测试通过率 | 0% | ≥90% | 高 |
| 超时失败率 | 100% | 0% | 高 |
| 用例执行时间 | 30-50秒 | 15-25秒 | 中 |
| 误报率 | 高 | 低 | 高 |
---
## 三、需求详情
### 3.1 功能需求
#### FR-1:优化超时配置
**需求描述:** 增加测试用例的超时时间,并支持动态配置
**详细说明:**
- 默认超时从30秒增加到60秒
- 在config中支持自定义超时时间
- 不同步骤可以有不同的超时设置
**验收标准:**
- ✅ 页面导航步骤超时设置为60秒
- ✅ 等待元素步骤超时可自定义(默认10秒)
- ✅ 断言步骤不设置超时,立即执行
#### FR-2:优化等待策略
**需求描述:** 使用更合理的等待策略替代`networkidle`
**详细说明:**
- 页面导航使用`domcontentloaded`替代`networkidle`
- 新增显式等待步骤,等待关键元素出现
- 删除不可靠的URL断言,改用元素等待
**验收标准:**
- ✅ 导航步骤不等待完全加载,只等待DOM就绪
- ✅ 后续步骤等待关键元素(如登录按钮、页面标题)
- ✅ 等待时间充足(10秒)
#### FR-3:优化元素选择器
**需求描述:** 使用更通用和稳定的选择器
**详细说明:**
- 避免使用`:nth-of-type`等不稳定选择器
- 使用属性选择器:`input[type='text']`
- 使用文本选择器:`text=登录`
- 添加选择器降级策略
**验收标准:**
- ✅ 选择器在多种场景下都能匹配
- ✅ 不依赖特定的DOM结构顺序
- ✅ 支持Element Plus等UI框架生成的类名
#### FR-4:关闭auto_login
**需求描述:** 登录测试不应依赖自动登录功能
**详细说明:**
- 登录功能测试自己控制登录流程
- auto_login会干扰登录测试的步骤执行
- 在config中设置`auto_login: false`
**验收标准:**
- ✅ 所有登录用例的config中auto_login=false
- ✅ 测试步骤完整执行,不被auto_login跳过
---
### 3.2 非功能需求
#### NFR-1:可维护性
- 用例步骤清晰,注释完整
- 参数配置化,易于修改
- 选择器可维护,有注释说明
#### NFR-2:可扩展性
- 支持添加新的测试场景
- 支持参数化测试
- 支持数据驱动
#### NFR-3:稳定性
- 连续执行3次,通过率≥90%
- 不同网络环境下执行稳定
- 不因页面加载慢而误报失败
---
## 四、技术方案
### 4.1 断言优化方案
**问题代码:**
```python
# 步骤8 - 原来的URL断言
{"action": "assert", "params": {"type": "url"}, "expected": "platform"}
```
**优化方案:**
```python
# 步骤7 - 改用元素等待
{"action": "wait", "params": {"selector": "text=统一管理平台", "timeout": 10000}}
```
**原因:**
- URL断言立即检查,不等待
- 元素等待会持续检查直到超时或成功
- "统一管理平台"文本在登录后的页面会出现
### 4.2 选择器优化方案
**问题选择器:**
```python
"selector": "input[placeholder*='用户']" # 可能匹配不到
"selector": "input[type='text']:nth-of-type(3)" # 依赖顺序
```
**优化选择器:**
```python
"selector": "input[type='text']" # 通用选择器
"selector": "input[type='password']" # 明确类型
"selector": "button" # 简单选择器
```
**降级策略:**
- 主选择器失败后尝试备选选择器
- 先尝试精确选择器,再尝试通用选择器
### 4.3 超时配置方案
**config配置:**
```json
{
"timeout": 60000, // 全局超时60
"retry": 0, // 不重试
"screenshot": true, // 自动截图
"auto_login": false // 关闭自动登录
}
```
**步骤级超时:**
```json
{"params": {"selector": "input", "timeout": 10000}} // 等待元素10
```
---
## 五、实施计划
### 5.1 第一阶段:用例优化(预计2小时)
**任务列表:**
1. 修改"登录功能-正常登录"用例(高优先级)
- 删除URL断言步骤
- 改用元素等待验证登录成功
- 关闭auto_login
2. 修改"登录功能-页面访问验证"用例(高优先级)
- 增加显式等待步骤
- 优化选择器
3. 修改其他登录用例(中优先级)
- 统一超时配置
- 优化选择器
### 5.2 第二阶段:代码优化(预计1小时)
**任务列表:**
1. 修改playwright_executor.py
- 改进断言逻辑,支持等待
- 添加选择器降级策略
2. 测试验证
- 执行所有登录用例
- 验证通过率≥90%
### 5.3 第三阶段:文档输出(预计0.5小时)
**输出文档:**
1. 需求文档(本文档)
2. 执行计划文档
3. 测试报告
---
## 六、验收标准
### 6.1 功能验收
| 验收项 | 验收标准 | 验收方式 |
|--------|---------|----------|
| 用例执行 | 连续执行3次,通过率≥90% | 自动化测试 |
| 超时问题 | 不再出现30秒超时失败 | 查看执行日志 |
| 选择器匹配 | 所有选择器都能正确匹配元素 | 手动验证 |
| auto_login | 所有登录用例不依赖auto_login | 查看config配置 |
### 6.2 文档验收
- ✅ 需求文档完整,包含问题分析、解决方案
- ✅ 执行计划详细,任务分解合理
- ✅ 测试报告真实反映优化效果
---
## 七、风险评估
| 风险项 | 影响程度 | 概率 | 应对措施 |
|--------|---------|------|----------|
| 页面元素变化导致选择器失效 | 高 | 中 | 使用多种选择器策略 |
| 网络慢导致页面加载慢 | 中 | 中 | 增加超时时间 |
| Element Plus版本更新 | 低 | 低 | 选择器不依赖特定类名 |
---
## 八、附录
### 8.1 相关文件
- `backend/app/executors/playwright_executor.py` - 执行器实现
- `backend/app/services/execution_service.py` - 执行服务
- `backend/scripts/recreate_login_cases.py` - 用例创建脚本
### 8.2 参考资料
- [Playwright等待策略](https://playwright.dev/python/docs/api/class-page#page-goto)
- [Element Plus选择器规范](https://element-plus.org/)
---
**文档版本:** v1.0
**创建日期:** 2026-07-12
**创建人:** Claude
# 执行计划 - 登录模块测试用例优化
## 一、执行概述
### 1.1 执行目标
优化登录模块测试用例,解决超时失败问题,提高测试通过率至90%以上。
### 1.2 执行范围
- 优化8个登录模块测试用例
- 修改playwright_executor.py代码
- 验证优化效果
### 1.3 执行时间
- 开始时间:2026-07-12
- 预计工期:3.5小时
- 完成时间:2026-07-12
---
## 二、执行任务分解
### 任务1:优化"登录功能-正常登录"用例 ⭐高优先级
**执行内容:**
- 删除步骤8的URL断言
- 修改步骤7为元素等待
- 确保config中auto_login=false
**详细步骤:**
```python
# 原步骤(8步):
步骤1: 打开登录页面 (navigate)
步骤2: 等待登录表单加载 (wait, 10)
步骤3: 输入用户名 (fill)
步骤4: 输入密码 (fill)
步骤5: 输入验证码 (fill)
步骤6: 点击登录按钮 (click)
步骤7: 等待页面跳转 (wait, 5)
步骤8: 验证登录成功 (assert url) 删除此步骤
# 优化后(7步):
步骤1: 打开登录页面 (navigate)
步骤2: 等待登录表单加载 (wait, 10)
步骤3: 输入用户名 (fill)
步骤4: 输入密码 (fill)
步骤5: 输入验证码 (fill)
步骤6: 点击登录按钮 (click)
步骤7: 等待页面跳转完成 (wait "text=统一管理平台", 10) 改用元素等待
```
**执行方式:**
- 使用API或脚本更新用例
**验收标准:**
- ✅ 用例只有7个步骤
- ✅ 最后一步是元素等待
- ✅ config中auto_login=false
- ✅ 执行不超时,结果准确
**预计耗时:** 30分钟
---
### 任务2:优化"登录功能-页面访问验证"用例 ⭐高优先级
**执行内容:**
- 增加显式等待步骤
- 优化元素选择器
- 确保config中timeout=60000
**详细步骤:**
```python
# 优化后(3步):
步骤1: 打开登录页面 (navigate, timeout=60)
步骤2: 等待页面加载 (wait "body", 10) 新增等待步骤
步骤3: 验证页面标题 (assert title)
```
**执行方式:**
- 使用API更新用例
**验收标准:**
- ✅ 有显式等待步骤
- ✅ config中timeout=60000
- ✅ 执行通过
**预计耗时:** 20分钟
---
### 任务3:优化其他登录用例 ⭐中优先级
**执行内容:**
- 统一超时配置为60秒
- 优化选择器,避免使用nth-of-type
- 确保所有用例auto_login=false
**涉及用例:**
1. 登录功能-用户名为空(5步)
2. 登录功能-密码为空(5步)
3. 登录功能-错误密码(7步)
4. 登录功能-记住密码勾选框(4步)
5. 登录功能-忘记密码链接(3步)
6. 登录功能-立即注册链接(3步)
**选择器优化清单:**
```python
# 优化前 → 优化后
"input[placeholder*='用户']" "input[type='text']"
"input[type='text']:nth-of-type(3)" "input[type='text']" (最后一个)
"text=忘记密码" "text=忘记密码" (保持)
```
**执行方式:**
- 批量更新脚本
**验收标准:**
- ✅ 所有用例config统一
- ✅ 选择器稳定
- ✅ 执行通过
**预计耗时:** 40分钟
---
### 任务4:修改playwright_executor.py代码 ⭐高优先级
**执行内容:**
- 改进URL断言逻辑,添加等待机制
- 添加选择器降级策略
- 优化错误提示
**代码修改点:**
**修改1:URL断言添加等待**
```python
# 文件:backend/app/executors/playwright_executor.py
# 行号:709-712
# 原代码:
elif assert_type == "url":
current_url = self._page.url
if expected_value not in current_url:
raise AssertionError(f"断言失败: 当前URL '{current_url}' 不包含 '{expected_value}'")
# 优化后:
elif assert_type == "url":
# 等待URL变化(最多等待5秒)
max_wait = 5000
start_time = datetime.now()
while (datetime.now() - start_time).total_seconds() * 1000 < max_wait:
current_url = self._page.url
if expected_value in current_url:
break
self._page.wait_for_timeout(500)
else:
current_url = self._page.url
if expected_value not in current_url:
raise AssertionError(f"断言失败: 当前URL '{current_url}' 不包含 '{expected_value}'")
```
**修改2:添加选择器降级策略**
```python
# 在 _do_click 方法中添加降级逻辑
def _do_click(self, params: dict) -> None:
selector = params.get("selector", "")
if not selector:
raise ValueError("选择器不能为空")
# 主选择器
try:
self._page.wait_for_selector(selector, state="visible", timeout=5000)
self._page.click(selector, timeout=5000)
return
except Exception:
pass
# 降级:尝试在iframe中查找
for frame in self._page.frames:
if frame != self._page.main_frame:
try:
frame.wait_for_selector(selector, state="visible", timeout=3000)
frame.click(selector, timeout=5000)
logger.debug(f"在iframe中点击成功: {selector}")
return
except Exception:
continue
raise Exception(f"无法点击元素: {selector}")
```
**验收标准:**
- ✅ URL断言有等待机制
- ✅ 选择器支持降级
- ✅ 错误提示清晰
**预计耗时:** 30分钟
---
### 任务5:测试验证 ⭐高优先级
**执行内容:**
- 执行所有优化后的登录用例
- 记录执行结果
- 分析失败原因(如有)
**验证步骤:**
1. 在前端界面选择所有登录用例
2. 点击"执行选中"
3. 观察浏览器执行过程
4. 查看执行结果报告
5. 连续执行3次,统计通过率
**验证清单:**
```
[ ] 用例1:登录功能-正常登录 - 通过/失败
[ ] 用例2:登录功能-页面访问验证 - 通过/失败
[ ] 用例3:登录功能-用户名为空 - 通过/失败
[ ] 用例4:登录功能-密码为空 - 通过/失败
[ ] 用例5:登录功能-错误密码 - 通过/失败
[ ] 用例6:登录功能-记住密码勾选框 - 通过/失败
[ ] 用例7:登录功能-忘记密码链接 - 通过/失败
[ ] 用例8:登录功能-立即注册链接 - 通过/失败
```
**验收标准:**
- ✅ 通过率≥90%(至少7/8通过)
- ✅ 无超时失败
- ✅ 执行时间合理(总计<3分钟)
**预计耗时:** 30分钟
---
### 任务6:输出文档 ⭐中优先级
**输出文档列表:**
1. ✅ 需求文档(已完成)
2. ✅ 执行计划文档(本文档)
3. 测试报告(待输出)
**测试报告内容:**
- 执行环境信息
- 测试结果统计
- 失败用例分析
- 优化效果对比
- 改进建议
**预计耗时:** 30分钟
---
## 三、执行进度跟踪
### 3.1 任务看板
| 任务 | 状态 | 负责人 | 开始时间 | 完成时间 | 备注 |
|------|------|--------|---------|---------|------|
| 任务1:优化正常登录用例 | ⏳进行中 | Claude | 2026-07-12 | - | - |
| 任务2:优化页面访问验证用例 | 待开始 | Claude | - | - | - |
| 任务3:优化其他登录用例 | 待开始 | Claude | - | - | - |
| 任务4:修改executor代码 | 待开始 | Claude | - | - | - |
| 任务5:测试验证 | 待开始 | Claude | - | - | - |
| 任务6:输出文档 | 进行中 | Claude | 2026-07-12 | - | 已完成需求文档和执行计划 |
### 3.2 里程碑
- **里程碑1**:完成所有用例优化(预计1.5小时)
- **里程碑2**:完成代码修改(预计2小时)
- **里程碑3**:完成测试验证(预计2.5小时)
- **里程碑4**:完成文档输出(预计3小时)
---
## 四、风险控制
### 4.1 风险识别
| 风险 | 概率 | 影响 | 应对措施 |
|------|------|------|----------|
| 用例执行仍失败 | 中 | 高 | 分析日志,进一步优化选择器 |
| 页面元素变化 | 低 | 高 | 更新选择器 |
| 时间不足 | 低 | 中 | 优先完成高优先级任务 |
### 4.2 回滚方案
如果优化后效果不佳,可:
1. 恢复原有用例
2. 使用git回滚代码
3. 重新分析问题
---
## 五、验收清单
### 5.1 功能验收
- [ ] 所有登录用例执行不超时
- [ ] 测试通过率≥90%
- [ ] 测试结果准确反映实际状态
- [ ] config配置正确
### 5.2 代码验收
- [ ] playwright_executor.py修改完成
- [ ] 代码可读性良好
- [ ] 有必要的注释
### 5.3 文档验收
- [x] 需求文档完整
- [x] 执行计划详细
- [ ] 测试报告真实
---
## 六、执行总结
### 6.1 完成情况
(待执行完成后填写)
### 6.2 效果对比
(待执行完成后填写)
### 6.3 经验总结
(待执行完成后填写)
---
**文档版本:** v1.0
**创建日期:** 2026-07-12
**创建人:** Claude
# 测试用例管理优化 - 第一阶段完成报告
## 执行日期
2026-07-12
---
## 一、完成情况
### ✅ 已完成任务
| 任务 | 状态 | 文件 | 说明 |
|------|------|------|------|
| #1 创建 CaseDependency 模型 | ✅ 完成 | `backend/app/models/case_dependency.py` | 用例依赖关系模型,支持硬依赖和软依赖 |
| #2 扩展 TestCase 模型 | ✅ 完成 | `backend/app/models/test_case.py` | 新增 depends_on, parameters, condition, order, version 字段 |
| #3 实现 DependencyService | ✅ 完成 | `backend/app/services/dependency_service.py` | 依赖管理服务,包含拓扑排序和循环依赖检测算法 |
| #4 实现 BatchService | ✅ 完成 | `backend/app/services/batch_service.py` | 批量操作服务,支持批量删除/更新/复制/标签/优先级 |
| #5 新增依赖管理 API | ✅ 完成 | `backend/app/routers/dependencies.py` | 依赖管理相关 API 接口 |
| #6 新增批量操作 API | ✅ 完成 | `backend/app/routers/batch.py` | 批量操作相关 API 接口 |
| #7 编写数据库迁移脚本 | ✅ 完成 | `backend/scripts/migration_add_dependencies.py` | 数据库迁移脚本,已成功执行 |
---
## 二、新增文件清单
### 数据模型层
- `backend/app/models/case_dependency.py` - 用例依赖关系模型
### Schema层
- `backend/app/schemas/dependency.py` - 依赖管理相关 Schema
- `backend/app/schemas/batch.py` - 批量操作相关 Schema
### 服务层
- `backend/app/services/dependency_service.py` - 依赖管理服务
- `backend/app/services/batch_service.py` - 批量操作服务
### API路由层
- `backend/app/routers/dependencies.py` - 依赖管理 API 路由
- `backend/app/routers/batch.py` - 批量操作 API 路由
### 脚本
- `backend/scripts/migration_add_dependencies.py` - 数据库迁移脚本
---
## 三、修改文件清单
| 文件 | 修改内容 |
|------|----------|
| `backend/app/models/test_case.py` | 添加新字段:depends_on, parameters, condition, order, version<br>添加关联关系:dependencies, dependents |
| `backend/app/main.py` | 注册新路由:batch, dependencies |
| `backend/app/database.py` | 数据库表结构自动创建(通过Base.metadata) |
---
## 四、核心功能实现
### 1. 依赖管理
**CaseDependency 模型:**
```python
class CaseDependency(Base):
id: str # 依赖关系ID
case_id: str # 用例ID(依赖方)
depends_on_case_id: str # 被依赖的用例ID
dependency_type: str # hard(强依赖)/soft(弱依赖)
order: int # 依赖顺序
created_at: datetime # 创建时间
```
**DependencyService 核心方法:**
- `get_case_dependencies()` - 获取用例依赖关系
- `set_case_dependencies()` - 设置用例依赖关系(含验证)
- `validate_dependencies()` - 验证依赖关系(检测循环)
- `topological_sort()` - 拓扑排序(Kahn算法)
- `get_execution_order()` - 计算执行顺序(拓扑+优先级)
- `check_dependency_status()` - 检查依赖状态
**关键算法:**
```python
# 拓扑排序(Kahn算法)
def topological_sort(case_ids):
# 构建依赖图
# 计算入度
# 零入度队列
# 在依赖级别内按优先级排序
# 检测循环依赖
return ordered_ids, cycle
```
### 2. 批量操作
**BatchService 核心方法:**
- `batch_delete()` - 批量删除用例
- `batch_update()` - 批量更新用例
- `batch_copy()` - 批量复制用例
- `batch_assign_tags()` - 批量分配标签
- `batch_set_priority()` - 批量设置优先级
- `batch_enable()` - 批量启用
- `batch_disable()` - 批量禁用
- `batch_move_to_module()` - 批量移动到模块
---
## 五、API 接口
### 依赖管理接口
| 方法 | 路径 | 功能 |
|------|------|------|
| GET | `/api/cases/{case_id}/dependencies` | 获取用例依赖关系 |
| POST | `/api/cases/{case_id}/dependencies` | 设置用例依赖关系 |
| POST | `/api/validate` | 验证依赖关系(检测循环) |
| POST | `/api/execution-order` | 计算执行顺序 |
| GET | `/api/cases/{case_id}/dependency-status` | 检查依赖状态 |
### 批量操作接口
| 方法 | 路径 | 功能 |
|------|------|------|
| POST | `/api/batch/delete` | 批量删除 |
| POST | `/api/batch/update` | 批量更新 |
| POST | `/api/batch/copy` | 批量复制 |
| POST | `/api/batch/assign-tags` | 批量分配标签 |
| POST | `/api/batch/set-priority` | 批量设置优先级 |
| POST | `/api/batch/enable` | 批量启用 |
| POST | `/api/batch/disable` | 批量禁用 |
---
## 六、数据库变更
### 新增表
**case_dependencies 表:**
- id (TEXT PRIMARY KEY)
- case_id (TEXT NOT NULL)
- depends_on_case_id (TEXT NOT NULL)
- dependency_type (TEXT DEFAULT 'hard')
- "order" (INTEGER DEFAULT 0)
- created_at (TIMESTAMP)
### 扩展字段
**test_cases 表:**
- depends_on (JSON DEFAULT '[]')
- parameters (JSON DEFAULT '{}')
- condition (JSON DEFAULT '{}')
- "order" (INTEGER DEFAULT 0)
- version (INTEGER DEFAULT 1)
**executions 表:**
- execution_mode (TEXT DEFAULT 'sequential')
- max_parallel (INTEGER DEFAULT 1)
- param_values (JSON DEFAULT '{}')
**case_results 表:**
- param_instance (JSON DEFAULT '{}')
- dependency_status (TEXT DEFAULT 'ok')
- condition_met (BOOLEAN DEFAULT 1)
---
## 七、使用示例
### 1. 设置用例依赖关系
```python
# 用例 case_002 依赖 case_001
POST /api/cases/case_002/dependencies
{
"dependencies": [
{
"depends_on_case_id": "case_001",
"dependency_type": "hard",
"order": 1
}
]
}
```
### 2. 验证依赖关系
```python
POST /api/validate?case_ids=case_001,case_002,case_003
# 响应
{
"valid": true,
"cycle": null,
"message": "依赖关系有效,无循环"
}
```
### 3. 计算执行顺序
```python
POST /api/execution-order?case_ids=case_001,case_002,case_003
# 响应
{
"ordered_case_ids": ["case_001", "case_002", "case_003"],
"has_cycle": false,
"cycle_path": null
}
```
### 4. 批量删除用例
```python
POST /api/batch/delete
{
"case_ids": ["case_001", "case_002", "case_003"]
}
# 响应
{
"success": 3,
"failed": 0,
"total": 3,
"message": "成功删除 3 个用例"
}
```
---
## 八、下一步计划
### 第二阶段:参数化和条件执行
**待完成任务:**
1. 创建 CaseParameter 模型
2. 创建 ParameterSet 模型
3. 实现 ParameterService(变量替换)
4. 实现 ConditionService(条件评估)
5. 改造 PlaywrightExecutor(支持参数化)
6. 扩展 ExecutionService(参数解析)
**预计工作量:** 3-5天
---
## 九、注意事项
### 已知问题
1. SQLite 的 order 字段是保留关键字,需要用双引号包围
2. 数据库迁移脚本中的 ALTER TABLE 可能因字段已存在而失败(已处理)
3. 循环依赖检测使用 DFS 算法,可能需要优化性能
### 建议
1. 在生产环境部署前,先在测试环境验证迁移脚本
2. 为 case_dependencies 表添加复合索引以提升查询性能
3. 考虑使用 Alembic 进行正式的数据库版本管理
---
## 十、测试建议
### 单元测试
```python
# tests/test_dependency_service.py
def test_topological_sort():
# 测试拓扑排序算法
pass
def test_detect_cycle():
# 测试循环依赖检测
pass
```
### 集成测试
```python
# tests/test_dependency_api.py
def test_set_dependencies():
# 测试设置依赖关系API
pass
def test_validate_dependencies():
# 测试验证依赖关系API
pass
```
---
**完成日期:** 2026-07-12
**下一阶段启动时间:** 待定
\ No newline at end of file
...@@ -134,7 +134,7 @@ class PlaywrightExecutor: ...@@ -134,7 +134,7 @@ class PlaywrightExecutor:
config (Optional[dict]): 执行配置 config (Optional[dict]): 执行配置
- timeout (int): 超时时间(毫秒),默认30000 - timeout (int): 超时时间(毫秒),默认30000
- screenshot (bool): 是否截图,默认True - screenshot (bool): 是否截图,默认True
- headless (bool): 是否无头模式,默认True - headless (bool): 是否无头模式,默认False
- screenshot_dir (str): 截图目录 - screenshot_dir (str): 截图目录
- retry (int): 失败重试次数,默认0 - retry (int): 失败重试次数,默认0
- ignore_https_errors (bool): 是否忽略 HTTPS 错误,默认True - ignore_https_errors (bool): 是否忽略 HTTPS 错误,默认True
...@@ -143,7 +143,7 @@ class PlaywrightExecutor: ...@@ -143,7 +143,7 @@ class PlaywrightExecutor:
self.config = config or {} self.config = config or {}
self.timeout: int = self.config.get("timeout", 30000) self.timeout: int = self.config.get("timeout", 30000)
self.screenshot_enabled: bool = self.config.get("screenshot", True) self.screenshot_enabled: bool = self.config.get("screenshot", True)
self.headless: bool = self.config.get("headless", True) self.headless: bool = self.config.get("headless", False)
self.screenshot_dir: str = self.config.get( self.screenshot_dir: str = self.config.get(
"screenshot_dir", settings.SCREENSHOT_DIR "screenshot_dir", settings.SCREENSHOT_DIR
) )
......
...@@ -27,7 +27,7 @@ from fastapi.responses import FileResponse ...@@ -27,7 +27,7 @@ from fastapi.responses import FileResponse
from app.config import settings from app.config import settings
from app.database import init_db from app.database import init_db
from app.routers import modules, cases, executions, recorder, stats, reports, cleanup from app.routers import modules, cases, executions, recorder, stats, reports, cleanup, batch, dependencies
# 配置日志 # 配置日志
logging.basicConfig( logging.basicConfig(
...@@ -128,6 +128,18 @@ app.include_router( ...@@ -128,6 +128,18 @@ app.include_router(
tags=["数据清理"] tags=["数据清理"]
) )
app.include_router(
batch.router,
prefix="/api/batch",
tags=["批量操作"]
)
app.include_router(
dependencies.router,
prefix="/api",
tags=["依赖管理"]
)
# ==================== 根路径 ==================== # ==================== 根路径 ====================
......
...@@ -12,10 +12,12 @@ from app.models.module import Module ...@@ -12,10 +12,12 @@ from app.models.module import Module
from app.models.test_case import TestCase from app.models.test_case import TestCase
from app.models.execution import Execution from app.models.execution import Execution
from app.models.case_result import CaseResult from app.models.case_result import CaseResult
from app.models.case_dependency import CaseDependency
__all__ = [ __all__ = [
"Module", "Module",
"TestCase", "TestCase",
"Execution", "Execution",
"CaseResult", "CaseResult",
"CaseDependency",
] ]
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:case_dependency.py
模块描述:用例依赖关系数据模型
作者:czj
创建日期:2026-07-12
"""
from datetime import datetime
from sqlalchemy import String, Integer, DateTime, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base
class CaseDependency(Base):
"""
用例依赖关系模型
定义测试用例之间的依赖关系,支持拓扑排序和条件执行。
Attributes:
id (str): 依赖关系唯一标识
case_id (str): 用例ID(依赖方)
depends_on_case_id (str): 被依赖的用例ID
dependency_type (str): 依赖类型 (hard=强依赖/soft=弱依赖)
order (int): 依赖顺序,当有多个依赖时决定优先级
created_at (datetime): 创建时间
依赖类型说明:
- hard: 强依赖,前置用例必须成功,否则跳过执行
- soft: 弱依赖,前置用例失败仅记录警告,仍继续执行
Example:
>>> dependency = CaseDependency(
... case_id="case_002",
... depends_on_case_id="case_001",
... dependency_type="hard"
... )
"""
__tablename__ = "case_dependencies"
id: Mapped[str] = mapped_column(String(32), primary_key=True, comment="依赖关系ID")
case_id: Mapped[str] = mapped_column(
String(32),
ForeignKey("test_cases.id", ondelete="CASCADE"),
nullable=False,
index=True,
comment="用例ID(依赖方)"
)
depends_on_case_id: Mapped[str] = mapped_column(
String(32),
ForeignKey("test_cases.id", ondelete="CASCADE"),
nullable=False,
index=True,
comment="被依赖的用例ID"
)
dependency_type: Mapped[str] = mapped_column(
String(20),
default="hard",
comment="依赖类型: hard(强依赖)/soft(弱依赖)"
)
order: Mapped[int] = mapped_column(
Integer,
default=0,
comment="依赖顺序(优先级)"
)
created_at: Mapped[datetime] = mapped_column(
DateTime,
default=datetime.utcnow,
comment="创建时间"
)
# 关联关系
case: Mapped["TestCase"] = relationship(
"TestCase",
foreign_keys=[case_id],
back_populates="dependencies"
)
depends_on_case: Mapped["TestCase"] = relationship(
"TestCase",
foreign_keys=[depends_on_case_id],
back_populates="dependents"
)
def __repr__(self) -> str:
"""字符串表示"""
return f"<CaseDependency(id={self.id}, case_id={self.case_id}, depends_on={self.depends_on_case_id})>"
def to_dict(self) -> dict:
"""
转换为字典
Returns:
dict: 依赖关系数据字典
"""
return {
"id": self.id,
"case_id": self.case_id,
"depends_on_case_id": self.depends_on_case_id,
"dependency_type": self.dependency_type,
"order": self.order,
"created_at": self.created_at.isoformat() if self.created_at else None,
}
...@@ -11,7 +11,7 @@ ...@@ -11,7 +11,7 @@
from datetime import datetime from datetime import datetime
from typing import Optional from typing import Optional
from sqlalchemy import String, Text, JSON, ForeignKey, DateTime from sqlalchemy import String, Text, JSON, ForeignKey, DateTime, Integer
from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base from app.database import Base
...@@ -72,6 +72,14 @@ class TestCase(Base): ...@@ -72,6 +72,14 @@ class TestCase(Base):
tags: Mapped[list] = mapped_column(JSON, default=list, comment="标签列表") tags: Mapped[list] = mapped_column(JSON, default=list, comment="标签列表")
steps: Mapped[list] = mapped_column(JSON, default=list, comment="测试步骤") steps: Mapped[list] = mapped_column(JSON, default=list, comment="测试步骤")
config: Mapped[dict] = mapped_column(JSON, default=dict, comment="执行配置") config: Mapped[dict] = mapped_column(JSON, default=dict, comment="执行配置")
# 新增字段:依赖管理和参数化支持
depends_on: Mapped[list] = mapped_column(JSON, default=list, comment="依赖用例ID列表")
parameters: Mapped[dict] = mapped_column(JSON, default=dict, comment="参数定义")
condition: Mapped[dict] = mapped_column(JSON, default=dict, comment="执行条件")
order: Mapped[int] = mapped_column(Integer, default=0, comment="执行顺序")
version: Mapped[int] = mapped_column(Integer, default=1, comment="版本号")
created_at: Mapped[datetime] = mapped_column( created_at: Mapped[datetime] = mapped_column(
DateTime, DateTime,
default=datetime.utcnow, default=datetime.utcnow,
...@@ -90,6 +98,22 @@ class TestCase(Base): ...@@ -90,6 +98,22 @@ class TestCase(Base):
back_populates="cases" back_populates="cases"
) )
# 关联依赖关系(一对多)
dependencies: Mapped[list["CaseDependency"]] = relationship(
"CaseDependency",
foreign_keys="CaseDependency.case_id",
back_populates="case",
cascade="all, delete-orphan"
)
# 关联被依赖关系(一对多)
dependents: Mapped[list["CaseDependency"]] = relationship(
"CaseDependency",
foreign_keys="CaseDependency.depends_on_case_id",
back_populates="depends_on_case",
cascade="all, delete-orphan"
)
def __repr__(self) -> str: def __repr__(self) -> str:
"""字符串表示""" """字符串表示"""
return f"<TestCase(id={self.id}, name={self.name}, status={self.status})>" return f"<TestCase(id={self.id}, name={self.name}, status={self.status})>"
...@@ -111,6 +135,11 @@ class TestCase(Base): ...@@ -111,6 +135,11 @@ class TestCase(Base):
"tags": self.tags or [], "tags": self.tags or [],
"steps": self.steps or [], "steps": self.steps or [],
"config": self.config or {}, "config": self.config or {},
"depends_on": self.depends_on or [],
"parameters": self.parameters or {},
"condition": self.condition or {},
"order": self.order,
"version": self.version,
"created_at": self.created_at.isoformat() if self.created_at else None, "created_at": self.created_at.isoformat() if self.created_at else None,
"updated_at": self.updated_at.isoformat() if self.updated_at else None, "updated_at": self.updated_at.isoformat() if self.updated_at else None,
} }
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:batch.py
模块描述:批量操作 API 路由
作者:czj
创建日期:2026-07-12
"""
import logging
from typing import List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.services.batch_service import BatchService
from app.schemas.batch import (
BatchDeleteRequest,
BatchUpdateRequest,
BatchCopyRequest,
BatchAssignTagsRequest,
BatchSetPriorityRequest,
BatchOperationResponse,
)
from app.schemas.test_case import TestCaseResponse
logger = logging.getLogger(__name__)
router = APIRouter()
def get_batch_service(db: AsyncSession = Depends(get_db)) -> BatchService:
"""获取批量操作服务"""
return BatchService(db)
@router.post("/delete", response_model=BatchOperationResponse)
async def batch_delete_cases(
request: BatchDeleteRequest,
service: BatchService = Depends(get_batch_service),
):
"""
批量删除测试用例
Args:
request: 批量删除请求,包含用例ID列表
Returns:
BatchOperationResponse: 操作结果统计
"""
try:
result = await service.batch_delete(request.case_ids)
return BatchOperationResponse(
success=result["success"],
failed=result["failed"],
total=result["total"],
message=f"成功删除 {result['success']} 个用例"
)
except Exception as e:
logger.error(f"批量删除失败: {str(e)}")
raise HTTPException(status_code=500, detail=f"批量删除失败: {str(e)}")
@router.post("/update", response_model=BatchOperationResponse)
async def batch_update_cases(
request: BatchUpdateRequest,
service: BatchService = Depends(get_batch_service),
):
"""
批量更新测试用例
Args:
request: 批量更新请求,包含用例ID列表和更新字段
Returns:
BatchOperationResponse: 操作结果统计
"""
try:
result = await service.batch_update(request.case_ids, request.updates)
return BatchOperationResponse(
success=result["success"],
failed=result["failed"],
total=result["total"],
message=f"成功更新 {result['success']} 个用例"
)
except ValueError as e:
logger.warning(f"批量更新参数错误: {str(e)}")
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
logger.error(f"批量更新失败: {str(e)}")
raise HTTPException(status_code=500, detail=f"批量更新失败: {str(e)}")
@router.post("/copy", response_model=List[TestCaseResponse])
async def batch_copy_cases(
request: BatchCopyRequest,
service: BatchService = Depends(get_batch_service),
):
"""
批量复制测试用例
Args:
request: 批量复制请求,包含用例ID列表和目标模块ID
Returns:
List[TestCaseResponse]: 新创建的用例列表
"""
try:
new_cases, stats = await service.batch_copy(
request.case_ids,
request.target_module_id
)
logger.info(f"批量复制完成: {stats}")
return [TestCaseResponse.model_validate(c) for c in new_cases]
except Exception as e:
logger.error(f"批量复制失败: {str(e)}")
raise HTTPException(status_code=500, detail=f"批量复制失败: {str(e)}")
@router.post("/assign-tags", response_model=BatchOperationResponse)
async def batch_assign_tags(
request: BatchAssignTagsRequest,
service: BatchService = Depends(get_batch_service),
):
"""
批量分配标签
Args:
request: 批量分配标签请求
Returns:
BatchOperationResponse: 操作结果统计
"""
try:
result = await service.batch_assign_tags(
request.case_ids,
request.tag_names
)
return BatchOperationResponse(
success=result["success"],
failed=result["failed"],
total=result["total"],
message=f"成功分配标签到 {result['success']} 个用例"
)
except Exception as e:
logger.error(f"批量分配标签失败: {str(e)}")
raise HTTPException(status_code=500, detail=f"批量分配标签失败: {str(e)}")
@router.post("/set-priority", response_model=BatchOperationResponse)
async def batch_set_priority(
request: BatchSetPriorityRequest,
service: BatchService = Depends(get_batch_service),
):
"""
批量设置优先级
Args:
request: 批量设置优先级请求
Returns:
BatchOperationResponse: 操作结果统计
"""
try:
result = await service.batch_set_priority(
request.case_ids,
request.priority
)
return BatchOperationResponse(
success=result["success"],
failed=result["failed"],
total=result["total"],
message=f"成功设置 {result['success']} 个用例的优先级"
)
except ValueError as e:
logger.warning(f"批量设置优先级参数错误: {str(e)}")
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
logger.error(f"批量设置优先级失败: {str(e)}")
raise HTTPException(status_code=500, detail=f"批量设置优先级失败: {str(e)}")
@router.post("/enable", response_model=BatchOperationResponse)
async def batch_enable_cases(
case_ids: List[str],
service: BatchService = Depends(get_batch_service),
):
"""
批量启用用例
Args:
case_ids: 用例ID列表
Returns:
BatchOperationResponse: 操作结果统计
"""
try:
result = await service.batch_enable(case_ids)
return BatchOperationResponse(
success=result["success"],
failed=result["failed"],
total=result["total"],
message=f"成功启用 {result['success']} 个用例"
)
except Exception as e:
logger.error(f"批量启用失败: {str(e)}")
raise HTTPException(status_code=500, detail=f"批量启用失败: {str(e)}")
@router.post("/disable", response_model=BatchOperationResponse)
async def batch_disable_cases(
case_ids: List[str],
service: BatchService = Depends(get_batch_service),
):
"""
批量禁用用例
Args:
case_ids: 用例ID列表
Returns:
BatchOperationResponse: 操作结果统计
"""
try:
result = await service.batch_disable(case_ids)
return BatchOperationResponse(
success=result["success"],
failed=result["failed"],
total=result["total"],
message=f"成功禁用 {result['success']} 个用例"
)
except Exception as e:
logger.error(f"批量禁用失败: {str(e)}")
raise HTTPException(status_code=500, detail=f"批量禁用失败: {str(e)}")
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:dependencies.py
模块描述:依赖管理 API 路由
作者:czj
创建日期:2026-07-12
"""
import logging
from typing import List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.services.dependency_service import DependencyService
from app.schemas.dependency import (
DependencyResponse,
SetDependenciesRequest,
ValidateDependenciesResponse,
ExecutionOrderResponse,
DependencyStatusResponse,
)
logger = logging.getLogger(__name__)
router = APIRouter()
def get_dependency_service(db: AsyncSession = Depends(get_db)) -> DependencyService:
"""获取依赖管理服务"""
return DependencyService(db)
@router.get("/cases/{case_id}/dependencies", response_model=List[DependencyResponse])
async def get_case_dependencies(
case_id: str,
service: DependencyService = Depends(get_dependency_service),
):
"""
获取用例的所有依赖关系
Args:
case_id: 用例ID
Returns:
List[DependencyResponse]: 依赖关系列表
"""
try:
dependencies = await service.get_case_dependencies(case_id)
return [DependencyResponse.model_validate(d) for d in dependencies]
except Exception as e:
logger.error(f"获取依赖关系失败: {str(e)}")
raise HTTPException(status_code=500, detail=f"获取依赖关系失败: {str(e)}")
@router.post("/cases/{case_id}/dependencies", response_model=dict)
async def set_case_dependencies(
case_id: str,
request: SetDependenciesRequest,
service: DependencyService = Depends(get_dependency_service),
):
"""
设置用例依赖关系
Args:
case_id: 用例ID
request: 依赖定义列表
Returns:
dict: 操作结果
"""
try:
# 转换为字典列表
dependencies = [dep.model_dump() for dep in request.dependencies]
success = await service.set_case_dependencies(case_id, dependencies)
return {
"success": success,
"message": f"成功设置 {len(dependencies)} 个依赖关系"
}
except ValueError as e:
logger.warning(f"设置依赖关系参数错误: {str(e)}")
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
logger.error(f"设置依赖关系失败: {str(e)}")
raise HTTPException(status_code=500, detail=f"设置依赖关系失败: {str(e)}")
@router.post("/validate", response_model=ValidateDependenciesResponse)
async def validate_dependencies(
case_ids: List[str],
service: DependencyService = Depends(get_dependency_service),
):
"""
验证依赖关系是否有效(检测循环依赖)
Args:
case_ids: 用例ID列表
Returns:
ValidateDependenciesResponse: 验证结果
"""
try:
result = await service.validate_dependencies(case_ids)
return ValidateDependenciesResponse(**result)
except Exception as e:
logger.error(f"验证依赖关系失败: {str(e)}")
raise HTTPException(status_code=500, detail=f"验证依赖关系失败: {str(e)}")
@router.post("/execution-order", response_model=ExecutionOrderResponse)
async def get_execution_order(
case_ids: List[str],
service: DependencyService = Depends(get_dependency_service),
):
"""
计算执行顺序(拓扑排序 + 优先级排序)
Args:
case_ids: 用例ID列表
Returns:
ExecutionOrderResponse: 执行顺序
"""
try:
ordered_ids, cycle = await service.topological_sort(case_ids, respect_priority=True)
return ExecutionOrderResponse(
ordered_case_ids=ordered_ids,
has_cycle=cycle is not None,
cycle_path=cycle
)
except Exception as e:
logger.error(f"计算执行顺序失败: {str(e)}")
raise HTTPException(status_code=500, detail=f"计算执行顺序失败: {str(e)}")
@router.get("/cases/{case_id}/dependency-status", response_model=DependencyStatusResponse)
async def check_dependency_status(
case_id: str,
execution_id: str,
service: DependencyService = Depends(get_dependency_service),
):
"""
检查用例依赖状态
Args:
case_id: 用例ID
execution_id: 执行ID
Returns:
DependencyStatusResponse: 依赖状态
"""
try:
result = await service.check_dependency_status(case_id, execution_id)
return DependencyStatusResponse(**result)
except Exception as e:
logger.error(f"检查依赖状态失败: {str(e)}")
raise HTTPException(status_code=500, detail=f"检查依赖状态失败: {str(e)}")
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:batch.py
模块描述:批量操作相关 Pydantic 模式定义
作者:czj
创建日期:2026-07-12
"""
from typing import List, Dict, Any, Optional
from pydantic import BaseModel, Field
class BatchDeleteRequest(BaseModel):
"""
批量删除请求模式
用于批量删除测试用例。
"""
case_ids: List[str] = Field(
...,
min_length=1,
description="用例ID列表"
)
class BatchUpdateRequest(BaseModel):
"""
批量更新请求模式
用于批量更新测试用例。
Attributes:
case_ids (List[str]): 用例ID列表
updates (Dict[str, Any]): 更新字段
"""
case_ids: List[str] = Field(
...,
min_length=1,
description="用例ID列表"
)
updates: Dict[str, Any] = Field(
...,
description="更新字段(支持: status, priority, tags, order, config)"
)
class BatchCopyRequest(BaseModel):
"""
批量复制请求模式
用于批量复制测试用例。
Attributes:
case_ids (List[str]): 用例ID列表
target_module_id (Optional[str]): 目标模块ID,不指定则复制到原模块
"""
case_ids: List[str] = Field(
...,
min_length=1,
description="用例ID列表"
)
target_module_id: Optional[str] = Field(
None,
description="目标模块ID,不指定则复制到原模块"
)
class BatchAssignTagsRequest(BaseModel):
"""
批量分配标签请求模式
用于批量给用例分配标签。
"""
case_ids: List[str] = Field(
...,
min_length=1,
description="用例ID列表"
)
tag_names: List[str] = Field(
...,
min_length=1,
description="标签名称列表"
)
class BatchSetPriorityRequest(BaseModel):
"""
批量设置优先级请求模式
用于批量设置用例优先级。
"""
case_ids: List[str] = Field(
...,
min_length=1,
description="用例ID列表"
)
priority: str = Field(
...,
description="优先级: high/medium/low"
)
class BatchOperationResponse(BaseModel):
"""
批量操作响应模式
用于返回批量操作的结果统计。
"""
success: int = Field(..., description="成功数量")
failed: int = Field(..., description="失败数量")
total: int = Field(..., description="总数量")
message: Optional[str] = Field(None, description="操作消息")
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:dependency.py
模块描述:依赖管理相关 Pydantic 模式定义
作者:czj
创建日期:2026-07-12
"""
from datetime import datetime
from typing import Optional, List
from pydantic import BaseModel, Field, ConfigDict
class DependencyDefinition(BaseModel):
"""
依赖定义模式
定义单个依赖关系。
Attributes:
depends_on_case_id (str): 被依赖的用例ID
dependency_type (str): 依赖类型 (hard=强依赖/soft=弱依赖)
order (int): 依赖顺序
"""
depends_on_case_id: str = Field(..., description="被依赖的用例ID")
dependency_type: str = Field(
default="hard",
description="依赖类型: hard(强依赖)/soft(弱依赖)"
)
order: int = Field(default=0, ge=0, description="依赖顺序")
class SetDependenciesRequest(BaseModel):
"""
设置依赖请求模式
用于设置用例的依赖关系。
"""
dependencies: List[DependencyDefinition] = Field(
default_factory=list,
description="依赖定义列表"
)
class DependencyResponse(BaseModel):
"""
依赖关系响应模式
用于返回依赖关系数据给客户端。
"""
model_config = ConfigDict(from_attributes=True)
id: str = Field(..., description="依赖关系ID")
case_id: str = Field(..., description="用例ID")
depends_on_case_id: str = Field(..., description="被依赖的用例ID")
dependency_type: str = Field(default="hard", description="依赖类型")
order: int = Field(default=0, description="依赖顺序")
created_at: Optional[datetime] = Field(None, description="创建时间")
class ValidateDependenciesResponse(BaseModel):
"""
验证依赖响应模式
用于返回依赖验证结果。
"""
valid: bool = Field(..., description="依赖关系是否有效")
cycle: Optional[List[str]] = Field(None, description="检测到的循环路径")
message: str = Field(..., description="验证消息")
class ExecutionOrderResponse(BaseModel):
"""
执行顺序响应模式
用于返回拓扑排序后的执行顺序。
"""
ordered_case_ids: List[str] = Field(..., description="排序后的用例ID列表")
has_cycle: bool = Field(..., description="是否存在循环依赖")
cycle_path: Optional[List[str]] = Field(None, description="循环路径")
class DependencyStatusResponse(BaseModel):
"""
依赖状态响应模式
用于返回用例的依赖状态检查结果。
"""
can_execute: bool = Field(..., description="是否可以执行")
status: str = Field(..., description="依赖状态: ok/blocked/skipped")
failed_dependencies: List[str] = Field(default_factory=list, description="失败的依赖用例ID列表")
message: str = Field(..., description="状态消息")
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:batch_service.py
模块描述:批量操作服务
作者:czj
创建日期:2026-07-12
"""
import logging
from typing import List, Dict, Any, Tuple
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, delete, update
from app.models.test_case import TestCase
from app.models.case_result import CaseResult
from app.utils.id_generator import generate_id
logger = logging.getLogger(__name__)
class BatchService:
"""
批量操作服务
负责测试用例的批量操作,包括:
- 批量删除
- 批量更新
- 批量复制
- 批量分配标签
- 批量设置优先级
Attributes:
db (AsyncSession): 数据库会话对象
Example:
>>> service = BatchService(db)
>>> result = await service.batch_delete(["case_001", "case_002"])
"""
def __init__(self, db: AsyncSession):
"""
初始化批量操作服务
Args:
db (AsyncSession): 数据库会话对象
"""
if db is None:
raise ValueError("数据库会话不能为空")
self.db = db
async def batch_delete(self, case_ids: List[str]) -> Dict[str, int]:
"""
批量删除用例
Args:
case_ids (List[str]): 用例ID列表
Returns:
Dict[str, int]: 操作结果
{"success": 成功数量, "failed": 失败数量}
"""
try:
# 检查用例是否存在
query = select(TestCase.id).where(TestCase.id.in_(case_ids))
result = await self.db.execute(query)
existing_ids = set(result.scalars().all())
# 找出不存在的用例
missing_ids = set(case_ids) - existing_ids
if missing_ids:
logger.warning(f"以下用例不存在: {missing_ids}")
# 删除存在的用例(级联删除依赖关系和执行结果)
delete_query = delete(TestCase).where(TestCase.id.in_(existing_ids))
delete_result = await self.db.execute(delete_query)
await self.db.flush()
success_count = delete_result.rowcount
failed_count = len(case_ids) - success_count
logger.info(f"批量删除完成: 成功={success_count}, 失败={failed_count}")
return {
"success": success_count,
"failed": failed_count,
"total": len(case_ids)
}
except Exception as e:
logger.error(f"批量删除失败: {str(e)}")
raise
async def batch_update(
self,
case_ids: List[str],
updates: Dict[str, Any]
) -> Dict[str, int]:
"""
批量更新用例
Args:
case_ids (List[str]): 用例ID列表
updates (Dict[str, Any]): 更新字段
{"priority": "high", "status": "active"}
Returns:
Dict[str, int]: 操作结果
{"success": 成功数量, "failed": 失败数量}
"""
try:
# 验证更新字段
allowed_fields = {"status", "priority", "tags", "order", "config"}
invalid_fields = set(updates.keys()) - allowed_fields
if invalid_fields:
raise ValueError(f"不允许批量更新以下字段: {invalid_fields}")
# 执行批量更新
update_query = (
update(TestCase)
.where(TestCase.id.in_(case_ids))
.values(**updates)
)
update_result = await self.db.execute(update_query)
await self.db.flush()
success_count = update_result.rowcount
failed_count = len(case_ids) - success_count
logger.info(f"批量更新完成: 成功={success_count}, 失败={failed_count}")
return {
"success": success_count,
"failed": failed_count,
"total": len(case_ids)
}
except Exception as e:
logger.error(f"批量更新失败: {str(e)}")
raise
async def batch_copy(
self,
case_ids: List[str],
target_module_id: str = None
) -> Tuple[List[TestCase], Dict[str, Any]]:
"""
批量复制用例
Args:
case_ids (List[str]): 用例ID列表
target_module_id (str, optional): 目标模块ID,不指定则复制到原模块
Returns:
Tuple[List[TestCase], Dict[str, Any]]: (新用例列表, 统计信息)
"""
try:
# 查询源用例
query = select(TestCase).where(TestCase.id.in_(case_ids))
result = await self.db.execute(query)
source_cases = list(result.scalars().all())
new_cases = []
failed_count = 0
for source in source_cases:
try:
# 创建新用例
new_case = TestCase(
id=generate_id("case"),
module_id=target_module_id or source.module_id,
name=f"{source.name}(副本)",
description=source.description,
status=source.status,
priority=source.priority,
tags=source.tags.copy() if source.tags else [],
steps=source.steps.copy() if source.steps else [],
config=source.config.copy() if source.config else {},
depends_on=source.depends_on.copy() if source.depends_on else [],
parameters=source.parameters.copy() if source.parameters else {},
condition=source.condition.copy() if source.condition else {},
order=source.order,
version=1, # 新用例版本为1
)
self.db.add(new_case)
new_cases.append(new_case)
except Exception as e:
logger.error(f"复制用例失败: {source.id}, 错误: {str(e)}")
failed_count += 1
await self.db.flush()
logger.info(f"批量复制完成: 成功={len(new_cases)}, 失败={failed_count}")
stats = {
"success": len(new_cases),
"failed": failed_count,
"total": len(case_ids)
}
return new_cases, stats
except Exception as e:
logger.error(f"批量复制失败: {str(e)}")
raise
async def batch_assign_tags(
self,
case_ids: List[str],
tag_names: List[str]
) -> Dict[str, int]:
"""
批量分配标签
Args:
case_ids (List[str]): 用例ID列表
tag_names (List[str]): 标签名称列表
Returns:
Dict[str, int]: 操作结果
{"success": 成功数量, "failed": 失败数量}
"""
try:
# 查询用例
query = select(TestCase).where(TestCase.id.in_(case_ids))
result = await self.db.execute(query)
cases = list(result.scalars().all())
success_count = 0
for case in cases:
# 合并标签(去重)
existing_tags = set(case.tags or [])
new_tags = existing_tags.union(set(tag_names))
case.tags = list(new_tags)
success_count += 1
await self.db.flush()
logger.info(f"批量分配标签完成: 成功={success_count}, 总数={len(case_ids)}")
return {
"success": success_count,
"failed": len(case_ids) - success_count,
"total": len(case_ids)
}
except Exception as e:
logger.error(f"批量分配标签失败: {str(e)}")
raise
async def batch_set_priority(
self,
case_ids: List[str],
priority: str
) -> Dict[str, int]:
"""
批量设置优先级
Args:
case_ids (List[str]): 用例ID列表
priority (str): 优先级 (high/medium/low)
Returns:
Dict[str, int]: 操作结果
"""
if priority not in ["high", "medium", "low"]:
raise ValueError(f"无效的优先级: {priority}")
return await self.batch_update(case_ids, {"priority": priority})
async def batch_enable(self, case_ids: List[str]) -> Dict[str, int]:
"""
批量启用用例
Args:
case_ids (List[str]): 用例ID列表
Returns:
Dict[str, int]: 操作结果
"""
return await self.batch_update(case_ids, {"status": "active"})
async def batch_disable(self, case_ids: List[str]) -> Dict[str, int]:
"""
批量禁用用例
Args:
case_ids (List[str]): 用例ID列表
Returns:
Dict[str, int]: 操作结果
"""
return await self.batch_update(case_ids, {"status": "disabled"})
async def batch_move_to_module(
self,
case_ids: List[str],
target_module_id: str
) -> Dict[str, int]:
"""
批量移动到指定模块
Args:
case_ids (List[str]): 用例ID列表
target_module_id (str): 目标模块ID
Returns:
Dict[str, int]: 操作结果
"""
from app.models.module import Module
# 验证目标模块是否存在
query = select(Module.id).where(Module.id == target_module_id)
result = await self.db.execute(query)
if not result.scalar_one_or_none():
raise ValueError(f"目标模块不存在: {target_module_id}")
return await self.batch_update(case_ids, {"module_id": target_module_id})
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:dependency_service.py
模块描述:用例依赖管理服务
作者:czj
创建日期:2026-07-12
"""
import logging
from typing import Optional, List, Dict, Any, Tuple
from collections import defaultdict, deque
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, delete
from sqlalchemy.orm import selectinload
from app.models.test_case import TestCase
from app.models.case_dependency import CaseDependency
from app.utils.id_generator import generate_id
logger = logging.getLogger(__name__)
class DependencyService:
"""
依赖管理服务
负责测试用例依赖关系的管理,包括:
- 依赖关系的设置和验证
- 拓扑排序算法
- 循环依赖检测
- 执行顺序计算
Attributes:
db (AsyncSession): 数据库会话对象
Example:
>>> service = DependencyService(db)
>>> dependencies = await service.get_case_dependencies("case_001")
>>> order = await service.topological_sort(["case_001", "case_002", "case_003"])
"""
def __init__(self, db: AsyncSession):
"""
初始化依赖管理服务
Args:
db (AsyncSession): 数据库会话对象
"""
if db is None:
raise ValueError("数据库会话不能为空")
self.db = db
async def get_case_dependencies(self, case_id: str) -> List[CaseDependency]:
"""
获取用例的所有依赖关系
Args:
case_id (str): 用例ID
Returns:
List[CaseDependency]: 依赖关系列表
"""
query = (
select(CaseDependency)
.where(CaseDependency.case_id == case_id)
.order_by(CaseDependency.order)
)
result = await self.db.execute(query)
return list(result.scalars().all())
async def set_case_dependencies(
self,
case_id: str,
dependencies: List[Dict[str, Any]]
) -> bool:
"""
设置用例依赖关系
先删除现有依赖,再创建新的依赖关系。
Args:
case_id (str): 用例ID
dependencies (List[Dict]): 依赖定义列表
[{"depends_on_case_id": "case_001", "dependency_type": "hard", "order": 1}]
Returns:
bool: 设置是否成功
Raises:
ValueError: 当依赖关系无效时抛出
"""
# 验证依赖用例是否存在
dep_case_ids = [d["depends_on_case_id"] for d in dependencies]
if dep_case_ids:
query = select(TestCase.id).where(TestCase.id.in_(dep_case_ids))
result = await self.db.execute(query)
existing_ids = set(result.scalars().all())
missing_ids = set(dep_case_ids) - existing_ids
if missing_ids:
raise ValueError(f"依赖的用例不存在: {missing_ids}")
# 验证是否包含自依赖
if case_id in dep_case_ids:
raise ValueError("用例不能依赖自己")
# 删除现有依赖
delete_query = delete(CaseDependency).where(CaseDependency.case_id == case_id)
await self.db.execute(delete_query)
# 创建新依赖
for dep_data in dependencies:
dependency = CaseDependency(
id=generate_id("dep"),
case_id=case_id,
depends_on_case_id=dep_data["depends_on_case_id"],
dependency_type=dep_data.get("dependency_type", "hard"),
order=dep_data.get("order", 0),
)
self.db.add(dependency)
await self.db.flush()
logger.info(f"设置用例依赖成功: case_id={case_id}, count={len(dependencies)}")
return True
async def validate_dependencies(self, case_ids: List[str]) -> Dict[str, Any]:
"""
验证依赖关系是否有效(检测循环依赖)
Args:
case_ids (List[str]): 用例ID列表
Returns:
Dict[str, Any]: 验证结果
{
"valid": bool,
"cycle": Optional[List[str]],
"message": str
}
"""
# 构建依赖图
graph = await self._build_dependency_graph(case_ids)
# 检测循环
cycle = self._detect_cycle(graph)
if cycle:
return {
"valid": False,
"cycle": cycle,
"message": f"检测到循环依赖: {' -> '.join(cycle)}"
}
else:
return {
"valid": True,
"cycle": None,
"message": "依赖关系有效,无循环"
}
async def topological_sort(
self,
case_ids: List[str],
respect_priority: bool = True
) -> Tuple[List[str], Optional[List[str]]]:
"""
拓扑排序用例ID列表(Kahn算法)
Args:
case_ids (List[str]): 用例ID列表
respect_priority (bool): 是否在依赖级别内按优先级排序,默认True
Returns:
Tuple[List[str], Optional[List[str]]]: (排序结果, 检测到的循环)
如果存在循环依赖,第二个元素为循环路径
"""
if not case_ids:
return [], None
# 构建依赖图
graph = await self._build_dependency_graph(case_ids)
# 计算入度
in_degree = {case_id: 0 for case_id in case_ids}
for case_id, deps in graph.items():
for dep in deps:
if dep in in_degree:
in_degree[case_id] += 1
# 零入度队列
queue = deque([case_id for case_id, degree in in_degree.items() if degree == 0])
result = []
# 获取用例优先级(如果需要)
priority_map = {}
if respect_priority:
query = select(TestCase.id, TestCase.priority).where(TestCase.id.in_(case_ids))
result_db = await self.db.execute(query)
priority_map = {row[0]: row[1] for row in result_db.all()}
while queue:
# 在同一层级内按优先级排序
if respect_priority and len(queue) > 1:
# 转换为列表排序后再转回deque
level_nodes = list(queue)
level_nodes.sort(key=lambda x: self._priority_to_int(priority_map.get(x, "medium")))
queue = deque(level_nodes)
node = queue.popleft()
result.append(node)
# 减少依赖节点的入度
for dependent in graph.get(node, []):
if dependent in in_degree:
in_degree[dependent] -= 1
if in_degree[dependent] == 0:
queue.append(dependent)
# 检测循环
if len(result) != len(case_ids):
cycle = [case_id for case_id in case_ids if case_id not in result]
logger.warning(f"检测到循环依赖: {cycle}")
return result, cycle
return result, None
async def get_execution_order(
self,
case_ids: List[str]
) -> List[str]:
"""
计算执行顺序(拓扑排序 + 优先级排序)
1. 先拓扑排序处理依赖
2. 在依赖级别内按优先级排序
3. 返回最终执行顺序
Args:
case_ids (List[str]): 用例ID列表
Returns:
List[str]: 排序后的用例ID列表
Raises:
ValueError: 当存在循环依赖时抛出
"""
sorted_ids, cycle = await self.topological_sort(case_ids, respect_priority=True)
if cycle:
raise ValueError(f"存在循环依赖,无法确定执行顺序: {cycle}")
return sorted_ids
async def check_dependency_status(
self,
case_id: str,
execution_id: str
) -> Dict[str, Any]:
"""
检查用例依赖状态,判断是否可以执行
Args:
case_id (str): 用例ID
execution_id (str): 执行ID
Returns:
Dict[str, Any]: 依赖状态
{
"can_execute": bool,
"status": str, # ok/blocked/skipped
"failed_dependencies": List[str],
"message": str
}
"""
from app.models.case_result import CaseResult
# 获取依赖关系
dependencies = await self.get_case_dependencies(case_id)
if not dependencies:
return {
"can_execute": True,
"status": "ok",
"failed_dependencies": [],
"message": "无依赖关系"
}
# 检查每个依赖的执行结果
failed_deps = []
for dep in dependencies:
# 查询依赖用例的执行结果
query = select(CaseResult).where(
CaseResult.execution_id == execution_id,
CaseResult.case_id == dep.depends_on_case_id
)
result = await self.db.execute(query)
case_result = result.scalar_one_or_none()
if not case_result:
# 依赖用例尚未执行
if dep.dependency_type == "hard":
return {
"can_execute": False,
"status": "blocked",
"failed_dependencies": [dep.depends_on_case_id],
"message": f"依赖用例 {dep.depends_on_case_id} 尚未执行"
}
elif case_result.status == "failed":
# 依赖用例执行失败
if dep.dependency_type == "hard":
failed_deps.append(dep.depends_on_case_id)
elif case_result.status == "skipped":
# 依赖用例被跳过
if dep.dependency_type == "hard":
failed_deps.append(dep.depends_on_case_id)
if failed_deps:
return {
"can_execute": False,
"status": "blocked",
"failed_dependencies": failed_deps,
"message": f"依赖用例执行失败或被跳过: {failed_deps}"
}
else:
return {
"can_execute": True,
"status": "ok",
"failed_dependencies": [],
"message": "所有依赖满足"
}
async def _build_dependency_graph(self, case_ids: List[str]) -> Dict[str, List[str]]:
"""
构建依赖关系图
Args:
case_ids (List[str]): 用例ID列表
Returns:
Dict[str, List[str]]: 依赖图
{case_id: [depends_on_case_id_1, depends_on_case_id_2, ...]}
"""
# 查询所有依赖关系
query = select(CaseDependency).where(
CaseDependency.case_id.in_(case_ids)
)
result = await self.db.execute(query)
dependencies = list(result.scalars().all())
# 构建图
graph = defaultdict(list)
for dep in dependencies:
if dep.depends_on_case_id in case_ids:
graph[dep.case_id].append(dep.depends_on_case_id)
# 确保所有case_id都在图中
for case_id in case_ids:
if case_id not in graph:
graph[case_id] = []
return dict(graph)
def _detect_cycle(self, graph: Dict[str, List[str]]) -> Optional[List[str]]:
"""
检测依赖图中的循环(DFS)
Args:
graph (Dict[str, List[str]]): 依赖图
Returns:
Optional[List[str]]: 检测到的循环路径,无循环时返回None
"""
visited = set()
rec_stack = set()
path = []
def dfs(node: str) -> Optional[List[str]]:
visited.add(node)
rec_stack.add(node)
path.append(node)
for neighbor in graph.get(node, []):
if neighbor not in visited:
cycle = dfs(neighbor)
if cycle:
return cycle
elif neighbor in rec_stack:
# 找到循环
cycle_start = path.index(neighbor)
return path[cycle_start:] + [neighbor]
path.pop()
rec_stack.remove(node)
return None
for node in graph:
if node not in visited:
cycle = dfs(node)
if cycle:
return cycle
return None
def _priority_to_int(self, priority: str) -> int:
"""
将优先级转换为整数(用于排序)
Args:
priority (str): 优先级 (high/medium/low)
Returns:
int: 优先级数值(数值越小优先级越高)
"""
priority_map = {
"high": 0,
"medium": 1,
"low": 2
}
return priority_map.get(priority, 1)
[pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
asyncio_mode = auto
filterwarnings =
ignore::DeprecationWarning
ignore::UserWarning
addopts = -v --tb=short
markers =
unit: Unit tests
integration: Integration tests
slow: Slow running tests
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
目标平台元素选择器分析脚本
分析目标平台的页面元素,提取更精确的选择器。
"""
from playwright.sync_api import sync_playwright
import json
import sys
BASE_URL = "https://192.168.5.44"
def analyze_page(page, page_name: str):
"""分析页面元素"""
print(f"\n{'='*60}")
print(f"Analyzing: {page_name}")
print(f"{'='*60}")
result = {
"page": page_name,
"url": page.url,
"inputs": [],
"buttons": [],
"forms": [],
"tables": [],
"important_elements": []
}
# 分析输入框
inputs = page.query_selector_all('input')
for inp in inputs:
try:
info = {
"type": inp.get_attribute('type') or 'text',
"placeholder": inp.get_attribute('placeholder') or '',
"name": inp.get_attribute('name') or '',
"id": inp.get_attribute('id') or '',
"class": (inp.get_attribute('class') or '').split()[:3],
}
result["inputs"].append(info)
print(f"Input: type={info['type']}, placeholder={info['placeholder'][:30]}")
except:
pass
# 分析按钮
buttons = page.query_selector_all('button, .el-button, [role="button"]')
for btn in buttons:
try:
text = btn.inner_text().strip()[:30]
if text:
info = {
"text": text,
"class": (btn.get_attribute('class') or '').split()[:3],
"type": btn.get_attribute('type') or 'button',
}
result["buttons"].append(info)
print(f"Button: text={info['text']}")
except:
pass
# 分析表单
forms = page.query_selector_all('form, .el-form')
for form in forms:
try:
info = {
"class": (form.get_attribute('class') or '').split()[:3],
"id": form.get_attribute('id') or '',
}
result["forms"].append(info)
except:
pass
# 分析表格
tables = page.query_selector_all('table, .el-table')
for table in tables:
try:
info = {
"class": (table.get_attribute('class') or '').split()[:3],
}
result["tables"].append(info)
except:
pass
return result
def main():
"""主函数"""
all_pages_info = {}
with sync_playwright() as p:
# 启动浏览器(忽略证书错误)
browser = p.chromium.launch(
headless=True,
args=['--ignore-certificate-errors']
)
context = browser.new_context(
ignore_https_errors=True,
viewport={"width": 1920, "height": 1080}
)
page = context.new_page()
page.set_default_timeout(30000)
try:
# 1. 登录页
print("\n[1] Visiting login page...")
page.goto(BASE_URL, wait_until="networkidle")
page.wait_for_timeout(2000)
all_pages_info["login"] = analyze_page(page, "Login Page")
# 执行登录(如果需要继续分析其他页面)
# 填写表单
inputs = page.query_selector_all('input:visible')
for inp in inputs:
inp_type = inp.get_attribute('type') or 'text'
inp_placeholder = inp.get_attribute('placeholder') or ''
if inp_type == 'text' and ('用户' in inp_placeholder or '手机' in inp_placeholder):
inp.fill('admin@xty')
elif inp_type == 'password':
inp.fill('Ubains@13579')
elif '验证' in inp_placeholder:
inp.fill('csba')
# 点击登录
buttons = page.query_selector_all('button')
for btn in buttons:
if '登录' in btn.inner_text():
btn.click()
break
page.wait_for_timeout(3000)
# 处理协议弹窗
try:
dialogs = page.query_selector_all('.el-dialog button, [role="dialog"] button')
for db in dialogs:
if '确定' in db.inner_text() and db.is_visible():
db.click()
break
except:
pass
page.wait_for_timeout(5000)
# 2. 首页
print("\n[2] Visiting home page...")
all_pages_info["home"] = analyze_page(page, "Home Page")
# 3. 会议预约页
print("\n[3] Visiting meeting reservation page...")
page.goto(f"{BASE_URL}/#/meeting/reservation", wait_until="networkidle")
page.wait_for_timeout(2000)
all_pages_info["meeting_reservation"] = analyze_page(page, "Meeting Reservation")
# 4. 会议室列表
print("\n[4] Visiting meeting room list...")
page.goto(f"{BASE_URL}/#/meeting/room", wait_until="networkidle")
page.wait_for_timeout(2000)
all_pages_info["meeting_room_list"] = analyze_page(page, "Meeting Room List")
# 5. 会议列表
print("\n[5] Visiting meeting list...")
page.goto(f"{BASE_URL}/#/meeting/list", wait_until="networkidle")
page.wait_for_timeout(2000)
all_pages_info["meeting_list"] = analyze_page(page, "Meeting List")
except Exception as e:
print(f"Error: {str(e)}")
finally:
browser.close()
# 输出汇总
print("\n" + "="*60)
print("Summary")
print("="*60)
for page_name, info in all_pages_info.items():
print(f"\n{page_name}:")
print(f" Inputs: {len(info['inputs'])}")
print(f" Buttons: {len(info['buttons'])}")
print(f" Forms: {len(info['forms'])}")
print(f" Tables: {len(info['tables'])}")
return all_pages_info
if __name__ == "__main__":
result = main()
# 保存结果
output_file = "element_selectors.json"
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(result, f, indent=2, ensure_ascii=False)
print(f"\nResults saved to: {output_file}")
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
清空测试数据脚本
"""
import sys
import asyncio
from pathlib import Path
# 添加项目路径
backend_dir = Path(__file__).parent.parent
sys.path.insert(0, str(backend_dir))
from sqlalchemy import text
from app.database import engine
async def clean_data():
"""清空所有测试数据"""
print("开始清空测试数据...")
async with engine.begin() as conn:
# 按外键依赖顺序删除
print("1. 删除依赖关系...")
result = await conn.execute(text("DELETE FROM case_dependencies"))
print(f" 删除了 {result.rowcount} 条依赖关系")
print("2. 删除执行结果...")
result = await conn.execute(text("DELETE FROM case_results"))
print(f" 删除了 {result.rowcount} 条执行结果")
print("3. 删除执行记录...")
result = await conn.execute(text("DELETE FROM executions"))
print(f" 删除了 {result.rowcount} 条执行记录")
print("4. 删除测试用例...")
result = await conn.execute(text("DELETE FROM test_cases"))
print(f" 删除了 {result.rowcount} 条测试用例")
print("5. 删除模块...")
result = await conn.execute(text("DELETE FROM modules"))
print(f" 删除了 {result.rowcount} 条模块")
print("")
print("验证清空结果...")
# 验证清空结果
async with engine.begin() as conn:
result = await conn.execute(text("SELECT COUNT(*) FROM test_cases"))
case_count = result.scalar()
result = await conn.execute(text("SELECT COUNT(*) FROM modules"))
module_count = result.scalar()
result = await conn.execute(text("SELECT COUNT(*) FROM executions"))
exec_count = result.scalar()
result = await conn.execute(text("SELECT COUNT(*) FROM case_dependencies"))
dep_count = result.scalar()
print(f" 模块数: {module_count}")
print(f" 用例数: {case_count}")
print(f" 执行记录: {exec_count}")
print(f" 依赖关系: {dep_count}")
print("")
print("✅ 数据清空完成!")
if __name__ == "__main__":
asyncio.run(clean_data())
\ No newline at end of file
#!/usr/bin/env python
"""批量创建测试用例"""
import requests
test_cases = [
{
'module_id': 'module_aabcbfc81c6f4ac9978e77a6642d854c',
'name': '平台首页-基础访问验证',
'description': '验证平台首页可以正常打开并显示常用功能',
'priority': 'high',
'steps': [
{'order': 1, 'name': '打开平台首页', 'action': 'navigate', 'params': {'url': 'https://192.168.5.44/'}},
{'order': 2, 'name': '等待页面加载', 'action': 'wait', 'params': {'selector': 'body', 'timeout': 10000}},
{'order': 3, 'name': '截图保存', 'action': 'screenshot', 'params': {}}
],
'config': {'timeout': 30000, 'screenshot': True}
},
{
'module_id': 'module_6d79f66611b842acb283ffe15e87a1ed',
'name': '管理后台-页面访问验证',
'priority': 'medium',
'steps': [
{'order': 1, 'name': '打开页面', 'action': 'navigate', 'params': {'url': 'https://192.168.5.44/'}},
{'order': 2, 'name': '等待加载', 'action': 'wait', 'params': {'selector': 'body', 'timeout': 10000}},
{'order': 3, 'name': '截图', 'action': 'screenshot', 'params': {}}
],
'config': {'timeout': 30000, 'screenshot': True}
},
]
modules_for_cases = [
('module_c17732d36604472cb5f771ecc867b641', '集控控制', 'high'),
('module_4c636e50394d4d1a81f9883679246636', '会议转录', 'medium'),
('module_83e49ea4e70848bd8ce81e58bfed02c3', '信息发布', 'medium'),
('module_3520418cd5474f8f896ec2df6d99c439', '终端管理', 'medium'),
('module_4ce4e97a2feb453caf99a83eee845e20', '预定数据', 'low'),
('module_589a128205cf45418d6fcd35fa02a91f', '使用数据', 'low'),
('module_554ebfaac37e49e4a7c066152fb7f89f', '故障数据', 'low'),
('module_0f4ead2110e748769b5541bcfe443324', '运维数据', 'low'),
('module_b72acf42c27c4d62beea4c43c2d3e1e3', '运维统计', 'low'),
('module_6bcc95a2adaf47119b1fc566bbca2175', '巡检报表', 'low'),
('module_96014b5cded542569472adec3f14113f', '会议统计', 'low'),
('module_79cacf2501874f999e8d7290fb1ac8db', '维护工单', 'medium'),
('module_e705cb2eedcd4d41ba51ea7c6fa25451', '运维区域', 'medium'),
('module_1bea60dba5894b6ab6e4c7faa7e911db', '运维设备', 'medium'),
('module_1159ea4e0f5b4d07b1366542b2ef9742', '视讯设备', 'medium'),
('module_cf415bb8fe034dd88e3db41657493dec', '登录界面', 'high'),
]
for mid, mname, pri in modules_for_cases:
test_cases.append({
'module_id': mid,
'name': f'{mname}-页面访问验证',
'description': f'验证{mname}页面可以正常访问',
'priority': pri,
'steps': [
{'order': 1, 'name': '打开页面', 'action': 'navigate', 'params': {'url': 'https://192.168.5.44/'}},
{'order': 2, 'name': '等待加载', 'action': 'wait', 'params': {'selector': 'body', 'timeout': 10000}},
{'order': 3, 'name': '截图', 'action': 'screenshot', 'params': {}}
],
'config': {'timeout': 30000, 'screenshot': True}
})
created = []
for case in test_cases:
try:
resp = requests.post('http://localhost:8001/api/cases', json=case)
if resp.status_code == 201:
data = resp.json()
created.append(data['id'])
print(f'OK: {case["name"]}')
else:
print(f'FAIL: {case["name"]} - {resp.status_code}')
except Exception as e:
print(f'ERROR: {case["name"]} - {e}')
print(f'\nTotal created: {len(created)} cases')
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:create_deep_interaction_cases.py
模块描述:创建深层交互测试用例
功能:
1. 登录流程完整测试(包含表单填写、按钮点击、协议确认)
2. 会议预约功能测试(包含表单填写、下拉选择、日期选择)
3. 会议室查询测试(包含搜索、筛选、分页)
4. 数据统计验证测试(包含数据读取、图表验证)
作者:czj
创建日期:2026-07-12
"""
import sys
import os
import asyncio
import json
from pathlib import Path
# 添加项目根目录到路径
sys.path.insert(0, str(Path(__file__).parent.parent))
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.database import async_session_maker, init_db
from app.models.module import Module
from app.models.test_case import TestCase
from app.utils.id_generator import generate_id
# ==================== 深层交互用例定义 ====================
DEEP_INTERACTION_CASES = [
# ==================== 登录流程测试 ====================
{
"module_name": "登录模块",
"name": "登录流程-完整登录验证",
"description": "测试完整的登录流程,包括表单填写、按钮点击、协议确认",
"priority": "high",
"tags": ["登录", "表单", "核心功能"],
"steps": [
{
"order": 1,
"name": "访问登录页面",
"action": "navigate",
"params": {"url": "https://192.168.5.44/"},
"expected": "显示登录表单"
},
{
"order": 2,
"name": "验证登录表单可见",
"action": "assert",
"params": {
"type": "visible",
"selector": "input[type='text'], input[placeholder*='用户']"
},
"expected": "用户名输入框可见"
},
{
"order": 3,
"name": "填写用户名",
"action": "fill",
"params": {
"selector": "input[type='text']",
"value": "admin@xty"
},
"expected": "用户名填写成功"
},
{
"order": 4,
"name": "填写密码",
"action": "fill",
"params": {
"selector": "input[type='password']",
"value": "Ubains@13579"
},
"expected": "密码填写成功"
},
{
"order": 5,
"name": "填写验证码",
"action": "fill",
"params": {
"selector": "input[placeholder*='验证']",
"value": "csba"
},
"expected": "验证码填写成功"
},
{
"order": 6,
"name": "点击登录按钮",
"action": "click",
"params": {
"selector": "button:has-text('登录')"
},
"expected": "点击登录按钮"
},
{
"order": 7,
"name": "等待页面跳转",
"action": "wait",
"params": {
"selector": "body",
"timeout": 5000
},
"expected": "页面跳转完成"
},
{
"order": 8,
"name": "处理协议弹窗",
"action": "click",
"params": {
"selector": ".el-dialog button:has-text('确定'), [role='dialog'] button:has-text('确定')"
},
"expected": "确认协议弹窗"
},
{
"order": 9,
"name": "验证登录成功",
"action": "assert",
"params": {
"type": "url",
"expected": "platform"
},
"expected": "URL包含platform,登录成功"
}
],
"config": {
"timeout": 30000,
"screenshot": True,
"auto_login": False # 登录测试不自动登录
}
},
# ==================== 会议预约测试 ====================
{
"module_name": "会议预约",
"name": "会议预约-新建会议预约",
"description": "测试新建会议预约的完整流程,包括日期选择、会议室选择、信息填写",
"priority": "high",
"tags": ["会议预约", "表单", "核心功能"],
"steps": [
{
"order": 1,
"name": "导航到会议预约页面",
"action": "navigate",
"params": {"url": "https://192.168.5.44/#/meeting/reservation"},
"expected": "进入会议预约模块"
},
{
"order": 2,
"name": "验证页面标题",
"action": "assert",
"params": {
"type": "contains",
"selector": "body",
"expected": "预约"
},
"expected": "页面包含预约相关内容"
},
{
"order": 3,
"name": "点击新建预约按钮",
"action": "click",
"params": {
"selector": "button:has-text('新建'), button:has-text('预约')"
},
"expected": "打开新建预约表单"
},
{
"order": 4,
"name": "等待表单加载",
"action": "wait",
"params": {
"selector": ".el-form, form",
"timeout": 5000
},
"expected": "表单加载完成"
},
{
"order": 5,
"name": "验证表单可见",
"action": "assert",
"params": {
"type": "visible",
"selector": ".el-form-item, form-item"
},
"expected": "表单项可见"
}
],
"config": {
"timeout": 30000,
"screenshot": True,
"auto_login": True # 需要登录态
}
},
# ==================== 会议室列表测试 ====================
{
"module_name": "会议室列表",
"name": "会议室列表-搜索筛选功能",
"description": "测试会议室列表的搜索、筛选、分页功能",
"priority": "high",
"tags": ["会议室", "搜索", "筛选"],
"steps": [
{
"order": 1,
"name": "导航到会议室列表",
"action": "navigate",
"params": {"url": "https://192.168.5.44/#/meeting/room"},
"expected": "进入会议室列表页面"
},
{
"order": 2,
"name": "验证列表加载",
"action": "assert",
"params": {
"type": "visible",
"selector": ".el-table, table, .list"
},
"expected": "列表组件可见"
},
{
"order": 3,
"name": "验证搜索框",
"action": "assert",
"params": {
"type": "visible",
"selector": "input[placeholder*='搜索'], input[placeholder*='查询'], .el-input"
},
"expected": "搜索框可见"
},
{
"order": 4,
"name": "填写搜索关键词",
"action": "fill",
"params": {
"selector": "input[placeholder*='搜索'], input[placeholder*='查询']",
"value": "会议室"
},
"expected": "填写搜索关键词"
},
{
"order": 5,
"name": "点击搜索按钮",
"action": "click",
"params": {
"selector": "button:has-text('搜索'), button:has-text('查询')"
},
"expected": "执行搜索"
},
{
"order": 6,
"name": "等待搜索结果",
"action": "wait",
"params": {
"selector": ".el-table, table",
"timeout": 5000
},
"expected": "搜索结果加载完成"
}
],
"config": {
"timeout": 30000,
"screenshot": True,
"auto_login": True
}
},
# ==================== 管理看板测试 ====================
{
"module_name": "管理看板",
"name": "管理看板-数据展示验证",
"description": "验证管理看板的数据统计和图表展示功能",
"priority": "medium",
"tags": ["看板", "数据", "图表"],
"steps": [
{
"order": 1,
"name": "导航到管理看板",
"action": "navigate",
"params": {"url": "https://192.168.5.44/#/dashboard/management"},
"expected": "进入管理看板"
},
{
"order": 2,
"name": "验证数据卡片",
"action": "assert",
"params": {
"type": "visible",
"selector": ".card, .data-card, .stat-card, .el-card"
},
"expected": "数据卡片可见"
},
{
"order": 3,
"name": "验证图表组件",
"action": "assert",
"params": {
"type": "visible",
"selector": ".chart, .echarts, canvas, [class*='chart']"
},
"expected": "图表组件可见"
},
{
"order": 4,
"name": "验证数据非空",
"action": "assert",
"params": {
"type": "count",
"selector": ".card, .stat-card",
"expected": "2"
},
"expected": "至少有2个数据卡片"
}
],
"config": {
"timeout": 30000,
"screenshot": True,
"auto_login": True
}
},
# ==================== 会议列表测试 ====================
{
"module_name": "会议列表",
"name": "会议列表-列表操作验证",
"description": "测试会议列表的展示、筛选、详情查看功能",
"priority": "medium",
"tags": ["会议列表", "筛选", "详情"],
"steps": [
{
"order": 1,
"name": "导航到会议列表",
"action": "navigate",
"params": {"url": "https://192.168.5.44/#/meeting/list"},
"expected": "进入会议列表页面"
},
{
"order": 2,
"name": "验证表格加载",
"action": "assert",
"params": {
"type": "visible",
"selector": ".el-table, table"
},
"expected": "表格可见"
},
{
"order": 3,
"name": "验证筛选区域",
"action": "assert",
"params": {
"type": "visible",
"selector": ".filter, .search, .el-form"
},
"expected": "筛选区域可见"
},
{
"order": 4,
"name": "滚动查看更多数据",
"action": "scroll",
"params": {
"direction": "down",
"distance": 500
},
"expected": "滚动页面"
}
],
"config": {
"timeout": 30000,
"screenshot": True,
"auto_login": True
}
},
# ==================== 系统设置测试 ====================
{
"module_name": "集控控制",
"name": "集控控制-页面操作验证",
"description": "验证集控控制页面是否正常加载和操作",
"priority": "low",
"tags": ["设置", "系统"],
"steps": [
{
"order": 1,
"name": "导航到集控控制页面",
"action": "navigate",
"params": {"url": "https://192.168.5.44/#/system/control"},
"expected": "进入集控控制页面"
},
{
"order": 2,
"name": "验证页面内容",
"action": "assert",
"params": {
"type": "visible",
"selector": ".content, .main, .el-card"
},
"expected": "页面内容可见"
}
],
"config": {
"timeout": 30000,
"screenshot": True,
"auto_login": True
}
}
]
async def create_deep_cases():
"""创建深层交互测试用例"""
print("=" * 60)
print("Creating deep interaction test cases...")
print("=" * 60)
# 初始化数据库
await init_db()
async with async_session_maker() as db:
created_count = 0
skipped_count = 0
updated_count = 0
for case_data in DEEP_INTERACTION_CASES:
module_name = case_data["module_name"]
case_name = case_data["name"]
# 查找模块(使用名称模糊匹配)
query = select(Module).where(Module.name.contains(module_name))
result = await db.execute(query)
module = result.scalar_one_or_none()
if not module:
print(f"[WARN] Module not found, skip: {module_name}")
skipped_count += 1
continue
# 检查用例是否已存在
existing_query = select(TestCase).where(
TestCase.module_id == module.id,
TestCase.name == case_name
)
existing_result = await db.execute(existing_query)
existing_case = existing_result.scalar_one_or_none()
if existing_case:
# 更新现有用例
existing_case.steps = case_data["steps"]
existing_case.config = case_data["config"]
existing_case.description = case_data["description"]
existing_case.priority = case_data["priority"]
existing_case.tags = case_data["tags"]
print(f"[UPDATE] Update case: {case_name} (module: {module.name})")
updated_count += 1
else:
# 创建新用例
new_case = TestCase(
id=generate_id("case"),
module_id=module.id,
name=case_name,
description=case_data.get("description", ""),
status="active",
priority=case_data.get("priority", "medium"),
tags=case_data.get("tags", []),
steps=case_data["steps"],
config=case_data.get("config", {}),
)
db.add(new_case)
print(f"[CREATE] Create case: {case_name} (module: {module.name})")
created_count += 1
await db.commit()
print("\n" + "=" * 60)
print("Creation completed")
print(f" - Created: {created_count}")
print(f" - Updated: {updated_count}")
print(f" - Skipped: {skipped_count}")
print("=" * 60)
if __name__ == "__main__":
asyncio.run(create_deep_cases())
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
创建登录模块和测试用例
"""
import requests
import json
BASE_URL = "http://localhost:8001/api"
# 1. 创建登录模块
print("创建登录模块...")
module_data = {
"name": "登录模块",
"description": "统一管理平台登录功能测试",
"order": 1
}
response = requests.post(f"{BASE_URL}/modules", json=module_data)
module = response.json()
print(f"模块ID: {module['id']}")
print(f"模块名称: {module['name']}")
module_id = module['id']
# 2. 创建测试用例
test_cases = [
{
"name": "登录功能-页面访问验证",
"description": "验证登录页面是否可以正常访问",
"priority": "high",
"steps": [
{"order": 1, "name": "打开登录页面", "action": "navigate", "params": {"url": "https://192.168.5.44"}},
{"order": 2, "name": "验证页面标题", "action": "assert", "params": {"type": "title"}, "expected": "统一管理平台"},
{"order": 3, "name": "验证登录按钮存在", "action": "assert", "params": {"type": "visible", "selector": "button:has-text('登录')"}}
]
},
{
"name": "登录功能-正常登录",
"description": "使用正确的用户名和密码登录",
"priority": "high",
"steps": [
{"order": 1, "name": "打开登录页面", "action": "navigate", "params": {"url": "https://192.168.5.44"}},
{"order": 2, "name": "输入用户名", "action": "fill", "params": {"selector": "input[placeholder*='用户']", "value": "admin@xty"}},
{"order": 3, "name": "输入密码", "action": "fill", "params": {"selector": "input[type='password']", "value": "Ubains@13579"}},
{"order": 4, "name": "输入验证码", "action": "fill", "params": {"selector": "input[placeholder*='验证']", "value": "csba"}},
{"order": 5, "name": "点击登录按钮", "action": "click", "params": {"selector": "button:has-text('登录')"}},
{"order": 6, "name": "等待页面跳转", "action": "wait", "params": {"selector": "body", "timeout": 5000}},
{"order": 7, "name": "验证登录成功", "action": "assert", "params": {"type": "url"}, "expected": "platform"}
]
},
{
"name": "登录功能-用户名为空",
"description": "用户名为空时点击登录",
"priority": "medium",
"steps": [
{"order": 1, "name": "打开登录页面", "action": "navigate", "params": {"url": "https://192.168.5.44"}},
{"order": 2, "name": "输入密码", "action": "fill", "params": {"selector": "input[type='password']", "value": "Ubains@13579"}},
{"order": 3, "name": "点击登录按钮", "action": "click", "params": {"selector": "button:has-text('登录')"}},
{"order": 4, "name": "验证错误提示", "action": "assert", "params": {"type": "visible", "selector": ".el-message--error"}}
]
},
{
"name": "登录功能-密码为空",
"description": "密码为空时点击登录",
"priority": "medium",
"steps": [
{"order": 1, "name": "打开登录页面", "action": "navigate", "params": {"url": "https://192.168.5.44"}},
{"order": 2, "name": "输入用户名", "action": "fill", "params": {"selector": "input[placeholder*='用户']", "value": "admin@xty"}},
{"order": 3, "name": "点击登录按钮", "action": "click", "params": {"selector": "button:has-text('登录')"}},
{"order": 4, "name": "验证错误提示", "action": "assert", "params": {"type": "visible", "selector": ".el-message--error"}}
]
},
{
"name": "登录功能-错误密码",
"description": "使用错误的密码登录",
"priority": "medium",
"steps": [
{"order": 1, "name": "打开登录页面", "action": "navigate", "params": {"url": "https://192.168.5.44"}},
{"order": 2, "name": "输入用户名", "action": "fill", "params": {"selector": "input[placeholder*='用户']", "value": "admin@xty"}},
{"order": 3, "name": "输入错误密码", "action": "fill", "params": {"selector": "input[type='password']", "value": "WrongPassword"}},
{"order": 4, "name": "输入验证码", "action": "fill", "params": {"selector": "input[placeholder*='验证']", "value": "csba"}},
{"order": 5, "name": "点击登录按钮", "action": "click", "params": {"selector": "button:has-text('登录')"}},
{"order": 6, "name": "验证错误提示", "action": "assert", "params": {"type": "visible", "selector": ".el-message--error"}}
]
},
{
"name": "登录功能-记住密码勾选框",
"description": "验证记住密码勾选框功能",
"priority": "low",
"steps": [
{"order": 1, "name": "打开登录页面", "action": "navigate", "params": {"url": "https://192.168.5.44"}},
{"order": 2, "name": "验证勾选框存在", "action": "assert", "params": {"type": "visible", "selector": "input[type='checkbox']"}},
{"order": 3, "name": "勾选记住密码", "action": "check", "params": {"selector": "input[type='checkbox']"}}
]
},
{
"name": "登录功能-忘记密码链接",
"description": "验证忘记密码链接是否可用",
"priority": "low",
"steps": [
{"order": 1, "name": "打开登录页面", "action": "navigate", "params": {"url": "https://192.168.5.44"}},
{"order": 2, "name": "验证忘记密码链接存在", "action": "assert", "params": {"type": "visible", "selector": "text=忘记密码"}}
]
},
{
"name": "登录功能-立即注册链接",
"description": "验证立即注册链接是否可用",
"priority": "low",
"steps": [
{"order": 1, "name": "打开登录页面", "action": "navigate", "params": {"url": "https://192.168.5.44"}},
{"order": 2, "name": "验证立即注册链接存在", "action": "assert", "params": {"type": "visible", "selector": "text=立即注册"}}
]
}
]
print(f"\n开始创建 {len(test_cases)} 个测试用例...")
created_cases = []
for case_data in test_cases:
case_payload = {
"module_id": module_id,
"name": case_data["name"],
"description": case_data["description"],
"priority": case_data["priority"],
"steps": case_data["steps"]
}
response = requests.post(f"{BASE_URL}/cases", json=case_payload)
if response.status_code == 200:
case = response.json()
created_cases.append(case)
print(f"[OK] {case_data['name']} (ID: {case['id']})")
else:
print(f"[FAIL] Created failed: {case_data['name']} - {response.text}")
print(f"\n[SUCCESS] Created {len(created_cases)} test cases")
print(f"Module ID: {module_id}")
print(f"Case IDs: {[c['id'] for c in created_cases]}")
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:migration_add_dependencies.py
模块描述:数据库迁移脚本 - 添加依赖管理相关表和字段
作者:czj
创建日期:2026-07-12
"""
import asyncio
import logging
import sys
import os
from pathlib import Path
# 添加项目根目录到Python路径
sys.path.insert(0, str(Path(__file__).parent.parent))
from sqlalchemy import text
from app.database import engine, Base
from app.models.test_case import TestCase
from app.models.case_dependency import CaseDependency
from app.models.case_result import CaseResult
from app.models.execution import Execution
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
async def migrate():
"""
执行数据库迁移
添加依赖管理相关的表和字段
"""
logger.info("开始数据库迁移...")
try:
# 创建所有新表
async with engine.begin() as conn:
# 创建 case_dependencies 表
logger.info("创建 case_dependencies 表...")
await conn.execute(text("""
CREATE TABLE IF NOT EXISTS case_dependencies (
id TEXT PRIMARY KEY,
case_id TEXT NOT NULL,
depends_on_case_id TEXT NOT NULL,
dependency_type TEXT DEFAULT 'hard',
"order" INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (case_id) REFERENCES test_cases(id) ON DELETE CASCADE,
FOREIGN KEY (depends_on_case_id) REFERENCES test_cases(id) ON DELETE CASCADE
)
"""))
# 为 case_dependencies 表创建索引
logger.info("创建索引...")
await conn.execute(text("""
CREATE INDEX IF NOT EXISTS ix_case_dependencies_case_id
ON case_dependencies(case_id)
"""))
await conn.execute(text("""
CREATE INDEX IF NOT EXISTS ix_case_dependencies_depends_on_case_id
ON case_dependencies(depends_on_case_id)
"""))
# 为 test_cases 表添加新字段
logger.info("扩展 test_cases 表...")
try:
await conn.execute(text("""
ALTER TABLE test_cases ADD COLUMN depends_on JSON DEFAULT '[]'
"""))
except Exception as e:
if "duplicate column" not in str(e).lower():
logger.warning(f"添加 depends_on 字段失败: {str(e)}")
try:
await conn.execute(text("""
ALTER TABLE test_cases ADD COLUMN parameters JSON DEFAULT '{}'
"""))
except Exception as e:
if "duplicate column" not in str(e).lower():
logger.warning(f"添加 parameters 字段失败: {str(e)}")
try:
await conn.execute(text("""
ALTER TABLE test_cases ADD COLUMN condition JSON DEFAULT '{}'
"""))
except Exception as e:
if "duplicate column" not in str(e).lower():
logger.warning(f"添加 condition 字段失败: {str(e)}")
try:
await conn.execute(text("""
ALTER TABLE test_cases ADD COLUMN "order" INTEGER DEFAULT 0
"""))
except Exception as e:
if "duplicate column" not in str(e).lower():
logger.warning(f"添加 order 字段失败: {str(e)}")
try:
await conn.execute(text("""
ALTER TABLE test_cases ADD COLUMN version INTEGER DEFAULT 1
"""))
except Exception as e:
if "duplicate column" not in str(e).lower():
logger.warning(f"添加 version 字段失败: {str(e)}")
# 为 executions 表添加新字段
logger.info("扩展 executions 表...")
try:
await conn.execute(text("""
ALTER TABLE executions ADD COLUMN execution_mode TEXT DEFAULT 'sequential'
"""))
except Exception as e:
if "duplicate column" not in str(e).lower():
logger.warning(f"添加 execution_mode 字段失败: {str(e)}")
try:
await conn.execute(text("""
ALTER TABLE executions ADD COLUMN max_parallel INTEGER DEFAULT 1
"""))
except Exception as e:
if "duplicate column" not in str(e).lower():
logger.warning(f"添加 max_parallel 字段失败: {str(e)}")
try:
await conn.execute(text("""
ALTER TABLE executions ADD COLUMN param_values JSON DEFAULT '{}'
"""))
except Exception as e:
if "duplicate column" not in str(e).lower():
logger.warning(f"添加 param_values 字段失败: {str(e)}")
# 为 case_results 表添加新字段
logger.info("扩展 case_results 表...")
try:
await conn.execute(text("""
ALTER TABLE case_results ADD COLUMN param_instance JSON DEFAULT '{}'
"""))
except Exception as e:
if "duplicate column" not in str(e).lower():
logger.warning(f"添加 param_instance 字段失败: {str(e)}")
try:
await conn.execute(text("""
ALTER TABLE case_results ADD COLUMN dependency_status TEXT DEFAULT 'ok'
"""))
except Exception as e:
if "duplicate column" not in str(e).lower():
logger.warning(f"添加 dependency_status 字段失败: {str(e)}")
try:
await conn.execute(text("""
ALTER TABLE case_results ADD COLUMN condition_met BOOLEAN DEFAULT 1
"""))
except Exception as e:
if "duplicate column" not in str(e).lower():
logger.warning(f"添加 condition_met 字段失败: {str(e)}")
logger.info("✅ 数据库迁移完成!")
except Exception as e:
logger.error(f"❌ 数据库迁移失败: {str(e)}")
raise
async def rollback():
"""
回滚数据库迁移
删除添加的表和字段
"""
logger.info("开始回滚数据库迁移...")
try:
async with engine.begin() as conn:
# 删除 case_dependencies 表
logger.info("删除 case_dependencies 表...")
await conn.execute(text("""
DROP TABLE IF EXISTS case_dependencies
"""))
# 注意:SQLite 不支持 DROP COLUMN,所以字段会保留
# 如需完全回滚,需要重新创建表
logger.info("✅ 回滚完成!")
except Exception as e:
logger.error(f"❌ 回滚失败: {str(e)}")
raise
if __name__ == "__main__":
import sys
if len(sys.argv) > 1 and sys.argv[1] == "rollback":
asyncio.run(rollback())
else:
asyncio.run(migrate())
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
重新创建优化的登录模块测试用例
"""
import requests
BASE_URL = "http://localhost:8001/api"
# 获取登录模块
print("获取登录模块...")
response = requests.get(f"{BASE_URL}/modules")
modules = response.json()['items']
login_module = None
for m in modules:
if 'login' in m['name'].lower() or '登录' in m['name']:
login_module = m
break
if not login_module:
print("创建登录模块...")
response = requests.post(f"{BASE_URL}/modules", json={
"name": "登录模块",
"description": "统一管理平台登录功能测试",
"order": 1
})
login_module = response.json()
module_id = login_module['id']
print(f"登录模块ID: {module_id}")
# 删除旧用例
print("\n删除旧用例...")
response = requests.get(f"{BASE_URL}/cases?module_id={module_id}")
old_cases = response.json()['items']
for case in old_cases:
requests.delete(f"{BASE_URL}/cases/{case['id']}")
print(f" 删除: {case['name']}")
print(f"\n创建了 {len(old_cases)} 个旧用例")
# 创建优化的测试用例
print("\n创建新的优化用例...")
test_cases = [
{
"name": "登录功能-页面访问验证",
"description": "验证登录页面是否可以正常访问",
"priority": "high",
"steps": [
{"order": 1, "name": "打开登录页面", "action": "navigate",
"params": {"url": "https://192.168.5.44"}},
{"order": 2, "name": "等待页面加载", "action": "wait",
"params": {"selector": "body", "timeout": 10000}},
{"order": 3, "name": "验证页面标题", "action": "assert",
"params": {"type": "title"}, "expected": "统一管理平台"}
],
"config": {"timeout": 60000, "retry": 0, "screenshot": True, "auto_login": False}
},
{
"name": "登录功能-正常登录",
"description": "使用正确的用户名和密码登录",
"priority": "high",
"steps": [
{"order": 1, "name": "打开登录页面", "action": "navigate",
"params": {"url": "https://192.168.5.44"}},
{"order": 2, "name": "等待登录表单加载", "action": "wait",
"params": {"selector": "input", "timeout": 10000}},
{"order": 3, "name": "输入用户名", "action": "fill",
"params": {"selector": "input[type='text']", "value": "admin@xty"}},
{"order": 4, "name": "输入密码", "action": "fill",
"params": {"selector": "input[type='password']", "value": "Ubains@13579"}},
{"order": 5, "name": "输入验证码", "action": "fill",
"params": {"selector": "input[type='text']:nth-of-type(3)", "value": "csba"}},
{"order": 6, "name": "点击登录按钮", "action": "click",
"params": {"selector": "button"}},
{"order": 7, "name": "等待页面跳转", "action": "wait",
"params": {"selector": "body", "timeout": 5000}},
{"order": 8, "name": "验证登录成功", "action": "assert",
"params": {"type": "url"}, "expected": "platform"}
],
"config": {"timeout": 60000, "retry": 0, "screenshot": True, "auto_login": False}
},
{
"name": "登录功能-用户名为空",
"description": "用户名为空时点击登录",
"priority": "medium",
"steps": [
{"order": 1, "name": "打开登录页面", "action": "navigate",
"params": {"url": "https://192.168.5.44"}},
{"order": 2, "name": "等待页面加载", "action": "wait",
"params": {"selector": "input[type='password']", "timeout": 10000}},
{"order": 3, "name": "输入密码", "action": "fill",
"params": {"selector": "input[type='password']", "value": "Ubains@13579"}},
{"order": 4, "name": "点击登录按钮", "action": "click",
"params": {"selector": "button"}},
{"order": 5, "name": "验证错误提示", "action": "wait",
"params": {"selector": ".el-message", "timeout": 5000}}
],
"config": {"timeout": 60000, "retry": 0, "screenshot": True, "auto_login": False}
},
{
"name": "登录功能-密码为空",
"description": "密码为空时点击登录",
"priority": "medium",
"steps": [
{"order": 1, "name": "打开登录页面", "action": "navigate",
"params": {"url": "https://192.168.5.44"}},
{"order": 2, "name": "等待页面加载", "action": "wait",
"params": {"selector": "input[type='text']", "timeout": 10000}},
{"order": 3, "name": "输入用户名", "action": "fill",
"params": {"selector": "input[type='text']", "value": "admin@xty"}},
{"order": 4, "name": "点击登录按钮", "action": "click",
"params": {"selector": "button"}},
{"order": 5, "name": "验证错误提示", "action": "wait",
"params": {"selector": ".el-message", "timeout": 5000}}
],
"config": {"timeout": 60000, "retry": 0, "screenshot": True, "auto_login": False}
},
{
"name": "登录功能-错误密码",
"description": "使用错误的密码登录",
"priority": "medium",
"steps": [
{"order": 1, "name": "打开登录页面", "action": "navigate",
"params": {"url": "https://192.168.5.44"}},
{"order": 2, "name": "等待页面加载", "action": "wait",
"params": {"selector": "input", "timeout": 10000}},
{"order": 3, "name": "输入用户名", "action": "fill",
"params": {"selector": "input[type='text']", "value": "admin@xty"}},
{"order": 4, "name": "输入错误密码", "action": "fill",
"params": {"selector": "input[type='password']", "value": "WrongPassword"}},
{"order": 5, "name": "输入验证码", "action": "fill",
"params": {"selector": "input[type='text']:nth-of-type(3)", "value": "csba"}},
{"order": 6, "name": "点击登录按钮", "action": "click",
"params": {"selector": "button"}},
{"order": 7, "name": "验证错误提示", "action": "wait",
"params": {"selector": ".el-message", "timeout": 5000}}
],
"config": {"timeout": 60000, "retry": 0, "screenshot": True, "auto_login": False}
},
{
"name": "登录功能-记住密码勾选框",
"description": "验证记住密码勾选框功能",
"priority": "low",
"steps": [
{"order": 1, "name": "打开登录页面", "action": "navigate",
"params": {"url": "https://192.168.5.44"}},
{"order": 2, "name": "等待页面加载", "action": "wait",
"params": {"selector": "input[type='checkbox']", "timeout": 10000}},
{"order": 3, "name": "验证勾选框存在", "action": "assert",
"params": {"type": "visible", "selector": "input[type='checkbox']"}},
{"order": 4, "name": "勾选记住密码", "action": "check",
"params": {"selector": "input[type='checkbox']"}}
],
"config": {"timeout": 60000, "retry": 0, "screenshot": True, "auto_login": False}
},
{
"name": "登录功能-忘记密码链接",
"description": "验证忘记密码链接是否可用",
"priority": "low",
"steps": [
{"order": 1, "name": "打开登录页面", "action": "navigate",
"params": {"url": "https://192.168.5.44"}},
{"order": 2, "name": "等待页面加载", "action": "wait",
"params": {"selector": "body", "timeout": 10000}},
{"order": 3, "name": "验证忘记密码链接存在", "action": "assert",
"params": {"type": "visible", "selector": "text=忘记密码"}}
],
"config": {"timeout": 60000, "retry": 0, "screenshot": True, "auto_login": False}
},
{
"name": "登录功能-立即注册链接",
"description": "验证立即注册链接是否可用",
"priority": "low",
"steps": [
{"order": 1, "name": "打开登录页面", "action": "navigate",
"params": {"url": "https://192.168.5.44"}},
{"order": 2, "name": "等待页面加载", "action": "wait",
"params": {"selector": "body", "timeout": 10000}},
{"order": 3, "name": "验证立即注册链接存在", "action": "assert",
"params": {"type": "visible", "selector": "text=立即注册"}}
],
"config": {"timeout": 60000, "retry": 0, "screenshot": True, "auto_login": False}
}
]
created_count = 0
for case_data in test_cases:
case_payload = {
"module_id": module_id,
"name": case_data["name"],
"description": case_data["description"],
"priority": case_data["priority"],
"steps": case_data["steps"],
"config": case_data["config"]
}
response = requests.post(f"{BASE_URL}/cases", json=case_payload)
if response.status_code == 200:
created_count += 1
print(f" [OK] {case_data['name']}")
else:
print(f" [FAIL] {case_data['name']} - {response.text}")
print(f"\n[SUCCESS] 创建了 {created_count}/{len(test_cases)} 个测试用例")
print(f"模块ID: {module_id}")
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
独立测试执行脚本
在独立进程中运行 Playwright 测试,避免 asyncio 环境冲突
用法:
python run_case_worker.py <execution_id> <case_id>
"""
import asyncio
import json
import os
import sys
from datetime import datetime
# 添加项目路径
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from playwright.sync_api import sync_playwright
from app.database import async_session_maker
from app.models.execution import Execution
from app.models.case_result import CaseResult
from app.models.test_case import TestCase
from app.executors.playwright_executor import PlaywrightExecutor
from sqlalchemy import select
async def run_case(execution_id: str, case_id: str):
"""执行单个测试用例"""
async with async_session_maker() as db:
# 获取用例结果
result_query = select(CaseResult).where(
CaseResult.case_id == case_id,
CaseResult.execution_id == execution_id
)
res = await db.execute(result_query)
case_result = res.scalar_one_or_none()
if not case_result:
print(f"Error: Case result not found for {case_id}/{execution_id}")
return
# 获取用例定义
case_query = select(TestCase).where(TestCase.id == case_id)
case_res = await db.execute(case_query)
case = case_res.scalar_one_or_none()
if not case:
case_result.status = "skipped"
case_result.error_message = "用例定义不存在"
await db.flush()
await db.commit()
return
# 更新状态为运行中
case_result.status = "running"
case_result.start_time = datetime.now()
await db.flush()
await db.commit()
# 创建执行器并执行
executor = PlaywrightExecutor({'headless': True, 'timeout': 30000, 'screenshot': True})
try:
result = executor.execute_case(case=case.to_dict())
# 更新结果
case_result.status = result.status
case_result.duration = result.duration
case_result.end_time = datetime.now()
case_result.retry_count = result.retry_count
case_result.error_message = result.error_message
case_result.steps_result = [
{
"order": s.order,
"name": s.name,
"action": s.action,
"status": s.status,
"duration": s.duration,
"screenshot": s.screenshot,
"log": s.log,
"error": s.error,
}
for s in result.steps_result
]
await db.flush()
await db.commit()
print(f"DONE: {case.name} - {result.status} ({result.duration:.2f}s)")
except Exception as e:
case_result.status = "failed"
case_result.error_message = f"执行异常: {str(e)}"
case_result.end_time = datetime.now()
await db.flush()
await db.commit()
print(f"ERROR: {case.name}: {e}")
finally:
executor.stop()
await db.close()
if __name__ == "__main__":
if len(sys.argv) < 3:
print("Usage: python run_case_worker.py <execution_id> <case_id>")
sys.exit(1)
execution_id = sys.argv[1]
case_id = sys.argv[2]
print(f"Running case {case_id} for execution {execution_id}")
asyncio.run(run_case(execution_id, case_id))
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
独立的测试用例执行脚本
完全独立于FastAPI,直接通过子进程运行Playwright
"""
import json
import os
import sys
import time
def run_test(case_steps, screenshots_dir):
"""运行Playwright测试步骤"""
from playwright.sync_api import sync_playwright
results = []
with sync_playwright() as p:
browser = p.chromium.launch(headless=True, args=['--no-sandbox', '--disable-setuid-sandbox'])
context = browser.new_context(ignore_https_errors=True, viewport={'width': 1920, 'height': 1080})
page = context.new_page()
page.set_default_timeout(30000)
os.makedirs(screenshots_dir, exist_ok=True)
for step in case_steps:
order = step['order']
name = step['name']
action = step['action']
params = step.get('params', {})
step_result = {
'order': order,
'name': name,
'action': action,
'status': 'running',
'log': '',
'error': None,
'screenshot': None,
'duration': 0
}
start = time.time()
try:
if action == 'navigate':
url = params.get('url', '')
page.goto(url, wait_until='networkidle', timeout=30000)
step_result['log'] = f'Nav to {url}'
step_result['status'] = 'passed'
elif action == 'click':
selector = params.get('selector', '')
page.click(selector, timeout=30000)
step_result['log'] = f'Clicked {selector}'
step_result['status'] = 'passed'
elif action == 'fill':
selector = params.get('selector', '')
value = params.get('value', '')
page.fill(selector, str(value))
step_result['log'] = f'Filled {selector} = {value}'
step_result['status'] = 'passed'
elif action == 'wait':
selector = params.get('selector', '')
timeout = params.get('timeout', 30000)
page.wait_for_selector(selector, timeout=timeout)
step_result['log'] = f'Waited for {selector}'
step_result['status'] = 'passed'
elif action == 'screenshot':
screenshot_path = os.path.join(screenshots_dir, f'step_{order}_{int(time.time())}.png')
page.screenshot(path=screenshot_path, full_page=False)
step_result['log'] = 'Screenshot taken'
step_result['screenshot'] = screenshot_path
step_result['status'] = 'passed'
elif action == 'assert':
assert_type = params.get('type', 'visible')
selector = params.get('selector', '')
expected = params.get('expected', '')
if assert_type == 'visible':
el = page.wait_for_selector(selector, state='visible', timeout=10000)
if not el:
raise Exception(f'Element not visible: {selector}')
elif assert_type == 'url':
if expected not in page.url:
raise Exception(f'URL does not contain {expected}')
elif assert_type == 'title':
if expected not in page.title():
raise Exception(f'Title does not contain {expected}')
step_result['log'] = f'Assert {assert_type} passed'
step_result['status'] = 'passed'
else:
raise Exception(f'Unknown action: {action}')
except Exception as e:
step_result['status'] = 'failed'
step_result['error'] = str(e)
step_result['log'] = f'Failed: {str(e)}'
finally:
step_result['duration'] = time.time() - start
results.append(step_result)
# 将后续步骤标记为 skipped
if step_result['status'] == 'failed':
for remaining in case_steps:
if remaining['order'] > step_result['order']:
results.append({
'order': remaining['order'],
'name': remaining['name'],
'action': remaining['action'],
'status': 'skipped',
'log': 'Skipped due to previous failure',
'error': None,
'screenshot': None,
'duration': 0
})
break
browser.close()
return results
if __name__ == '__main__':
# 从stdin读取JSON数据
input_data = json.loads(sys.stdin.read())
case_steps = input_data['steps']
screenshots_dir = input_data.get('screenshots_dir', './data/screenshots')
try:
results = run_test(case_steps, screenshots_dir)
print(json.dumps({'status': 'ok', 'results': results}, ensure_ascii=False))
except Exception as e:
print(json.dumps({'status': 'error', 'error': str(e)}, ensure_ascii=False))
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
数据清理功能测试脚本
测试清理服务的各个功能。
"""
import asyncio
import httpx
import json
from datetime import datetime
API_BASE = "http://localhost:8001"
async def test_cleanup():
"""测试清理功能"""
async with httpx.AsyncClient(timeout=30.0) as client:
print("=" * 60)
print("Cleanup Service Test")
print("=" * 60)
# 1. 获取存储统计
print("\n[Step 1] Getting storage stats...")
response = await client.get(f"{API_BASE}/api/cleanup/stats")
stats = response.json()["data"]
print(f" Screenshots: {stats['screenshots']['count']} files, {stats['screenshots']['size']/1024/1024:.2f} MB")
print(f" Reports: {stats['reports']['count']} files, {stats['reports']['size']/1024/1024:.2f} MB")
print(f" Database: {stats['database']['size']/1024:.1f} KB")
print(f" Executions: {stats['database']['executions']}")
print(f" Case Results: {stats['database']['case_results']}")
# 2. 模拟清理
print("\n[Step 2] Testing dry run cleanup (days=7)...")
response = await client.post(
f"{API_BASE}/api/cleanup/cleanup",
params={"days": 7, "dry_run": True}
)
result = response.json()
print(f" Success: {result['success']}")
print(f" Message: {result['message']}")
print(f" Data: {json.dumps(result['data'], indent=2)}")
# 3. 模拟清理孤儿截图
print("\n[Step 3] Testing orphan screenshots cleanup (dry run)...")
response = await client.post(
f"{API_BASE}/api/cleanup/cleanup/orphan-screenshots",
params={"dry_run": True}
)
result = response.json()
print(f" Success: {result['success']}")
print(f" Message: {result['message']}")
print(f" Orphan screenshots: {result['data']['count']}")
# 4. 实际清理(保留30天)
print("\n[Step 4] Performing actual cleanup (days=30)...")
response = await client.post(
f"{API_BASE}/api/cleanup/cleanup",
params={"days": 30, "dry_run": False}
)
result = response.json()
print(f" Success: {result['success']}")
print(f" Message: {result['message']}")
print(f" Screenshots deleted: {result['data']['screenshots']}")
print(f" Reports deleted: {result['data']['reports']}")
print(f" Executions deleted: {result['data']['executions']}")
print(f" Space freed: {result['data']['freed_space']/1024/1024:.2f} MB")
# 5. 获取清理后的统计
print("\n[Step 5] Getting stats after cleanup...")
response = await client.get(f"{API_BASE}/api/cleanup/stats")
stats_after = response.json()["data"]
print(f" Screenshots: {stats_after['screenshots']['count']} files, {stats_after['screenshots']['size']/1024/1024:.2f} MB")
print(f" Reports: {stats_after['reports']['count']} files, {stats_after['reports']['size']/1024/1024:.2f} MB")
print(f" Executions: {stats_after['database']['executions']}")
print(f" Case Results: {stats_after['database']['case_results']}")
print("\n" + "=" * 60)
print("Test completed!")
print("=" * 60)
if __name__ == "__main__":
asyncio.run(test_cleanup())
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 """
WebSocket 客户端测试脚本
用于验证后端 WebSocket 实时推送功能是否正常工作。
运行方式:
python test_websocket_client.py <execution_id>
示例:
python test_websocket_client.py exec_abc123
"""
import asyncio
import json
import sys
import websockets
from datetime import datetime
async def test_websocket(execution_id: str):
"""测试 WebSocket 连接"""
uri = f"ws://localhost:8001/api/executions/ws/{execution_id}"
print(f"[{datetime.now().strftime('%H:%M:%S')}] Connecting to WebSocket: {uri}")
try:
async with websockets.connect(uri) as ws:
print(f"[{datetime.now().strftime('%H:%M:%S')}] Connected successfully!")
# 发送心跳消息
await ws.send(json.dumps({"type": "heartbeat"}))
print(f"[{datetime.now().strftime('%H:%M:%S')}] Sent heartbeat")
# 监听消息(最多等待 60 秒)
message_count = 0
try:
while True:
message = await asyncio.wait_for(ws.recv(), timeout=60.0)
message_count += 1
try:
data = json.loads(message)
msg_type = data.get("type", "unknown")
timestamp = data.get("timestamp", "")
print(f"[{datetime.now().strftime('%H:%M:%S')}] Received message #{message_count}:")
print(f" Type: {msg_type}")
print(f" Timestamp: {timestamp}")
print(f" Data: {json.dumps(data.get('data', {}), indent=2, ensure_ascii=False)}")
# 如果收到执行完成消息,退出
if msg_type in ("execution_complete", "execution_cancelled"):
print(f"[{datetime.now().strftime('%H:%M:%S')}] Execution finished, closing connection...")
break
except json.JSONDecodeError:
print(f"[{datetime.now().strftime('%H:%M:%S')}] Raw message: {message[:200]}")
except asyncio.TimeoutError:
print(f"[{datetime.now().strftime('%H:%M:%S')}] Timeout (60s), closing connection...")
print(f"[{datetime.now().strftime('%H:%M:%S')}] Total messages received: {message_count}")
except websockets.exceptions.InvalidURI:
print(f"Error: Invalid WebSocket URI: {uri}")
except websockets.exceptions.InvalidHandshake:
print(f"Error: WebSocket handshake failed")
except Exception as e:
print(f"Error: {str(e)}")
def main():
"""主函数"""
if len(sys.argv) < 2:
print("Usage: python test_websocket_client.py <execution_id>")
print("Example: python test_websocket_client.py exec_abc123")
sys.exit(1)
execution_id = sys.argv[1]
asyncio.run(test_websocket(execution_id))
if __name__ == "__main__":
main()
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
WebSocket 实时推送验证脚本
功能:
1. 查询可用的测试用例
2. 创建执行任务
3. 连接 WebSocket 监听实时推送
4. 触发执行
5. 验证 WebSocket 消息是否正常接收
运行方式:
python test_websocket_flow.py
"""
import asyncio
import json
import sys
import httpx
import websockets
from datetime import datetime
from pathlib import Path
# 添加项目路径
sys.path.insert(0, str(Path(__file__).parent.parent))
# API 配置
API_BASE_URL = "http://localhost:8001"
WS_BASE_URL = "ws://localhost:8001"
async def get_test_cases():
"""获取可用的测试用例"""
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.get(f"{API_BASE_URL}/api/cases", params={"limit": 5})
data = response.json()
return data.get("items", [])
async def create_execution(case_ids: list):
"""创建执行任务"""
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{API_BASE_URL}/api/executions",
json={
"case_ids": case_ids,
"name": f"WebSocket测试-{datetime.now().strftime('%H%M%S')}",
"environment": "test",
"config": {"headless": True, "auto_login": True}
}
)
data = response.json()
return data.get("id")
async def run_execution(execution_id: str):
"""触发执行"""
async with httpx.AsyncClient(timeout=300.0) as client:
response = await client.post(
f"{API_BASE_URL}/api/executions/{execution_id}/run",
json={"headless": True, "auto_login": True}
)
data = response.json()
return data
async def listen_websocket(execution_id: str):
"""监听 WebSocket 推送"""
uri = f"{WS_BASE_URL}/api/executions/ws/{execution_id}"
messages_received = []
print(f"\n[{datetime.now().strftime('%H:%M:%S')}] [WS] Connecting to: {uri}")
try:
async with websockets.connect(uri) as ws:
print(f"[{datetime.now().strftime('%H:%M:%S')}] [WS] Connected!")
# 发送心跳
await ws.send(json.dumps({"type": "heartbeat"}))
# 监听消息
while True:
try:
message = await asyncio.wait_for(ws.recv(), timeout=120.0)
data = json.loads(message)
msg_type = data.get("type", "unknown")
messages_received.append(data)
print(f"[{datetime.now().strftime('%H:%M:%S')}] [WS] Received: {msg_type}")
# 详细打印关键消息
if msg_type in ("execution_start", "case_start", "case_complete", "execution_complete"):
msg_data = data.get("data", {})
if msg_type == "execution_start":
print(f" Execution: {msg_data.get('name')}, Total cases: {msg_data.get('total_cases')}")
elif msg_type == "case_start":
print(f" Case: {msg_data.get('case_name')}")
elif msg_type == "case_complete":
print(f" Case: {msg_data.get('case_name')}, Status: {msg_data.get('status')}")
elif msg_type == "execution_complete":
print(f" Passed: {msg_data.get('passed')}, Failed: {msg_data.get('failed')}")
print(f" Pass rate: {msg_data.get('pass_rate')}%")
# 执行完成,退出
if msg_type in ("execution_complete", "execution_cancelled"):
break
except asyncio.TimeoutError:
print(f"[{datetime.now().strftime('%H:%M:%S')}] [WS] Timeout waiting for message")
break
except Exception as e:
print(f"[{datetime.now().strftime('%H:%M:%S')}] [WS] Error: {str(e)}")
return messages_received
async def main():
"""主流程"""
print("=" * 70)
print("WebSocket Real-time Push Validation Test")
print("=" * 70)
# Step 1: 获取测试用例
print(f"\n[{datetime.now().strftime('%H:%M:%S')}] [Step 1] Getting test cases...")
cases = await get_test_cases()
if not cases:
print("Error: No test cases found!")
return
# 选择需要登录态的用例(auto_login=True)
suitable_cases = [c for c in cases if c.get("config", {}).get("auto_login", False)]
if not suitable_cases:
suitable_cases = cases[:2] # 任意选择前两个
case_ids = [c["id"] for c in suitable_cases[:2]]
print(f" Selected cases: {[c['name'] for c in suitable_cases[:2]]}")
print(f" Case IDs: {case_ids}")
# Step 2: 创建执行任务
print(f"\n[{datetime.now().strftime('%H:%M:%S')}] [Step 2] Creating execution...")
execution_id = await create_execution(case_ids)
print(f" Execution ID: {execution_id}")
if not execution_id:
print("Error: Failed to create execution!")
return
# Step 3: 启动 WebSocket 监听(后台任务)
print(f"\n[{datetime.now().strftime('%H:%M:%S')}] [Step 3] Starting WebSocket listener...")
websocket_task = asyncio.create_task(listen_websocket(execution_id))
# 等待 WebSocket 连接建立
await asyncio.sleep(2)
# Step 4: 触发执行
print(f"\n[{datetime.now().strftime('%H:%M:%S')}] [Step 4] Triggering execution...")
result = await run_execution(execution_id)
print(f" Execution result: {result}")
# Step 5: 等待 WebSocket 完成
print(f"\n[{datetime.now().strftime('%H:%M:%S')}] [Step 5] Waiting for WebSocket messages...")
messages = await websocket_task
# 验证结果
print("\n" + "=" * 70)
print("Validation Results")
print("=" * 70)
print(f" Total messages received: {len(messages)}")
# 检查关键消息
message_types = [m.get("type") for m in messages]
expected_types = ["execution_start", "case_complete", "execution_complete"]
found_types = [t for t in expected_types if t in message_types]
print(f" Expected message types: {expected_types}")
print(f" Found message types: {found_types}")
# 验证是否所有关键消息都收到了
if all(t in message_types for t in expected_types):
print("\n[SUCCESS] WebSocket real-time push is working correctly!")
elif len(messages) > 0:
print(f"\n[PARTIAL] Received {len(messages)} messages, but some expected types missing")
print(f" Missing: {[t for t in expected_types if t not in message_types]}")
else:
print("\n[FAILED] No WebSocket messages received!")
print("=" * 70)
if __name__ == "__main__":
asyncio.run(main())
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
更新登录测试用例 - 修复断言超时问题
"""
import requests
BASE_URL = "http://localhost:8001/api"
# 更新正常登录用例
case_id = "case_39aab8d4284c4b508465e0eae871d801"
updated_steps = [
{"order": 1, "name": "打开登录页面", "action": "navigate",
"params": {"url": "https://192.168.5.44"}, "expected": ""},
{"order": 2, "name": "等待登录表单加载", "action": "wait",
"params": {"selector": "input", "timeout": 10000}, "expected": ""},
{"order": 3, "name": "输入用户名", "action": "fill",
"params": {"selector": "input[type='text']", "value": "admin@xty"}, "expected": ""},
{"order": 4, "name": "输入密码", "action": "fill",
"params": {"selector": "input[type='password']", "value": "Ubains@13579"}, "expected": ""},
{"order": 5, "name": "输入验证码", "action": "fill",
"params": {"selector": "input[type='text']:nth-of-type(3)", "value": "csba"}, "expected": ""},
{"order": 6, "name": "点击登录按钮", "action": "click",
"params": {"selector": "button"}, "expected": ""},
{"order": 7, "name": "等待页面跳转完成", "action": "wait",
"params": {"selector": "text=统一管理平台", "timeout": 10000}, "expected": ""}
]
response = requests.put(
f"{BASE_URL}/cases/{case_id}",
json={"steps": updated_steps}
)
if response.status_code == 200:
print("用例更新成功!")
print("修改内容:")
print(" - 删除了URL断言步骤(会立即检查导致超时)")
print(" - 改为等待页面元素'统一管理平台'出现")
print(" - 等待时间增加到10秒")
else:
print(f"更新失败: {response.text}")
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
更新测试用例选择器脚本
使用更精确的选择器更新现有测试用例。
"""
import asyncio
import json
from pathlib import Path
import sys
sys.path.insert(0, str(Path(__file__).parent.parent))
from app.database import async_session_maker, init_db
from app.models.test_case import TestCase
from app.utils.selector_mapper import SELECTORS, get_smart_selector
from sqlalchemy import select
# 选择器优化映射
SELECTOR_IMPROVEMENTS = {
# 登录相关
"用户名输入框": "input[placeholder*='用户']",
"密码输入框": "input[type='password']",
"验证码输入框": "input[placeholder*='验证']",
"登录按钮": "button:has-text('登录')",
"协议确认按钮": ".el-dialog button:has-text('确定')",
# 通用
"搜索输入框": "input[placeholder*='搜索']",
"搜索按钮": "button:has-text('搜索')",
"新建按钮": "button:has-text('新建')",
"确认按钮": "button:has-text('确定')",
"取消按钮": "button:has-text('取消')",
# 列表
"表格": ".el-table",
"列表": ".el-table",
"表单": ".el-form",
}
async def update_case_selectors():
"""更新测试用例的选择器"""
await init_db()
async with async_session_maker() as db:
query = select(TestCase)
result = await db.execute(query)
cases = list(result.scalars().all())
updated_count = 0
for case in cases:
steps = case.steps
if not steps:
continue
has_update = False
for step in steps:
params = step.get("params", {})
action = step.get("action", "")
step_name = step.get("name", "")
old_selector = params.get("selector", "")
# 优化选择器
if action in ["click", "fill", "wait", "assert", "hover"]:
if old_selector in ["body", ".el-button", "button", ".el-input", "input"]:
# 根据步骤名称推断更精确的选择器
for key, new_selector in SELECTOR_IMPROVEMENTS.items():
if key in step_name:
params["selector"] = new_selector
has_update = True
print(f"[{case.name}] Updated selector: {old_selector} -> {new_selector}")
break
# 优化断言选择器
if action == "assert":
assert_type = params.get("type", "")
if assert_type == "visible" and old_selector == "body":
if "表单" in step_name:
params["selector"] = ".el-form, form"
has_update = True
elif "表格" in step_name or "列表" in step_name:
params["selector"] = ".el-table, table"
has_update = True
elif "卡片" in step_name:
params["selector"] = ".card, .el-card"
has_update = True
elif "图表" in step_name:
params["selector"] = ".chart, canvas"
has_update = True
if has_update:
case.steps = steps
updated_count += 1
await db.commit()
print(f"\nUpdated {updated_count} test cases")
if __name__ == "__main__":
asyncio.run(update_case_selectors())
\ No newline at end of file
"""
Tests package
"""
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
API 路由测试
"""
import pytest
from httpx import AsyncClient
from httpx._transports.asgi import ASGITransport
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
from app.main import app
@pytest.fixture
async def client():
"""创建测试客户端"""
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
yield ac
class TestHealthCheck:
"""健康检查测试"""
@pytest.mark.asyncio
async def test_root(self, client):
"""测试根路径"""
response = await client.get("/")
assert response.status_code == 200
data = response.json()
assert data["status"] == "ok"
assert "message" in data
@pytest.mark.asyncio
async def test_health(self, client):
"""测试健康检查"""
response = await client.get("/health")
assert response.status_code == 200
data = response.json()
assert data["status"] == "healthy"
class TestModulesAPI:
"""模块 API 测试"""
@pytest.mark.asyncio
async def test_list_modules(self, client):
"""测试获取模块列表"""
response = await client.get("/api/modules")
assert response.status_code == 200
data = response.json()
assert "total" in data
assert "items" in data
assert isinstance(data["items"], list)
class TestCasesAPI:
"""用例 API 测试"""
@pytest.mark.asyncio
async def test_list_cases(self, client):
"""测试获取用例列表"""
response = await client.get("/api/cases")
assert response.status_code == 200
data = response.json()
assert "total" in data
assert "items" in data
class TestExecutionsAPI:
"""执行 API 测试"""
@pytest.mark.asyncio
async def test_list_executions(self, client):
"""测试获取执行记录列表"""
response = await client.get("/api/executions")
assert response.status_code == 200
data = response.json()
assert "total" in data
assert "items" in data
class TestReportsAPI:
"""报告 API 测试"""
@pytest.mark.asyncio
async def test_list_reports(self, client):
"""测试获取报告列表"""
response = await client.get("/api/reports/list")
assert response.status_code == 200
data = response.json()
assert "total" in data
assert "items" in data
class TestCleanupAPI:
"""清理 API 测试"""
@pytest.mark.asyncio
async def test_get_stats(self, client):
"""测试获取存储统计"""
response = await client.get("/api/cleanup/stats")
assert response.status_code == 200
data = response.json()
assert data["success"] == True
assert "data" in data
@pytest.mark.asyncio
async def test_dry_run_cleanup(self, client):
"""测试模拟清理"""
response = await client.post(
"/api/cleanup/cleanup",
params={"days": 30, "dry_run": True}
)
assert response.status_code == 200
data = response.json()
assert data["success"] == True
assert data["data"]["dry_run"] == True
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论