diff --git a/Dockerfile.fastapi b/Dockerfile.fastapi new file mode 100644 index 0000000..aaac0b0 --- /dev/null +++ b/Dockerfile.fastapi @@ -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"] diff --git a/OFFLINE_DEPLOYMENT_SUMMARY.md b/OFFLINE_DEPLOYMENT_SUMMARY.md deleted file mode 100644 index 22f978c..0000000 --- a/OFFLINE_DEPLOYMENT_SUMMARY.md +++ /dev/null @@ -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. **设置备份** - 建立定期数据备份机制 diff --git a/ORGANIZATION_SUMMARY.md b/ORGANIZATION_SUMMARY.md deleted file mode 100644 index c100b68..0000000 --- a/ORGANIZATION_SUMMARY.md +++ /dev/null @@ -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] 项目结构文档创建 - -现在项目结构更加专业和易于管理！🎯 diff --git a/create-compose-package-fastapi.sh b/create-compose-package-fastapi.sh new file mode 100755 index 0000000..2c67a00 --- /dev/null +++ b/create-compose-package-fastapi.sh @@ -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部署包创建完成！" diff --git a/ddms_compliance_suite/input_parser/parser.py b/ddms_compliance_suite/input_parser/parser.py index 98aef45..7e1d08a 100644 --- a/ddms_compliance_suite/input_parser/parser.py +++ b/ddms_compliance_suite/input_parser/parser.py @@ -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 diff --git a/ddms_compliance_suite/test_orchestrator.py b/ddms_compliance_suite/test_orchestrator.py index 7fc4037..a9a6ed4 100644 --- a/ddms_compliance_suite/test_orchestrator.py +++ b/ddms_compliance_suite/test_orchestrator.py @@ -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，测试终止。") diff --git a/docker-compose.fastapi.yml b/docker-compose.fastapi.yml new file mode 100644 index 0000000..67dc3cd --- /dev/null +++ b/docker-compose.fastapi.yml @@ -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: diff --git a/docs/FastAPI_Pagination_Summary.md b/docs/FastAPI_Pagination_Summary.md new file mode 100644 index 0000000..e51ce28 --- /dev/null +++ b/docs/FastAPI_Pagination_Summary.md @@ -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文档支持！ diff --git a/docs/FastAPI_Server_Guide.md b/docs/FastAPI_Server_Guide.md new file mode 100644 index 0000000..d947b5f --- /dev/null +++ b/docs/FastAPI_Server_Guide.md @@ -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. **测试策略** + - 使用小分页大小进行快速测试 + - 利用交互式文档进行手动测试 + - 编写自动化测试脚本 diff --git a/fastapi_server.py b/fastapi_server.py new file mode 100644 index 0000000..82af2df --- /dev/null +++ b/fastapi_server.py @@ -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" + ) diff --git a/requirements_fastapi.txt b/requirements_fastapi.txt new file mode 100644 index 0000000..de32673 --- /dev/null +++ b/requirements_fastapi.txt @@ -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 diff --git a/run_api_tests.py b/run_api_tests.py index 5f4d2b8..d4fc900 100644 --- a/run_api_tests.py +++ b/run_api_tests.py @@ -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("未获取到分页信息") diff --git a/start_fastapi.sh b/start_fastapi.sh new file mode 100755 index 0000000..00dda97 --- /dev/null +++ b/start_fastapi.sh @@ -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 diff --git a/test-deployment.sh b/test-deployment.sh deleted file mode 100755 index 754c0c0..0000000 --- a/test-deployment.sh +++ /dev/null @@ -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 "$@" diff --git a/test_api_server.py b/test_api_server.py deleted file mode 100644 index 69e81a0..0000000 --- a/test_api_server.py +++ /dev/null @@ -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()) diff --git a/test_fastapi.py b/test_fastapi.py new file mode 100644 index 0000000..4f5a403 --- /dev/null +++ b/test_fastapi.py @@ -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测试完成！") diff --git a/test_network.py b/test_network.py deleted file mode 100644 index 0d789a7..0000000 --- a/test_network.py +++ /dev/null @@ -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() diff --git a/test_pagination.py b/test_pagination.py deleted file mode 100644 index 61a15d7..0000000 --- a/test_pagination.py +++ /dev/null @@ -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("测试完成") diff --git a/test_single_page.py b/test_single_page.py new file mode 100644 index 0000000..58d4819 --- /dev/null +++ b/test_single_page.py @@ -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("测试完成") diff --git a/test_with_curl.sh b/test_with_curl.sh deleted file mode 100755 index 7cb57db..0000000 --- a/test_with_curl.sh +++ /dev/null @@ -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/" diff --git a/test_with_mock.py b/test_with_mock.py deleted file mode 100644 index 363677e..0000000 --- a/test_with_mock.py +++ /dev/null @@ -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()) diff --git a/test_with_python.py b/test_with_python.py deleted file mode 100755 index 13e2e97..0000000 --- a/test_with_python.py +++ /dev/null @@ -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())