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

feat: 新增自动化部署测试/性能测试/功能测试报告三大模块

新增自动化部署测试模块(SSH远程执行+交互式应答+WebSocket实时日志)
新增性能测试模块(系统指标监控+图表展示)
新增功能测试报告模块(Excel解析→数据分析→图表→Word/MD报告→ERP上传)
修复执行引擎死锁(Playwright线程+uvloop兼容+headless默认值)
修复元素映射表服务(v1.2扩展至220元素)
新增ERP配置项、性能监控模型、数据库注册
前端:部署/性能/功能测试报告三套页面+菜单+路由
文档:PRD+计划执行+Handoff完整文档
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 09346c8e
# PRD 需求文档 — 自动化部署测试模块
> **文档类型**: PRD 需求文档
> **创建日期**: 2026-08-12
> **作者**: czj
> **优先级**: P0
> **版本**: v1.0
> **状态**: 待评审
> **所属模块**: 自动化部署测试(部署运维)
---
## 一、需求背景
### 1.1 业务背景
当前平台已覆盖 UI 自动化测试、安全测试、性能测试、设备模拟等模块,测试能力日趋完善。但随着**新统一平台**的持续迭代,每次版本发布都需要人工通过 SSH 登录目标服务器执行部署脚本,部署过程存在以下痛点:
1. **操作繁琐**:需要运维/测试人员手动打开终端、SSH 连接多台服务器、逐条执行命令。
2. **交互式输入无法自动化**:部署脚本执行过程中常包含交互式提示(输入数据库密码、确认 Y/N、选择部署模式等),人工值守耗时且易出错。
3. **过程不可追溯**:部署过程中的输出日志散落在各终端,无法统一留存、回看、排障。
4. **无法实时监测**:部署发起后无法集中查看进度,失败原因排查困难。
因此,需要在本测试平台中内置**自动化部署测试模块**,通过 Web 界面一键触发远程部署,自动完成脚本交互输入,实时流式展示部署日志,并留存历史执行记录。
### 1.2 目标用户
| 用户角色 | 使用场景 | 核心需求 |
|---------|---------|---------|
| 测试工程师 | 新统一平台版本发布前执行部署验证 | 一键执行部署、实时查看部署进度、确认部署成功 |
| 开发工程师 | 联调环境部署 / 脚本调试 | 配置服务器与部署脚本、查看完整交互日志 |
| 运维人员 | 多环境发布与回滚 | 集中管理多台目标服务器、查看历史部署记录 |
### 1.3 需求目标
1. **远程部署执行**:通过 SSH 连接**远程目标服务器**(非本机),一键执行目标服务器上已有的部署脚本
2. **交互式输入自动完成**:部署脚本执行期间的交互式提示(如输入密码、确认、选择模式)自动应答
3. **实时进度监测**:部署过程通过 WebSocket 实时流式展示日志输出
4. **执行记录留存**:部署历史记录与完整日志持久化存储,支持回看
5. **服务器与脚本管理**:平台内维护目标服务器列表与部署脚本配置
---
## 二、功能需求
### 2.1 功能模块总览
```
自动化部署测试(父级模块,侧边栏独立菜单)
├── 服务器管理 — 目标服务器 SSH 连接配置管理(IP、端口、凭据、连接测试)
├── 部署脚本 — 部署脚本配置(远程命令、工作目录、交互式应答规则)
└── 部署执行 — 一键执行部署 + 实时日志监测 + 历史记录回看
```
### 2.2 服务器管理
#### 功能描述
管理平台可 SSH 连接的**远程目标服务器**信息,是部署执行的基础。服务器凭据加密存储,支持密码/密钥两种认证方式。
#### 功能清单
| 功能 | 说明 | 优先级 |
|------|------|--------|
| 新增服务器 | 添加目标服务器(名称、IP、端口、用户名、密码/私钥、认证方式) | P0 |
| 编辑服务器 | 修改服务器配置(密码留空时不修改) | P0 |
| 删除服务器 | 删除服务器(有关联部署脚本时禁止删除) | P0 |
| 连接测试 | 通过 SSH 实际连接测试服务器连通性 | P0 |
| 服务器列表 | 展示所有服务器,含连接状态标识(未知/在线/离线) | P0 |
#### 服务器字段
| 字段 | 类型 | 说明 |
|------|------|------|
| 名称 | string | 服务器标识名称,如"生产环境-60" |
| 主机地址 | string | IP 或域名 |
| 端口 | integer | SSH 端口,默认 22 |
| 用户名 | string | SSH 登录用户名 |
| 密码 | string | SSH 密码(**加密存储**,编辑时留空表示不修改) |
| 认证方式 | enum | password(密码)/ key(私钥) |
| 私钥 | text | SSH 私钥内容(认证方式为 key 时填写) |
| 状态 | enum | unknown(未知)/ online(在线)/ offline(离线) |
| 描述 | text | 备注说明 |
### 2.3 部署脚本
#### 功能描述
配置部署脚本的远程执行命令与**交互式应答规则**。部署脚本已存在于目标服务器上,本模块仅配置其执行路径、工作目录与交互应答。
#### 功能清单
| 功能 | 说明 | 优先级 |
|------|------|--------|
| 新增脚本配置 | 配置脚本名称、目标服务器、远程命令、工作目录、超时时间 | P0 |
| 交互应答配置 | 配置多条"触发提示 → 应答内容"规则,自动完成交互输入 | P0 |
| 编辑脚本配置 | 修改脚本配置与交互应答规则 | P0 |
| 删除脚本配置 | 删除脚本配置 | P0 |
| 脚本列表 | 展示所有脚本,可按服务器筛选 | P0 |
#### 脚本字段
| 字段 | 类型 | 说明 |
|------|------|------|
| 名称 | string | 脚本配置名称,如"统一平台全量部署" |
| 描述 | text | 说明 |
| 目标服务器 | FK | 关联的服务器 |
| 远程命令 | string | 远程执行命令,如 `bash /opt/deploy/deploy.sh` |
| 工作目录 | string | 远程工作目录(可选,执行前先 cd) |
| 超时时间 | integer | 整体执行超时(秒),默认 3600 |
#### 交互式应答规则(核心)
每条规则定义"当输出中出现某提示时,自动发送某应答":
| 字段 | 类型 | 说明 |
|------|------|------|
| 触发提示 | string | 匹配的提示文本,如"请输入数据库密码:" |
| 匹配方式 | enum | contains(包含匹配,默认)/ regex(正则匹配) |
| 应答内容 | string | 匹配后自动发送的内容(自动追加换行) |
| 超时 | integer | 等待该提示出现的最大时间(秒) |
| 顺序 | integer | 匹配顺序,防止重复匹配 |
| 敏感 | boolean | 是否敏感信息(日志中显示为 `***`) |
| 描述 | string | 该应答规则的说明 |
### 2.4 部署执行
#### 功能描述
一键执行部署的核心页面。选择脚本 → 点击【执行部署】→ 实时查看部署日志 → 完成/失败/取消。
#### 功能清单
| 功能 | 说明 | 优先级 |
|------|------|--------|
| 发起执行 | 选择部署脚本,一键【执行部署】 | P0 |
| 实时日志 | WebSocket 实时流式展示部署输出(终端风格,自动滚动) | P0 |
| 自动应答提示 | 日志中展示自动应答已发送(敏感内容隐藏) | P0 |
| 执行状态 | 状态徽标:待执行/执行中/成功/失败/已取消 + 耗时 | P0 |
| 取消执行 | 执行中可一键取消(关闭 SSH 通道) | P0 |
| 执行记录列表 | 展示历史执行记录(脚本、服务器、状态、耗时、触发人) | P0 |
| 执行详情 | 查看单次执行完整日志(历史回看) | P0 |
| 日志落库 | 每次执行日志持久化到数据库 | P0 |
#### 执行状态机
```
pending(待执行) → running(执行中) → completed(成功)
→ failed(失败)
→ cancelled(已取消)
```
#### WebSocket 消息类型
| 类型 | 说明 | 主要字段 |
|------|------|---------|
| `deploy_start` | 部署开始 | execution_id, script_name, server, command |
| `deploy_output` | 标准输出 | execution_id, line, timestamp |
| `deploy_stderr` | 错误输出 | execution_id, line, timestamp |
| `deploy_answer_sent` | 自动应答已发送 | execution_id, trigger, answer, order |
| `deploy_complete` | 部署完成 | execution_id, status, exit_code, duration |
| `deploy_error` | 执行错误 | execution_id, error_message |
---
## 三、非功能需求
| 类别 | 需求 |
|------|------|
| 安全 | 服务器密码加密存储(Fernet 加密);敏感应答在日志/接口中隐藏 |
| 性能 | 日志流式推送,不阻塞其他请求;部署执行在独立线程运行 |
| 可靠性 | SSH 断连、脚本超时、非零退出码均能正确标记失败状态 |
| 兼容性 | 支持密码/私钥两种 SSH 认证;支持 SQLite/MySQL 数据库 |
| 可追溯 | 执行记录含脚本/服务器快照,删除脚本/服务器不影响历史记录 |
## 四、约束与依赖
1. **SSH 库**:paramiko(需新增依赖,现有 `deploy_to_560.py` 已使用同款)
2. **执行模型**:长耗时任务沿用 `execution_service.py` 的专用 `threading.Thread` + `loop.run_in_executor` 模式(Windows 平台避免 asyncio 死锁)
3. **实时推送**:复用现有 WebSocket `ConnectionManager`(按 execution_id 分组广播)
4. **前端约束**:axios `baseURL` 留空、`useRoute()` 在 setup 顶层调用(遵守项目踩坑记录)
\ No newline at end of file
# 执行计划:自动化部署测试模块
> **版本**:v1.0
> **创建日期**:2026-08-12
> **来源**:`_PRD_需求文档_自动化部署测试模块.md`
> **状态**:待执行
---
## 一、任务分解
| 阶段 | 任务 | 预计文件数 | 优先级 |
|------|------|-----------|--------|
| Phase 1 | 后端依赖 + 数据模型(4 模型 + 注册) | 5 文件 | P0 |
| Phase 2 | Pydantic Schemas | 1 文件 | P0 |
| Phase 3 | SSH 执行器(核心交互式输入处理) | 1 文件 | P0 |
| Phase 4 | 部署服务(业务逻辑 + 线程管理) | 1 文件 | P0 |
| Phase 5 | API 路由(CRUD + WebSocket) + main.py 注册 | 2 文件 | P0 |
| Phase 6 | 前端类型定义 + API 模块 | 2 文件 | P0 |
| Phase 7 | 前端路由 + 侧边栏菜单 | 2 文件 | P0 |
| Phase 8 | 前端视图(4 页面) | 5 文件 | P0 |
| Phase 9 | 验证:启动服务 + 冒烟测试 | — | P1 |
---
## 二、详细实现步骤
### Phase 1: 依赖 + 数据模型
**文件**`backend/requirements.txt`(修改)
- 追加 `paramiko==3.4.0`
**文件**`backend/app/models/deploy_server.py`(新建)
- `DeployServer` 模型,表名 `deploy_servers`
- 字段:id(String64 PK, `generate_id("srv")`), name, host, port(default 22), username, password_encrypted, auth_type, private_key(nullable), status, description, created_at, updated_at
- 密码加密:`encrypt_password()` / `decrypt_password()` 方法(Fernet)
**文件**`backend/app/models/deploy_script.py`(新建)
- `DeployScript` 模型,表名 `deploy_scripts`
- 字段:id(String64 PK), name, description, server_id(FK → deploy_servers), remote_command, work_dir(nullable), timeout(default 3600), expect_answers(JSON), created_at, updated_at
- `server` 关系:`relationship("DeployServer", back_populates="scripts")`
**文件**`backend/app/models/deploy_execution.py`(新建)
- `DeployExecution` 模型,表名 `deploy_executions`
- 快照字段:script_id, server_id, script_name, server_name, server_host, remote_command
- 状态字段:status(pending/running/completed/failed/cancelled), start_time, end_time, duration, exit_code(nullable), error_message(nullable Text), trigger_by, created_at
- `script` 关系:`relationship("DeployScript")`
**文件**`backend/app/models/deploy_log.py`(新建)
- `DeployLog` 模型,表名 `deploy_logs`
- 字段:id(String64 PK), execution_id(FK → deploy_executions), timestamp, level, message(Text), is_sensitive
**文件**`backend/app/models/__init__.py`(修改)
- 导入 4 个新模型,加入 `__all__`
### Phase 2: Pydantic Schemas
**文件**`backend/app/schemas/deploy.py`(新建)
- `DeployServerBase/Create/Update/Response/ListResponse`
- `ExpectAnswer`(单条规则:trigger, trigger_type, answer, timeout, order, sensitive, description)
- `DeployScriptBase/Create/Update/Response/ListResponse`
- `DeployExecutionCreate/Response/ListResponse`
- `DeployLogResponse/ListResponse`
### Phase 3: SSH 执行器
**文件**`backend/app/executors/deploy_executor.py`(新建)
- `DeployExecutor`
- `__init__(server_config, script_config)`:接收纯 dict 配置
- `execute(execution_id, ev_loop, broadcast_fn)`:主执行方法
- SSH 连接(paramiko.SSHClient + AutoAddPolicy)
- 打开交互式 PTY(`invoke_shell(term='xterm')` + `get_pty=True`
- 单线程轮询循环:
- `channel.recv(4096)` 读取输出,追加到 buffer
- 每行输出通过 `asyncio.run_coroutine_threadsafe(broadcast_fn(...), ev_loop)` 广播
- 按 order 遍历未匹配的 `expect_answers`,匹配触发词则发送应答
- `time.sleep(0.05)` 防忙等
- 完成检测:`channel.exit_status_ready()``recv_exit_status()`
- 超时保护:总执行时间超 timeout 时退出
- `cancel()`:设 `_cancelled=True``channel.close()`
- `close()`:清理 SSH 连接
### Phase 4: 部署服务
**文件**`backend/app/services/deploy_service.py`(新建)
- `DeployService`
- 服务器 CRUD:`list_servers`, `get_server`, `create_server`(加密密码), `update_server`(密码留空不修改), `delete_server`, `test_server_connection`
- 脚本 CRUD:`list_scripts`, `get_script`, `create_script`, `update_script`, `delete_script`
- 执行 CRUD:`list_executions`, `get_execution`, `get_execution_logs`, `delete_execution`
- 核心执行:
- `create_execution`:建执行记录,状态 pending
- `run_execution`:状态→running,起专用线程,`loop.run_in_executor(None, join)`
- `cancel_execution`:调用 executor.cancel()
- 日志双通道:WebSocket 实时广播 + 逐条/批量落库
### Phase 5: API 路由
**文件**`backend/app/routers/deploy.py`(新建)
| 方法 | 路径 | 说明 |
|------|------|------|
| GET | /api/deploy/servers | 服务器列表 |
| POST | /api/deploy/servers | 新增服务器 |
| GET | /api/deploy/servers/{id} | 服务器详情 |
| PUT | /api/deploy/servers/{id} | 编辑服务器 |
| DELETE | /api/deploy/servers/{id} | 删除服务器 |
| POST | /api/deploy/servers/{id}/test | 测试 SSH 连接 |
| GET | /api/deploy/scripts | 脚本列表(支持 server_id 筛选) |
| POST | /api/deploy/scripts | 新增脚本配置 |
| GET | /api/deploy/scripts/{id} | 脚本详情 |
| PUT | /api/deploy/scripts/{id} | 编辑脚本 |
| DELETE | /api/deploy/scripts/{id} | 删除脚本 |
| GET | /api/deploy/executions | 执行记录列表 |
| POST | /api/deploy/executions | 创建执行 |
| POST | /api/deploy/executions/{id}/run | 执行部署 |
| POST | /api/deploy/executions/{id}/cancel | 取消执行 |
| GET | /api/deploy/executions/{id} | 执行详情 |
| DELETE | /api/deploy/executions/{id} | 删除执行记录 |
| GET | /api/deploy/executions/{id}/logs | 执行日志 |
| WS | /api/deploy/ws/executions/{id} | WebSocket 实时日志 |
**文件**`backend/app/main.py`(修改)
- 导入 `from app.routers import deploy`
- 注册 `app.include_router(deploy.router, prefix="/api/deploy", tags=["自动化部署测试"])`
### Phase 6: 前端类型 + API
**文件**`frontend/src/types/deploy.ts`(新建)
- `DeployServer`, `ExpectAnswer`, `DeployScript`, `DeployExecution`, `DeployLog`
**文件**`frontend/src/api/deploy.ts`(新建)
- `deployApi` 对象,全量 API 调用
### Phase 7: 前端路由 + 菜单
**文件**`frontend/src/router/index.ts`(修改)
- 新增 `/deploy` 父路由 + 子路由
**文件**`frontend/src/App.vue`(修改)
- 新增 `el-sub-menu` 侧边栏菜单,图标导入,pageTitle 映射
### Phase 8: 前端视图
**文件**`frontend/src/views/deploy/index.vue`(新建)
- 父容器 `<router-view/>`
**文件**`frontend/src/views/deploy/DeployServers.vue`(新建)
- 服务器列表 + 新增/编辑对话框(含密码加密、认证方式切换、连接测试)
**文件**`frontend/src/views/deploy/DeployScripts.vue`(新建)
- 脚本列表 + 编辑对话框(脚本信息 + 交互式应答编辑器)
**文件**`frontend/src/views/deploy/DeployExecutions.vue`(新建)
- 执行中心:选择脚本 → 执行 → 实时日志终端(深色主题、自动滚动)+ 历史记录列表
**文件**`frontend/src/views/deploy/DeployExecutionDetail.vue`(新建)
- 执行详情 + 全量日志回看
---
## 三、关键设计决策
| 决策 | 选择 | 理由 |
|------|------|------|
| SSH 库 | paramiko(同步) | 项目的 `deploy_to_560.py` 已在使用,同步 API 避免 asyncio 问题 |
| 交互式输入 | 预设触发词→应答 | 用户在规划阶段已确认此方案 |
| 执行线程 | 专用 `threading.Thread` | 复用 `execution_service.py` 已验证的模式,避免 ThreadPoolExecutor 死锁 |
| 日志实时推送 | WebSocket(ConnectionManager) | 复用现有基础设施,按 execution_id 分组广播 |
| 密码加密 | Fernet 对称加密 | 复现 `device_sim_service.py` 的加密方案 |
| 脚本来源 | 目标服务器已有脚本 | 用户在规划阶段已确认此方案 |
---
## 四、验收标准
1. 后端 `/docs` 自动文档出现 `/api/deploy` 路由组
2. 数据库自动创建 `deploy_servers`, `deploy_scripts`, `deploy_executions`, `deploy_logs` 四张表
3. 侧边栏出现"自动化部署测试"父菜单,含 3 个子菜单
4. 新增服务器 → 测试连接成功
5. 新建脚本(含交互应答规则)→ 执行部署 → 实时日志流式显示 → 自动应答发送 → 执行完成
6. 执行中可取消,历史记录可回看
7. `frontend npm run build` 类型检查 + 打包通过
\ No newline at end of file
...@@ -16,7 +16,9 @@ ...@@ -16,7 +16,9 @@
- Git 提交协作 - Git 提交协作
**当前多窗口状态**:本项目有多个窗口分工开发,交接文档分散在: **当前多窗口状态**:本项目有多个窗口分工开发,交接文档分散在:
- `HANDOFF.md`(本文档)— **UI自动化用例录入优化 + 执行引擎死锁修复 + P1端到端验证通过**(最新,本次会话) - `HANDOFF_自动化部署测试.md` — 自动化部署测试模块(最新,本次会话)
- `HANDOFF_功能测试报告.md` — 功能测试报告模块
- `HANDOFF.md`(本文档)— **UI自动化用例录入优化 + 执行引擎死锁修复 + P1端到端验证通过**
- `HANDOFF_V2部署升级.md` — V2 部署详细交接 - `HANDOFF_V2部署升级.md` — V2 部署详细交接
- `HANDOFF_UI自动化.md` — UI 自动化测试模块 - `HANDOFF_UI自动化.md` — UI 自动化测试模块
- `HANDOFF_安全测试.md` — 安全测试模块 - `HANDOFF_安全测试.md` — 安全测试模块
......
此差异已折叠。
此差异已折叠。
...@@ -67,6 +67,13 @@ class Settings: ...@@ -67,6 +67,13 @@ class Settings:
CLAUDE_MODEL: str = os.getenv("CLAUDE_MODEL", "claude-sonnet-5") # 默认模型 CLAUDE_MODEL: str = os.getenv("CLAUDE_MODEL", "claude-sonnet-5") # 默认模型
CLAUDE_MAX_CANDIDATES: int = int(os.getenv("CLAUDE_MAX_CANDIDATES", "5")) # 最大候选数 CLAUDE_MAX_CANDIDATES: int = int(os.getenv("CLAUDE_MAX_CANDIDATES", "5")) # 最大候选数
# ERP 配置
ERP_API_KEY: str = os.getenv("ERP_API_KEY", "9adc1ce611544fae9bbbc5f6b1d6ce87")
ERP_BASE_URL: str = os.getenv("ERP_BASE_URL", "https://office.ubainsyun.com:5082/api/uerp")
ERP_DEVELOPTESTING_ID: int = int(os.getenv("ERP_DEVELOPTESTING_ID", "397"))
ERP_TYPE_ID: int = int(os.getenv("ERP_TYPE_ID", "2"))
ERP_IMAGE_URL_PREFIX: str = os.getenv("ERP_IMAGE_URL_PREFIX", "https://office.ubainsyun.com:5015/")
class Config: class Config:
"""Pydantic 配置""" """Pydantic 配置"""
env_file = ".env" env_file = ".env"
......
...@@ -80,22 +80,22 @@ ...@@ -80,22 +80,22 @@
"description": "邮箱登录选项卡(第3个.el-tabs__item)" "description": "邮箱登录选项卡(第3个.el-tabs__item)"
}, },
"登录页-账号输入框": { "登录页-账号输入框": {
"selector": "input.el-input__inner", "selector": "input[placeholder*=\"手机号\"]",
"type": "css", "type": "css",
"tier": "登录页", "tier": "登录页",
"description": "手机号/用户名/邮箱输入框(第1个input)" "description": "手机号/用户名/邮箱输入框(第1个input,精确匹配避免覆盖其他字段)"
}, },
"登录页-密码输入框": { "登录页-密码输入框": {
"selector": "input.el-input__inner", "selector": "input[type=\"password\"]",
"type": "css", "type": "css",
"tier": "登录页", "tier": "登录页",
"description": "密码输入框(第2个input,type=password)" "description": "密码输入框(第2个input,type=password,精确匹配避免覆盖其他字段)"
}, },
"登录页-验证码输入框": { "登录页-验证码输入框": {
"selector": "input.el-input__inner", "selector": "input[placeholder*=\"\"]",
"type": "css", "type": "css",
"tier": "登录页", "tier": "登录页",
"description": "图形验证码输入框(第3个input)" "description": "图形验证码输入框(第3个input,精确匹配避免覆盖其他字段)"
}, },
"登录页-验证码图片": { "登录页-验证码图片": {
"selector": "img", "selector": "img",
...@@ -782,10 +782,10 @@ ...@@ -782,10 +782,10 @@
"description": "创建会议 - 顶部保存模板按钮(saveTemplate)" "description": "创建会议 - 顶部保存模板按钮(saveTemplate)"
}, },
"创建会议-新建会议": { "创建会议-新建会议": {
"selector": ".btn", "selector": "div:visible:has-text(\"新建会议\")",
"type": "css", "type": "css",
"tier": "会议预约-创建会议", "tier": "会议预约-创建会议",
"description": "创建会议 - 顶部新建会议按钮(addMeeting),第2个.btn" "description": "创建会议 - 顶部新建会议按钮(addMeeting),文本定位避免与其他.btn冲突"
}, },
"创建会议-会议名称": { "创建会议-会议名称": {
"selector": ".meeting_list .meeting_name .el-input__inner", "selector": ".meeting_list .meeting_name .el-input__inner",
......
...@@ -122,6 +122,7 @@ async def _ensure_columns(conn) -> None: ...@@ -122,6 +122,7 @@ async def _ensure_columns(conn) -> None:
# 导入模型以确保表被创建(必须导入才会注册到 Base.metadata) # 导入模型以确保表被创建(必须导入才会注册到 Base.metadata)
from app.models import device_sim # noqa: F401 from app.models import device_sim # noqa: F401
from app.models import performance # noqa: F401
# 需要补齐的字段:[(表名, 列名, 列定义SQL)] # 需要补齐的字段:[(表名, 列名, 列定义SQL)]
columns_to_add = [ columns_to_add = [
......
此差异已折叠。
此差异已折叠。
...@@ -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, batch, dependencies, security, ai_locator, device_sim, system, element_locator, login_template, smart_locate from app.routers import modules, cases, executions, recorder, stats, reports, cleanup, batch, dependencies, security, ai_locator, device_sim, system, element_locator, login_template, smart_locate, performance, deploy, functional_report
# 配置日志 # 配置日志
logging.basicConfig( logging.basicConfig(
...@@ -212,6 +212,24 @@ app.include_router( ...@@ -212,6 +212,24 @@ app.include_router(
tags=["登录模板"] tags=["登录模板"]
) )
app.include_router(
performance.router,
prefix="/api/performance",
tags=["性能测试"]
)
app.include_router(
deploy.router,
prefix="/api/deploy",
tags=["自动化部署测试"]
)
app.include_router(
functional_report.router,
prefix="/api/functional-report",
tags=["功能测试报告"]
)
# ==================== 根路径 ==================== # ==================== 根路径 ====================
......
...@@ -17,6 +17,11 @@ from app.models.security_config import SecurityConfig ...@@ -17,6 +17,11 @@ from app.models.security_config import SecurityConfig
from app.models.vulnerability_result import VulnerabilityResult from app.models.vulnerability_result import VulnerabilityResult
from app.models.device_sim import EnvConfig, DeviceSimulator, ReportLog from app.models.device_sim import EnvConfig, DeviceSimulator, ReportLog
from app.models.login_template import LoginTemplate from app.models.login_template import LoginTemplate
from app.models.deploy_server import DeployServer
from app.models.deploy_script import DeployScript
from app.models.deploy_execution import DeployExecution
from app.models.deploy_log import DeployLog
from app.models.performance import PerformanceTask, PerformanceSnapshot
__all__ = [ __all__ = [
"Module", "Module",
...@@ -30,4 +35,10 @@ __all__ = [ ...@@ -30,4 +35,10 @@ __all__ = [
"DeviceSimulator", "DeviceSimulator",
"ReportLog", "ReportLog",
"LoginTemplate", "LoginTemplate",
"DeployServer",
"DeployScript",
"DeployExecution",
"DeployLog",
"PerformanceTask",
"PerformanceSnapshot",
] ]
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:deploy_execution.py
模块描述:部署执行记录数据库模型定义
作者:czj
创建日期:2026-08-12
最后修改:2026-08-12
"""
from datetime import datetime
from typing import Optional
from sqlalchemy import String, Integer, Float, Text, DateTime, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base
from app.utils.id_generator import generate_id
class DeployExecution(Base):
"""
部署执行记录
记录每次部署执行的整体信息,包含脚本/服务器快照。
Attributes:
id (str): 执行记录唯一标识
script_id (str): 关联的脚本配置 ID
server_id (str): 关联的目标服务器 ID
script_name (str): 快照:脚本名称
server_name (str): 快照:服务器名称
server_host (str): 快照:服务器地址
remote_command (str): 快照:执行的命令
status (str): 状态 (pending/running/completed/failed/cancelled)
start_time (datetime): 开始时间
end_time (datetime): 结束时间
duration (float): 耗时(秒)
exit_code (int): 退出码
error_message (str): 错误信息
trigger_by (str): 触发人
created_at (datetime): 创建时间
"""
__tablename__ = "deploy_executions"
id: Mapped[str] = mapped_column(
String(64), primary_key=True, default=lambda: generate_id("deploy_exec"),
comment="执行记录唯一标识"
)
script_id: Mapped[str] = mapped_column(
String(64), ForeignKey("deploy_scripts.id", ondelete="SET NULL"),
nullable=False, comment="关联的脚本配置 ID"
)
server_id: Mapped[str] = mapped_column(
String(64), nullable=False, comment="关联的目标服务器 ID"
)
script_name: Mapped[str] = mapped_column(
String(200), default="", comment="快照:脚本名称"
)
server_name: Mapped[str] = mapped_column(
String(100), default="", comment="快照:服务器名称"
)
server_host: Mapped[str] = mapped_column(
String(255), default="", comment="快照:服务器地址"
)
remote_command: Mapped[str] = mapped_column(
String(1000), default="", comment="快照:执行的命令"
)
status: Mapped[str] = mapped_column(
String(20), default="pending",
comment="状态: pending/running/completed/failed/cancelled"
)
start_time: Mapped[Optional[datetime]] = mapped_column(
DateTime, nullable=True, comment="开始时间"
)
end_time: Mapped[Optional[datetime]] = mapped_column(
DateTime, nullable=True, comment="结束时间"
)
duration: Mapped[float] = mapped_column(
Float, default=0.0, comment="耗时(秒)"
)
exit_code: Mapped[Optional[int]] = mapped_column(
Integer, nullable=True, comment="退出码"
)
error_message: Mapped[Optional[str]] = mapped_column(
Text, nullable=True, comment="错误信息"
)
trigger_by: Mapped[str] = mapped_column(
String(100), default="", comment="触发人"
)
created_at: Mapped[datetime] = mapped_column(
DateTime, default=datetime.utcnow, comment="创建时间"
)
# 关联脚本配置
script: Mapped[Optional["DeployScript"]] = relationship(
"DeployScript", back_populates="executions"
)
# 关联执行日志(一对多)
logs: Mapped[list["DeployLog"]] = relationship(
"DeployLog",
back_populates="execution",
cascade="all, delete-orphan"
)
def __repr__(self) -> str:
return f"<DeployExecution(id={self.id}, status={self.status})>"
def to_dict(self) -> dict:
return {
"id": self.id,
"script_id": self.script_id,
"server_id": self.server_id,
"script_name": self.script_name,
"server_name": self.server_name,
"server_host": self.server_host,
"remote_command": self.remote_command,
"status": self.status,
"start_time": self.start_time.isoformat() if self.start_time else None,
"end_time": self.end_time.isoformat() if self.end_time else None,
"duration": self.duration,
"exit_code": self.exit_code,
"error_message": self.error_message,
"trigger_by": self.trigger_by,
"created_at": self.created_at.isoformat() if self.created_at else None,
}
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:deploy_log.py
模块描述:部署执行日志数据库模型定义
作者:czj
创建日期:2026-08-12
最后修改:2026-08-12
"""
from datetime import datetime
from sqlalchemy import String, Integer, Text, DateTime, Boolean, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base
from app.utils.id_generator import generate_id
class DeployLog(Base):
"""
部署执行日志
记录部署执行过程中的每一行输出,支持实时流式推送和持久化回看。
Attributes:
id (str): 日志唯一标识
execution_id (str): 关联的执行记录 ID
timestamp (datetime): 日志时间
level (str): 日志级别 (INFO/WARN/ERROR/INPUT/OUTPUT)
message (str): 日志内容
is_sensitive (bool): 是否敏感信息
"""
__tablename__ = "deploy_logs"
id: Mapped[str] = mapped_column(
String(64), primary_key=True, default=lambda: generate_id("deploy_log"),
comment="日志唯一标识"
)
execution_id: Mapped[str] = mapped_column(
String(64), ForeignKey("deploy_executions.id", ondelete="CASCADE"),
nullable=False, index=True, comment="关联的执行记录 ID"
)
timestamp: Mapped[datetime] = mapped_column(
DateTime, default=datetime.utcnow, comment="日志时间"
)
level: Mapped[str] = mapped_column(
String(10), default="INFO", comment="日志级别: INFO/WARN/ERROR/INPUT/OUTPUT"
)
message: Mapped[str] = mapped_column(
Text, default="", comment="日志内容"
)
is_sensitive: Mapped[bool] = mapped_column(
Boolean, default=False, comment="是否敏感信息"
)
# 关联执行记录
execution: Mapped["DeployExecution"] = relationship(
"DeployExecution", back_populates="logs"
)
def __repr__(self) -> str:
return f"<DeployLog(id={self.id}, level={self.level})>"
def to_dict(self) -> dict:
return {
"id": self.id,
"execution_id": self.execution_id,
"timestamp": self.timestamp.isoformat() if self.timestamp else None,
"level": self.level,
"message": self.message,
"is_sensitive": self.is_sensitive,
}
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:deploy_script.py
模块描述:部署脚本配置数据库模型定义
作者:czj
创建日期:2026-08-12
最后修改:2026-08-12
"""
from datetime import datetime
from typing import Optional
from sqlalchemy import String, Integer, Text, JSON, DateTime, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base
from app.utils.id_generator import generate_id
class DeployScript(Base):
"""
部署脚本配置
配置远程服务器上部署脚本的执行命令与交互式应答规则。
脚本已存在于目标服务器上,本配置仅定义如何执行及如何应答交互提示。
Attributes:
id (str): 脚本配置唯一标识
name (str): 脚本配置名称
description (str): 描述
server_id (str): 关联的目标服务器 ID
remote_command (str): 远程执行命令
work_dir (str): 远程工作目录
timeout (int): 执行超时时间(秒)
expect_answers (dict): 交互式应答规则配置
created_at (datetime): 创建时间
updated_at (datetime): 更新时间
"""
__tablename__ = "deploy_scripts"
id: Mapped[str] = mapped_column(
String(64), primary_key=True, default=lambda: generate_id("script"),
comment="脚本配置唯一标识"
)
name: Mapped[str] = mapped_column(
String(200), nullable=False, comment="脚本配置名称"
)
description: Mapped[Optional[str]] = mapped_column(
Text, nullable=True, comment="描述"
)
server_id: Mapped[str] = mapped_column(
String(64), ForeignKey("deploy_servers.id", ondelete="CASCADE"),
nullable=False, comment="关联的目标服务器 ID"
)
remote_command: Mapped[str] = mapped_column(
String(1000), nullable=False, comment="远程执行命令"
)
work_dir: Mapped[Optional[str]] = mapped_column(
String(500), nullable=True, comment="远程工作目录"
)
timeout: Mapped[int] = mapped_column(
Integer, default=3600, comment="执行超时时间(秒)"
)
expect_answers: Mapped[dict] = mapped_column(
JSON, default=list, comment="交互式应答规则配置"
)
created_at: Mapped[datetime] = mapped_column(
DateTime, default=datetime.utcnow, comment="创建时间"
)
updated_at: Mapped[datetime] = mapped_column(
DateTime, default=datetime.utcnow, onupdate=datetime.utcnow,
comment="更新时间"
)
# 关联目标服务器
server: Mapped["DeployServer"] = relationship(
"DeployServer", back_populates="scripts"
)
# 关联执行记录
executions: Mapped[list["DeployExecution"]] = relationship(
"DeployExecution",
back_populates="script",
cascade="all, delete-orphan"
)
def __repr__(self) -> str:
return f"<DeployScript(id={self.id}, name={self.name})>"
def to_dict(self) -> dict:
return {
"id": self.id,
"name": self.name,
"description": self.description,
"server_id": self.server_id,
"server_name": self.server.name if self.server else "",
"remote_command": self.remote_command,
"work_dir": self.work_dir,
"timeout": self.timeout,
"expect_answers": self.expect_answers or [],
"created_at": self.created_at.isoformat() if self.created_at else None,
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
}
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:deploy_server.py
模块描述:部署目标服务器配置数据库模型定义
作者:czj
创建日期:2026-08-12
最后修改:2026-08-12
"""
import logging
from datetime import datetime
from typing import Optional
from cryptography.fernet import Fernet
from sqlalchemy import String, Integer, Text, DateTime
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base
from app.config import settings
from app.utils.id_generator import generate_id
logger = logging.getLogger(__name__)
# 加密密钥(与 device_sim_service.py 保持一致)
_ENCRYPTION_KEY = None
def _get_encryption_key() -> bytes:
"""获取 Fernet 加密密钥"""
global _ENCRYPTION_KEY
if _ENCRYPTION_KEY is None:
key = getattr(settings, 'ENCRYPTION_KEY', None)
if key:
_ENCRYPTION_KEY = key.encode() if isinstance(key, str) else key
else:
_ENCRYPTION_KEY = Fernet.generate_key()
return _ENCRYPTION_KEY
def encrypt_password(password: str) -> str:
"""加密密码"""
f = Fernet(_get_encryption_key())
return f.encrypt(password.encode()).decode()
def decrypt_password(password_encrypted: str) -> str:
"""解密密码"""
try:
f = Fernet(_get_encryption_key())
return f.decrypt(password_encrypted.encode()).decode()
except Exception as e:
logger.error(f"密码解密失败: {e}")
return ""
class DeployServer(Base):
"""
部署目标服务器配置
记录可 SSH 连接的远程服务器信息,用于自动化部署执行。
Attributes:
id (str): 服务器唯一标识
name (str): 服务器名称
host (str): 主机地址(IP 或域名)
port (int): SSH 端口
username (str): SSH 登录用户名
password_encrypted (str): 加密后的密码
auth_type (str): 认证类型 (password/key)
private_key (str): SSH 私钥内容(key 认证时使用)
status (str): 连接状态 (unknown/online/offline)
description (str): 备注描述
created_at (datetime): 创建时间
updated_at (datetime): 更新时间
"""
__tablename__ = "deploy_servers"
id: Mapped[str] = mapped_column(
String(64), primary_key=True, default=lambda: generate_id("srv"),
comment="服务器唯一标识"
)
name: Mapped[str] = mapped_column(
String(100), nullable=False, comment="服务器名称"
)
host: Mapped[str] = mapped_column(
String(255), nullable=False, comment="主机地址"
)
port: Mapped[int] = mapped_column(
Integer, default=22, comment="SSH 端口"
)
username: Mapped[str] = mapped_column(
String(100), nullable=False, comment="SSH 登录用户名"
)
password_encrypted: Mapped[Optional[str]] = mapped_column(
String(500), nullable=True, comment="加密后的密码"
)
auth_type: Mapped[str] = mapped_column(
String(20), default="password", comment="认证类型: password/key"
)
private_key: Mapped[Optional[str]] = mapped_column(
Text, nullable=True, comment="SSH 私钥内容"
)
status: Mapped[str] = mapped_column(
String(20), default="unknown", comment="连接状态: unknown/online/offline"
)
description: Mapped[Optional[str]] = mapped_column(
Text, nullable=True, comment="备注描述"
)
created_at: Mapped[datetime] = mapped_column(
DateTime, default=datetime.utcnow, comment="创建时间"
)
updated_at: Mapped[datetime] = mapped_column(
DateTime, default=datetime.utcnow, onupdate=datetime.utcnow,
comment="更新时间"
)
# 关联部署脚本(一对多)
scripts: Mapped[list["DeployScript"]] = relationship(
"DeployScript",
back_populates="server",
cascade="all, delete-orphan"
)
def __repr__(self) -> str:
return f"<DeployServer(id={self.id}, name={self.name}, host={self.host}:{self.port})>"
def to_dict(self) -> dict:
return {
"id": self.id,
"name": self.name,
"host": self.host,
"port": self.port,
"username": self.username,
"auth_type": self.auth_type,
"status": self.status,
"description": self.description,
"created_at": self.created_at.isoformat() if self.created_at else None,
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
}
\ No newline at end of file
此差异已折叠。
...@@ -2,20 +2,5 @@ ...@@ -2,20 +2,5 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
模块名称:__init__.py 模块名称:__init__.py
模块描述:API 路由包初始化 模块描述:路由包初始化
作者:czj
创建日期:2026-07-09
""" """
\ No newline at end of file
from app.routers import modules
from app.routers import cases
from app.routers import executions
from app.routers import recorder
__all__ = [
"modules",
"cases",
"executions",
"recorder",
]
\ No newline at end of file
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:functional_report.py
模块描述:功能测试报告 Pydantic 数据模型
作者:czj
创建日期:2026-08-12
最后修改:2026-08-12
"""
from typing import Dict, List, Optional
from pydantic import BaseModel, Field
class UploadResponse(BaseModel):
"""上传Excel文件响应"""
session_id: str = Field(..., description="会话ID")
testcase_path: str = Field(..., description="测试用例文件路径")
buglist_path: str = Field(..., description="BUG列表文件路径")
message: str = Field("文件上传成功", description="提示信息")
class GenerateRequest(BaseModel):
"""生成报告请求"""
session_id: str = Field(..., description="会话ID")
project_name: str = Field("", description="项目名称")
report_format: str = Field("docx", description="报告格式: docx/md/both")
class GenerateResponse(BaseModel):
"""生成报告响应"""
session_id: str = Field(..., description="会话ID")
project_name: str = Field(..., description="项目名称")
case_analysis: Dict = Field(..., description="用例分析结果")
bug_analysis: Dict = Field(..., description="BUG分析结果")
case_bug_link_count: int = Field(0, description="用例-BUG关联数")
charts: Dict[str, str] = Field(default_factory=dict, description="图表文件路径")
reports: List[str] = Field(default_factory=list, description="报告文件路径列表")
class DownloadResponse(BaseModel):
"""下载报告响应"""
report_path: str = Field(..., description="报告文件路径")
filename: str = Field(..., description="文件名")
class UploadToErpRequest(BaseModel):
"""上传到ERP请求"""
session_id: str = Field(..., description="会话ID")
report_filename: str = Field(..., description="报告文件名(从generate接口返回的reports列表中选择)")
developtesting_id: Optional[int] = Field(None, description="测试单ID,默认使用系统配置")
copyuser_names: List[str] = Field(default_factory=list, description="抄送人姓名列表")
createuser_name: Optional[str] = Field(None, description="创建人姓名")
class UploadToErpResponse(BaseModel):
"""上传到ERP响应"""
success: bool = Field(..., description="是否成功")
report_id: Optional[int] = Field(None, description="ERP报告ID")
message: str = Field(..., description="提示信息")
class UploadToProjectRequest(BaseModel):
"""上传文件到ERP项目资料请求"""
session_id: str = Field(..., description="会话ID")
report_filename: str = Field(..., description="要上传的报告文件名")
project_id: int = Field(..., description="关联的项目ID")
name: Optional[str] = Field(None, description="文件名称(默认取文件名)")
class UploadToCooperationRequest(BaseModel):
"""上传文件到ERP协作文档请求"""
session_id: str = Field(..., description="会话ID")
report_filename: str = Field(..., description="要上传的报告文件名")
project_id: int = Field(..., description="关联的项目ID")
name: Optional[str] = Field(None, description="文件名称(默认取文件名)")
group_id: Optional[int] = Field(None, description="分组ID(可选)")
class UploadFileToErpResponse(BaseModel):
"""上传文件到ERP(项目资料/协作文档)响应"""
success: bool = Field(..., description="是否成功")
file_id: Optional[int] = Field(None, description="ERP文件ID")
message: str = Field(..., description="提示信息")
class ErpConfigResponse(BaseModel):
"""ERP配置响应"""
api_key: str = Field("", description="API密钥")
base_url: str = Field("", description="基础URL")
developtesting_id: int = Field(0, description="默认测试单ID")
type_id: int = Field(2, description="报告类型ID")
image_url_prefix: str = Field("", description="图片URL前缀")
class ErpConfigUpdateRequest(BaseModel):
"""ERP配置更新请求"""
api_key: Optional[str] = Field(None, description="API密钥")
base_url: Optional[str] = Field(None, description="基础URL")
developtesting_id: Optional[int] = Field(None, description="默认测试单ID")
type_id: Optional[int] = Field(None, description="报告类型ID")
image_url_prefix: Optional[str] = Field(None, description="图片URL前缀")
class ReportInfoResponse(BaseModel):
"""报告信息响应"""
session_id: str = Field(..., description="会话ID")
reports: List[Dict] = Field(default_factory=list, description="报告文件列表")
charts: List[Dict] = Field(default_factory=list, description="图表文件列表")
analysis: Dict = Field(default_factory=dict, description="分析结果缓存")
\ No newline at end of file
此差异已折叠。
此差异已折叠。
此差异已折叠。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:__init__.py
模块描述:功能测试报告生成子包,移植自独立工具 FunctionalTestReportGeneration
作者:czj
创建日期:2026-08-12
"""
__all__ = [
"config",
"excel_reader",
"data_analyzer",
"data_extractor",
"chart_generator",
"word_generator",
"markdown_generator",
]
\ No newline at end of file
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
...@@ -9,6 +9,13 @@ python-multipart==0.0.6 ...@@ -9,6 +9,13 @@ python-multipart==0.0.6
playwright==1.40.0 playwright==1.40.0
websockets==12.0 websockets==12.0
requests==2.31.0 requests==2.31.0
aiohttp>=3.9.0
pycryptodome>=3.19.0
paho-mqtt==2.1.0 paho-mqtt==2.1.0
openpyxl==3.1.2 openpyxl==3.1.2
jieba>=0.42.1 jieba>=0.42.1
paramiko==3.4.0
python-docx>=1.0.0
mammoth>=1.8.0
Pillow>=10.0.0
matplotlib>=3.7.0
\ No newline at end of file
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论