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

feat(security): 安全测试 ERP 配置/上传/任务创建全链路对接

- 新增 ERP 配置管理页面(ErpConfig.vue),支持查看/修改 ERP 连接参数
- 新增安全报告 ERP 上传流程:支持上传到 ERP 测试单、项目资料、协作文档
- 新增安全报告 ERP 任务创建:预览+创建,含报告解析、紧急程度计算、关联需求搜索
- 前端 Reports.vue 安全报告弹窗新增 ERP 上传/创建任务按钮及对话框
- 新增 4 份需求文档与计划执行文档
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 ce2c75de
# 安全测试ERP配置 · 需求文档(PRD)
> **文档状态**:初稿 · **版本**:v1.0 · **最后更新**:2026-08-13
> **维护者**:czj · **所属模块**:安全测试
> **参考实现**:功能测试报告模块 ERP 配置子菜单
---
## 一、需求背景与目标
### 1.1 背景
当前安全测试模块已作为一级菜单独立运行,包含「用例管理」「执行中心」「报告中心」三个子页面。安全测试报告生成后,需要将报告上传到 ERP 系统进行归档和管理,但目前安全测试模块下缺少 ERP 配置入口。
功能测试报告模块已实现 ERP 配置功能(API Key、Base URL、测试单 ID、Type ID、图片URL前缀),通过 `erp_configs` 数据表持久化。安全测试模块可以直接复用同一套 ERP 配置数据与 API,仅需在前端新增配置页面入口。
### 1.2 目标
在安全测试模块下新增「ERP 配置」子菜单,提供 ERP 连接配置管理功能:
1. **ERP 配置管理**:配置 API Key、Base URL、测试单 ID、Type ID、图片 URL 前缀
2. **配置持久化**:保存到数据库,服务重启后保留
3. **配置回退**:数据库无记录时自动回退到环境变量默认值
### 1.3 范围
| 范围 | 包含 | 不包含 |
|------|------|--------|
| ERP 配置表单 | API Key、Base URL、测试单 ID、Type ID、图片 URL 前缀 | 其他 ERP 字段 |
| 配置持久化 | 数据库读写,重启保留 | 多租户配置隔离 |
| 配置回退 | 数据库无记录时回退环境变量 | 配置版本管理 |
| 后端 API | 复用 `erp_configs` 表的 GET/PUT 接口 | 新建数据表 |
---
## 二、功能需求
### 2.1 功能清单
| 编号 | 功能模块 | 子功能 | 优先级 | 说明 |
|------|---------|--------|--------|------|
| F1 | ERP 配置 | 配置获取 | P0 | 页面加载时自动获取当前 ERP 配置 |
| F2 | ERP 配置 | 配置保存 | P0 | 修改后保存到数据库,重启后持久化 |
| F3 | ERP 配置 | 配置重置 | P1 | 重置回加载时的原始值 |
| F4 | ERP 配置 | 表单校验 | P0 | API Key 和 Base URL 必填校验 |
### 2.2 功能详细描述
#### 2.2.1 配置获取(F1)
- 触发时机:页面加载时自动调用
- 数据来源:优先从数据库 `erp_configs` 表读取,无记录时回退到环境变量默认值
- 字段映射:
| 表单字段 | 数据库字段 | 环境变量 | 默认值 |
|---------|-----------|---------|--------|
| API Key | `api_key` | `ERP_API_KEY` | 空字符串 |
| Base URL | `base_url` | `ERP_BASE_URL` | 空字符串 |
| 测试单 ID | `developtesting_id` | `ERP_DEVELOPTESTING_ID` | 397 |
| Type ID | `type_id` | `ERP_TYPE_ID` | 2 |
| 图片 URL 前缀 | `image_url_prefix` | `ERP_IMAGE_URL_PREFIX` | 空字符串 |
#### 2.2.2 配置保存(F2)
- 所有字段均支持修改
- 保存成功后弹出成功提示
- 更新数据库记录,同时更新运行时环境变量(`settings` 对象)
- 保存后更新原始快照,使"重置"按钮回到最新保存值
#### 2.2.3 配置重置(F3)
- 点击"重置"按钮,表单恢复到加载时的原始值
- 清除表单校验错误状态
- 如果未保存过修改,重置到加载时的值
#### 2.2.4 表单校验(F4)
| 字段 | 规则 | 提示 |
|------|------|------|
| API Key | 必填 | 请输入 ERP API 密钥 |
| Base URL | 必填 | 请输入 ERP 基础 URL |
---
## 三、非功能需求
### 3.1 性能要求
- ERP 配置读取应在 1s 内返回
- 配置保存应在 2s 内返回
### 3.2 可用性要求
- 表单提交时按钮显示加载状态,防止重复提交
- 保存失败时显示错误信息,不丢失已填数据
### 3.3 兼容性要求
- 复用功能测试报告模块的 `erp_configs` 表,不新增数据表
- 复用后端 ERP 配置 API(`GET/PUT /api/functional-report/config`
---
## 四、数据模型
### 4.1 复用现有模型
复用 `backend/app/models/erp_config.py` 中的 `ErpConfig` 模型,表名 `erp_configs`,固定 ID=1 的单行记录:
```
erp_configs
├── id: int (PK, 固定为 1)
├── api_key: varchar(255) — ERP API 密钥
├── base_url: varchar(500) — ERP 基础 URL
├── developtesting_id: int — 默认测试单 ID
├── type_id: int — 报告类型 ID
├── image_url_prefix: varchar(500) — 图片 URL 前缀
└── updated_at: datetime — 更新时间
```
### 4.2 API 设计
| 方法 | 路径 | 说明 | 请求体 | 响应 |
|------|------|------|--------|------|
| GET | `/api/security/erp-config` | 获取 ERP 配置 | — | `{ api_key, base_url, developtesting_id, type_id, image_url_prefix }` |
| PUT | `/api/security/erp-config` | 更新 ERP 配置 | `{ api_key?, base_url?, developtesting_id?, type_id?, image_url_prefix? }` | 同上 |
---
## 五、用户界面
### 5.1 页面布局
- 安全测试 → ERP 配置(子菜单)
- 页面内容:一个卡片容器,包含表单
- 表单宽度:最大 600px,标签右对齐,标签宽度 140px
- 表单字段:API Key(密码框)、Base URL(输入框)、测试单 ID(数字框)、Type ID(数字框)、图片 URL 前缀(输入框)
- 操作按钮:保存配置 + 重置
### 5.2 菜单结构变更
```
安全测试(改前) 安全测试(改后)
├── 用例管理 ├── 用例管理
├── 执行中心 ├── 执行中心
└── 报告中心 ├── 报告中心
└── ERP 配置 ← 新增
```
---
## 六、风险与依赖
| 风险 | 影响 | 缓解措施 |
|------|------|---------|
| 与功能测试报告模块 ERP 配置共用数据表 | 两个模块的配置同步 | 共享同一套 ERP 配置属于预期行为,降低维护成本 |
| 环境变量未配置 | 配置为空 | 数据库无记录时回退环境变量,环境变量无值时回退空字符串 |
---
*本文档为安全测试 ERP 配置功能需求规格说明,供开发和测试参考。*
\ No newline at end of file
此差异已折叠。
......@@ -27,7 +27,7 @@ from fastapi.responses import FileResponse
from app.config import settings
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, performance, performance_output, api_preset, deploy, functional_report, report_template
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
# 配置日志
logging.basicConfig(
......@@ -176,6 +176,25 @@ app.include_router(
tags=["安全测试"]
)
app.include_router(
security_erp.router,
prefix="/api/security",
tags=["安全测试-ERP配置"]
)
app.include_router(
security_erp_upload.router,
prefix="/api/security",
tags=["安全测试-ERP上传"]
)
# 安全测试-ERP任务创建
app.include_router(
security_erp_task.router,
prefix="/api/security",
tags=["安全测试-ERP任务创建"]
)
app.include_router(
ai_locator.router,
prefix="/api/ai",
......@@ -251,6 +270,13 @@ app.include_router(
tags=["功能测试报告"]
)
# 接口测试路由
app.include_router(
api_test.router,
prefix="/api/api-test",
tags=["接口测试"]
)
# ==================== 根路径 ====================
......
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:security_erp.py
模块描述:安全测试 ERP 配置 API 路由,提供获取/更新 ERP 连接配置接口
复用功能测试报告的 erp_configs 表与环境变量配置,共用同一套 ERP 配置
作者:czj
创建日期:2026-08-13
最后修改:2026-08-13
"""
import logging
from fastapi import APIRouter, Depends
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.config import settings
from app.models.erp_config import ErpConfig
from app.schemas.security_erp import (
SecurityErpConfigResponse,
SecurityErpConfigUpdateRequest,
)
logger = logging.getLogger(__name__)
router = APIRouter()
@router.get("/erp-config", response_model=SecurityErpConfigResponse, summary="获取安全测试ERP配置")
async def get_erp_config(db: AsyncSession = Depends(get_db)):
"""
获取安全测试 ERP 配置
优先从数据库读取,若数据库无记录则回退到环境变量默认值。
Returns:
SecurityErpConfigResponse: 当前 ERP 配置
"""
result = await db.execute(select(ErpConfig).where(ErpConfig.id == 1))
config = result.scalar_one_or_none()
if config:
# 数据库有记录,但空字段回退到环境变量默认值(保证配置自动填充)
return SecurityErpConfigResponse(
api_key=config.api_key or settings.ERP_API_KEY or "",
base_url=config.base_url or settings.ERP_BASE_URL or "",
developtesting_id=config.developtesting_id or settings.ERP_DEVELOPTESTING_ID or 0,
type_id=config.type_id or settings.ERP_TYPE_ID or 2,
image_url_prefix=config.image_url_prefix or settings.ERP_IMAGE_URL_PREFIX or "",
)
# 数据库无记录,回退到环境变量
return SecurityErpConfigResponse(
api_key=settings.ERP_API_KEY or "",
base_url=settings.ERP_BASE_URL or "",
developtesting_id=settings.ERP_DEVELOPTESTING_ID or 0,
type_id=settings.ERP_TYPE_ID or 2,
image_url_prefix=settings.ERP_IMAGE_URL_PREFIX or "",
)
@router.put("/erp-config", response_model=SecurityErpConfigResponse, summary="更新安全测试ERP配置")
async def update_erp_config(request: SecurityErpConfigUpdateRequest, db: AsyncSession = Depends(get_db)):
"""
更新安全测试 ERP 配置(持久化到数据库)
更新后的配置会持久化到数据库,服务重启后依然有效,
与功能测试报告模块共享同一套 ERP 配置。
Args:
request: 更新请求(字段可为空,仅更新传入字段)
Returns:
SecurityErpConfigResponse: 更新后的配置
"""
result = await db.execute(select(ErpConfig).where(ErpConfig.id == 1))
config = result.scalar_one_or_none()
if not config:
# 新记录:先用环境变量默认值填充,再用请求值覆盖
config = ErpConfig(
id=1,
api_key=settings.ERP_API_KEY or "",
base_url=settings.ERP_BASE_URL or "",
developtesting_id=settings.ERP_DEVELOPTESTING_ID or 0,
type_id=settings.ERP_TYPE_ID or 2,
image_url_prefix=settings.ERP_IMAGE_URL_PREFIX or "",
)
db.add(config)
# 更新字段(同时同步运行时环境变量,保证本次进程内立即生效)
if request.api_key is not None:
config.api_key = request.api_key
settings.ERP_API_KEY = request.api_key
if request.base_url is not None:
config.base_url = request.base_url
settings.ERP_BASE_URL = request.base_url
if request.developtesting_id is not None:
config.developtesting_id = request.developtesting_id
settings.ERP_DEVELOPTESTING_ID = request.developtesting_id
if request.type_id is not None:
config.type_id = request.type_id
settings.ERP_TYPE_ID = request.type_id
if request.image_url_prefix is not None:
config.image_url_prefix = request.image_url_prefix
settings.ERP_IMAGE_URL_PREFIX = request.image_url_prefix
await db.commit()
await db.refresh(config)
logger.info("安全测试ERP配置已更新(持久化到数据库,与功能测试报告共享)")
return SecurityErpConfigResponse(
api_key=config.api_key or settings.ERP_API_KEY or "",
base_url=config.base_url or settings.ERP_BASE_URL or "",
developtesting_id=config.developtesting_id or settings.ERP_DEVELOPTESTING_ID or 0,
type_id=config.type_id or settings.ERP_TYPE_ID or 2,
image_url_prefix=config.image_url_prefix or settings.ERP_IMAGE_URL_PREFIX or "",
)
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:security_erp_task.py
模块描述:安全测试 ERP 任务创建 API 路由,提供任务创建预览和创建接口
参考实现:ApiSecurityTest/utils/task_creator.py + report_parser.py
作者:czj
创建日期:2026-08-17
最后修改:2026-08-17
"""
import logging
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.schemas.security_erp_task import (
TaskPreviewResponse,
CreateTaskRequest,
CreateTaskResponse,
)
from app.services.security_erp_task_service import SecurityErpTaskService
logger = logging.getLogger(__name__)
router = APIRouter()
@router.get(
"/executions/{execution_id}/task-preview",
response_model=TaskPreviewResponse,
summary="获取安全测试ERP任务创建预览",
)
async def get_task_preview(
execution_id: str,
db: AsyncSession = Depends(get_db),
):
"""
获取安全测试 ERP 任务创建预览数据
在创建任务前调用,返回:
- 报告摘要(测试目标、时间、漏洞统计)
- 自动生成的任务名称
- 自动计算的紧急程度
- ERP 基础数据(任务类型列表、状态列表、人员列表)
- 搜索到的关联需求
- 默认任务配置
前端基于此数据展示对话框,允许用户修改参数后创建任务。
"""
try:
service = SecurityErpTaskService(db)
preview = await service.get_task_preview(execution_id)
return preview
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
except Exception as e:
logger.error(f"获取任务预览失败: {e}")
raise HTTPException(status_code=500, detail=f"获取任务预览失败: {str(e)}")
@router.post(
"/executions/{execution_id}/create-task",
response_model=CreateTaskResponse,
summary="创建安全测试ERP任务",
)
async def create_task(
execution_id: str,
request: CreateTaskRequest,
db: AsyncSession = Depends(get_db),
):
"""
创建安全测试 ERP 任务
基于已生成的报告,构建任务数据并调用 ERP 接口创建任务。
支持自定义任务参数(类型、状态、责任人、紧急程度、截止时间等)。
Args:
execution_id: 安全测试执行 ID
request: 任务配置参数
Returns:
CreateTaskResponse: 创建结果
"""
try:
service = SecurityErpTaskService(db)
result = await service.create_task(
execution_id=execution_id,
config=request.model_dump(),
)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
except Exception as e:
logger.error(f"创建 ERP 任务失败: {e}")
raise HTTPException(status_code=500, detail=f"创建 ERP 任务失败: {str(e)}")
\ No newline at end of file
此差异已折叠。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:security_erp.py
模块描述:安全测试 ERP 配置 Pydantic 数据模型
作者:czj
创建日期:2026-08-13
最后修改:2026-08-13
"""
from typing import Optional
from pydantic import BaseModel, Field
class SecurityErpConfigResponse(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 SecurityErpConfigUpdateRequest(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前缀")
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:security_erp_task.py
模块描述:安全测试 ERP 任务创建 Pydantic Schema
作者:czj
创建日期:2026-08-17
最后修改:2026-08-17
"""
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field
class TaskStaffInfo(BaseModel):
"""ERP 人员信息"""
id: int = Field(..., description="人员ID")
name: str = Field("", description="人员姓名")
class TaskTypeInfo(BaseModel):
"""ERP 任务类型"""
id: int = Field(..., description="类型ID")
name: str = Field("", description="类型名称")
class TaskStatusInfo(BaseModel):
"""ERP 任务状态"""
id: int = Field(..., description="状态ID")
name: str = Field("", description="状态名称")
class RequirementInfo(BaseModel):
"""ERP 需求信息"""
id: int = Field(..., description="需求ID")
name: str = Field("", description="需求名称")
class TaskDefaultConfig(BaseModel):
"""任务默认配置"""
type_id: int = Field(3, description="任务类型ID")
status_id: int = Field(1, description="任务状态ID")
level: int = Field(4, description="紧急程度(1-5)")
excepthour: int = Field(21, description="期望工时(小时)")
user_id: int = Field(101, description="责任人ID")
createuser_id: int = Field(441, description="创建人ID")
deadline_days: int = Field(7, description="截止天数")
class ReportSummary(BaseModel):
"""报告摘要"""
target: str = Field("", description="测试目标URL")
server_ip: str = Field("", description="服务器IP")
start_time: str = Field("", description="测试开始时间")
end_time: str = Field("", description="测试结束时间")
total: int = Field(0, description="测试用例总数")
high: int = Field(0, description="高危漏洞数")
medium: int = Field(0, description="中危漏洞数")
low: int = Field(0, description="低危漏洞数")
info: int = Field(0, description="信息类数")
safe: int = Field(0, description="安全项数")
class TaskPreviewResponse(BaseModel):
"""任务创建预览响应"""
summary: ReportSummary = Field(..., description="报告摘要")
task_name: str = Field("", description="自动生成的任务名称")
calculated_level: int = Field(4, description="自动计算的紧急程度")
task_types: List[TaskTypeInfo] = Field(default_factory=list, description="任务类型列表")
task_statuses: List[TaskStatusInfo] = Field(default_factory=list, description="任务状态列表")
staff_list: List[TaskStaffInfo] = Field(default_factory=list, description="人员列表")
requirement: Optional[RequirementInfo] = Field(None, description="关联需求")
default_config: TaskDefaultConfig = Field(default_factory=TaskDefaultConfig, description="默认配置")
class CreateTaskRequest(BaseModel):
"""创建 ERP 任务请求"""
type_id: int = Field(3, description="任务类型ID")
status_id: int = Field(1, description="任务状态ID")
level: int = Field(4, description="紧急程度(1-5)")
excepthour: int = Field(21, description="期望工时(小时)")
user_id: int = Field(101, description="责任人ID")
createuser_id: int = Field(441, description="创建人ID")
deadline_days: int = Field(7, description="截止天数(从当前时间+N天)")
develop_id: Optional[int] = Field(None, description="关联需求ID(可选)")
task_name: Optional[str] = Field(None, description="任务名称(可选,自动生成)")
class CreateTaskResponse(BaseModel):
"""创建 ERP 任务响应"""
success: bool = Field(..., description="是否成功")
task_id: Optional[int] = Field(None, description="ERP任务ID")
task_name: str = Field("", description="任务名称")
message: str = Field("", description="提示信息")
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:erp_task_client.py
模块描述:ERP 任务 API 客户端,封装任务类型/状态/人员/需求/任务创建接口
含缓存和重试机制,参考实现:ApiSecurityTest/utils/erp_client.py
作者:czj
创建日期:2026-08-17
最后修改:2026-08-17
"""
import logging
import time
from typing import Dict, List, Optional, Any
import requests
logger = logging.getLogger(__name__)
# API 端点常量
TASK_TYPE_ENDPOINT = "/openclaw/tasktype"
TASK_STATUS_ENDPOINT = "/openclaw/taskstatus"
STAFF_ENDPOINT = "/openclaw/stuff"
DEVELOP_ENDPOINT = "/openclaw/develop"
TASK_ENDPOINT = "/openclaw/task"
class ERPTaskClient:
"""
ERP 任务 API 客户端
缓存策略:
- 任务类型、任务状态、人员列表缓存 1 小时
- 需求列表、任务创建不缓存
"""
def __init__(
self,
base_url: str,
api_key: str,
timeout: int = 30,
max_retries: int = 3,
retry_interval: int = 2,
cache_duration: int = 3600,
):
"""
Args:
base_url: ERP 基础 URL(如 https://office.ubainsyun.com:5082/api/uerp)
api_key: ERP API 密钥
timeout: 请求超时时间(秒)
max_retries: 最大重试次数
retry_interval: 重试间隔(秒)
cache_duration: 基础数据缓存时长(秒)
"""
self.base_url = base_url.rstrip("/")
self.api_key = api_key
self.timeout = timeout
self.max_retries = max_retries
self.retry_interval = retry_interval
self.cache_duration = cache_duration
# 缓存
self._cache: Dict[str, Any] = {}
self._cache_timestamps: Dict[str, float] = {}
# ==================== 公开 API ====================
def get_task_types(self) -> Optional[List[Dict]]:
"""
获取任务类型列表
Returns:
[{"id": 1, "name": "UI设计"}, ...],失败返回 None
"""
result = self._get_with_cache("task_types", lambda: self._make_request("GET", TASK_TYPE_ENDPOINT))
if result and isinstance(result, dict) and "data" in result:
return result["data"]
return result
def get_task_statuses(self) -> Optional[List[Dict]]:
"""
获取任务状态列表
Returns:
[{"id": 1, "name": "新建"}, ...],失败返回 None
"""
result = self._get_with_cache("task_statuses", lambda: self._make_request("GET", TASK_STATUS_ENDPOINT))
if result and isinstance(result, dict) and "data" in result:
return result["data"]
return result
def get_staff_list(self) -> Optional[List[Dict]]:
"""
获取人员列表
Returns:
[{"id": 1, "name": "超管员", ...}, ...],失败返回 None
"""
result = self._get_with_cache("staff_list", lambda: self._make_request("GET", STAFF_ENDPOINT))
if result and isinstance(result, dict) and "data" in result:
return result["data"]
return result
def search_develop(
self,
search_str: str,
page_num: int = 1,
page_size: int = 10,
) -> Optional[List[Dict]]:
"""
搜索需求列表
Args:
search_str: 搜索关键词
page_num: 页码
page_size: 每页数量
Returns:
需求列表,格式 [{"id": 128, "name": "需求名称", ...}, ...]
失败返回 None
"""
params = {"pageNum": page_num, "pageSize": page_size}
if search_str:
params["search_str"] = search_str
result = self._make_request("GET", DEVELOP_ENDPOINT, params=params)
if result is None:
return None
# 响应可能是 dict 带 data 字段,也可能是直接列表
if isinstance(result, dict) and "data" in result:
data = result["data"]
if isinstance(data, list):
return data
elif isinstance(result, list):
return result
return None
def create_task(self, task_data: Dict) -> Optional[Dict]:
"""
创建 ERP 任务
Args:
task_data: 任务数据字典
Returns:
ERP 响应结果,失败返回 None
"""
return self._make_request("POST", TASK_ENDPOINT, json=task_data)
def clear_cache(self):
"""清空所有缓存"""
self._cache.clear()
self._cache_timestamps.clear()
logger.info("ERP 任务客户端缓存已清空")
# ==================== 内部方法 ====================
def _get_with_cache(self, cache_key: str, fetch_func) -> Optional[Any]:
"""
带缓存的数据获取
Args:
cache_key: 缓存键名
fetch_func: 获取数据的函数
Returns:
缓存或新获取的数据
"""
now = time.time()
if cache_key in self._cache:
timestamp = self._cache_timestamps.get(cache_key, 0)
if now - timestamp < self.cache_duration:
logger.debug(f"使用缓存数据: {cache_key}")
return self._cache[cache_key]
data = fetch_func()
if data is not None:
self._cache[cache_key] = data
self._cache_timestamps[cache_key] = now
return data
def _make_request(
self,
method: str,
endpoint: str,
**kwargs,
) -> Optional[Dict]:
"""
统一请求方法(含重试机制)
Args:
method: HTTP 方法(GET/POST)
endpoint: API 端点路径
**kwargs: 其他请求参数(params, json 等)
Returns:
响应数据字典,失败返回 None
"""
url = f"{self.base_url}{endpoint}"
headers = {"X-Api-Key": self.api_key, "Content-Type": "application/json"}
# 分离参数
json_data = kwargs.pop("json", None)
params = kwargs.pop("params", None)
for attempt in range(self.max_retries):
try:
logger.info(f"ERP API 请求: {method} {endpoint}")
if method == "GET":
response = requests.get(
url, headers=headers, params=params, timeout=self.timeout
)
elif method == "POST":
request_kwargs = {}
if json_data:
request_kwargs["json"] = json_data
if params:
request_kwargs["params"] = params
response = requests.post(
url, headers=headers, timeout=self.timeout, **request_kwargs
)
else:
logger.error(f"不支持的 HTTP 方法: {method}")
return None
if response.status_code == 200:
result = response.json()
if result.get("success") == 1:
logger.info(f"✓ ERP API 请求成功: {endpoint}")
return result
else:
error_code = result.get("error", 0)
msg = result.get("msg", "未知错误")
if 400 <= error_code < 500:
logger.error(f"✗ ERP API 客户端错误 ({error_code}): {msg}")
return None
if attempt < self.max_retries - 1:
logger.warning(
f"ERP API 服务端错误,{self.retry_interval}s 后重试 ({attempt+1}/{self.max_retries}): {msg}"
)
time.sleep(self.retry_interval)
else:
logger.error(f"✗ ERP API 请求失败 (已达最大重试次数): {msg}")
return None
else:
if attempt < self.max_retries - 1 and response.status_code >= 500:
logger.warning(
f"ERP API HTTP {response.status_code},{self.retry_interval}s 后重试 ({attempt+1}/{self.max_retries})"
)
time.sleep(self.retry_interval)
else:
logger.error(f"✗ ERP API HTTP 错误: {response.status_code}")
return None
except requests.exceptions.Timeout:
if attempt < self.max_retries - 1:
logger.warning(
f"ERP API 请求超时,{self.retry_interval}s 后重试 ({attempt+1}/{self.max_retries})"
)
time.sleep(self.retry_interval)
else:
logger.error("✗ ERP API 请求超时 (已达最大重试次数)")
return None
except requests.exceptions.RequestException as e:
logger.error(f"✗ ERP API 请求异常: {e}")
return None
except Exception as e:
logger.error(f"✗ ERP API 未知异常: {e}")
return None
return None
\ No newline at end of file
此差异已折叠。
......@@ -68,6 +68,7 @@
<el-menu-item index="/cases/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="/security/erp-config">ERP 配置</el-menu-item>
</el-sub-menu>
<!-- 性能测试(父菜单,含三个子页面) -->
......@@ -82,6 +83,17 @@
<el-menu-item index="/performance/presets">接口预设</el-menu-item>
</el-sub-menu>
<!-- 接口测试(一级菜单,含用例管理/执行中心/报告中心) -->
<el-sub-menu index="/api-test">
<template #title>
<el-icon><Connection /></el-icon>
<span>接口测试</span>
</template>
<el-menu-item index="/api-test/cases">用例管理</el-menu-item>
<el-menu-item index="/api-test/execution">执行中心</el-menu-item>
<el-menu-item index="/api-test/report">报告中心</el-menu-item>
</el-sub-menu>
<!-- 自动化部署测试(父菜单) -->
<el-sub-menu index="/deploy">
<template #title>
......@@ -202,7 +214,8 @@ import {
Tools,
Lock,
Setting,
EditPen
EditPen,
Connection
} from '@element-plus/icons-vue'
// ==================== 响应式数据 ====================
......@@ -220,6 +233,13 @@ const CASE_TYPE_LABELS: Record<string, string> = {
security: '安全自动化测试',
}
/** 接口测试子页面标签映射 */
const API_TEST_LABELS: Record<string, string> = {
cases: '用例管理',
execution: '执行中心',
report: '报告中心',
}
/** 设备模拟子页面标签映射 */
const DEVICE_SIM_LABELS: Record<string, string> = {
settings: '环境配置',
......@@ -271,13 +291,18 @@ const DOCUMENT_LABELS: Record<string, string> = {
translate: '文档翻译',
}
/** 安全测试子页面标签映射(ERP配置等独立子路由) */
const SECURITY_SUB_LABELS: Record<string, string> = {
'erp-config': 'ERP 配置',
}
/** 当前激活菜单项(无 type 时回退到默认子项以保持高亮) */
const currentRoute = computed(() => {
const seg = route.path.split('/').filter(Boolean)
const root = '/' + (seg[0] || '')
// 无 type 参数时回退到默认子项,保证菜单高亮
if (seg.length === 1 && ['/cases', '/execution', '/reports', '/system', '/deploy', '/device-sim', '/performance', '/security', '/document'].includes(root)) {
const defaultType = root === '/system' ? 'settings' : root === '/deploy' ? 'servers' : root === '/device-sim' ? 'settings' : root === '/performance' ? 'tasks' : root === '/security' ? 'security' : root === '/document' ? 'optimize' : 'ui'
if (seg.length === 1 && ['/cases', '/execution', '/reports', '/system', '/deploy', '/device-sim', '/performance', '/security', '/document', '/api-test'].includes(root)) {
const defaultType = root === '/system' ? 'settings' : root === '/deploy' ? 'servers' : root === '/device-sim' ? 'settings' : root === '/performance' ? 'tasks' : root === '/security' ? 'security' : root === '/document' ? 'optimize' : root === '/api-test' ? 'cases' : 'ui'
return `${root}/${defaultType}`
}
return route.path
......@@ -302,6 +327,7 @@ const pageTitle = computed(() => {
'/device-sim': '设备模拟',
'/system': '系统管理',
'/document': '文档管理',
'/api-test': '接口测试',
}
const base = baseMap[root] || '测试管理平台'
const type = seg[1]
......@@ -319,7 +345,9 @@ const pageTitle = computed(() => {
if (root === '/system') return SYSTEM_LABELS[type]
if (root === '/deploy') return DEPLOY_LABELS[type]
if (root === '/performance') return PERFORMANCE_LABELS[type]
if (root === '/security') return SECURITY_SUB_LABELS[type]
if (root === '/document') return DOCUMENT_LABELS[type]
if (root === '/api-test') return API_TEST_LABELS[type]
return ''
})()
......
......@@ -94,10 +94,8 @@ export function getSecuritySummary(executionId: string) {
}
/** 获取安全测试报告内容(Markdown) */
export function getSecurityReport(executionId: string) {
return request.get<{ execution_id: string; content: string; format: string }>(
`/api/security/executions/${executionId}/report`
)
export function getSecurityReport(executionId: string): Promise<{ execution_id: string; content: string; format: string }> {
return request.get(`/api/security/executions/${executionId}/report`) as any
}
/** 安全测试报告下载地址 */
......@@ -105,6 +103,68 @@ export function getSecurityReportDownloadUrl(executionId: string): string {
return `/api/security/executions/${executionId}/report/download`
}
// ==================== ERP 上传 ====================
/** 上传安全测试报告到 ERP 测试单 */
export function uploadSecurityReportToErp(
executionId: string,
data: {
developtesting_id?: number
copyuser_names?: string[]
createuser_name?: string
}
): Promise<{ success: boolean; report_id?: number; message: string }> {
return request.post(`/api/security/executions/${executionId}/upload-to-erp`, data) as any
}
/** 上传安全测试报告到 ERP 项目资料 */
export function uploadSecurityReportToProject(
executionId: string,
data: {
project_id: number
name?: string
}
): Promise<{ success: boolean; file_id?: number; message: string }> {
return request.post(`/api/security/executions/${executionId}/upload-to-project`, data) as any
}
/** 上传安全测试报告到 ERP 协作文档 */
export function uploadSecurityReportToCooperation(
executionId: string,
data: {
project_id: number
name?: string
group_id?: number
}
): Promise<{ success: boolean; file_id?: number; message: string }> {
return request.post(`/api/security/executions/${executionId}/upload-to-cooperation`, data) as any
}
// ==================== ERP 任务创建 ====================
/** 获取安全测试 ERP 任务创建预览数据 */
export function getSecurityTaskPreview(executionId: string): Promise<any> {
return request.get(`/api/security/executions/${executionId}/task-preview`) as any
}
/** 创建安全测试 ERP 任务 */
export function createSecurityTask(
executionId: string,
data: {
type_id: number
status_id: number
level: number
excepthour: number
user_id: number
createuser_id: number
deadline_days: number
develop_id?: number
task_name?: string
}
): Promise<{ success: boolean; task_id?: number; task_name: string; message: string }> {
return request.post(`/api/security/executions/${executionId}/create-task`, data) as any
}
// ==================== 知识库 ====================
/** 获取安全知识库 */
......
/**
* 安全测试 ERP 配置 API 封装
*
* @author czj
* @date 2026-08-13
*/
import request from '@/utils/request'
export const securityErpApi = {
/**
* 获取 ERP 配置
* 优先从数据库读取,无记录时回退到环境变量默认值
*/
async getConfig(): Promise<{
api_key: string
base_url: string
developtesting_id: number
type_id: number
image_url_prefix: string
}> {
const response = await request.get('/api/security/erp-config')
return response as any
},
/**
* 更新 ERP 配置
* 持久化到数据库,同时更新运行时环境变量
*/
async updateConfig(config: {
api_key?: string
base_url?: string
developtesting_id?: number
type_id?: number
image_url_prefix?: string
}): Promise<any> {
const response = await request.put('/api/security/erp-config', config)
return response as any
}
}
\ No newline at end of file
......@@ -183,6 +183,34 @@ const routes: RouteRecordRaw[] = [
}
]
},
// 接口测试模块(一级菜单,含用例管理/执行中心/报告中心)
{
path: '/api-test',
name: 'ApiTest',
component: () => import('@/views/api-test/index.vue'),
redirect: '/api-test/cases',
meta: { title: '接口测试' },
children: [
{
path: 'cases',
name: 'ApiTestCases',
component: () => import('@/views/api-test/ApiCaseList.vue'),
meta: { title: '用例管理' }
},
{
path: 'execution',
name: 'ApiTestExecution',
component: () => import('@/views/Execution.vue'),
meta: { title: '执行中心' }
},
{
path: 'report',
name: 'ApiTestReport',
component: () => import('@/views/Reports.vue'),
meta: { title: '报告中心' }
}
]
},
// 重定向:旧版路径 → 性能测试模块
{
path: '/cases/performance',
......@@ -196,6 +224,13 @@ const routes: RouteRecordRaw[] = [
path: '/reports/performance',
redirect: '/performance'
},
// 安全测试 ERP 配置(独立子页面,复用 erp_configs 表,与功能测试报告共享配置)
{
path: '/security/erp-config',
name: 'SecurityErpConfig',
component: () => import('@/views/security/ErpConfig.vue'),
meta: { title: 'ERP配置' }
},
// 系统管理(父菜单)
{
path: '/system',
......
此差异已折叠。
<!--
组件名称:ErpConfig.vue
组件描述:安全测试 ERP 配置管理页面
@author czj
@date 2026-08-13
-->
<template>
<div class="erp-config-page">
<el-card shadow="never">
<template #header>
<div class="card-header">
<span>ERP 配置</span>
</div>
</template>
<el-form
ref="formRef"
:model="form"
:rules="rules"
label-width="140px"
label-position="right"
style="max-width: 600px"
>
<el-form-item label="API Key" prop="api_key">
<el-input
v-model="form.api_key"
type="password"
show-password
placeholder="请输入 ERP API 密钥"
clearable
/>
</el-form-item>
<el-form-item label="Base URL" prop="base_url">
<el-input
v-model="form.base_url"
placeholder="请输入 ERP 基础 URL"
clearable
/>
</el-form-item>
<el-form-item label="测试单 ID" prop="developtesting_id">
<el-input-number
v-model="form.developtesting_id"
:min="0"
:max="999999"
:step="1"
style="width: 100%"
/>
</el-form-item>
<el-form-item label="Type ID" prop="type_id">
<el-input-number
v-model="form.type_id"
:min="0"
:max="999"
:step="1"
style="width: 100%"
/>
</el-form-item>
<el-form-item label="图片 URL 前缀" prop="image_url_prefix">
<el-input
v-model="form.image_url_prefix"
placeholder="请输入图片 URL 前缀"
clearable
/>
</el-form-item>
<el-form-item>
<el-button type="primary" :loading="saving" @click="handleSave">
{{ saving ? '保存中...' : '保存配置' }}
</el-button>
<el-button @click="handleReset">重置</el-button>
</el-form-item>
</el-form>
</el-card>
</div>
</template>
<script setup lang="ts">
/**
* 安全测试 ERP 配置管理页面
*
* 功能:
* 1. 加载时自动获取当前 ERP 配置(从数据库 / 环境变量回退)
* 2. 修改后保存到数据库,重启后持久化
* 3. 与功能测试报告模块共用同一套 ERP 配置(erp_configs 表)
*/
import { reactive, ref, onMounted } from 'vue'
import { ElMessage } from 'element-plus'
import { securityErpApi } from '@/api/securityErp'
// ==================== 类型定义 ====================
interface ErpConfigForm {
api_key: string
base_url: string
developtesting_id: number
type_id: number
image_url_prefix: string
}
// ==================== 响应式数据 ====================
const formRef = ref()
const saving = ref(false)
const form = reactive<ErpConfigForm>({
api_key: '',
base_url: '',
developtesting_id: 397,
type_id: 2,
image_url_prefix: '',
})
/** 原始数据快照,用于重置 */
let originalSnapshot = ''
// ==================== 表单校验规则 ====================
const rules = {
api_key: [{ required: true, message: '请输入 ERP API 密钥', trigger: 'blur' }],
base_url: [{ required: true, message: '请输入 ERP 基础 URL', trigger: 'blur' }],
}
// ==================== 方法 ====================
/** 加载配置 */
async function loadConfig() {
try {
const res = await securityErpApi.getConfig()
Object.assign(form, res)
originalSnapshot = JSON.stringify(res)
} catch (err: any) {
ElMessage.error('加载 ERP 配置失败:' + (err?.message || '未知错误'))
}
}
/** 保存配置 */
async function handleSave() {
const valid = await formRef.value?.validate().catch(() => false)
if (!valid) return
saving.value = true
try {
await securityErpApi.updateConfig({
api_key: form.api_key,
base_url: form.base_url,
developtesting_id: form.developtesting_id,
type_id: form.type_id,
image_url_prefix: form.image_url_prefix,
})
ElMessage.success('ERP 配置已保存')
originalSnapshot = JSON.stringify({ ...form })
} catch (err: any) {
ElMessage.error('保存 ERP 配置失败:' + (err?.message || '未知错误'))
} finally {
saving.value = false
}
}
/** 重置表单到加载时的值 */
function handleReset() {
const snapshot = JSON.parse(originalSnapshot)
Object.assign(form, snapshot)
formRef.value?.clearValidate()
}
// ==================== 生命周期 ====================
onMounted(() => {
loadConfig()
})
</script>
<style scoped>
.erp-config-page {
max-width: 800px;
}
.card-header {
font-size: 16px;
font-weight: 600;
}
</style>
\ No newline at end of file
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论