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

feat(security): 新增安全测试定时任务模块

新增安全测试定时任务 API、调度分流和前端管理页面。\n复用 scheduled_tasks 表并支持 interval/daily/weekly 周期、测试配置关联和自动报告。\n补充需求文档、计划执行文档及部署验证记录。\n\nCo-Authored-By: Claude <noreply@anthropic.com>
上级 1a3d52ad
# HANDOFF — 安全测试模块会话交接文档 # HANDOFF — 安全测试模块会话交接文档
> **生成时间**: 2026-07-22 > **生成时间**: 2026-07-22
> **最后更新**: 2026-08-17 > **最后更新**: 2026-08-19
> **当前分支**: `platform-auto-test` > **当前分支**: `platform-auto-test`
> **开发窗口**: 安全测试模块 > **开发窗口**: 安全测试模块
> **最近提交**: `5b5b35de` docs(security): 同步安全测试 HANDOFF 提交状态 > **最近提交**: `1a3d52ad` fix(executor): 修复执行中无法取消(MySQL 1205 行锁)并加固 SUT URL 校验
> **状态**: ✅ 安全测试 P1 全部完成 + P2 执行验证完成 + 菜单升级 + ERP 配置子菜单 + ERP 上传流程对接完成 + ERP 任务创建对接完成,全部**已提交并部署 5.60** > **状态**: ✅ 安全测试 P1 全部完成 + P2 执行验证完成 + 菜单升级 + ERP 配置子菜单 + ERP 上传流程对接完成 + ERP 任务创建对接完成 + **安全测试定时任务模块完成并部署 5.60**,全部已实现
--- ---
...@@ -26,6 +26,40 @@ ...@@ -26,6 +26,40 @@
## 一、本次任务记录 ## 一、本次任务记录
### 本次任务(2026-08-19):安全测试新增定时任务模块 + 部署 5.60 验证通过
**背景**:安全测试模块补充定时任务能力(设置执行周期 interval/daily/weekly + 执行时间),参考 UI 自动化定时任务模块实现,复用同一张 `scheduled_tasks` 表和调度引擎,按 `case_type` 分流调度。
**流程**:需求文档 → 计划执行文档 → 后端 → 前端 → 构建验证 → 部署 5.60 服务器 + 验证 全链路完成。
**核心成果**
1. ✅ 按规范输出两份文档(`Docs/PRD/功能测试/需求文档/_PRD_功能测试模块读取系统配置被测试系统URL.md` 同目录规范):
- `Docs/PRD/需求文档/安全测试/_PRD_安全测试定时任务_需求文档.md`
- `Docs/PRD/需求文档/安全测试/_PRD_安全测试定时任务_计划执行.md`
2.**后端实现**
- `backend/app/models/scheduled_task.py``scheduled_tasks` 表新增 `case_type VARCHAR(20) DEFAULT 'ui'` + `config_id VARCHAR(64) DEFAULT NULL` 字段(`_ensure_columns()` 迁移,重启自动加列)
- `backend/app/services/scheduler_service.py``run_task_once()``task.case_type` 分流,新增 `_run_security_task_once()`:收集 `collect_module_cases(db, module_ids, "security")``SecurityService.create_execution(config_id, module_ids, case_ids)` → 覆盖 `trigger_type="scheduled"` / `trigger_by=task.name``run_execution()` → 可选 `SecurityReportService.generate_report()` → 更新 last_run_at/run_count/next_run_at
- `backend/app/routers/security_scheduled_tasks.py`(新增)— 7 个端点:GET 列表(仅 `case_type=="security"`,含 config_name 映射)/ POST 创建(config_id 必填 422 / 配置不存在 404)/ PUT 更新(拒绝 case_type 变更 422)/ DELETE / POST toggle / POST run(asyncio.create_task)/ GET runs(按 trigger_type+trigger_by 过滤)
- `backend/app/main.py` — 注册 `security_scheduled_tasks.router`,prefix `/api/security/scheduled-tasks`
3.**前端实现**
- `frontend/src/api/securityScheduledTasks.ts`(新增)— 完整 API 封装,baseURL 空字符串 + 完整 `/api/security/scheduled-tasks` 路径 + `as any` 返回类型模式
- `frontend/src/views/security/ScheduledTasks.vue`(新增)— 任务表格(名称/模块/测试配置 config_name 标签/周期/自动报告/状态开关/上次/下次执行/运行次数/操作)+ 创建/编辑对话框(测试配置下拉 + 周期模式单选 interval/daily/weekly + 自动报告提示)+ 运行记录 drawer 跳转报告中心
- `frontend/src/router/index.ts` — 新增 `/security/scheduled-tasks` 独立路由
- `frontend/src/App.vue` — 安全测试子菜单新增「定时任务」+ `SECURITY_SUB_LABELS` 增加 `scheduled-tasks: '定时任务'`
4.**构建验证**`npm run build` 通过(修复了 `getSecurityConfigs` 带类型返回值导致的 TS2339:改用 `(res as any).data || res` 解包模式)
5.**部署 5.60 服务器 + 全部验证通过**
- SFTP 上传 6 个后端文件(router/service/model/schema/database/main.py)+ 85/86 个前端 dist 文件,`docker restart plat-auto-test-app` 容器恢复 healthy
- `GET /api/security/scheduled-tasks``{"total":0,"items":[]}`(路由加载正常)
- DB 列验证:`scheduled_tasks.case_type`(varchar(20) 默认 ui)+ `config_id`(varchar(64) 默认 NULL)均已迁移成功
- 前端 chunk `ScheduledTasks-*.js/.css` 已部署
- **端到端冒烟测试**:POST 创建安全定时任务(1 分钟间隔、禁用、关联 `sec_cfg_default`)→ 返回完整对象含 next_run_at 计算 → 列表可见 → DELETE 清理成功
**注意事项**
- 安全测试无父路由结构,定时任务复用 UI 定时任务的独立路由风格,采用 `/security/scheduled-tasks` 独立路由
- `scheduler_service.py` 已按 case_type 分流:`ui` 走原逻辑,`security` 走新 `_run_security_task_once`,互不影响
- 安全定时任务执行前检查 `_has_running_security_execution()`,有 pending/running 的安全执行时跳过本轮,避免 MySQL 行锁/并发冲突(与 UI 定时任务行为一致)
- 因对话在部署验证中途中断,本次已在后续会话补齐全部部署验证(创建/删除冒烟测试通过后方记录)
### 本次任务(2026-08-13):安全测试新增 ERP 配置子菜单 ### 本次任务(2026-08-13):安全测试新增 ERP 配置子菜单
**核心成果** **核心成果**
...@@ -321,6 +355,7 @@ GET /api/security/executions/{id}/report/download → 下载 .md 文件 ...@@ -321,6 +355,7 @@ GET /api/security/executions/{id}/report/download → 下载 .md 文件
| **P2.5** | ✅ ~~前端报告类型标签修复~~ | 报告类型显示"UI自动化"应为"安全测试" | | **P2.5** | ✅ ~~前端报告类型标签修复~~ | 报告类型显示"UI自动化"应为"安全测试" |
| **P3** | ✅ ~~安全报告上传 ERP 流程对接~~ | ERP 配置入口已完成,ERP 上传流程已完成(2026-08-17) | | **P3** | ✅ ~~安全报告上传 ERP 流程对接~~ | ERP 配置入口已完成,ERP 上传流程已完成(2026-08-17) |
| **P3.5** | ✅ ~~安全测试 ERP 任务创建对接~~ | 任务创建预览 + 创建接口 + 前端对话框已完成(2026-08-17);**已部署 5.60 并实测 task-preview 通过**(真实执行 ID 返回正确的任务名/紧急程度/ERP 基础数据),create-task 待用户在界面实测 | | **P3.5** | ✅ ~~安全测试 ERP 任务创建对接~~ | 任务创建预览 + 创建接口 + 前端对话框已完成(2026-08-17);**已部署 5.60 并实测 task-preview 通过**(真实执行 ID 返回正确的任务名/紧急程度/ERP 基础数据),create-task 待用户在界面实测 |
| **P4** | ✅ ~~安全测试定时任务模块~~ | 周期调度(interval/daily/weekly)+ 自动报告可选 + 独立页面已完成(2026-08-19);**已部署 5.60 并通过创建/删除冒烟测试**,定时调度触发待真实执行验证 |
--- ---
...@@ -342,6 +377,9 @@ GET /api/security/executions/{id}/report/download → 下载 .md 文件 ...@@ -342,6 +377,9 @@ GET /api/security/executions/{id}/report/download → 下载 .md 文件
| S12 | `TS2339: Property 'content' does not exist on type 'AxiosResponse<...>'` | `request.get<T>()` 泛型与实际运行时类型(解包后)不匹配 | API 函数统一用 `as any` 返回类型,绕过类型推断(已修复) | | S12 | `TS2339: Property 'content' does not exist on type 'AxiosResponse<...>'` | `request.get<T>()` 泛型与实际运行时类型(解包后)不匹配 | API 函数统一用 `as any` 返回类型,绕过类型推断(已修复) |
| S13 | `create_security_cases.py` 用同步引擎连 MySQL 报 aiomysql 兼容错误 | 脚本复制了异步引擎 URL,同步连接用 `mysql+aiomysql://` 前缀冲突 | URL 自动转换:`mysql+aiomysql://``mysql+pymysql://`(已修复) | | S13 | `create_security_cases.py` 用同步引擎连 MySQL 报 aiomysql 兼容错误 | 脚本复制了异步引擎 URL,同步连接用 `mysql+aiomysql://` 前缀冲突 | URL 自动转换:`mysql+aiomysql://``mysql+pymysql://`(已修复) |
| S14 | 端口 8001 旧 uvicorn 进程无法 taskkill 杀掉 | uvicorn `--reload` 残留父进程,MSYS bash 的 taskkill 权限/编码问题 | 用 `powershell Stop-Process -Id <pid> -Force`,或换端口 8002 起新实例验证 | | S14 | 端口 8001 旧 uvicorn 进程无法 taskkill 杀掉 | uvicorn `--reload` 残留父进程,MSYS bash 的 taskkill 权限/编码问题 | 用 `powershell Stop-Process -Id <pid> -Force`,或换端口 8002 起新实例验证 |
| S15 | 前端构建时 `getSecurityConfigs``items` 类型错误 | Axios 类型声明是 `AxiosResponse`,运行时拦截器已解包为 body | 调用处使用 `const data = (res as any).data || res`,再访问 `data.items` |
| S16 | 服务器远程命令通过 paramiko 读取超时 | 长时间 `curl`/MySQL 命令在 SSH channel 上未及时结束,PipeTimeout | 使用短命令 + `head -c`;健康检查优先使用容器内 `wget -qO- --timeout=10 http://127.0.0.1/health`;必要时对读取异常做容错 |
| S17 | 安全定时任务接口未返回/表字段缺失 | 新字段未迁移或容器未加载新代码 | 上传绑定挂载目录后执行 `docker restart plat-auto-test-app`;启动时 `init_db()` 自动运行 `_ensure_columns()`,并验证 `case_type/config_id` |
--- ---
...@@ -406,6 +444,20 @@ PYTHONIOENCODING=utf-8 python scripts/create_security_cases.py ...@@ -406,6 +444,20 @@ PYTHONIOENCODING=utf-8 python scripts/create_security_cases.py
- 文档:`Docs/PRD/需求文档/安全测试/` 下两份 ERP 任务创建文档(新增) - 文档:`Docs/PRD/需求文档/安全测试/` 下两份 ERP 任务创建文档(新增)
- **已提交并推送****已部署至 5.60**,task-preview 接口实测通过(真实执行 ID 返回正确数据),create-task 待用户在界面端到端实测 - **已提交并推送****已部署至 5.60**,task-preview 接口实测通过(真实执行 ID 返回正确数据),create-task 待用户在界面端到端实测
**2026-08-19 安全测试定时任务模块(已实现,已部署至 5.60 已验证)**
- `backend/app/routers/security_scheduled_tasks.py`(新增,7 个 API 端点,prefix `/api/security/scheduled-tasks`
- `backend/app/services/scheduler_service.py`(修改,`run_task_once` 按 case_type 分流 + `_run_security_task_once` 新函数)
- `backend/app/models/scheduled_task.py`(修改,`_ensure_columns` 新增 `case_type` / `config_id` 字段)
- `backend/app/schemas/scheduled_task.py`(修改,Pydantic 模型对齐新字段)
- `backend/app/database.py`(修改,`_ensure_columns` 支持安全定时任务表的列迁移)
- `backend/app/main.py`(修改,注册 `security_scheduled_tasks` 路由)
- `frontend/src/api/securityScheduledTasks.ts`(新增,API 封装)
- `frontend/src/views/security/ScheduledTasks.vue`(新增,完整页面:表格 + 创建/编辑对话框 + 周期配置 + 运行记录 drawer)
- `frontend/src/router/index.ts`(修改,新增 `/security/scheduled-tasks` 独立路由)
- `frontend/src/App.vue`(修改,安全测试子菜单新增「定时任务」+ 页面标题映射)
- 文档:`Docs/PRD/需求文档/安全测试/` 下两份安全定时任务文档(新增)
- **已部署至 5.60**,通过创建/删除冒烟测试验证,定时调度触发待真实执行验证
--- ---
*本文档由安全测试开发窗口生成,供下一次会话快速恢复上下文。会议管理窗口的交接见 `HANDOFF.md`。* *本文档由安全测试开发窗口生成,供下一次会话快速恢复上下文。会议管理窗口的交接见 `HANDOFF.md`。*
...@@ -145,6 +145,18 @@ async def _ensure_columns(conn) -> None: ...@@ -145,6 +145,18 @@ async def _ensure_columns(conn) -> None:
("report_templates", "description", "VARCHAR(200) DEFAULT ''"), ("report_templates", "description", "VARCHAR(200) DEFAULT ''"),
# 钉钉配置:被测系统URL(旧库升级) # 钉钉配置:被测系统URL(旧库升级)
("dingtalk_configs", "target_url", "VARCHAR(500) DEFAULT ''"), ("dingtalk_configs", "target_url", "VARCHAR(500) DEFAULT ''"),
# 钉钉配置:登录接口路径(旧库升级)
("dingtalk_configs", "login_path", "VARCHAR(200) DEFAULT '/platform/api/auth/login'"),
# 钉钉配置:验证码接口路径(旧库升级)
("dingtalk_configs", "captcha_path", "VARCHAR(200) DEFAULT '/platform/api/code'"),
# 性能测试:自定义登录用户名(旧库升级)
("performance_tasks", "login_username", "VARCHAR(200) DEFAULT NULL"),
# 性能测试:自定义登录密码(旧库升级)
("performance_tasks", "login_password", "VARCHAR(200) DEFAULT NULL"),
# 定时任务:用例类型(旧库升级,区分 UI/安全测试)
("scheduled_tasks", "case_type", "VARCHAR(20) DEFAULT 'ui'"),
# 定时任务:安全测试配置ID(旧库升级)
("scheduled_tasks", "config_id", "VARCHAR(64) DEFAULT NULL"),
] ]
def _do_ensure(sync_conn) -> None: def _do_ensure(sync_conn) -> None:
......
...@@ -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, security_erp, security_erp_upload, security_erp_task, ai_locator, device_sim, system, element_locator, login_template, smart_locate, performance, performance_output, api_preset, deploy, functional_report, report_template, api_test, debug, scheduled_tasks from app.routers import modules, cases, executions, recorder, stats, reports, cleanup, batch, dependencies, security, security_erp, security_erp_upload, security_erp_task, ai_locator, device_sim, system, element_locator, login_template, smart_locate, performance, performance_output, api_preset, deploy, functional_report, report_template, api_test, debug, scheduled_tasks, security_scheduled_tasks
# 配置日志 # 配置日志
logging.basicConfig( logging.basicConfig(
...@@ -70,7 +70,7 @@ async def lifespan(app: FastAPI): ...@@ -70,7 +70,7 @@ async def lifespan(app: FastAPI):
except Exception as e: except Exception as e:
logger.warning(f"默认登录模板初始化失败(非致命): {e}") logger.warning(f"默认登录模板初始化失败(非致命): {e}")
# 加载被测系统 URL(从数据库持久化值,修复服务重启后回退到环境变量默认值的问题) # 加载被测系统 URL 与认证路径(从数据库持久化值,修复服务重启后回退到环境变量默认值的问题)
try: try:
from app.models.dingtalk_config import DingTalkConfig from app.models.dingtalk_config import DingTalkConfig
from sqlalchemy import select from sqlalchemy import select
...@@ -82,6 +82,14 @@ async def lifespan(app: FastAPI): ...@@ -82,6 +82,14 @@ async def lifespan(app: FastAPI):
if cfg and cfg.target_url: if cfg and cfg.target_url:
settings.TARGET_URL = cfg.target_url settings.TARGET_URL = cfg.target_url
logger.info(f"被测系统 URL 已从数据库加载: {settings.TARGET_URL}") logger.info(f"被测系统 URL 已从数据库加载: {settings.TARGET_URL}")
if cfg and cfg.login_path:
settings.AUTH_LOGIN_PATH = cfg.login_path
if cfg and cfg.captcha_path:
settings.AUTH_CAPTCHA_PATH = cfg.captcha_path
if cfg and (cfg.login_path or cfg.captcha_path):
logger.info(
f"认证接口路径已从数据库加载: login={settings.AUTH_LOGIN_PATH}, captcha={settings.AUTH_CAPTCHA_PATH}"
)
except Exception as e: except Exception as e:
logger.warning(f"加载被测系统 URL 失败(使用环境变量默认值): {e}") logger.warning(f"加载被测系统 URL 失败(使用环境变量默认值): {e}")
...@@ -324,6 +332,13 @@ app.include_router( ...@@ -324,6 +332,13 @@ app.include_router(
tags=["定时任务"] tags=["定时任务"]
) )
# 安全测试定时任务管理
app.include_router(
security_scheduled_tasks.router,
prefix="/api/security/scheduled-tasks",
tags=["安全测试-定时任务"]
)
# ==================== 根路径 ==================== # ==================== 根路径 ====================
......
...@@ -2,7 +2,7 @@ ...@@ -2,7 +2,7 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
模块名称:scheduled_task.py 模块名称:scheduled_task.py
模块描述:UI 自动化定时任务数据库模型定义 模块描述:定时任务数据库模型定义(UI 自动化 + 安全测试共用)
作者:czj 作者:czj
创建日期:2026-08-19 创建日期:2026-08-19
...@@ -19,15 +19,18 @@ from app.database import Base ...@@ -19,15 +19,18 @@ from app.database import Base
class ScheduledTask(Base): class ScheduledTask(Base):
""" """
UI 自动化定时任务数据库模型 定时任务数据库模型
存储定时任务的定义:选择哪些模块执行、周期设置、启用状态等。 存储定时任务的定义:选择哪些模块执行、周期设置、启用状态等。
每次到期触发会创建一条 Execution 记录(trigger_type=scheduled, 每次到期触发会创建一条 Execution 记录(trigger_type=scheduled,
trigger_by=任务名),执行完成后自动生成报告到报告中心。 trigger_by=任务名),执行完成后自动生成报告到报告中心。
通过 case_type 字段区分 UI 自动化定时任务与安全测试定时任务。
Attributes: Attributes:
id (str): 任务唯一标识 id (str): 任务唯一标识
name (str): 任务名称(作为执行记录的 trigger_by) name (str): 任务名称(作为执行记录的 trigger_by)
case_type (str): 用例类型 (ui/security)
config_id (str): 安全测试配置ID(case_type=security 时使用)
enabled (bool): 是否启用 enabled (bool): 是否启用
module_ids (list): 选中的模块 ID 列表 module_ids (list): 选中的模块 ID 列表
schedule_type (str): 周期模式 (interval/daily/weekly) schedule_type (str): 周期模式 (interval/daily/weekly)
...@@ -45,8 +48,10 @@ class ScheduledTask(Base): ...@@ -45,8 +48,10 @@ class ScheduledTask(Base):
Example: Example:
>>> task = ScheduledTask( >>> task = ScheduledTask(
... id="scheduled_abc123", ... id="scheduled_abc123",
... name="每日回归", ... name="每日安全扫描",
... module_ids=["module_001", "module_002"], ... case_type="security",
... config_id="sec_cfg_default",
... module_ids=["sec_api01"],
... schedule_type="daily", ... schedule_type="daily",
... run_time="22:00", ... run_time="22:00",
... ) ... )
...@@ -56,6 +61,12 @@ class ScheduledTask(Base): ...@@ -56,6 +61,12 @@ class ScheduledTask(Base):
id: Mapped[str] = mapped_column(String(64), primary_key=True, comment="任务ID") id: Mapped[str] = mapped_column(String(64), primary_key=True, comment="任务ID")
name: Mapped[str] = mapped_column(String(100), nullable=False, comment="任务名称") name: Mapped[str] = mapped_column(String(100), nullable=False, comment="任务名称")
case_type: Mapped[str] = mapped_column(
String(20), default="ui", comment="用例类型: ui/security"
)
config_id: Mapped[Optional[str]] = mapped_column(
String(64), nullable=True, default=None, comment="安全测试配置ID(security时使用)"
)
enabled: Mapped[bool] = mapped_column( enabled: Mapped[bool] = mapped_column(
Boolean, default=True, comment="是否启用" Boolean, default=True, comment="是否启用"
) )
...@@ -116,6 +127,8 @@ class ScheduledTask(Base): ...@@ -116,6 +127,8 @@ class ScheduledTask(Base):
return { return {
"id": self.id, "id": self.id,
"name": self.name, "name": self.name,
"case_type": self.case_type,
"config_id": self.config_id,
"enabled": self.enabled, "enabled": self.enabled,
"module_ids": self.module_ids or [], "module_ids": self.module_ids or [],
"schedule_type": self.schedule_type, "schedule_type": self.schedule_type,
......
此差异已折叠。
...@@ -23,6 +23,8 @@ class ScheduledTaskCreate(BaseModel): ...@@ -23,6 +23,8 @@ class ScheduledTaskCreate(BaseModel):
weekdays: List[int] = Field([], description="星期 [0-6](weekly)") weekdays: List[int] = Field([], description="星期 [0-6](weekly)")
auto_report: bool = Field(True, description="执行完成后自动生成报告") auto_report: bool = Field(True, description="执行完成后自动生成报告")
enabled: bool = Field(True, description="是否启用") enabled: bool = Field(True, description="是否启用")
case_type: str = Field("ui", description="用例类型: ui/security")
config_id: Optional[str] = Field(None, description="安全测试配置ID(security时必填)")
class ScheduledTaskUpdate(BaseModel): class ScheduledTaskUpdate(BaseModel):
...@@ -36,6 +38,8 @@ class ScheduledTaskUpdate(BaseModel): ...@@ -36,6 +38,8 @@ class ScheduledTaskUpdate(BaseModel):
weekdays: Optional[List[int]] = Field(None, description="星期 [0-6]") weekdays: Optional[List[int]] = Field(None, description="星期 [0-6]")
auto_report: Optional[bool] = Field(None, description="自动生成报告") auto_report: Optional[bool] = Field(None, description="自动生成报告")
enabled: Optional[bool] = Field(None, description="是否启用") enabled: Optional[bool] = Field(None, description="是否启用")
case_type: Optional[str] = Field(None, description="用例类型: ui/security")
config_id: Optional[str] = Field(None, description="安全测试配置ID")
class ToggleRequest(BaseModel): class ToggleRequest(BaseModel):
......
...@@ -28,6 +28,7 @@ from sqlalchemy import select ...@@ -28,6 +28,7 @@ from sqlalchemy import select
from app.database import async_session_maker from app.database import async_session_maker
from app.models.scheduled_task import ScheduledTask from app.models.scheduled_task import ScheduledTask
from app.models.test_case import TestCase from app.models.test_case import TestCase
from app.models.execution import Execution
from app.services.execution_service import ExecutionService, is_execution_running from app.services.execution_service import ExecutionService, is_execution_running
from app.services.report_service import ReportService from app.services.report_service import ReportService
from app.utils.id_generator import generate_id from app.utils.id_generator import generate_id
...@@ -103,13 +104,18 @@ def compute_next_run( ...@@ -103,13 +104,18 @@ def compute_next_run(
# ==================== 触发流程 ==================== # ==================== 触发流程 ====================
async def collect_module_cases(db, module_ids: List[str]) -> List[TestCase]: async def collect_module_cases(
db,
module_ids: List[str],
case_type: str = "ui",
) -> List[TestCase]:
""" """
收集选中模块下所有启用的 UI 测试用例(按模块分组 + order 排序) 收集选中模块下所有启用的测试用例(按模块分组 + order 排序)
Args: Args:
db: 数据库会话 db: 异步数据库会话
module_ids (List[str]): 模块 ID 列表 module_ids (List[str]): 模块 ID 列表
case_type (str): 用例类型(ui/security)
Returns: Returns:
List[TestCase]: 用例列表 List[TestCase]: 用例列表
...@@ -121,7 +127,7 @@ async def collect_module_cases(db, module_ids: List[str]) -> List[TestCase]: ...@@ -121,7 +127,7 @@ async def collect_module_cases(db, module_ids: List[str]) -> List[TestCase]:
.where( .where(
TestCase.module_id.in_(module_ids), TestCase.module_id.in_(module_ids),
TestCase.status == "active", TestCase.status == "active",
TestCase.case_type == "ui", TestCase.case_type == case_type,
) )
.order_by(TestCase.module_id, TestCase.order, TestCase.created_at) .order_by(TestCase.module_id, TestCase.order, TestCase.created_at)
) )
...@@ -129,11 +135,112 @@ async def collect_module_cases(db, module_ids: List[str]) -> List[TestCase]: ...@@ -129,11 +135,112 @@ async def collect_module_cases(db, module_ids: List[str]) -> List[TestCase]:
return list(result.scalars().all()) return list(result.scalars().all())
async def _has_running_security_execution(db) -> bool:
"""检查是否已有安全测试执行在运行,避免定时任务并发压垮目标系统。"""
result = await db.execute(
select(Execution.id).where(
Execution.case_type == "security",
Execution.status.in_(["pending", "running"]),
).limit(1)
)
return result.scalar_one_or_none() is not None
async def _run_security_task_once(
task: ScheduledTask,
db,
) -> Optional[str]:
"""
安全测试定时任务触发流程
收集安全测试用例 → 创建执行记录(SecurityService)→ 运行 → 自动生成报告 → 推进调度
Args:
task (ScheduledTask): 定时任务对象(已从数据库加载)
db: 异步数据库会话
Returns:
Optional[str]: 本次创建的执行记录 ID;跳过/失败时返回 None
"""
from app.services.security_service import SecurityService
from app.services.security_report_service import SecurityReportService
# 1. 收集安全测试用例
cases = await collect_module_cases(db, task.module_ids or [], "security")
if not cases:
logger.warning(
f"[定时任务] 安全任务「{task.name}」选中模块下无启用安全用例,跳过本次"
)
task.next_run_at = compute_next_run(task)
await db.flush()
return None
# 2. 执行锁预检:已有安全测试执行在运行 → 跳过
if await _has_running_security_execution(db):
logger.info(
f"[定时任务] 安全任务「{task.name}」到点但已有安全执行在运行,"
f"跳过本次,下次周期再触发"
)
task.next_run_at = compute_next_run(task)
await db.flush()
return None
# 3. 创建执行记录(通过 SecurityService)
case_ids = [c.id for c in cases]
security_service = SecurityService(db)
run_name = f"{task.name}-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
execution = await security_service.create_execution(
config_id=task.config_id,
module_ids=task.module_ids,
case_ids=case_ids,
name=run_name,
)
# 覆盖 trigger_type / trigger_by(SecurityService 默认 manual)
execution.trigger_type = "scheduled"
execution.trigger_by = task.name
await db.commit()
logger.info(
f"[定时任务] 安全任务「{task.name}」触发执行: {execution.id}, "
f"用例 {len(case_ids)} 个"
)
# 4. 运行执行(阻塞等待完成)
run_ok = True
try:
await security_service.run_execution(execution.id)
except Exception as e:
run_ok = False
logger.error(f"[定时任务] 安全任务「{task.name}」执行异常: {e}")
# 5. 自动生成报告(执行成功后才生成)
if task.auto_report and run_ok:
try:
report_service = SecurityReportService()
await report_service.generate_report(execution.id, db)
logger.info(
f"[定时任务] 安全任务「{task.name}」报告已生成: {execution.id}"
)
except Exception as e:
logger.error(
f"[定时任务] 安全任务「{task.name}」生成报告失败: {e}"
)
# 6. 推进调度状态
task.last_run_at = datetime.now()
task.run_count = (task.run_count or 0) + 1
task.next_run_at = compute_next_run(task)
await db.flush()
await db.commit()
return execution.id
async def run_task_once(task_id: str) -> Optional[str]: async def run_task_once(task_id: str) -> Optional[str]:
""" """
执行一次定时任务(创建执行记录 → 运行 → 自动生成报告 → 推进下次调度) 执行一次定时任务(创建执行记录 → 运行 → 自动生成报告 → 推进下次调度)
供后台调度循环与「立即执行」接口共用。 供后台调度循环与「立即执行」接口共用。
按任务 case_type 分流:ui 走原 Playwright 执行路径,security 走安全测试执行器路径。
Args: Args:
task_id (str): 定时任务 ID task_id (str): 定时任务 ID
...@@ -156,7 +263,11 @@ async def run_task_once(task_id: str) -> Optional[str]: ...@@ -156,7 +263,11 @@ async def run_task_once(task_id: str) -> Optional[str]:
logger.warning(f"[定时任务] 任务不存在: {task_id}") logger.warning(f"[定时任务] 任务不存在: {task_id}")
return None return None
# 1. 收集模块用例 # 按 case_type 分流
if task.case_type == "security":
return await _run_security_task_once(task, db)
# ===== 以下为 UI 自动化定时任务原有逻辑 =====
cases = await collect_module_cases(db, task.module_ids or []) cases = await collect_module_cases(db, task.module_ids or [])
if not cases: if not cases:
logger.warning( logger.warning(
......
...@@ -75,6 +75,7 @@ ...@@ -75,6 +75,7 @@
<el-menu-item index="/execution/security">执行中心</el-menu-item> <el-menu-item index="/execution/security">执行中心</el-menu-item>
<el-menu-item index="/reports/security">报告中心</el-menu-item> <el-menu-item index="/reports/security">报告中心</el-menu-item>
<el-menu-item index="/security/erp-config">ERP 配置</el-menu-item> <el-menu-item index="/security/erp-config">ERP 配置</el-menu-item>
<el-menu-item index="/security/scheduled-tasks">定时任务</el-menu-item>
</el-sub-menu> </el-sub-menu>
<!-- 性能测试(父菜单,含三个子页面) --> <!-- 性能测试(父菜单,含三个子页面) -->
...@@ -301,6 +302,7 @@ const DOCUMENT_LABELS: Record<string, string> = { ...@@ -301,6 +302,7 @@ const DOCUMENT_LABELS: Record<string, string> = {
/** 安全测试子页面标签映射(ERP配置等独立子路由) */ /** 安全测试子页面标签映射(ERP配置等独立子路由) */
const SECURITY_SUB_LABELS: Record<string, string> = { const SECURITY_SUB_LABELS: Record<string, string> = {
'erp-config': 'ERP 配置', 'erp-config': 'ERP 配置',
'scheduled-tasks': '定时任务',
} }
/** 当前激活菜单项(无 type 时回退到默认子项以保持高亮) */ /** 当前激活菜单项(无 type 时回退到默认子项以保持高亮) */
......
/**
* 安全测试定时任务管理 API 接口
*
* @module api/securityScheduledTasks
* @author czj
* @date 2026-08-19
*/
import request from '@/utils/request'
/** 安全测试定时任务数据结构 */
export interface SecurityScheduledTask {
id: string
name: string
enabled: boolean
module_ids: string[]
module_names: string[]
config_id: string
config_name: string
schedule_type: string
interval_value: number
interval_unit: string
run_time: string
weekdays: number[]
auto_report: boolean
schedule_description: string
last_run_at: string | null
next_run_at: string | null
run_count: number
created_at: string | null
updated_at: string | null
}
/** 创建/编辑安全测试定时任务请求体 */
export interface SecurityScheduledTaskPayload {
name: string
module_ids: string[]
config_id: string
schedule_type: 'interval' | 'daily' | 'weekly'
interval_value?: number
interval_unit?: 'minutes' | 'hours' | 'days'
run_time?: string
weekdays?: number[]
auto_report?: boolean
enabled?: boolean
}
/**
* 安全测试定时任务管理 API 服务
*/
export const securityScheduledTaskApi = {
/** 获取安全测试定时任务列表 */
async list(): Promise<{ total: number; items: SecurityScheduledTask[] }> {
const response = await request.get('/api/security/scheduled-tasks')
return response as any
},
/** 创建安全测试定时任务 */
async create(data: SecurityScheduledTaskPayload): Promise<SecurityScheduledTask> {
const response = await request.post('/api/security/scheduled-tasks', data)
return response as any
},
/** 更新安全测试定时任务 */
async update(id: string, data: Partial<SecurityScheduledTaskPayload>): Promise<SecurityScheduledTask> {
const response = await request.put(`/api/security/scheduled-tasks/${id}`, data)
return response as any
},
/** 删除安全测试定时任务(不影响已产生的执行记录与报告) */
async remove(id: string): Promise<void> {
await request.delete(`/api/security/scheduled-tasks/${id}`)
},
/** 立即执行 */
async run(id: string): Promise<{ message: string; task_id: string }> {
const response = await request.post(`/api/security/scheduled-tasks/${id}/run`)
return response as any
},
/** 启用 / 停用 */
async toggle(id: string, enabled: boolean): Promise<SecurityScheduledTask> {
const response = await request.post(`/api/security/scheduled-tasks/${id}/toggle`, { enabled })
return response as any
},
/** 运行历史(该任务触发的执行记录) */
async getRuns(id: string, params?: { skip?: number; limit?: number }): Promise<{ total: number; items: any[] }> {
const response = await request.get(`/api/security/scheduled-tasks/${id}/runs`, { params })
return response as any
},
}
\ No newline at end of file
...@@ -237,6 +237,13 @@ const routes: RouteRecordRaw[] = [ ...@@ -237,6 +237,13 @@ const routes: RouteRecordRaw[] = [
component: () => import('@/views/security/ErpConfig.vue'), component: () => import('@/views/security/ErpConfig.vue'),
meta: { title: 'ERP配置' } meta: { title: 'ERP配置' }
}, },
// 安全测试定时任务(复用 scheduled_tasks 表,case_type=security 分流调度)
{
path: '/security/scheduled-tasks',
name: 'SecurityScheduledTasks',
component: () => import('@/views/security/ScheduledTasks.vue'),
meta: { title: '定时任务' }
},
// 系统管理(父菜单) // 系统管理(父菜单)
{ {
path: '/system', path: '/system',
...@@ -266,27 +273,28 @@ const routes: RouteRecordRaw[] = [ ...@@ -266,27 +273,28 @@ const routes: RouteRecordRaw[] = [
] ]
}, },
// 文档管理(父菜单,含文档校正优化和文档翻译两个子页面) // 文档管理(父菜单,含文档校正优化和文档翻译两个子页面)
{ // TODO: 文档管理模块正在开发中,解锁后取消注释
path: '/document', // {
name: 'Document', // path: '/document',
component: () => import('@/views/document/index.vue'), // name: 'Document',
redirect: '/document/optimize', // component: () => import('@/views/document/index.vue'),
meta: { title: '文档管理' }, // redirect: '/document/optimize',
children: [ // meta: { title: '文档管理' },
{ // children: [
path: 'optimize', // {
name: 'DocumentOptimize', // path: 'optimize',
component: () => import('@/views/document/DocumentOptimize.vue'), // name: 'DocumentOptimize',
meta: { title: '文档校正优化' } // component: () => import('@/views/document/DocumentOptimize.vue'),
}, // meta: { title: '文档校正优化' }
{ // },
path: 'translate', // {
name: 'DocumentTranslate', // path: 'translate',
component: () => import('@/views/document/DocumentTranslate.vue'), // name: 'DocumentTranslate',
meta: { title: '文档翻译' } // component: () => import('@/views/document/DocumentTranslate.vue'),
} // meta: { title: '文档翻译' }
] // }
}, // ]
// },
// 旧 /settings 路由 → 重定向到 /system/settings // 旧 /settings 路由 → 重定向到 /system/settings
{ {
path: '/settings', path: '/settings',
......
此差异已折叠。
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论