add single page

This commit is contained in:
gongwenxin 2025-08-19 14:44:57 +08:00
parent 336913fbd0
commit fc2b64ccc4
22 changed files with 2309 additions and 1431 deletions

62
Dockerfile.fastapi Normal file
View File

@ -0,0 +1,62 @@
# DMS合规性测试工具 - FastAPI版本 Dockerfile
# 多阶段构建,优化镜像大小
# 第一阶段:构建阶段
FROM python:3.9-alpine AS builder
# 安装构建依赖
RUN apk update && \
apk add --no-cache \
gcc \
musl-dev \
linux-headers \
libffi-dev \
openssl-dev \
cargo \
rust && \
rm -rf /var/cache/apk/*
# 设置工作目录
WORKDIR /app
# 复制requirements并安装Python依赖
COPY requirements_fastapi.txt .
RUN pip install --upgrade pip setuptools wheel && \
pip install --no-cache-dir --user -r requirements_fastapi.txt
# 第二阶段:运行时镜像
FROM python:3.9-alpine
# 安装运行时依赖
RUN apk update && \
apk add --no-cache \
curl \
bash && \
rm -rf /var/cache/apk/*
# 从构建阶段复制Python包
COPY --from=builder /root/.local /root/.local
# 设置工作目录
WORKDIR /app
# 复制应用代码
COPY . .
# 创建必要目录
RUN mkdir -p /app/logs /app/test_reports /app/uploads
# 设置环境变量
ENV PATH=/root/.local/bin:$PATH
ENV PYTHONPATH=/app
ENV PYTHONUNBUFFERED=1
# 暴露端口
EXPOSE 5050
# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \
CMD curl -f http://localhost:5050/ || exit 1
# 启动FastAPI服务器
CMD ["python3", "fastapi_server.py", "--host", "0.0.0.0", "--port", "5050"]

View File

@ -1,242 +0,0 @@
# DMS合规性测试工具 - 离线部署方案总结
## 概述
已成功创建了DMS合规性测试工具的完整离线部署方案可以在没有网络连接的环境中部署和运行。
## 生成的文件
### 主要部署包
- **`dms-compliance-complete-offline-20250813-074148.tar.gz`** (577MB)
- 完整的离线部署包,包含所有必要组件
### 脚本文件
- **`create-offline-package.sh`** - 创建完整离线部署包的主脚本
- **`docker-export.sh`** - 导出Docker镜像的脚本
- **`install-docker.sh`** - 在目标环境安装Docker的脚本
- **`test-deployment.sh`** - 验证部署包完整性的测试脚本
### 文档文件
- **`DEPLOYMENT_GUIDE.md`** - 详细的部署指南
- **`OFFLINE_DEPLOYMENT_SUMMARY.md`** - 本总结文档
## 部署包内容
### Docker镜像
- **DMS应用镜像** (922MB) - 包含完整的DMS合规性测试工具
- **Nginx镜像** (52MB) - 用于反向代理(可选)
### 部署脚本
- **`install.sh`** - 一键安装脚本(推荐使用)
- **`deploy.sh`** - 应用部署脚本
- **`stop.sh`** - 停止服务脚本
- **`uninstall.sh`** - 完全卸载脚本
- **`install-docker.sh`** - Docker安装脚本
### 配置文件
- **`docker-compose.yml`** - Docker服务编排配置
- **`nginx/`** - Nginx配置目录如果存在
- **`config/`** - 应用配置目录(如果存在)
### 数据目录
- **`data/test_reports/`** - 测试报告存储(包含现有报告)
- **`data/uploads/`** - 上传文件存储
- **`data/logs/`** - 应用日志存储
## 快速部署流程
### 1. 在有网络的环境中准备
```bash
# 创建离线部署包
./create-offline-package.sh
# 验证部署包完整性(可选)
./test-deployment.sh
```
### 2. 传输到目标环境
```bash
# 将生成的tar.gz文件传输到目标服务器
scp dms-compliance-complete-offline-*.tar.gz user@target:/path/
```
### 3. 在目标环境中部署
```bash
# 解压部署包
tar -xzf dms-compliance-complete-offline-*.tar.gz
# 进入部署目录
cd dms-compliance-offline
# 一键安装
./install.sh
```
### 4. 验证部署
访问以下地址确认服务正常:
- **API服务器**: http://localhost:5050
- **历史查看器**: http://localhost:5051
## 系统要求
### 最低配置
- **CPU**: 2核心
- **内存**: 2GB
- **磁盘**: 5GB可用空间
- **操作系统**: Ubuntu 18.04+, CentOS 7+, RHEL 7+, Debian 9+
### 推荐配置
- **CPU**: 4核心
- **内存**: 4GB
- **磁盘**: 10GB可用空间
## 特性和优势
### 完全离线
- ✅ 包含所有必要的Docker镜像
- ✅ 包含Docker和Docker Compose安装脚本
- ✅ 无需任何网络连接即可部署
### 自动化部署
- ✅ 一键安装脚本,自动检测环境
- ✅ 自动安装Docker如果需要
- ✅ 自动配置和启动服务
### 数据持久化
- ✅ 测试报告持久化存储
- ✅ 上传文件持久化存储
- ✅ 应用日志持久化存储
### 服务管理
- ✅ 简单的启动/停止脚本
- ✅ 健康检查和自动重启
- ✅ 完整的卸载功能
### 可扩展性
- ✅ 支持Nginx反向代理
- ✅ 支持自定义配置
- ✅ 支持多环境部署
## 已解决的问题
### 1. Docker容器启动问题
- **问题**: supervisord权限配置错误导致容器重启
- **解决**: 修复supervisord配置移除root用户设置调整日志路径
### 2. 服务端口和绑定问题
- **问题**: API服务器绑定到127.0.0.1:5002无法从容器外访问
- **解决**: 修改为绑定到0.0.0.0:5050
### 3. 健康检查失败问题
- **问题**: API服务器没有根路径路由健康检查返回404
- **解决**: 添加健康检查端点 `GET /`
### 4. 离线部署依赖问题
- **问题**: 需要网络连接下载Docker镜像和安装包
- **解决**: 预先打包所有必要组件包括Docker安装脚本
## 测试验证
### 部署包完整性测试
- ✅ 所有必需文件存在
- ✅ 脚本权限正确
- ✅ Docker镜像文件完整
- ✅ 配置文件语法正确
- ✅ 数据目录结构正确
### 功能测试
- ✅ Docker镜像可以正常加载
- ✅ 服务可以正常启动
- ✅ API服务器响应正常
- ✅ 历史查看器显示正常
- ✅ 健康检查通过
## 使用场景
### 适用环境
- 🎯 内网隔离环境
- 🎯 安全要求高的环境
- 🎯 网络条件受限的环境
- 🎯 需要快速部署的环境
### 典型用户
- 🎯 企业内部IT部门
- 🎯 政府机构
- 🎯 金融机构
- 🎯 医疗机构
## 维护和支持
### 日常维护
```bash
# 查看服务状态
docker-compose ps
# 查看日志
docker-compose logs
# 重启服务
docker-compose restart
# 备份数据
tar -czf backup-$(date +%Y%m%d).tar.gz data/
```
### 故障排除
- 详细的故障排除指南在 `DEPLOYMENT_GUIDE.md`
- 包含常见问题和解决方案
- 提供日志查看和诊断方法
### 更新升级
- 支持数据备份和恢复
- 支持滚动更新
- 保持配置和数据的连续性
## 安全考虑
### 网络安全
- 默认只绑定到localhost
- 支持防火墙配置
- 可配置Nginx反向代理
### 数据安全
- 数据持久化存储
- 支持定期备份
- 权限控制
### 系统安全
- 非root用户运行
- 容器隔离
- 最小权限原则
## 性能优化
### 资源使用
- 优化的Docker镜像大小
- 合理的内存和CPU配置
- 高效的日志管理
### 存储优化
- 压缩的部署包
- 增量备份支持
- 日志轮转配置
## 总结
DMS合规性测试工具的离线部署方案已经完全就绪具备以下特点
1. **完整性** - 包含所有必要组件,无需外部依赖
2. **易用性** - 一键安装,自动化部署
3. **可靠性** - 经过完整测试验证
4. **可维护性** - 提供完整的管理工具
5. **安全性** - 符合企业级安全要求
部署包大小为577MB包含了完整的应用程序、Docker镜像、配置文件和历史测试数据。可以在任何支持Docker的Linux环境中快速部署和运行。
## 下一步行动
1. **传输部署包** - 将 `dms-compliance-complete-offline-20250813-074148.tar.gz` 传输到目标环境
2. **执行部署** - 在目标环境运行 `./install.sh`
3. **验证功能** - 访问服务地址确认正常运行
4. **配置网络** - 根据需要配置防火墙和反向代理
5. **设置备份** - 建立定期数据备份机制

View File

@ -1,144 +0,0 @@
# 项目整理总结
## 🎯 整理目标
将项目文件按功能分类整理,减少根目录混乱,提高项目的可维护性和专业性。
## ✅ 完成的整理工作
### 1. 创建了专门的目录结构
```
compliance/
├── docker/ # Docker相关文件
├── tests/ # 测试脚本
├── docs/ # 文档文件
├── nginx/ # Nginx配置
├── memory-bank/ # 项目上下文
└── ... # 其他现有目录
```
### 2. 文件移动和分类
#### Docker相关文件 → `docker/`
- `Dockerfile.service``docker/Dockerfile.service`
- `Dockerfile.simple``docker/Dockerfile.simple`
- `supervisord.conf``docker/supervisord.conf`
- `start_services.sh``docker/start_services.sh`
#### 测试脚本 → `tests/`
- `test_*.py``tests/test_*.py`
- `test-docker.sh``tests/test-docker.sh`
#### 文档文件 → `docs/`
- `*_Guide.md``docs/`
- `*_Summary.md``docs/`
- `*_Reference.md``docs/`
- `example_usage.py``docs/`
### 3. 更新了路径引用
#### docker-build.sh
- 更新Dockerfile路径`docker/Dockerfile.service`
#### docker-compose.yml
- 更新dockerfile路径`docker/Dockerfile.service`
#### Docker文件内部路径
- 更新配置文件复制路径
- 确保构建时能正确找到文件
### 4. 完善了.gitignore文件
添加了完整的Python、Docker、IDE等相关的忽略规则
- Python缓存和虚拟环境
- 测试报告和上传文件
- 日志和临时文件
- IDE和系统文件
- 敏感配置文件
### 5. 更新了README.md
- 简化了项目说明
- 添加了清晰的使用指南
- 包含了项目结构说明
- 保留了重要的TODO项目
## 📁 整理后的根目录
现在根目录更加简洁,只包含最重要的文件:
```
compliance/
├── api_server.py # 主要服务
├── history_viewer.py # 历史查看器
├── run_api_tests.py # 命令行工具
├── docker-build.sh # Docker构建脚本
├── docker-compose.yml # Docker编排
├── requirements.txt # 依赖文件
├── README.md # 项目说明
├── .gitignore # Git忽略规则
├── docker/ # Docker相关文件
├── tests/ # 测试脚本
├── docs/ # 文档
└── ... # 其他现有目录
```
## 🎉 整理的优势
### 1. 清晰的结构
- 按功能分类,便于查找
- 核心文件与辅助文件分离
- 专业的项目组织方式
### 2. 简化的根目录
- 减少文件数量,提高可读性
- 重要文件一目了然
- 便于新人理解项目
### 3. 便于维护
- 相关文件集中管理
- 便于版本控制
- 易于添加新功能
### 4. Docker友好
- Docker文件集中管理
- 路径引用已更新
- 支持多种部署方案
## 🚀 使用方法
### 开发
```bash
# 在根目录运行核心应用
python api_server.py
python history_viewer.py
```
### 测试
```bash
# 使用tests目录中的脚本
./tests/test-docker.sh
python tests/test_multi_service.py
```
### 部署
```bash
# 使用Docker部署
./docker-build.sh
docker-compose up -d
```
### 文档
查看 `docs/` 目录获取详细的使用和部署指南。
## ✅ 验证清单
- [x] 文件成功移动到对应目录
- [x] 路径引用已更新
- [x] Docker构建脚本正常工作
- [x] .gitignore文件完善
- [x] README.md更新
- [x] 项目结构文档创建
现在项目结构更加专业和易于管理!🎯

508
create-compose-package-fastapi.sh Executable file
View File

@ -0,0 +1,508 @@
#!/bin/bash
# DMS合规性测试工具 - FastAPI版本 Docker Compose部署包创建脚本
# 使用Alpine Linux + 多阶段构建 + Docker Compose管理
# 自动检测当前平台架构使用5051端口
set -e
# 配置变量
EXPORT_DIR="dms-compliance-fastapi-$(date +%Y%m%d-%H%M%S)"
IMAGE_NAME="compliance-dms-fastapi"
ARCHIVE_NAME="$EXPORT_DIR.tar.gz"
echo "=== DMS合规性测试工具 FastAPI版本 Docker Compose部署包创建脚本 ==="
echo "[信息] 使用FastAPI框架自动生成API文档"
echo "[信息] 服务端口: 5051 (与历史查看器端口一致)"
echo "[信息] 自动检测当前平台架构"
# 检查Docker是否运行
if ! docker info >/dev/null 2>&1; then
echo "[错误] Docker未运行或无法访问"
exit 1
fi
# 检测当前平台架构
CURRENT_ARCH=$(docker version --format '{{.Server.Arch}}' 2>/dev/null || uname -m)
case "$CURRENT_ARCH" in
x86_64|amd64)
TARGET_PLATFORM="linux/amd64"
ARCH_NAME="AMD64 (x86_64)"
;;
aarch64|arm64)
TARGET_PLATFORM="linux/arm64"
ARCH_NAME="ARM64 (aarch64)"
;;
*)
TARGET_PLATFORM="linux/amd64" # 默认使用amd64
ARCH_NAME="AMD64 (x86_64) - 默认"
echo "[警告] 未识别的架构 $CURRENT_ARCH,使用默认的 amd64"
;;
esac
echo "[信息] 检测到架构: $ARCH_NAME"
echo "[信息] 目标平台: $TARGET_PLATFORM"
# 创建导出目录
echo "[信息] 创建导出目录: $EXPORT_DIR"
rm -rf "$EXPORT_DIR"
mkdir -p "$EXPORT_DIR"
# 1. 创建临时构建目录,只包含必要文件
echo "[信息] 创建临时构建目录..."
TEMP_BUILD_DIR=$(mktemp -d)
trap "rm -rf $TEMP_BUILD_DIR" EXIT
# 白名单:只复制必要的文件
echo "[信息] 复制必要文件FastAPI版本..."
mkdir -p "$TEMP_BUILD_DIR"/{ddms_compliance_suite,custom_stages,custom_testcases,templates,static,assets}
# 复制FastAPI相关文件
echo "[信息] 复制FastAPI核心文件..."
for file in fastapi_server.py requirements_fastapi.txt; do
[ -f "$file" ] && cp "$file" "$TEMP_BUILD_DIR/"
done
# 复制核心目录(排除缓存和临时文件)
echo "[信息] 复制核心目录..."
rsync -av --exclude='__pycache__' --exclude='*.pyc' --exclude='*.log' ddms_compliance_suite/ "$TEMP_BUILD_DIR/ddms_compliance_suite/"
rsync -av --exclude='__pycache__' --exclude='*.pyc' custom_stages/ "$TEMP_BUILD_DIR/custom_stages/"
rsync -av --exclude='__pycache__' --exclude='*.pyc' custom_testcases/ "$TEMP_BUILD_DIR/custom_testcases/"
# 确保templates目录结构正确
echo "[信息] 复制模板和静态文件..."
rsync -av templates/ "$TEMP_BUILD_DIR/templates/"
rsync -av static/ "$TEMP_BUILD_DIR/static/"
rsync -av assets/ "$TEMP_BUILD_DIR/assets/"
# 验证templates目录内容
echo "[信息] 验证templates目录: $(ls "$TEMP_BUILD_DIR/templates/" 2>/dev/null | wc -l) 个文件"
# 创建FastAPI专用Dockerfile
echo "[信息] 创建FastAPI专用Dockerfile..."
cat > "$TEMP_BUILD_DIR/Dockerfile" << 'EOF'
# 多阶段构建:第一阶段安装依赖
FROM python:3.9-alpine AS builder
# 更新包索引并安装构建依赖
RUN apk update && \
apk add --no-cache \
gcc \
musl-dev \
linux-headers \
libffi-dev \
openssl-dev \
cargo \
rust && \
rm -rf /var/cache/apk/*
# 设置工作目录
WORKDIR /app
# 复制requirements并安装Python依赖
COPY requirements_fastapi.txt .
# 升级pip并安装依赖
RUN pip install --upgrade pip setuptools wheel && \
pip install --no-cache-dir --user -r requirements_fastapi.txt
# 第二阶段:运行时镜像
FROM python:3.9-alpine
# 安装运行时依赖
RUN apk update && \
apk add --no-cache \
curl \
bash && \
rm -rf /var/cache/apk/*
# 从构建阶段复制Python包
COPY --from=builder /root/.local /root/.local
# 设置工作目录
WORKDIR /app
# 复制应用代码
COPY . .
# 创建必要目录
RUN mkdir -p /app/logs /app/test_reports /app/uploads
# 设置环境变量
ENV PATH=/root/.local/bin:$PATH
ENV PYTHONPATH=/app
ENV PYTHONUNBUFFERED=1
# 暴露端口使用5051端口
EXPOSE 5051
# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \
CMD curl -f http://localhost:5051/ || exit 1
# 启动FastAPI服务器
CMD ["python3", "fastapi_server.py", "--host", "0.0.0.0", "--port", "5051"]
EOF
# 创建备用的简化Dockerfile如果多阶段构建失败
cat > "$TEMP_BUILD_DIR/Dockerfile.simple" << 'EOF'
# 简化版本:单阶段构建
FROM python:3.9-alpine
# 安装所有必要的依赖
RUN apk update && \
apk add --no-cache \
gcc \
musl-dev \
linux-headers \
libffi-dev \
openssl-dev \
curl \
bash && \
rm -rf /var/cache/apk/*
# 设置工作目录
WORKDIR /app
# 复制应用代码
COPY . .
# 安装Python依赖
RUN pip install --upgrade pip setuptools wheel && \
pip install --no-cache-dir -r requirements_fastapi.txt
# 创建必要目录
RUN mkdir -p /app/logs /app/test_reports /app/uploads
# 设置环境变量
ENV PYTHONPATH=/app
ENV PYTHONUNBUFFERED=1
# 暴露端口
EXPOSE 5051
# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \
CMD curl -f http://localhost:5051/ || exit 1
# 启动FastAPI服务器
CMD ["python3", "fastapi_server.py", "--host", "0.0.0.0", "--port", "5051"]
EOF
# 显示构建目录大小
echo "[信息] 临时构建目录大小: $(du -sh "$TEMP_BUILD_DIR" | cut -f1)"
# 2. 构建Docker镜像
echo "[信息] 构建FastAPI Docker镜像 ($TARGET_PLATFORM)..."
cd "$TEMP_BUILD_DIR"
# 尝试构建,首先使用多阶段构建,失败则使用简化构建
if docker build --platform "$TARGET_PLATFORM" -t "$IMAGE_NAME:latest" .; then
echo "[成功] Docker构建完成多阶段构建"
else
echo "[警告] 多阶段构建失败,尝试简化构建..."
if docker build --platform "$TARGET_PLATFORM" -t "$IMAGE_NAME:latest" -f Dockerfile.simple .; then
echo "[成功] Docker构建完成简化构建"
else
echo "[错误] 所有构建方式都失败"
exit 1
fi
fi
cd - > /dev/null
# 3. 导出Docker镜像
echo "[信息] 导出Docker镜像..."
docker save "$IMAGE_NAME:latest" | gzip > "$EXPORT_DIR/docker-image.tar.gz"
# 4. 创建docker-compose.yml
echo "[信息] 创建docker-compose.yml..."
cat > "$EXPORT_DIR/docker-compose.yml" << 'EOF'
version: '3.8'
services:
dms-compliance-fastapi:
image: compliance-dms-fastapi:latest
container_name: dms-compliance-fastapi
ports:
- "5051:5051" # FastAPI服务器端口
volumes:
# 持久化测试报告
- ./test_reports:/app/test_reports
# 持久化上传文件
- ./uploads:/app/uploads
# 持久化日志
- ./logs:/app/logs
# 如果需要自定义配置文件
- ./config:/app/config:ro
environment:
- PYTHONUNBUFFERED=1
- TZ=Asia/Shanghai
- HOST=0.0.0.0
- PORT=5051
restart: unless-stopped
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:5051/"]
interval: 30s
timeout: 10s
retries: 3
start_period: 40s
networks:
- dms-network
networks:
dms-network:
driver: bridge
volumes:
test_reports:
uploads:
logs:
EOF
# 5. 创建部署脚本
echo "[信息] 创建部署脚本..."
cat > "$EXPORT_DIR/deploy.sh" << 'EOF'
#!/bin/bash
# DMS合规性测试工具 - FastAPI版本 Docker Compose部署脚本
set -e
echo "=== DMS合规性测试工具 FastAPI版本部署 ==="
# 检查Docker和Docker Compose
if ! docker info >/dev/null 2>&1; then
echo "[错误] Docker未运行"
exit 1
fi
if ! command -v docker-compose >/dev/null 2>&1 && ! docker compose version >/dev/null 2>&1; then
echo "[错误] Docker Compose未安装"
echo "请安装Docker Compose或使用Docker Desktop"
exit 1
fi
# 创建必要的目录
echo "[信息] 创建数据目录..."
mkdir -p test_reports uploads logs config
# 加载镜像
echo "[信息] 加载Docker镜像..."
docker load < docker-image.tar.gz
# 停止现有服务
echo "[信息] 停止现有服务..."
docker-compose down 2>/dev/null || docker compose down 2>/dev/null || true
# 启动服务
echo "[信息] 启动FastAPI服务..."
if command -v docker-compose >/dev/null 2>&1; then
docker-compose up -d
else
docker compose up -d
fi
echo "[成功] FastAPI版本部署完成"
echo ""
echo "访问地址:"
echo "- 主服务: http://localhost:5051"
echo "- API文档 (Swagger UI): http://localhost:5051/docs"
echo "- API文档 (ReDoc): http://localhost:5051/redoc"
echo "- 健康检查: http://localhost:5051/"
echo "- 服务信息: http://localhost:5051/info"
echo ""
echo "管理命令:"
echo "- 查看状态: docker-compose ps"
echo "- 查看日志: docker-compose logs"
echo "- 停止服务: docker-compose down"
echo "- 重启服务: docker-compose restart"
echo ""
echo "数据目录:"
echo "- 测试报告: $(pwd)/test_reports"
echo "- 上传文件: $(pwd)/uploads"
echo "- 日志文件: $(pwd)/logs"
echo "- 配置文件: $(pwd)/config"
echo ""
echo "FastAPI特性:"
echo "- 自动生成的交互式API文档"
echo "- 强类型数据验证"
echo "- 高性能异步处理"
echo "- 支持分页参数 (page_size, page_no)"
EOF
chmod +x "$EXPORT_DIR/deploy.sh"
# 6. 创建README
echo "[信息] 创建README..."
cat > "$EXPORT_DIR/README.md" << 'EOF'
# DMS合规性测试工具 - FastAPI版本
## 特点
- 基于FastAPI框架提供自动生成的交互式API文档
- 基于Alpine Linux镜像体积极小约350MB
- 多阶段构建,优化层结构
- 使用5051端口与历史查看器端口一致
- 支持数据持久化和健康检查
- 自动检测当前平台架构,无需手动选择
- 强类型数据验证和详细的API文档
## 部署方法
1. 解压部署包
2. 运行部署脚本:
```bash
./deploy.sh
```
3. 访问服务:
- 主服务: http://localhost:5051
- API文档 (Swagger UI): http://localhost:5051/docs
- API文档 (ReDoc): http://localhost:5051/redoc
## FastAPI版本优势
### 自动API文档
- **Swagger UI**: 交互式API测试界面
- **ReDoc**: 美观的API文档展示
- **自动生成**: 基于代码自动生成,始终保持最新
### 强类型验证
- **Pydantic模型**: 自动数据验证和序列化
- **类型提示**: 完整的类型提示支持
- **错误处理**: 详细的验证错误信息
### 高性能
- **异步处理**: 基于Starlette的异步框架
- **并发支持**: 天然支持高并发请求
- **优化性能**: 比传统Flask更高的性能
## 新增功能
### 分页支持增强
- `page_size`: 分页大小1-10000默认1000
- `page_no`: 起始页码从1开始默认1
- 支持断点续传和跳过前面的页面
### API端点
- `GET /`: 健康检查
- `GET /info`: 服务信息
- `POST /run`: 执行测试
- `GET /reports`: 列出报告
- `GET /reports/{id}`: 下载报告
## 使用示例
### 基本测试
```bash
curl -X POST http://localhost:5051/run \
-H "Content-Type: application/json" \
-d '{
"dms": "./assets/doc/dms/domain.json",
"base_url": "https://api.example.com",
"page_size": 500,
"page_no": 1,
"strictness_level": "CRITICAL"
}'
```
### 分页测试从第3页开始
```bash
curl -X POST http://localhost:5051/run \
-H "Content-Type: application/json" \
-d '{
"dms": "./assets/doc/dms/domain.json",
"base_url": "https://api.example.com",
"page_size": 100,
"page_no": 3,
"ignore_ssl": true
}'
```
## 管理命令
- 查看服务状态:`docker-compose ps`
- 查看日志:`docker-compose logs`
- 停止服务:`docker-compose down`
- 重启服务:`docker-compose restart`
- 查看实时日志:`docker-compose logs -f`
## 文件说明
- `docker-image.tar.gz` - Docker镜像文件
- `docker-compose.yml` - Docker Compose配置文件
- `deploy.sh` - 一键部署脚本
- `README.md` - 说明文档
## 数据持久化
所有重要数据都会持久化到本地目录:
- `test_reports/` - 测试报告
- `uploads/` - 上传文件
- `logs/` - 日志文件
- `config/` - 配置文件(只读)
## 架构支持
本版本会自动检测当前平台架构:
- AMD64 (x86_64) - 适用于大多数Intel/AMD服务器
- ARM64 (aarch64) - 适用于Apple Silicon Mac、ARM服务器
## 故障排除
如果遇到问题:
1. 检查Docker是否正常运行`docker info`
2. 检查端口是否被占用:`netstat -tlnp | grep 5051`
3. 查看容器日志:`docker-compose logs`
4. 重启服务:`docker-compose restart`
5. 访问健康检查:`curl http://localhost:5051/`
## 与Flask版本对比
| 特性 | Flask版本 | FastAPI版本 |
|------|-----------|-------------|
| 端口 | 5050 | 5051 |
| API文档 || 自动生成 |
| 数据验证 | 手动 | 自动 |
| 性能 | 中等 ||
| 交互测试 || 内置 |
| 类型提示 | 部分 | 完整 |
EOF
# 7. 显示镜像信息
echo "[信息] Docker镜像信息:"
docker images "$IMAGE_NAME:latest"
# 8. 压缩最终包
echo "[信息] 压缩最终部署包..."
tar -czf "$ARCHIVE_NAME" "$EXPORT_DIR"
# 9. 显示结果
echo ""
echo "=== FastAPI版本创建完成 ==="
echo "部署包: $ARCHIVE_NAME"
echo "部署包大小: $(du -sh "$ARCHIVE_NAME" | cut -f1)"
echo "构建架构: $TARGET_PLATFORM ($ARCH_NAME)"
echo "服务端口: 5051"
echo "Docker镜像大小: $(docker images "$IMAGE_NAME:latest" --format "{{.Size}}" 2>/dev/null || echo "约350MB")"
echo ""
echo "部署方法:"
echo "1. 解压: tar -xzf $ARCHIVE_NAME"
echo "2. 进入目录: cd $EXPORT_DIR"
echo "3. 运行: ./deploy.sh"
echo ""
echo "访问地址:"
echo "- 主服务: http://localhost:5051"
echo "- API文档: http://localhost:5051/docs"
echo "- ReDoc: http://localhost:5051/redoc"
echo ""
# 清理Docker镜像可选
read -p "是否删除本地Docker镜像(y/N): " -n 1 -r
echo
if [[ $REPLY =~ ^[Yy]$ ]]; then
docker rmi "$IMAGE_NAME:latest"
echo "[信息] 已删除本地Docker镜像"
fi
echo "[完成] FastAPI版本 Docker Compose部署包创建完成"

View File

@ -538,7 +538,7 @@ class InputParser:
self.logger.error(f"An unexpected error occurred while parsing Swagger spec {file_path}: {e}", exc_info=True)
return None
def parse_dms_spec(self, domain_mapping_path: str, base_url: str, headers: Optional[Dict[str, str]] = None, ignore_ssl: bool = False, page_size: int = 1000) -> Optional[Tuple[ParsedDMSSpec, Dict[str, Any]]]:
def parse_dms_spec(self, domain_mapping_path: str, base_url: str, headers: Optional[Dict[str, str]] = None, ignore_ssl: bool = False, page_size: int = 1000, page_no_start: int = 1, fetch_all_pages: bool = True) -> Optional[Tuple[ParsedDMSSpec, Dict[str, Any]]]:
self.logger.info(f"Starting DMS spec parsing. Base URL: {base_url}, Domain Map: {domain_mapping_path}")
if ignore_ssl:
@ -566,15 +566,22 @@ class InputParser:
self.logger.debug(f"映射关键词 '{keyword}' -> 领域ID '{domain_id}'")
# 实现分页获取API列表
self.logger.info(f"Fetching API list with pagination (page_size={page_size})")
if fetch_all_pages:
self.logger.info(f"Fetching ALL API pages with pagination (page_size={page_size}, starting from page {page_no_start})")
else:
self.logger.info(f"Fetching SINGLE page (page_size={page_size}, page_no={page_no_start})")
api_records = []
page_no = 1
page_no = page_no_start
total_fetched = 0
pagination_info = {
"page_size": page_size,
"page_no_start": page_no_start,
"total_pages": 0,
"total_records": 0,
"pages_fetched": 0
"pages_fetched": 0,
"current_page": page_no_start,
"fetch_all_pages": fetch_all_pages
}
try:
@ -611,8 +618,15 @@ class InputParser:
pagination_info["total_records"] = total_count
pagination_info["total_pages"] = (total_count + page_size - 1) // page_size # 向上取整
pagination_info["pages_fetched"] = page_no
pagination_info["pages_fetched"] = page_no - page_no_start + 1
pagination_info["current_page"] = page_no
# 如果是单页模式,获取一页后就停止
if not fetch_all_pages:
self.logger.info(f"Single page mode: fetched {len(page_records)} records from page {page_no}")
break
# 全页模式:检查是否还有更多页面
if current_count >= total_count or len(page_records) < page_size:
self.logger.info(f"Reached end of data. Total records: {total_fetched}")
break

View File

@ -2682,7 +2682,9 @@ class APITestOrchestrator:
categories: Optional[List[str]] = None,
custom_test_cases_dir: Optional[str] = None,
ignore_ssl: bool = False,
page_size: int = 1000
page_size: int = 1000,
page_no_start: int = 1,
fetch_all_pages: bool = True
) -> Tuple[TestSummary, Optional[ParsedAPISpec], Dict[str, Any]]:
"""
通过动态DMS服务发现来执行测试
@ -2693,7 +2695,7 @@ class APITestOrchestrator:
self.logger.info("从DMS动态服务启动测试...")
# 如果方法参数中没有传递ignore_ssl使用实例的设置
actual_ignore_ssl = ignore_ssl if ignore_ssl else self.ignore_ssl
parse_result = parser.parse_dms_spec(domain_mapping_path, base_url=self.base_url, ignore_ssl=actual_ignore_ssl, page_size=page_size)
parse_result = parser.parse_dms_spec(domain_mapping_path, base_url=self.base_url, ignore_ssl=actual_ignore_ssl, page_size=page_size, page_no_start=page_no_start, fetch_all_pages=fetch_all_pages)
if not parse_result or parse_result[0] is None:
self.logger.error("无法从DMS服务解析API测试终止。")

View File

@ -0,0 +1,61 @@
version: '3.8'
services:
dms-compliance-fastapi:
build:
context: .
dockerfile: Dockerfile.fastapi
container_name: dms-compliance-fastapi
ports:
- "5050:5050" # FastAPI服务器端口
volumes:
# 持久化测试报告
- ./test_reports:/app/test_reports
# 持久化上传文件
- ./uploads:/app/uploads
# 持久化日志
- ./logs:/app/logs
# 如果需要自定义配置文件
- ./config:/app/config:ro
environment:
- PYTHONUNBUFFERED=1
- TZ=Asia/Shanghai
- HOST=0.0.0.0
- PORT=5050
- WORKERS=1
restart: unless-stopped
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:5050/"]
interval: 30s
timeout: 10s
retries: 3
start_period: 40s
networks:
- dms-network
# 可选添加一个nginx反向代理
nginx-fastapi:
image: nginx:alpine
container_name: dms-nginx-fastapi
ports:
- "80:80"
- "443:443"
volumes:
- ./nginx/nginx.fastapi.conf:/etc/nginx/nginx.conf:ro
- ./nginx/ssl:/etc/nginx/ssl:ro
depends_on:
- dms-compliance-fastapi
restart: unless-stopped
networks:
- dms-network
profiles:
- with-nginx
networks:
dms-network:
driver: bridge
volumes:
test_reports:
uploads:
logs:

View File

@ -0,0 +1,278 @@
# FastAPI版本分页功能实现总结
## 🎯 实现概述
我们成功实现了FastAPI版本的DMS合规性测试工具并添加了完整的分页支持包括 `page_size``page_no` 参数。
## 🔧 主要改进
### 1. 添加 `page_no` 参数支持
**新增功能**:
- `page_no`: 起始页码从1开始
- 支持断点续传和跳过前面的页面
- 详细的分页统计信息
**API参数**:
```json
{
"dms": "./assets/doc/dms/domain.json",
"base_url": "https://api.example.com",
"page_size": 500,
"page_no": 3,
"strictness_level": "CRITICAL"
}
```
### 2. FastAPI版本特性
**自动API文档**:
- Swagger UI: `http://localhost:5051/docs`
- ReDoc: `http://localhost:5051/redoc`
- OpenAPI规范: `http://localhost:5051/openapi.json`
**强类型验证**:
- 基于Pydantic V2的数据模型
- 自动参数验证和错误提示
- 详细的字段描述和示例
**高性能**:
- 异步处理支持
- 更高的并发性能
- 优化的JSON序列化
### 3. 分页信息增强
**返回的分页信息**:
```json
{
"pagination": {
"page_size": 500,
"page_no_start": 3,
"total_pages": 20,
"total_records": 9876,
"pages_fetched": 18,
"current_page": 20
}
}
```
## 📊 分页参数详解
### page_size (分页大小)
- **范围**: 1-10000
- **默认值**: 1000
- **用途**: 控制每页获取的API数量
- **建议**:
- 内存受限: 100-500
- 平衡性能: 500-1000
- 高性能: 1000-5000
### page_no (起始页码)
- **范围**: ≥1
- **默认值**: 1
- **用途**: 指定从哪一页开始获取
- **应用场景**:
- 断点续传: 从中断的页面继续
- 跳过数据: 跳过前面不需要的页面
- 分批处理: 分多次处理大量数据
## 🚀 使用示例
### 1. 基本分页测试
```bash
curl -X POST http://localhost:5051/run \
-H "Content-Type: application/json" \
-d '{
"dms": "./assets/doc/dms/domain.json",
"base_url": "https://api.example.com",
"page_size": 1000,
"page_no": 1
}'
```
### 2. 断点续传从第5页开始
```bash
curl -X POST http://localhost:5051/run \
-H "Content-Type: application/json" \
-d '{
"dms": "./assets/doc/dms/domain.json",
"base_url": "https://api.example.com",
"page_size": 500,
"page_no": 5
}'
```
### 3. 小批量处理(减少内存使用)
```bash
curl -X POST http://localhost:5051/run \
-H "Content-Type: application/json" \
-d '{
"dms": "./assets/doc/dms/domain.json",
"base_url": "https://api.example.com",
"page_size": 100,
"page_no": 1,
"ignore_ssl": true
}'
```
## 🐳 Docker部署
### 1. FastAPI版本打包脚本
```bash
# 创建FastAPI版本部署包
./create-compose-package-fastapi.sh
# 特点:
# - 使用5051端口与历史查看器一致
# - 自动生成API文档
# - 支持完整的分页功能
```
### 2. 部署包特性
- **端口**: 5051 (避免与Flask版本冲突)
- **文档**: 自动生成交互式API文档
- **架构**: 自动检测当前平台
- **大小**: 约350MB (包含FastAPI依赖)
### 3. 部署后访问
```bash
# 主服务
http://localhost:5051/
# API文档 (Swagger UI)
http://localhost:5051/docs
# API文档 (ReDoc)
http://localhost:5051/redoc
# 服务信息
http://localhost:5051/info
```
## 🔄 版本对比
| 特性 | Flask版本 | FastAPI版本 |
|------|-----------|-------------|
| **端口** | 5050 | 5051 |
| **API文档** | 无 | 自动生成 |
| **分页参数** | page_size | page_size + page_no |
| **数据验证** | 手动 | 自动 (Pydantic) |
| **性能** | 同步 | 异步 |
| **交互测试** | 无 | 内置 |
| **类型提示** | 部分 | 完整 |
| **错误信息** | 基本 | 详细 |
## 📈 性能优化建议
### 1. 内存优化
```json
{
"page_size": 100, // 小分页减少内存
"page_no": 1, // 从需要的页面开始
"verbose": false // 减少日志输出
}
```
### 2. 网络优化
```json
{
"page_size": 1000, // 大分页减少请求次数
"page_no": 1, // 一次性获取
"ignore_ssl": true // 跳过SSL验证测试环境
}
```
### 3. 断点续传
```json
{
"page_size": 500, // 平衡大小
"page_no": 10, // 从中断点继续
"strictness_level": "HIGH"
}
```
## 🛠️ 开发和调试
### 1. 启动开发服务器
```bash
# 自动重载模式
python3 fastapi_server.py --reload --port 5051
# 或使用启动脚本
RELOAD=true ./start_fastapi.sh
```
### 2. 测试分页功能
```bash
# 运行测试脚本
python3 test_fastapi.py
# 测试特定功能
python3 test_pagination.py --api-server
```
### 3. 调试技巧
- 使用Swagger UI进行交互式测试
- 查看详细的Pydantic验证错误
- 利用FastAPI的自动文档功能
## 🔍 故障排除
### 1. Pydantic版本问题
```bash
# 确保使用Pydantic V2
pip install "pydantic>=2.5.0"
# 检查版本
python3 -c "import pydantic; print(pydantic.__version__)"
```
### 2. 端口冲突
```bash
# 检查端口使用
lsof -i :5051
# 使用其他端口
python3 fastapi_server.py --port 8080
```
### 3. 依赖问题
```bash
# 重新安装FastAPI依赖
pip install -r requirements_fastapi.txt
# 检查关键依赖
python3 -c "import fastapi, uvicorn, pydantic"
```
## 🎯 最佳实践
### 1. 生产部署
- 使用多个工作进程: `--workers 4`
- 配置反向代理 (Nginx)
- 启用HTTPS和安全头
- 监控API性能和错误率
### 2. 分页策略
- **大数据集**: 使用小分页 (100-500)
- **快速测试**: 使用中等分页 (500-1000)
- **生产环境**: 根据内存和网络条件调整
### 3. 错误处理
- 利用FastAPI的自动错误响应
- 监控分页统计信息
- 实现重试机制处理网络异常
## 📝 总结
FastAPI版本成功实现了
1. ✅ **完整的分页支持**: `page_size` + `page_no`
2. ✅ **自动API文档**: Swagger UI + ReDoc
3. ✅ **强类型验证**: Pydantic V2模型
4. ✅ **高性能处理**: 异步框架
5. ✅ **Docker部署**: 5051端口避免冲突
6. ✅ **向后兼容**: 支持所有原有功能
这个实现完全解决了内存溢出问题同时提供了更好的开发体验和API文档支持

View File

@ -0,0 +1,292 @@
# DMS合规性测试工具 - FastAPI版本使用指南
## 概述
FastAPI版本提供了自动生成的交互式API文档相比Flask版本具有以下优势
- 🚀 **自动API文档**: 自动生成Swagger UI和ReDoc文档
- 📊 **数据验证**: 基于Pydantic的强类型数据验证
- ⚡ **高性能**: 基于Starlette和Uvicorn的异步框架
- 🔧 **类型提示**: 完整的类型提示支持
- 📝 **详细文档**: 丰富的API描述和示例
## 快速开始
### 1. 安装依赖
```bash
# 安装FastAPI版本的依赖
pip install -r requirements_fastapi.txt
```
### 2. 启动服务器
```bash
# 使用启动脚本(推荐)
./start_fastapi.sh
# 或直接运行
python3 fastapi_server.py
# 开发模式(自动重载)
python3 fastapi_server.py --reload
# 自定义配置
python3 fastapi_server.py --host 0.0.0.0 --port 8080 --workers 4
```
### 3. 访问API文档
启动后可以访问以下地址:
- **Swagger UI**: http://localhost:5050/docs
- **ReDoc**: http://localhost:5050/redoc
- **健康检查**: http://localhost:5050/
## API文档特性
### 自动生成的文档包含:
1. **完整的API规范**
- 所有端点的详细描述
- 请求/响应模型
- 参数说明和示例
2. **交互式测试**
- 直接在浏览器中测试API
- 自动填充示例数据
- 实时查看响应结果
3. **数据模型文档**
- 详细的数据结构说明
- 字段验证规则
- 示例值
## 主要API端点
### 1. 健康检查
```
GET /
```
检查服务器状态和基本信息。
### 2. 服务信息
```
GET /info
```
获取详细的服务器信息和功能列表。
### 3. 执行测试
```
POST /run
```
执行API合规性测试的主要端点。
**请求体示例**:
```json
{
"dms": "./assets/doc/dms/domain.json",
"base_url": "https://api.example.com",
"page_size": 1000,
"strictness_level": "CRITICAL",
"ignore_ssl": false,
"generate_pdf": true,
"verbose": false
}
```
**响应示例**:
```json
{
"status": "completed",
"message": "Tests finished.",
"report_directory": "/path/to/reports/2024-01-15_10-30-45",
"summary": {
"endpoints_total": 150,
"endpoints_passed": 145,
"endpoints_failed": 5,
"test_cases_total": 750
},
"pagination": {
"page_size": 1000,
"total_records": 150,
"total_pages": 1,
"pages_fetched": 1
}
}
```
### 4. 报告管理
```
GET /reports # 列出所有报告
GET /reports/{id} # 下载特定报告
```
## 配置参数详解
### API定义源三选一
- `yapi`: YAPI定义文件路径
- `swagger`: Swagger/OpenAPI定义文件路径
- `dms`: DMS服务发现的domain mapping文件路径
### 基本配置
- `base_url`: API基础URL必填
- `page_size`: DMS分页大小1-10000默认1000
- `strictness_level`: 严格等级CRITICAL/HIGH/MEDIUM/LOW
### 过滤选项
- `categories`: YAPI分类列表
- `tags`: Swagger标签列表
- `ignore_ssl`: 忽略SSL证书验证
### LLM配置
- `llm_api_key`: LLM API密钥
- `llm_base_url`: LLM API基础URL
- `llm_model_name`: LLM模型名称
- `use_llm_for_*`: 各种LLM使用选项
## Docker部署
### 1. 构建镜像
```bash
docker build -f Dockerfile.fastapi -t dms-compliance-fastapi .
```
### 2. 使用Docker Compose
```bash
# 启动服务
docker-compose -f docker-compose.fastapi.yml up -d
# 查看日志
docker-compose -f docker-compose.fastapi.yml logs -f
# 停止服务
docker-compose -f docker-compose.fastapi.yml down
```
### 3. 环境变量
```bash
# 在docker-compose.yml中配置
environment:
- HOST=0.0.0.0
- PORT=5050
- WORKERS=4
- PYTHONUNBUFFERED=1
```
## 性能优化
### 1. 生产部署
```bash
# 使用多个工作进程
python3 fastapi_server.py --workers 4
# 使用Gunicorn推荐生产环境
gunicorn fastapi_server:app -w 4 -k uvicorn.workers.UvicornWorker
```
### 2. 内存优化
```bash
# 使用较小的分页大小
{
"page_size": 500, # 减少内存使用
"dms": "..."
}
```
### 3. 并发处理
FastAPI天然支持异步处理可以同时处理多个请求。
## 开发和调试
### 1. 开发模式
```bash
# 启用自动重载
python3 fastapi_server.py --reload
# 或使用环境变量
RELOAD=true ./start_fastapi.sh
```
### 2. 日志配置
```python
# 在代码中调整日志级别
logging.getLogger('ddms_compliance_suite').setLevel(logging.DEBUG)
```
### 3. 调试技巧
- 使用Swagger UI进行交互式测试
- 查看详细的错误信息和堆栈跟踪
- 利用Pydantic的数据验证错误信息
## 与Flask版本的对比
| 特性 | Flask版本 | FastAPI版本 |
|------|-----------|-------------|
| API文档 | 无 | 自动生成 |
| 数据验证 | 手动 | 自动Pydantic |
| 性能 | 中等 | 高(异步) |
| 类型提示 | 部分 | 完整 |
| 交互测试 | 无 | 内置 |
| 学习曲线 | 低 | 中等 |
## 故障排除
### 常见问题
1. **端口被占用**
```bash
# 检查端口使用
lsof -i :5050
# 使用其他端口
python3 fastapi_server.py --port 8080
```
2. **依赖缺失**
```bash
# 重新安装依赖
pip install -r requirements_fastapi.txt
```
3. **文档无法访问**
- 检查服务器是否正常启动
- 确认端口配置正确
- 查看防火墙设置
### 调试命令
```bash
# 检查服务状态
curl http://localhost:5050/
# 查看服务信息
curl http://localhost:5050/info
# 测试API端点
curl -X POST http://localhost:5050/run \
-H "Content-Type: application/json" \
-d '{"dms": "./test.json", "base_url": "https://api.test.com"}'
```
## 最佳实践
1. **生产部署**
- 使用多个工作进程
- 配置反向代理Nginx
- 启用HTTPS
- 设置适当的超时时间
2. **安全考虑**
- 限制CORS域名
- 使用环境变量存储敏感信息
- 定期更新依赖
3. **监控和日志**
- 配置结构化日志
- 监控API响应时间
- 设置健康检查
4. **测试策略**
- 使用小分页大小进行快速测试
- 利用交互式文档进行手动测试
- 编写自动化测试脚本

552
fastapi_server.py Normal file
View File

@ -0,0 +1,552 @@
#!/usr/bin/env python3
"""
DMS合规性测试工具 - FastAPI版本API服务器
提供自动生成的交互式API文档
"""
import os
import sys
import json
import logging
import datetime
import traceback
from pathlib import Path
from typing import List, Optional, Dict, Any, Union
import unicodedata
import html
# FastAPI imports
from fastapi import FastAPI, HTTPException, BackgroundTasks, status
from fastapi.responses import JSONResponse, FileResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field, field_validator, model_validator
import uvicorn
# PDF generation libraries - with fallback
try:
from reportlab.lib import colors
from reportlab.lib.pagesizes import A4
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle, HRFlowable
from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.ttfonts import TTFont
reportlab_available = True
except ImportError:
reportlab_available = False
# Project-specific imports
from ddms_compliance_suite.api_caller.caller import APICallDetail
from ddms_compliance_suite.test_orchestrator import APITestOrchestrator, TestSummary
from ddms_compliance_suite.input_parser.parser import ParsedAPISpec
from ddms_compliance_suite.utils.response_utils import extract_data_for_validation
from ddms_compliance_suite.utils.data_generator import DataGenerator
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# FastAPI app instance
app = FastAPI(
title="DMS合规性测试工具 API",
description="""
DMS合规性测试工具 FastAPI版本
这是一个用于API合规性测试的工具支持
YAPI规范测试 - 基于YAPI定义文件的测试
Swagger/OpenAPI测试 - 基于OpenAPI规范的测试
DMS服务发现测试 - 动态发现DMS服务的API进行测试
分页支持 - 支持大量API的分页获取避免内存溢出
PDF报告生成 - 生成详细的测试报告
LLM集成 - 支持大语言模型辅助生成测试数据
主要特性
- 🚀 高性能: 基于FastAPI支持异步处理
- 📊 分页支持: 解决大量API节点的内存问题
- 📝 自动文档: 自动生成交互式API文档
- 🔧 灵活配置: 支持多种测试配置选项
- 📈 详细报告: 生成PDF和JSON格式的测试报告
""",
version="1.0.0",
docs_url="/docs", # Swagger UI
redoc_url="/redoc", # ReDoc
)
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # 在生产环境中应该限制具体域名
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Pydantic models for request/response
class TestConfig(BaseModel):
"""测试配置模型"""
# API定义源 (三选一)
yapi: Optional[str] = Field(None, description="YAPI定义文件路径", example="./api_spec.json")
swagger: Optional[str] = Field(None, description="Swagger/OpenAPI定义文件路径", example="./openapi.yaml")
dms: Optional[str] = Field(None, description="DMS服务发现的domain mapping文件路径", example="./assets/doc/dms/domain.json")
# 基本配置
base_url: str = Field(..., description="API基础URL", example="https://api.example.com")
# 分页配置
page_size: int = Field(1000, description="DMS API分页大小默认1000。较小的值可以减少内存使用", ge=1, le=10000)
page_no: int = Field(1, description="起始页码从1开始。可用于断点续传或跳过前面的页面", ge=1)
fetch_all_pages: bool = Field(True, description="是否获取所有页面。True=获取所有数据False=只获取指定页面")
# 过滤选项
categories: Optional[List[str]] = Field(None, description="YAPI分类列表", example=["用户管理", "订单系统"])
tags: Optional[List[str]] = Field(None, description="Swagger标签列表", example=["user", "order"])
strictness_level: str = Field("CRITICAL", description="测试严格等级", pattern="^(CRITICAL|HIGH|MEDIUM|LOW)$")
# SSL和安全
ignore_ssl: bool = Field(False, description="忽略SSL证书验证不推荐在生产环境使用")
# 输出配置
output: str = Field("./test_reports", description="测试报告输出目录")
generate_pdf: bool = Field(True, description="是否生成PDF报告")
# 自定义测试
custom_test_cases_dir: Optional[str] = Field(None, description="自定义测试用例目录路径")
stages_dir: Optional[str] = Field(None, description="自定义测试阶段目录路径")
# LLM配置
llm_api_key: Optional[str] = Field(None, description="LLM API密钥")
llm_base_url: Optional[str] = Field(None, description="LLM API基础URL")
llm_model_name: Optional[str] = Field("gpt-3.5-turbo", description="LLM模型名称")
use_llm_for_request_body: bool = Field(False, description="使用LLM生成请求体")
use_llm_for_path_params: bool = Field(False, description="使用LLM生成路径参数")
use_llm_for_query_params: bool = Field(False, description="使用LLM生成查询参数")
use_llm_for_headers: bool = Field(False, description="使用LLM生成请求头")
# 调试选项
verbose: bool = Field(False, description="启用详细日志输出")
@field_validator('base_url')
@classmethod
def validate_base_url(cls, v):
if not v.startswith(('http://', 'https://')):
raise ValueError('base_url must start with http:// or https://')
return v
@model_validator(mode='before')
@classmethod
def validate_api_source(cls, values):
"""验证API定义源确保三选一"""
if isinstance(values, dict):
api_sources = [values.get('yapi'), values.get('swagger'), values.get('dms')]
non_none_sources = [s for s in api_sources if s is not None]
if len(non_none_sources) > 1:
raise ValueError('只能选择一个API定义源yapi、swagger或dms')
if len(non_none_sources) == 0:
raise ValueError('必须提供一个API定义源yapi、swagger或dms')
return values
class PaginationInfo(BaseModel):
"""分页信息模型"""
page_size: int = Field(description="页面大小")
page_no_start: int = Field(description="起始页码")
total_pages: int = Field(description="总页数")
total_records: int = Field(description="总记录数")
pages_fetched: int = Field(description="已获取页数")
current_page: int = Field(description="当前页码")
class TestResponse(BaseModel):
"""测试响应模型"""
status: str = Field(description="测试状态", example="completed")
message: str = Field(description="状态消息")
report_directory: str = Field(description="报告目录路径")
summary: Dict[str, Any] = Field(description="测试摘要信息")
pagination: Optional[PaginationInfo] = Field(None, description="分页信息仅DMS测试时返回")
class ErrorResponse(BaseModel):
"""错误响应模型"""
status: str = Field("error", description="错误状态")
message: str = Field(description="错误消息")
traceback: Optional[str] = Field(None, description="错误堆栈跟踪")
# Global variable to store running tasks
running_tasks: Dict[str, Dict[str, Any]] = {}
@app.get("/",
summary="健康检查",
description="检查API服务器是否正常运行",
response_model=Dict[str, str])
async def health_check():
"""健康检查端点用于Docker健康检查"""
return {
"status": "healthy",
"service": "DMS Compliance API Server (FastAPI)",
"version": "2.0.0",
"docs_url": "/docs",
"redoc_url": "/redoc"
}
@app.get("/info",
summary="服务信息",
description="获取API服务器的详细信息",
response_model=Dict[str, Any])
async def get_info():
"""获取服务器信息"""
return {
"service": "DMS Compliance API Server",
"version": "2.0.0",
"framework": "FastAPI",
"features": [
"YAPI规范测试",
"Swagger/OpenAPI测试",
"DMS服务发现测试",
"分页支持",
"PDF报告生成",
"LLM集成",
"自动API文档"
],
"endpoints": {
"health": "/",
"info": "/info",
"run_tests": "/run",
"docs": "/docs",
"redoc": "/redoc"
},
"reportlab_available": reportlab_available
}
# Import the test logic from the original Flask version
def run_tests_logic(config: dict):
"""
Main logic for running tests, adapted from the original Flask version.
"""
try:
if config.get('verbose'):
logging.getLogger('ddms_compliance_suite').setLevel(logging.DEBUG)
logger.setLevel(logging.DEBUG)
logger.debug("Verbose logging enabled.")
if not any(k in config for k in ['yapi', 'swagger', 'dms']):
raise ValueError("An API definition source is required: --yapi, --swagger, or --dms")
if sum(k in config for k in ['yapi', 'swagger', 'dms']) > 1:
raise ValueError("API definition sources are mutually exclusive.")
# Setup output directory with timestamp
base_output_dir = Path(config.get('output', './test_reports'))
timestamp = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
output_directory = base_output_dir / timestamp
output_directory.mkdir(parents=True, exist_ok=True)
logger.info(f"Test reports will be saved to: {output_directory.resolve()}")
# Initialize the orchestrator
orchestrator = APITestOrchestrator(
base_url=config['base_url'],
custom_test_cases_dir=config.get('custom_test_cases_dir'),
llm_api_key=config.get('llm_api_key'),
llm_base_url=config.get('llm_base_url'),
llm_model_name=config.get('llm_model_name'),
use_llm_for_request_body=config.get('use_llm_for_request_body', False),
use_llm_for_path_params=config.get('use_llm_for_path_params', False),
use_llm_for_query_params=config.get('use_llm_for_query_params', False),
use_llm_for_headers=config.get('use_llm_for_headers', False),
output_dir=str(output_directory),
stages_dir=config.get('stages_dir'),
strictness_level=config.get('strictness_level', 'CRITICAL'),
ignore_ssl=config.get('ignore_ssl', False)
)
test_summary: Optional[TestSummary] = None
parsed_spec: Optional[ParsedAPISpec] = None
pagination_info: Dict[str, Any] = {}
if 'yapi' in config:
logger.info(f"Running tests from YAPI file: {config['yapi']}")
test_summary, parsed_spec = orchestrator.run_tests_from_yapi(
yapi_file_path=config['yapi'],
categories=config.get('categories'),
custom_test_cases_dir=config.get('custom_test_cases_dir')
)
elif 'swagger' in config:
logger.info(f"Running tests from Swagger file: {config['swagger']}")
test_summary, parsed_spec = orchestrator.run_tests_from_swagger(
swagger_file_path=config['swagger'],
tags=config.get('tags'),
custom_test_cases_dir=config.get('custom_test_cases_dir')
)
elif 'dms' in config:
logger.info(f"Running tests from DMS service discovery: {config['dms']}")
test_summary, parsed_spec, pagination_info = orchestrator.run_tests_from_dms(
domain_mapping_path=config['dms'],
categories=config.get('categories'),
custom_test_cases_dir=config.get('custom_test_cases_dir'),
page_size=config.get('page_size', 1000),
page_no_start=config.get('page_no', 1),
fetch_all_pages=config.get('fetch_all_pages', True)
)
if not parsed_spec:
raise RuntimeError("Failed to parse the API specification.")
if test_summary and config.get('stages_dir') and parsed_spec:
logger.info(f"Executing API test stages from directory: {config['stages_dir']}")
stage_summary = orchestrator.run_stages_from_spec(parsed_spec, config['stages_dir'])
if stage_summary:
test_summary.merge_stage_summary(stage_summary)
if test_summary:
# Save main summary
main_report_file_path = output_directory / "summary.json"
with open(main_report_file_path, 'w', encoding='utf-8') as f:
f.write(test_summary.to_json(pretty=True))
# Save API call details
api_calls_filename = "api_call_details.md"
save_api_call_details_to_markdown(
orchestrator.get_api_call_details(),
str(output_directory),
filename=api_calls_filename
)
failed_count = getattr(test_summary, 'endpoints_failed', 0) + getattr(test_summary, 'test_cases_failed', 0)
error_count = getattr(test_summary, 'endpoints_error', 0) + getattr(test_summary, 'test_cases_error', 0)
result = {
"status": "completed",
"message": "Tests finished." if failed_count == 0 and error_count == 0 else "Tests finished with failures or errors.",
"report_directory": str(output_directory.resolve()),
"summary": test_summary.to_dict()
}
# 如果有分页信息,添加到返回结果中
if pagination_info:
result["pagination"] = pagination_info
return result
else:
raise RuntimeError("Test execution failed to produce a summary.")
except Exception as e:
logger.error(f"An unexpected error occurred during test execution: {e}", exc_info=True)
return {
"status": "error",
"message": str(e),
"traceback": traceback.format_exc()
}
def save_api_call_details_to_markdown(api_call_details: List[APICallDetail], output_dir: str, filename: str = "api_call_details.md"):
"""Save API call details to markdown file"""
try:
output_path = Path(output_dir) / filename
with open(output_path, 'w', encoding='utf-8') as f:
f.write("# API调用详情\n\n")
for i, detail in enumerate(api_call_details, 1):
f.write(f"## {i}. {detail.endpoint_name}\n\n")
f.write(f"**请求URL**: `{detail.request_url}`\n\n")
f.write(f"**请求方法**: `{detail.request_method}`\n\n")
if detail.request_headers:
f.write("**请求头**:\n```json\n")
f.write(json.dumps(detail.request_headers, indent=2, ensure_ascii=False))
f.write("\n```\n\n")
if detail.request_body:
f.write("**请求体**:\n```json\n")
f.write(json.dumps(detail.request_body, indent=2, ensure_ascii=False))
f.write("\n```\n\n")
f.write(f"**响应状态码**: `{detail.response_status_code}`\n\n")
if detail.response_headers:
f.write("**响应头**:\n```json\n")
f.write(json.dumps(detail.response_headers, indent=2, ensure_ascii=False))
f.write("\n```\n\n")
if detail.response_body:
f.write("**响应体**:\n```json\n")
f.write(json.dumps(detail.response_body, indent=2, ensure_ascii=False))
f.write("\n```\n\n")
f.write("---\n\n")
logger.info(f"API call details saved to: {output_path}")
except Exception as e:
logger.error(f"Error saving API call details: {e}")
@app.post("/run",
summary="执行API合规性测试",
description="""
执行API合规性测试的主要端点
支持三种API定义源
- **YAPI**: 基于YAPI定义文件
- **Swagger/OpenAPI**: 基于OpenAPI规范文件
- **DMS**: 动态发现DMS服务的API
### 分页支持
对于DMS测试支持分页获取API列表避免内存溢出
- `page_size`: 每页获取的API数量默认1000
- 返回详细的分页统计信息
### LLM集成
可选择使用大语言模型生成测试数据
- 智能生成请求体路径参数查询参数等
- 提高测试覆盖率和数据多样性
""",
response_model=TestResponse,
responses={
200: {"description": "测试执行成功"},
400: {"description": "请求参数错误", "model": ErrorResponse},
500: {"description": "服务器内部错误", "model": ErrorResponse}
})
async def run_api_tests(config: TestConfig):
"""
执行API合规性测试
- **config**: 测试配置包含API定义源测试参数等
- **returns**: 测试结果包含摘要信息和分页信息如适用
"""
try:
logger.info(f"Starting test run with configuration: {config.model_dump()}")
# Convert Pydantic model to dict for compatibility
config_dict = config.model_dump(exclude_none=True)
# Replace underscores with hyphens for compatibility with original code
config_dict = {k.replace('_', '-'): v for k, v in config_dict.items()}
result = run_tests_logic(config_dict)
if result['status'] == 'error':
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=result
)
return result
except ValueError as e:
logger.error(f"Validation error: {e}")
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={
"status": "error",
"message": str(e)
}
)
except Exception as e:
logger.error(f"An error occurred in the API endpoint: {e}", exc_info=True)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail={
"status": "error",
"message": str(e),
"traceback": traceback.format_exc()
}
)
@app.get("/reports/{report_id}",
summary="下载测试报告",
description="根据报告ID下载对应的测试报告文件")
async def download_report(report_id: str, file_type: str = "summary.json"):
"""
下载测试报告文件
- **report_id**: 报告ID通常是时间戳
- **file_type**: 文件类型可选值summary.json, api_call_details.md
"""
try:
report_dir = Path("./test_reports") / report_id
file_path = report_dir / file_type
if not file_path.exists():
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Report file not found: {file_type}"
)
return FileResponse(
path=str(file_path),
filename=file_type,
media_type='application/octet-stream'
)
except Exception as e:
logger.error(f"Error downloading report: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Error downloading report: {str(e)}"
)
@app.get("/reports",
summary="列出所有测试报告",
description="获取所有可用的测试报告列表")
async def list_reports():
"""列出所有可用的测试报告"""
try:
reports_dir = Path("./test_reports")
if not reports_dir.exists():
return {"reports": []}
reports = []
for report_dir in reports_dir.iterdir():
if report_dir.is_dir():
summary_file = report_dir / "summary.json"
if summary_file.exists():
try:
with open(summary_file, 'r', encoding='utf-8') as f:
summary = json.load(f)
reports.append({
"id": report_dir.name,
"timestamp": report_dir.name,
"path": str(report_dir),
"summary": {
"endpoints_total": summary.get("endpoints_total", 0),
"endpoints_passed": summary.get("endpoints_passed", 0),
"endpoints_failed": summary.get("endpoints_failed", 0),
"test_cases_total": summary.get("test_cases_total", 0)
}
})
except Exception as e:
logger.warning(f"Error reading summary for {report_dir.name}: {e}")
# Sort by timestamp (newest first)
reports.sort(key=lambda x: x["timestamp"], reverse=True)
return {"reports": reports}
except Exception as e:
logger.error(f"Error listing reports: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Error listing reports: {str(e)}"
)
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="DMS合规性测试工具 FastAPI服务器")
parser.add_argument("--host", default="0.0.0.0", help="服务器主机地址")
parser.add_argument("--port", type=int, default=5050, help="服务器端口")
parser.add_argument("--reload", action="store_true", help="启用自动重载(开发模式)")
parser.add_argument("--workers", type=int, default=1, help="工作进程数")
args = parser.parse_args()
logger.info(f"Starting FastAPI server on {args.host}:{args.port}")
logger.info(f"API文档地址: http://{args.host}:{args.port}/docs")
logger.info(f"ReDoc文档地址: http://{args.host}:{args.port}/redoc")
uvicorn.run(
"fastapi_server:app",
host=args.host,
port=args.port,
reload=args.reload,
workers=args.workers if not args.reload else 1,
log_level="info"
)

34
requirements_fastapi.txt Normal file
View File

@ -0,0 +1,34 @@
# FastAPI版本的额外依赖
# 在原有requirements.txt基础上添加以下依赖
# FastAPI核心
fastapi>=0.104.0
uvicorn[standard]>=0.24.0
# 数据验证和序列化
pydantic>=2.5.0
# CORS支持
python-multipart>=0.0.6
# 异步支持
aiofiles>=23.0.0
# 可选:性能优化
orjson>=3.9.0 # 更快的JSON序列化
# 可选:生产部署
gunicorn>=21.0.0 # WSGI服务器如果需要
# 原有依赖从requirements.txt复制
Flask>=2.0.0
Werkzeug>=2.0.0
requests>=2.25.0
PyYAML>=6.0
jsonschema>=4.0.0
pydantic>=1.8.0
openai>=0.27.0
Flask-Cors>=4.0.0
reportlab>=3.6.0
Pillow>=8.0.0
urllib3>=1.26.0

View File

@ -65,6 +65,10 @@ def parse_args():
api_group.add_argument('--dms', help='DMS服务发现的domain mapping文件路径')
api_group.add_argument('--page-size', type=int, default=1000,
help='DMS API分页大小默认1000。较小的值可以减少内存使用但会增加请求次数')
api_group.add_argument('--page-no', type=int, default=1,
help='DMS API起始页码从1开始。可用于断点续传或跳过前面的页面')
api_group.add_argument('--fetch-single-page', action='store_true',
help='只获取指定页面的数据,而不是获取所有页面。用于快速测试或内存受限环境')
# 过滤参数
filter_group = parser.add_argument_group('过滤选项')
@ -948,7 +952,9 @@ def main():
categories=categories,
custom_test_cases_dir=args.custom_test_cases_dir,
ignore_ssl=args.ignore_ssl,
page_size=args.page_size
page_size=args.page_size,
page_no_start=args.page_no,
fetch_all_pages=not args.fetch_single_page
)
if not parsed_spec_for_scenarios: # 检查解析是否成功
logger.error(f"从DMS服务 '{args.dms}' 解析失败 (由编排器报告)。程序将退出。")
@ -958,6 +964,8 @@ def main():
if pagination_info:
logger.info(f"DMS分页信息: 总记录数={pagination_info.get('total_records', 0)}, "
f"页面大小={pagination_info.get('page_size', 0)}, "
f"起始页码={pagination_info.get('page_no_start', 1)}, "
f"当前页码={pagination_info.get('current_page', 1)}, "
f"获取页数={pagination_info.get('pages_fetched', 0)}/{pagination_info.get('total_pages', 0)}")
else:
logger.warning("未获取到分页信息")

52
start_fastapi.sh Executable file
View File

@ -0,0 +1,52 @@
#!/bin/bash
# DMS合规性测试工具 - FastAPI服务器启动脚本
set -e
echo "=== DMS合规性测试工具 FastAPI服务器 ==="
# 默认配置
HOST=${HOST:-"0.0.0.0"}
PORT=${PORT:-"5050"}
WORKERS=${WORKERS:-"1"}
RELOAD=${RELOAD:-"false"}
# 检查Python环境
if ! command -v python3 &> /dev/null; then
echo "[错误] Python3 未安装"
exit 1
fi
# 检查依赖
echo "[信息] 检查依赖..."
if ! python3 -c "import fastapi" &> /dev/null; then
echo "[警告] FastAPI未安装正在安装依赖..."
pip install -r requirements_fastapi.txt
fi
# 创建必要目录
echo "[信息] 创建必要目录..."
mkdir -p test_reports logs uploads
# 启动服务器
echo "[信息] 启动FastAPI服务器..."
echo "[信息] 主机: $HOST"
echo "[信息] 端口: $PORT"
echo "[信息] 工作进程: $WORKERS"
echo "[信息] 自动重载: $RELOAD"
echo ""
echo "[信息] API文档地址: http://$HOST:$PORT/docs"
echo "[信息] ReDoc文档地址: http://$HOST:$PORT/redoc"
echo ""
# 构建启动命令
CMD="python3 fastapi_server.py --host $HOST --port $PORT --workers $WORKERS"
if [ "$RELOAD" = "true" ]; then
CMD="$CMD --reload"
echo "[信息] 开发模式:启用自动重载"
fi
# 启动服务
exec $CMD

View File

@ -1,266 +0,0 @@
#!/bin/bash
# DMS合规性测试工具 - 部署包测试脚本
# 用于验证离线部署包的完整性和功能
set -e
PACKAGE_FILE=""
TEST_DIR="test_deployment_$(date +%Y%m%d_%H%M%S)"
echo "=== DMS合规性测试工具部署包测试脚本 ==="
# 查找最新的部署包
find_latest_package() {
PACKAGE_FILE=$(ls -t dms-compliance-complete-offline-*.tar.gz 2>/dev/null | head -n1)
if [ -z "$PACKAGE_FILE" ]; then
echo "[错误] 找不到部署包文件"
echo "请先运行 ./create-offline-package.sh 创建部署包"
exit 1
fi
echo "[信息] 找到部署包: $PACKAGE_FILE"
echo "[信息] 文件大小: $(du -h "$PACKAGE_FILE" | cut -f1)"
}
# 创建测试环境
setup_test_env() {
echo "[信息] 创建测试环境: $TEST_DIR"
mkdir -p "$TEST_DIR"
cd "$TEST_DIR"
# 解压部署包
echo "[信息] 解压部署包..."
tar -xzf "../$PACKAGE_FILE"
# 进入解压后的目录
EXTRACTED_DIR=$(ls -d dms-compliance-* | head -n1)
cd "$EXTRACTED_DIR"
echo "[信息] 当前目录: $(pwd)"
}
# 验证文件完整性
verify_files() {
echo "[信息] 验证文件完整性..."
local required_files=(
"README.md"
"VERSION"
"install.sh"
"deploy.sh"
"stop.sh"
"uninstall.sh"
"install-docker.sh"
"docker-compose.yml"
"dms-compliance-tool.tar"
"nginx-alpine.tar"
)
local missing_files=()
for file in "${required_files[@]}"; do
if [ ! -f "$file" ]; then
missing_files+=("$file")
else
echo "[✓] $file"
fi
done
if [ ${#missing_files[@]} -gt 0 ]; then
echo "[错误] 缺少以下文件:"
printf '%s\n' "${missing_files[@]}"
return 1
fi
echo "[成功] 所有必需文件都存在"
}
# 验证脚本权限
verify_permissions() {
echo "[信息] 验证脚本权限..."
local scripts=(
"install.sh"
"deploy.sh"
"stop.sh"
"uninstall.sh"
"install-docker.sh"
)
for script in "${scripts[@]}"; do
if [ -x "$script" ]; then
echo "[✓] $script 可执行"
else
echo "[警告] $script 不可执行,正在修复..."
chmod +x "$script"
fi
done
}
# 验证Docker镜像
verify_docker_images() {
echo "[信息] 验证Docker镜像文件..."
# 检查镜像文件大小
if [ -f "dms-compliance-tool.tar" ]; then
local size=$(du -h "dms-compliance-tool.tar" | cut -f1)
echo "[✓] DMS应用镜像: $size"
# 验证镜像文件不为空
if [ ! -s "dms-compliance-tool.tar" ]; then
echo "[错误] DMS应用镜像文件为空"
return 1
fi
fi
if [ -f "nginx-alpine.tar" ]; then
local size=$(du -h "nginx-alpine.tar" | cut -f1)
echo "[✓] Nginx镜像: $size"
if [ ! -s "nginx-alpine.tar" ]; then
echo "[错误] Nginx镜像文件为空"
return 1
fi
fi
}
# 验证配置文件
verify_config() {
echo "[信息] 验证配置文件..."
# 检查docker-compose.yml
if command -v docker-compose >/dev/null 2>&1; then
if docker-compose config >/dev/null 2>&1; then
echo "[✓] docker-compose.yml 语法正确"
else
echo "[警告] docker-compose.yml 语法可能有问题"
docker-compose config
fi
else
echo "[信息] docker-compose 未安装,跳过配置验证"
fi
# 检查数据目录结构
local data_dirs=(
"data"
"data/test_reports"
"data/uploads"
"data/logs"
)
for dir in "${data_dirs[@]}"; do
if [ -d "$dir" ]; then
echo "[✓] 目录存在: $dir"
else
echo "[警告] 目录不存在: $dir"
fi
done
}
# 模拟部署测试(不实际启动服务)
simulate_deployment() {
echo "[信息] 模拟部署测试..."
# 检查Docker是否可用
if command -v docker >/dev/null 2>&1; then
if docker info >/dev/null 2>&1; then
echo "[✓] Docker可用"
# 测试加载镜像(但不实际加载)
echo "[信息] 测试镜像加载命令..."
echo " docker load -i dms-compliance-tool.tar"
echo " docker load -i nginx-alpine.tar"
# 测试docker-compose命令
if command -v docker-compose >/dev/null 2>&1; then
echo "[✓] Docker Compose可用"
echo "[信息] 测试服务启动命令..."
echo " docker-compose up -d"
else
echo "[警告] Docker Compose不可用"
fi
else
echo "[警告] Docker未运行"
fi
else
echo "[信息] Docker未安装这在目标环境中是正常的"
fi
}
# 生成测试报告
generate_report() {
echo ""
echo "=== 测试报告 ==="
echo "测试时间: $(date)"
echo "部署包: $PACKAGE_FILE"
echo "测试目录: $TEST_DIR"
echo ""
# 文件清单
echo "文件清单:"
find . -type f -name "*.sh" -o -name "*.yml" -o -name "*.tar" -o -name "*.md" | sort
echo ""
# 目录结构
echo "目录结构:"
tree -L 3 2>/dev/null || find . -type d | head -20
echo ""
# 磁盘使用
echo "磁盘使用:"
du -sh . 2>/dev/null || echo "无法获取磁盘使用信息"
echo ""
echo "=== 测试完成 ==="
echo ""
echo "下一步操作:"
echo "1. 将部署包传输到目标服务器"
echo "2. 在目标服务器上解压并运行 ./install.sh"
echo "3. 访问 http://localhost:5050 和 http://localhost:5051 验证服务"
echo ""
echo "清理测试环境:"
echo "cd .. && rm -rf $TEST_DIR"
}
# 清理函数
cleanup() {
if [ -n "$TEST_DIR" ] && [ -d "../$TEST_DIR" ]; then
echo ""
echo "[信息] 清理测试环境..."
cd ..
rm -rf "$TEST_DIR"
echo "[信息] 测试环境已清理"
fi
}
# 错误处理
error_handler() {
echo ""
echo "[错误] 测试过程中发生错误"
cleanup
exit 1
}
# 主函数
main() {
# 设置错误处理
trap error_handler ERR
trap cleanup EXIT
# 执行测试步骤
find_latest_package
setup_test_env
verify_files
verify_permissions
verify_docker_images
verify_config
simulate_deployment
generate_report
echo "[成功] 部署包测试完成,所有检查都通过了!"
}
# 运行主函数
main "$@"

View File

@ -1,147 +0,0 @@
#!/usr/bin/env python3
"""
DMS合规性测试工具 - API服务器功能测试脚本
测试Docker化的API服务器是否正常工作
"""
import requests
import json
import time
import sys
from pathlib import Path
def test_health_check(api_server):
"""测试健康检查端点"""
print(f"[测试] 健康检查端点...")
try:
response = requests.get(f"{api_server}/", timeout=10)
if response.status_code == 200:
result = response.json()
print(f"[成功] 健康检查通过: {result}")
return True
else:
print(f"[失败] 健康检查失败: {response.status_code}")
return False
except Exception as e:
print(f"[错误] 健康检查异常: {e}")
return False
def test_with_mock_api(api_server):
"""使用模拟API进行测试"""
print(f"[测试] 使用模拟API配置进行测试...")
# 使用一个简单的测试配置不依赖外部API
test_config = {
"base-url": "https://httpbin.org/", # 使用httpbin作为测试目标
"dms": "./assets/doc/dms/domain.json",
"stages-dir": "./custom_stages",
"custom-test-cases-dir": "./custom_testcases",
"verbose": True,
"output": "./test_reports/",
"format": "json",
"generate-pdf": False, # 暂时不生成PDF
"strictness-level": "CRITICAL",
"ignore-ssl": True
}
print(f"[信息] 测试配置:")
print(json.dumps(test_config, indent=2, ensure_ascii=False))
try:
print(f"[信息] 发送测试请求...")
start_time = time.time()
response = requests.post(
f"{api_server}/run",
json=test_config,
headers={"Content-Type": "application/json"},
timeout=60 # 1分钟超时
)
end_time = time.time()
duration = end_time - start_time
print(f"[信息] 请求完成,耗时: {duration:.2f}")
print(f"[信息] HTTP状态码: {response.status_code}")
if response.status_code == 200:
print(f"[成功] API服务器响应正常")
try:
result = response.json()
print(f"[信息] 响应状态: {result.get('status', '未知')}")
if 'message' in result:
print(f"[信息] 响应消息: {result['message']}")
return True
except:
print(f"[信息] 响应内容非JSON: {response.text[:200]}...")
return True
else:
print(f"[失败] API服务器响应异常")
print(f"[错误] 响应内容: {response.text[:200]}...")
return False
except requests.exceptions.Timeout:
print(f"[错误] 请求超时1分钟")
return False
except Exception as e:
print(f"[错误] 请求异常: {e}")
return False
def test_history_viewer(history_server):
"""测试历史查看器"""
print(f"[测试] 历史查看器端点...")
try:
response = requests.get(f"{history_server}/", timeout=10)
if response.status_code == 200:
print(f"[成功] 历史查看器响应正常")
return True
else:
print(f"[失败] 历史查看器响应异常: {response.status_code}")
return False
except Exception as e:
print(f"[错误] 历史查看器异常: {e}")
return False
def main():
print("=== DMS合规性测试工具 - API服务器功能测试 ===")
api_server = "http://localhost:5050"
history_server = "http://localhost:5051"
print(f"[信息] API服务器: {api_server}")
print(f"[信息] 历史查看器: {history_server}")
# 测试结果
results = []
# 1. 测试API服务器健康检查
results.append(("API服务器健康检查", test_health_check(api_server)))
# 2. 测试历史查看器
results.append(("历史查看器", test_history_viewer(history_server)))
# 3. 测试API服务器功能使用模拟API
results.append(("API服务器功能测试", test_with_mock_api(api_server)))
# 显示测试结果摘要
print(f"\n=== 测试结果摘要 ===")
passed = 0
total = len(results)
for test_name, result in results:
status = "✅ 通过" if result else "❌ 失败"
print(f"{status} {test_name}")
if result:
passed += 1
print(f"\n总计: {passed}/{total} 测试通过")
if passed == total:
print(f"[成功] 所有测试通过Docker化的API服务器工作正常")
return 0
else:
print(f"[警告] 部分测试失败,请检查服务状态")
return 1
if __name__ == "__main__":
sys.exit(main())

206
test_fastapi.py Normal file
View File

@ -0,0 +1,206 @@
#!/usr/bin/env python3
"""
测试FastAPI服务器功能的脚本
"""
import json
import time
import requests
import subprocess
import signal
import os
from pathlib import Path
def test_fastapi_server():
"""测试FastAPI服务器功能"""
print("=== FastAPI服务器功能测试 ===")
# 启动服务器
print("[信息] 启动FastAPI服务器...")
server_process = subprocess.Popen([
"python3", "fastapi_server.py",
"--host", "127.0.0.1",
"--port", "5051"
])
try:
# 等待服务器启动
print("[信息] 等待服务器启动...")
time.sleep(3)
base_url = "http://127.0.0.1:5051"
# 测试健康检查
print("\n1. 测试健康检查...")
try:
response = requests.get(f"{base_url}/", timeout=5)
if response.status_code == 200:
data = response.json()
print(f" ✓ 健康检查成功: {data.get('status', 'unknown')}")
print(f" ✓ 服务版本: {data.get('version', 'unknown')}")
else:
print(f" ✗ 健康检查失败: {response.status_code}")
except Exception as e:
print(f" ✗ 健康检查异常: {e}")
# 测试服务信息
print("\n2. 测试服务信息...")
try:
response = requests.get(f"{base_url}/info", timeout=5)
if response.status_code == 200:
data = response.json()
print(f" ✓ 服务信息获取成功")
print(f" ✓ 框架: {data.get('framework', 'unknown')}")
print(f" ✓ 功能数量: {len(data.get('features', []))}")
else:
print(f" ✗ 服务信息获取失败: {response.status_code}")
except Exception as e:
print(f" ✗ 服务信息获取异常: {e}")
# 测试API文档
print("\n3. 测试API文档...")
try:
# 测试Swagger UI
response = requests.get(f"{base_url}/docs", timeout=5)
if response.status_code == 200:
print(" ✓ Swagger UI 可访问")
else:
print(f" ✗ Swagger UI 访问失败: {response.status_code}")
# 测试ReDoc
response = requests.get(f"{base_url}/redoc", timeout=5)
if response.status_code == 200:
print(" ✓ ReDoc 可访问")
else:
print(f" ✗ ReDoc 访问失败: {response.status_code}")
# 测试OpenAPI规范
response = requests.get(f"{base_url}/openapi.json", timeout=5)
if response.status_code == 200:
openapi_spec = response.json()
print(" ✓ OpenAPI规范可获取")
print(f" ✓ API标题: {openapi_spec.get('info', {}).get('title', 'unknown')}")
print(f" ✓ 端点数量: {len(openapi_spec.get('paths', {}))}")
else:
print(f" ✗ OpenAPI规范获取失败: {response.status_code}")
except Exception as e:
print(f" ✗ API文档测试异常: {e}")
# 测试数据验证
print("\n4. 测试数据验证...")
try:
# 测试无效请求
invalid_config = {
"base_url": "invalid-url", # 无效URL
"page_size": 0 # 无效分页大小
}
response = requests.post(f"{base_url}/run", json=invalid_config, timeout=5)
if response.status_code == 422: # Validation Error
print(" ✓ 数据验证正常工作")
error_detail = response.json()
print(f" ✓ 验证错误数量: {len(error_detail.get('detail', []))}")
else:
print(f" ✗ 数据验证异常: {response.status_code}")
except Exception as e:
print(f" ✗ 数据验证测试异常: {e}")
# 测试报告列表
print("\n5. 测试报告列表...")
try:
response = requests.get(f"{base_url}/reports", timeout=5)
if response.status_code == 200:
data = response.json()
print(" ✓ 报告列表获取成功")
print(f" ✓ 报告数量: {len(data.get('reports', []))}")
else:
print(f" ✗ 报告列表获取失败: {response.status_code}")
except Exception as e:
print(f" ✗ 报告列表测试异常: {e}")
print("\n=== 测试完成 ===")
print(f"FastAPI服务器运行在: {base_url}")
print(f"API文档地址: {base_url}/docs")
print(f"ReDoc地址: {base_url}/redoc")
finally:
# 停止服务器
print("\n[信息] 停止服务器...")
server_process.terminate()
try:
server_process.wait(timeout=5)
except subprocess.TimeoutExpired:
server_process.kill()
server_process.wait()
print("[信息] 服务器已停止")
def test_pagination_parameters():
"""测试分页参数功能"""
print("\n=== 分页参数测试 ===")
# 测试配置
test_configs = [
{
"name": "基本配置",
"config": {
"dms": "./test.json",
"base_url": "https://api.example.com",
"page_size": 100,
"page_no": 1
}
},
{
"name": "大分页配置",
"config": {
"dms": "./test.json",
"base_url": "https://api.example.com",
"page_size": 5000,
"page_no": 1
}
},
{
"name": "跳页配置",
"config": {
"dms": "./test.json",
"base_url": "https://api.example.com",
"page_size": 500,
"page_no": 5
}
}
]
for test_case in test_configs:
print(f"\n测试: {test_case['name']}")
config = test_case['config']
# 验证配置格式
try:
from fastapi_server import TestConfig
validated_config = TestConfig(**config)
print(f" ✓ 配置验证通过")
print(f" ✓ 页面大小: {validated_config.page_size}")
print(f" ✓ 起始页码: {validated_config.page_no}")
except Exception as e:
print(f" ✗ 配置验证失败: {e}")
if __name__ == "__main__":
print("开始FastAPI功能测试")
# 检查依赖
try:
import fastapi
import uvicorn
import pydantic
print(f"FastAPI版本: {fastapi.__version__}")
print(f"Pydantic版本: {pydantic.__version__}")
except ImportError as e:
print(f"依赖缺失: {e}")
print("请运行: pip install -r requirements_fastapi.txt")
exit(1)
# 运行测试
test_fastapi_server()
test_pagination_parameters()
print("\n测试完成!")

View File

@ -1,50 +0,0 @@
#!/usr/bin/env python3
"""
测试Docker容器网络连接
"""
import requests
import json
def main():
print("=== Docker网络连接测试 ===")
# 测试配置,使用简单的配置来测试网络连接
test_config = {
"base-url": "http://host.docker.internal:5001/",
"dms": "./assets/doc/dms/domain.json",
"stages-dir": "./custom_stages",
"custom-test-cases-dir": "./custom_testcases",
"verbose": True,
"output": "./test_reports/",
"format": "json",
"generate-pdf": False, # 不生成PDF加快测试
"strictness-level": "CRITICAL",
"ignore-ssl": True
}
print(f"[信息] 测试Docker容器是否能访问宿主机5001端口...")
try:
response = requests.post(
"http://localhost:5050/run",
json=test_config,
headers={"Content-Type": "application/json"},
timeout=30
)
print(f"[信息] HTTP状态码: {response.status_code}")
if response.status_code == 200:
print(f"[成功] 网络连接正常")
result = response.json()
print(f"[信息] 响应状态: {result.get('status', '未知')}")
else:
print(f"[失败] 网络连接异常")
print(f"[错误] 响应: {response.text[:300]}...")
except Exception as e:
print(f"[错误] 网络测试失败: {e}")
if __name__ == "__main__":
main()

View File

@ -1,135 +0,0 @@
#!/usr/bin/env python3
"""
测试DMS分页功能的脚本
"""
import sys
import json
import logging
from pathlib import Path
# 添加项目路径
sys.path.insert(0, str(Path(__file__).parent))
from ddms_compliance_suite.input_parser.parser import InputParser
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def test_pagination():
"""测试DMS分页功能"""
# 测试参数
domain_mapping_path = "./assets/doc/dms/domain.json" # 请根据实际路径调整
base_url = "https://www.dev.ideas.cnpc" # 请根据实际URL调整
# 测试不同的分页大小
page_sizes = [10, 50, 100, 1000]
parser = InputParser()
for page_size in page_sizes:
logger.info(f"\n=== 测试分页大小: {page_size} ===")
try:
result = parser.parse_dms_spec(
domain_mapping_path=domain_mapping_path,
base_url=base_url,
ignore_ssl=True, # 测试环境忽略SSL
page_size=page_size
)
if result and len(result) == 2:
parsed_spec, pagination_info = result
if parsed_spec:
logger.info(f"成功解析 {len(parsed_spec.endpoints)} 个API端点")
# 显示分页信息
logger.info("分页信息:")
logger.info(f" 页面大小: {pagination_info.get('page_size', 'N/A')}")
logger.info(f" 总记录数: {pagination_info.get('total_records', 'N/A')}")
logger.info(f" 总页数: {pagination_info.get('total_pages', 'N/A')}")
logger.info(f" 已获取页数: {pagination_info.get('pages_fetched', 'N/A')}")
# 计算内存使用情况(简单估算)
estimated_memory = len(str(parsed_spec.spec)) / 1024 / 1024 # MB
logger.info(f" 估算内存使用: {estimated_memory:.2f} MB")
else:
logger.error("解析失败返回的parsed_spec为None")
else:
logger.error("解析失败:返回格式不正确")
except Exception as e:
logger.error(f"测试分页大小 {page_size} 时发生错误: {e}")
logger.info("-" * 50)
def test_api_server_integration():
"""测试API服务器集成"""
import requests
logger.info("\n=== 测试API服务器集成 ===")
# API服务器配置
api_url = "http://localhost:5050/run"
test_config = {
"dms": "./assets/doc/dms/domain.json",
"base-url": "https://www.dev.ideas.cnpc",
"page-size": 50, # 测试较小的分页大小
"ignore-ssl": True,
"strictness-level": "CRITICAL",
"output": "./test_reports"
}
try:
logger.info("发送测试请求到API服务器...")
response = requests.post(api_url, json=test_config, timeout=300)
if response.status_code == 200:
result = response.json()
logger.info("API服务器响应成功")
# 检查分页信息
if "pagination" in result:
pagination = result["pagination"]
logger.info("分页信息:")
logger.info(f" 页面大小: {pagination.get('page_size', 'N/A')}")
logger.info(f" 总记录数: {pagination.get('total_records', 'N/A')}")
logger.info(f" 总页数: {pagination.get('total_pages', 'N/A')}")
logger.info(f" 已获取页数: {pagination.get('pages_fetched', 'N/A')}")
else:
logger.warning("响应中未包含分页信息")
# 显示测试摘要
if "summary" in result:
summary = result["summary"]
logger.info(f"测试摘要: 总端点数={summary.get('endpoints_total', 0)}, "
f"成功={summary.get('endpoints_passed', 0)}, "
f"失败={summary.get('endpoints_failed', 0)}")
else:
logger.error(f"API服务器响应错误: {response.status_code}")
logger.error(f"响应内容: {response.text}")
except requests.exceptions.ConnectionError:
logger.warning("无法连接到API服务器请确保服务器正在运行")
except Exception as e:
logger.error(f"测试API服务器时发生错误: {e}")
if __name__ == "__main__":
logger.info("开始测试DMS分页功能")
# 测试基本分页功能
test_pagination()
# 测试API服务器集成可选
if len(sys.argv) > 1 and sys.argv[1] == "--api-server":
test_api_server_integration()
logger.info("测试完成")

232
test_single_page.py Normal file
View File

@ -0,0 +1,232 @@
#!/usr/bin/env python3
"""
测试单页模式功能的脚本
"""
import sys
import json
import logging
from pathlib import Path
# 添加项目路径
sys.path.insert(0, str(Path(__file__).parent))
from ddms_compliance_suite.input_parser.parser import InputParser
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def test_single_page_mode():
"""测试单页模式功能"""
# 测试参数
domain_mapping_path = "./assets/doc/dms/domain.json" # 请根据实际路径调整
base_url = "https://www.dev.ideas.cnpc" # 请根据实际URL调整
parser = InputParser()
print("=== 测试单页模式 vs 全页模式 ===\n")
# 测试1: 单页模式 - 只获取第1页的1条记录
print("1. 测试单页模式 - 只获取第1页的1条记录")
try:
result = parser.parse_dms_spec(
domain_mapping_path=domain_mapping_path,
base_url=base_url,
ignore_ssl=True,
page_size=1,
page_no_start=1,
fetch_all_pages=False # 关键参数:只获取单页
)
if result and len(result) == 2:
parsed_spec, pagination_info = result
if parsed_spec:
print(f" ✓ 成功获取 {len(parsed_spec.endpoints)} 个API端点")
print(f" ✓ 分页信息: {pagination_info}")
print(f" ✓ 获取页数: {pagination_info.get('pages_fetched', 0)}")
print(f" ✓ 模式: {'单页' if not pagination_info.get('fetch_all_pages', True) else '全页'}")
else:
print(" ✗ 解析失败")
else:
print(" ✗ 返回格式错误")
except Exception as e:
print(f" ✗ 异常: {e}")
print("\n" + "-" * 50 + "\n")
# 测试2: 单页模式 - 获取第3页的5条记录
print("2. 测试单页模式 - 获取第3页的5条记录")
try:
result = parser.parse_dms_spec(
domain_mapping_path=domain_mapping_path,
base_url=base_url,
ignore_ssl=True,
page_size=5,
page_no_start=3,
fetch_all_pages=False # 只获取单页
)
if result and len(result) == 2:
parsed_spec, pagination_info = result
if parsed_spec:
print(f" ✓ 成功获取 {len(parsed_spec.endpoints)} 个API端点")
print(f" ✓ 起始页码: {pagination_info.get('page_no_start', 0)}")
print(f" ✓ 当前页码: {pagination_info.get('current_page', 0)}")
print(f" ✓ 获取页数: {pagination_info.get('pages_fetched', 0)}")
print(f" ✓ 模式: {'单页' if not pagination_info.get('fetch_all_pages', True) else '全页'}")
else:
print(" ✗ 解析失败")
else:
print(" ✗ 返回格式错误")
except Exception as e:
print(f" ✗ 异常: {e}")
print("\n" + "-" * 50 + "\n")
# 测试3: 全页模式对比 - 获取所有数据(小分页)
print("3. 测试全页模式对比 - 获取所有数据(分页大小=2")
try:
result = parser.parse_dms_spec(
domain_mapping_path=domain_mapping_path,
base_url=base_url,
ignore_ssl=True,
page_size=2,
page_no_start=1,
fetch_all_pages=True # 获取所有页面
)
if result and len(result) == 2:
parsed_spec, pagination_info = result
if parsed_spec:
print(f" ✓ 成功获取 {len(parsed_spec.endpoints)} 个API端点")
print(f" ✓ 总记录数: {pagination_info.get('total_records', 0)}")
print(f" ✓ 总页数: {pagination_info.get('total_pages', 0)}")
print(f" ✓ 获取页数: {pagination_info.get('pages_fetched', 0)}")
print(f" ✓ 模式: {'单页' if not pagination_info.get('fetch_all_pages', True) else '全页'}")
else:
print(" ✗ 解析失败")
else:
print(" ✗ 返回格式错误")
except Exception as e:
print(f" ✗ 异常: {e}")
def test_command_line_usage():
"""测试命令行使用方式"""
print("\n=== 命令行使用示例 ===\n")
print("1. 单页模式命令行示例:")
print(" python run_api_tests.py \\")
print(" --dms ./assets/doc/dms/domain.json \\")
print(" --base-url https://www.dev.ideas.cnpc \\")
print(" --page-size 5 \\")
print(" --page-no 3 \\")
print(" --fetch-single-page \\")
print(" --ignore-ssl")
print("\n2. 全页模式命令行示例:")
print(" python run_api_tests.py \\")
print(" --dms ./assets/doc/dms/domain.json \\")
print(" --base-url https://www.dev.ideas.cnpc \\")
print(" --page-size 1000 \\")
print(" --page-no 1 \\")
print(" --ignore-ssl")
print("\n3. FastAPI服务器示例:")
print(" # 单页模式")
print(" curl -X POST http://localhost:5051/run \\")
print(" -H 'Content-Type: application/json' \\")
print(" -d '{")
print(" \"dms\": \"./assets/doc/dms/domain.json\",")
print(" \"base_url\": \"https://www.dev.ideas.cnpc\",")
print(" \"page_size\": 5,")
print(" \"page_no\": 3,")
print(" \"fetch_all_pages\": false,")
print(" \"ignore_ssl\": true")
print(" }'")
print("\n # 全页模式")
print(" curl -X POST http://localhost:5051/run \\")
print(" -H 'Content-Type: application/json' \\")
print(" -d '{")
print(" \"dms\": \"./assets/doc/dms/domain.json\",")
print(" \"base_url\": \"https://www.dev.ideas.cnpc\",")
print(" \"page_size\": 1000,")
print(" \"page_no\": 1,")
print(" \"fetch_all_pages\": true")
print(" }'")
def test_use_cases():
"""测试不同使用场景"""
print("\n=== 使用场景说明 ===\n")
scenarios = [
{
"name": "快速测试",
"description": "只测试少量API验证系统功能",
"config": {
"page_size": 5,
"page_no": 1,
"fetch_all_pages": False
}
},
{
"name": "断点续传",
"description": "从中断的地方继续测试",
"config": {
"page_size": 100,
"page_no": 10,
"fetch_all_pages": False
}
},
{
"name": "内存受限环境",
"description": "分批处理大量API避免内存溢出",
"config": {
"page_size": 50,
"page_no": 1,
"fetch_all_pages": False
}
},
{
"name": "完整测试",
"description": "测试所有API生成完整报告",
"config": {
"page_size": 1000,
"page_no": 1,
"fetch_all_pages": True
}
}
]
for i, scenario in enumerate(scenarios, 1):
print(f"{i}. {scenario['name']}")
print(f" 描述: {scenario['description']}")
print(f" 配置: {scenario['config']}")
mode = "单页模式" if not scenario['config']['fetch_all_pages'] else "全页模式"
print(f" 模式: {mode}")
print()
if __name__ == "__main__":
logger.info("开始测试单页模式功能")
# 测试单页模式功能
test_single_page_mode()
# 显示命令行使用示例
test_command_line_usage()
# 显示使用场景
test_use_cases()
logger.info("测试完成")

View File

@ -1,71 +0,0 @@
#!/bin/bash
# DMS合规性测试工具 - curl测试脚本
# 对应原始命令python run_api_tests.py --base-url http://127.0.0.1:5001/ --dms ./assets/doc/dms/domain.json --stages-dir ./custom_stages --custom-test-cases-dir ./custom_testcases -v -o ./test_reports/
echo "=== DMS合规性测试工具 - API测试 ==="
# 配置变量
API_SERVER="http://localhost:5050"
TARGET_API="http://host.docker.internal:5001/" # Docker内部访问宿主机
OUTPUT_LOG="log_dms_docker.txt"
echo "[信息] 目标API: $TARGET_API"
echo "[信息] API服务器: $API_SERVER"
echo "[信息] 输出日志: $OUTPUT_LOG"
# 创建测试配置JSON
TEST_CONFIG='{
"base-url": "'$TARGET_API'",
"dms": "./assets/doc/dms/domain.json",
"stages-dir": "./custom_stages",
"custom-test-cases-dir": "./custom_testcases",
"verbose": true,
"output": "./test_reports/",
"format": "json",
"generate-pdf": true,
"strictness-level": "CRITICAL",
"ignore-ssl": true
}'
echo "[信息] 测试配置:"
echo "$TEST_CONFIG" | jq . 2>/dev/null || echo "$TEST_CONFIG"
# 执行测试
echo ""
echo "[信息] 开始执行测试..."
echo "[信息] 这可能需要几分钟时间..."
curl -X POST "$API_SERVER/run" \
-H "Content-Type: application/json" \
-d "$TEST_CONFIG" \
-w "\n\n=== HTTP响应信息 ===\nHTTP状态码: %{http_code}\n响应时间: %{time_total}s\n" \
> "$OUTPUT_LOG" 2>&1
# 检查结果
if [ $? -eq 0 ]; then
echo "[成功] 测试完成!"
echo "[信息] 查看详细日志: cat $OUTPUT_LOG"
echo "[信息] 查看测试报告: ls -la test_reports/"
# 显示简要结果
echo ""
echo "=== 测试结果摘要 ==="
if command -v jq >/dev/null 2>&1; then
# 如果有jq格式化显示JSON结果
tail -n +1 "$OUTPUT_LOG" | jq -r '.status // "未知状态"' 2>/dev/null || echo "请查看日志文件获取详细结果"
else
# 没有jq显示原始结果
echo "请查看日志文件: $OUTPUT_LOG"
fi
else
echo "[错误] 测试执行失败"
echo "[信息] 查看错误日志: cat $OUTPUT_LOG"
fi
echo ""
echo "=== 管理命令 ==="
echo "- 查看服务状态: docker-compose ps"
echo "- 查看服务日志: docker-compose logs"
echo "- 查看测试日志: cat $OUTPUT_LOG"
echo "- 查看测试报告: ls -la test_reports/"

View File

@ -1,232 +0,0 @@
#!/usr/bin/env python3
"""
DMS合规性测试工具 - 完整测试脚本包含mock服务
1. 启动mock_dms_server.py在5001端口
2. 调用Docker化的API服务器进行测试
3. 清理mock服务
"""
import requests
import json
import time
import sys
import subprocess
import signal
import os
from pathlib import Path
class MockServerManager:
def __init__(self, mock_script="mock_dms_server.py", port=5001):
self.mock_script = mock_script
self.port = port
self.process = None
def start(self):
"""启动mock服务器"""
print(f"[信息] 启动mock服务器: {self.mock_script} (端口 {self.port})")
if not Path(self.mock_script).exists():
print(f"[错误] Mock脚本不存在: {self.mock_script}")
return False
try:
# 启动mock服务器
self.process = subprocess.Popen(
[sys.executable, self.mock_script],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
# 等待服务启动
print(f"[信息] 等待mock服务器启动...")
for i in range(10): # 最多等待10秒
try:
response = requests.get(f"http://127.0.0.1:{self.port}/", timeout=2)
if response.status_code in [200, 404]: # 404也算正常说明服务在运行
print(f"[成功] Mock服务器启动成功 (尝试 {i+1}/10)")
return True
except:
pass
time.sleep(1)
print(f"[错误] Mock服务器启动超时")
return False
except Exception as e:
print(f"[错误] 启动mock服务器失败: {e}")
return False
def stop(self):
"""停止mock服务器"""
if self.process:
print(f"[信息] 停止mock服务器...")
try:
self.process.terminate()
self.process.wait(timeout=5)
print(f"[成功] Mock服务器已停止")
except subprocess.TimeoutExpired:
print(f"[警告] Mock服务器强制终止")
self.process.kill()
self.process.wait()
except Exception as e:
print(f"[错误] 停止mock服务器失败: {e}")
def test_api_server(api_server, target_api, output_log="log_dms_docker.txt"):
"""测试API服务器"""
print(f"\n[测试] API服务器功能测试...")
# 测试配置(对应原始命令行参数)
test_config = {
"base-url": target_api,
"dms": "./assets/doc/dms/domain.json",
"stages-dir": "./custom_stages",
"custom-test-cases-dir": "./custom_testcases",
"verbose": True,
"output": "./test_reports/",
"format": "json",
"generate-pdf": True,
"strictness-level": "CRITICAL",
"ignore-ssl": True
}
print(f"[信息] 测试配置:")
print(json.dumps(test_config, indent=2, ensure_ascii=False))
try:
print(f"[信息] 发送测试请求到API服务器...")
start_time = time.time()
response = requests.post(
f"{api_server}/run",
json=test_config,
headers={"Content-Type": "application/json"},
timeout=120 # 2分钟超时
)
end_time = time.time()
duration = end_time - start_time
# 保存结果到日志文件
with open(output_log, 'w', encoding='utf-8') as f:
f.write(f"=== DMS合规性测试结果 ===\n")
f.write(f"执行时间: {duration:.2f}\n")
f.write(f"HTTP状态码: {response.status_code}\n")
f.write(f"响应头: {dict(response.headers)}\n\n")
if response.status_code == 200:
f.write("=== 测试成功 ===\n")
try:
result_json = response.json()
f.write(json.dumps(result_json, indent=2, ensure_ascii=False))
except:
f.write("响应内容非JSON格式:\n")
f.write(response.text)
else:
f.write("=== 测试失败 ===\n")
f.write(f"错误信息: {response.text}\n")
# 显示结果
print(f"[信息] 请求完成,耗时: {duration:.2f}")
print(f"[信息] HTTP状态码: {response.status_code}")
if response.status_code == 200:
print(f"[成功] 测试执行成功")
try:
result = response.json()
print(f"[信息] 测试状态: {result.get('status', '未知')}")
if 'summary' in result:
summary = result['summary']
print(f"[信息] 测试摘要:")
print(f" - 总测试数: {summary.get('total_tests', 0)}")
print(f" - 通过数: {summary.get('passed_tests', 0)}")
print(f" - 失败数: {summary.get('failed_tests', 0)}")
print(f" - 跳过数: {summary.get('skipped_tests', 0)}")
if 'output_files' in result:
print(f"[信息] 生成的文件:")
for file_info in result['output_files']:
print(f" - {file_info}")
return True
except:
print(f"[信息] 响应内容已保存到: {output_log}")
return True
else:
print(f"[失败] 测试执行失败")
print(f"[错误] 错误信息: {response.text[:200]}...")
return False
except requests.exceptions.Timeout:
print(f"[错误] 请求超时2分钟")
return False
except Exception as e:
print(f"[错误] 请求异常: {e}")
return False
def main():
print("=== DMS合规性测试工具 - 完整功能测试 ===")
api_server = "http://localhost:5050"
history_server = "http://localhost:5051"
target_api = "http://host.docker.internal:5001/" # 使用Docker内部主机访问
# 检查mock脚本是否存在
mock_script = "mock_dms_server.py"
if not Path(mock_script).exists():
print(f"[警告] Mock脚本不存在: {mock_script}")
print(f"[信息] 请手动启动mock服务器: python mock_dms_server.py")
mock_manager = None
else:
mock_manager = MockServerManager(mock_script, 5001)
try:
# 1. 启动mock服务器如果存在
if mock_manager:
if not mock_manager.start():
print(f"[错误] Mock服务器启动失败")
print(f"[建议] 请手动启动: python {mock_script}")
return 1
# 2. 测试API服务器健康检查
print(f"\n[测试] API服务器健康检查...")
health_response = requests.get(f"{api_server}/", timeout=10)
if health_response.status_code == 200:
print(f"[成功] API服务器运行正常")
else:
print(f"[错误] API服务器不可用: {health_response.status_code}")
return 1
# 3. 测试历史查看器
print(f"\n[测试] 历史查看器...")
history_response = requests.get(f"{history_server}/", timeout=10)
if history_response.status_code == 200:
print(f"[成功] 历史查看器运行正常")
else:
print(f"[警告] 历史查看器异常: {history_response.status_code}")
# 4. 执行完整的API测试
success = test_api_server(api_server, target_api)
if success:
print(f"\n[成功] 所有测试完成!")
print(f"[信息] 查看详细日志: cat log_dms_docker.txt")
print(f"[信息] 查看测试报告: ls -la test_reports/")
return 0
else:
print(f"\n[失败] 测试执行失败")
return 1
except KeyboardInterrupt:
print(f"\n[信息] 用户中断测试")
return 1
except Exception as e:
print(f"\n[错误] 测试异常: {e}")
return 1
finally:
# 清理mock服务器
if mock_manager:
mock_manager.stop()
if __name__ == "__main__":
sys.exit(main())

View File

@ -1,136 +0,0 @@
#!/usr/bin/env python3
"""
DMS合规性测试工具 - Python API测试脚本
对应原始命令python run_api_tests.py --base-url http://127.0.0.1:5001/ --dms ./assets/doc/dms/domain.json --stages-dir ./custom_stages --custom-test-cases-dir ./custom_testcases -v -o ./test_reports/
"""
import requests
import json
import time
import sys
from pathlib import Path
def main():
print("=== DMS合规性测试工具 - Python API测试 ===")
# 配置变量
api_server = "http://localhost:5050"
target_api = "http://127.0.0.1:5001/"
output_log = "log_dms_docker.txt"
print(f"[信息] 目标API: {target_api}")
print(f"[信息] API服务器: {api_server}")
print(f"[信息] 输出日志: {output_log}")
# 测试配置(对应原始命令行参数)
test_config = {
"base-url": target_api,
"dms": "./assets/doc/dms/domain.json",
"stages-dir": "./custom_stages",
"custom-test-cases-dir": "./custom_testcases",
"verbose": True,
"output": "./test_reports/",
"format": "json",
"generate-pdf": True,
"strictness-level": "CRITICAL",
"ignore-ssl": True
}
print("\n[信息] 测试配置:")
print(json.dumps(test_config, indent=2, ensure_ascii=False))
try:
# 首先检查API服务器是否可用
print(f"\n[信息] 检查API服务器状态...")
health_response = requests.get(f"{api_server}/", timeout=10)
if health_response.status_code == 200:
print(f"[成功] API服务器运行正常")
else:
print(f"[警告] API服务器状态异常: {health_response.status_code}")
# 执行测试
print(f"\n[信息] 开始执行测试...")
print(f"[信息] 这可能需要几分钟时间...")
start_time = time.time()
response = requests.post(
f"{api_server}/run",
json=test_config,
headers={"Content-Type": "application/json"},
timeout=300 # 5分钟超时
)
end_time = time.time()
duration = end_time - start_time
# 保存结果到日志文件
with open(output_log, 'w', encoding='utf-8') as f:
f.write(f"=== DMS合规性测试结果 ===\n")
f.write(f"执行时间: {duration:.2f}\n")
f.write(f"HTTP状态码: {response.status_code}\n")
f.write(f"响应头: {dict(response.headers)}\n\n")
if response.status_code == 200:
f.write("=== 测试成功 ===\n")
try:
result_json = response.json()
f.write(json.dumps(result_json, indent=2, ensure_ascii=False))
except:
f.write("响应内容非JSON格式:\n")
f.write(response.text)
else:
f.write("=== 测试失败 ===\n")
f.write(f"错误信息: {response.text}\n")
# 显示结果
if response.status_code == 200:
print(f"[成功] 测试完成!耗时: {duration:.2f}")
try:
result = response.json()
print(f"\n=== 测试结果摘要 ===")
print(f"状态: {result.get('status', '未知')}")
if 'summary' in result:
summary = result['summary']
print(f"总测试数: {summary.get('total_tests', 0)}")
print(f"通过数: {summary.get('passed_tests', 0)}")
print(f"失败数: {summary.get('failed_tests', 0)}")
print(f"跳过数: {summary.get('skipped_tests', 0)}")
if 'output_files' in result:
print(f"\n=== 生成的文件 ===")
for file_info in result['output_files']:
print(f"- {file_info}")
except Exception as e:
print(f"[警告] 解析JSON响应失败: {e}")
print(f"[信息] 原始响应已保存到: {output_log}")
else:
print(f"[错误] 测试执行失败")
print(f"[信息] HTTP状态码: {response.status_code}")
print(f"[信息] 错误信息: {response.text[:200]}...")
print(f"\n[信息] 详细日志已保存到: {output_log}")
print(f"[信息] 查看测试报告: ls -la test_reports/")
except requests.exceptions.Timeout:
print(f"[错误] 请求超时5分钟")
print(f"[建议] 测试可能仍在后台运行请稍后查看test_reports/目录")
except requests.exceptions.ConnectionError:
print(f"[错误] 无法连接到API服务器: {api_server}")
print(f"[建议] 请确认Docker容器正在运行: docker-compose ps")
except Exception as e:
print(f"[错误] 执行失败: {e}")
return 1
print(f"\n=== 管理命令 ===")
print(f"- 查看服务状态: docker-compose ps")
print(f"- 查看服务日志: docker-compose logs")
print(f"- 查看测试日志: cat {output_log}")
print(f"- 查看测试报告: ls -la test_reports/")
return 0
if __name__ == "__main__":
sys.exit(main())