From a0ee9aa312eeea8bfc61359cf4f63640a62ddbc9 Mon Sep 17 00:00:00 2001 From: gongwenxin Date: Wed, 27 Aug 2025 16:29:44 +0800 Subject: [PATCH 1/2] fix:fastapi --- create-compose-package-multiplatform.sh | 165 +++++++--------- fastapi_server.py | 103 +++++----- test-directory-fix.sh | 179 ----------------- test-multiplatform-script.sh | 253 ------------------------ 4 files changed, 114 insertions(+), 586 deletions(-) delete mode 100644 test-directory-fix.sh delete mode 100755 test-multiplatform-script.sh diff --git a/create-compose-package-multiplatform.sh b/create-compose-package-multiplatform.sh index 041609f..583913d 100755 --- a/create-compose-package-multiplatform.sh +++ b/create-compose-package-multiplatform.sh @@ -282,56 +282,40 @@ cd "$TEMP_BUILD_DIR" if [[ "$SELECTED_SERVICE_ARCH" == "dual" ]]; then # 双服务架构 - 使用supervisor管理两个服务 cat > "Dockerfile" << 'EOF' -# 多阶段构建 - 构建阶段 -FROM python:3.11-alpine AS builder +# 使用稳定的Python基础镜像 +FROM python:3.11-alpine -# 安装构建依赖 -RUN apk add --no-cache \ +# 安装系统依赖 +RUN apk update && apk add --no-cache \ gcc \ musl-dev \ libffi-dev \ openssl-dev \ - cargo \ - rust - -# 设置工作目录 -WORKDIR /app - -# 复制依赖文件 -COPY requirements.txt . - -# 创建虚拟环境并安装依赖 -RUN python -m venv /opt/venv -ENV PATH="/opt/venv/bin:$PATH" -RUN pip install --no-cache-dir --upgrade pip setuptools wheel -RUN pip install --no-cache-dir -r requirements.txt - -# 运行阶段 -FROM python:3.11-alpine AS runtime - -# 安装运行时依赖 -RUN apk add --no-cache \ + python3-dev \ + build-base \ + linux-headers \ supervisor \ curl \ bash \ - tzdata - -# 从构建阶段复制虚拟环境 -COPY --from=builder /opt/venv /opt/venv -ENV PATH="/opt/venv/bin:$PATH" + tzdata && \ + rm -rf /var/cache/apk/* # 设置工作目录 WORKDIR /app +# 复制依赖文件并安装Python包 +COPY requirements.txt . +RUN pip install --no-cache-dir --upgrade pip setuptools wheel && \ + pip install --no-cache-dir -r requirements.txt + # 复制应用代码 COPY . . -# 创建supervisor配置 -RUN mkdir -p /etc/supervisor/conf.d -COPY supervisord.conf /etc/supervisor/conf.d/ +# 创建supervisor配置目录 +RUN mkdir -p /etc/supervisor/conf.d /var/log/supervisor /app/logs /app/test_reports /app/uploads -# 创建必要目录 -RUN mkdir -p /var/log/supervisor /app/logs /app/test_reports /app/uploads +# 复制supervisor配置 +COPY supervisord.conf /etc/supervisor/conf.d/ # 创建非root用户 RUN addgroup -g 1000 appuser && \ @@ -346,7 +330,7 @@ ENV FLASK_ENV=production ENV PYTHONUNBUFFERED=1 # 健康检查 -HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \ +HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \ CMD curl -f http://localhost:5050/ || exit 1 # 暴露端口 @@ -409,50 +393,38 @@ EOF elif [[ "$SELECTED_SERVICE_ARCH" == "fastapi" ]]; then cat > "Dockerfile" << 'EOF' -# 多阶段构建 - 构建阶段 -FROM python:3.11-alpine AS builder +# 使用稳定的Python基础镜像 +FROM python:3.11-alpine -# 安装构建依赖 -RUN apk add --no-cache \ +# 安装系统依赖 +RUN apk update && apk add --no-cache \ gcc \ musl-dev \ libffi-dev \ openssl-dev \ - cargo \ - rust - -# 设置工作目录 -WORKDIR /app - -# 复制依赖文件 -COPY requirements.txt . - -# 创建虚拟环境并安装依赖 -RUN python -m venv /opt/venv -ENV PATH="/opt/venv/bin:$PATH" -RUN pip install --no-cache-dir --upgrade pip setuptools wheel -RUN pip install --no-cache-dir -r requirements.txt -RUN pip install --no-cache-dir fastapi uvicorn[standard] - -# 运行阶段 -FROM python:3.11-alpine AS runtime - -# 安装运行时依赖 -RUN apk add --no-cache \ + python3-dev \ + build-base \ + linux-headers \ curl \ bash \ - tzdata - -# 从构建阶段复制虚拟环境 -COPY --from=builder /opt/venv /opt/venv -ENV PATH="/opt/venv/bin:$PATH" + tzdata && \ + rm -rf /var/cache/apk/* # 设置工作目录 WORKDIR /app +# 复制依赖文件并安装Python包 +COPY requirements.txt . +RUN pip install --no-cache-dir --upgrade pip setuptools wheel && \ + pip install --no-cache-dir -r requirements.txt && \ + pip install --no-cache-dir fastapi uvicorn[standard] + # 复制应用代码 COPY . . +# 创建必要目录 +RUN mkdir -p /app/logs /app/uploads /app/reports + # 创建非root用户 RUN addgroup -g 1000 appuser && \ adduser -D -u 1000 -G appuser appuser && \ @@ -460,8 +432,12 @@ RUN addgroup -g 1000 appuser && \ USER appuser +# 设置环境变量 +ENV PYTHONPATH=/app +ENV PYTHONUNBUFFERED=1 + # 健康检查 -HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \ +HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \ CMD curl -f http://localhost:5051/health || exit 1 # 暴露端口 @@ -827,50 +803,38 @@ EOF else # Flask版本的Dockerfile cat > "Dockerfile" << 'EOF' -# 多阶段构建 - 构建阶段 -FROM python:3.11-alpine AS builder +# 使用稳定的Python基础镜像 +FROM python:3.11-alpine -# 安装构建依赖 -RUN apk add --no-cache \ +# 安装系统依赖 +RUN apk update && apk add --no-cache \ gcc \ musl-dev \ libffi-dev \ openssl-dev \ - cargo \ - rust - -# 设置工作目录 -WORKDIR /app - -# 复制依赖文件 -COPY requirements.txt . - -# 创建虚拟环境并安装依赖 -RUN python -m venv /opt/venv -ENV PATH="/opt/venv/bin:$PATH" -RUN pip install --no-cache-dir --upgrade pip setuptools wheel -RUN pip install --no-cache-dir -r requirements.txt -RUN pip install --no-cache-dir flask gunicorn - -# 运行阶段 -FROM python:3.11-alpine AS runtime - -# 安装运行时依赖 -RUN apk add --no-cache \ + python3-dev \ + build-base \ + linux-headers \ curl \ bash \ - tzdata - -# 从构建阶段复制虚拟环境 -COPY --from=builder /opt/venv /opt/venv -ENV PATH="/opt/venv/bin:$PATH" + tzdata && \ + rm -rf /var/cache/apk/* # 设置工作目录 WORKDIR /app +# 复制依赖文件并安装Python包 +COPY requirements.txt . +RUN pip install --no-cache-dir --upgrade pip setuptools wheel && \ + pip install --no-cache-dir -r requirements.txt && \ + pip install --no-cache-dir flask gunicorn + # 复制应用代码 COPY . . +# 创建必要目录 +RUN mkdir -p /app/logs /app/uploads /app/reports + # 创建非root用户 RUN addgroup -g 1000 appuser && \ adduser -D -u 1000 -G appuser appuser && \ @@ -878,9 +842,14 @@ RUN addgroup -g 1000 appuser && \ USER appuser +# 设置环境变量 +ENV PYTHONPATH=/app +ENV FLASK_ENV=production +ENV PYTHONUNBUFFERED=1 + # 健康检查 -HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \ - CMD curl -f http://localhost:5050/health || exit 1 +HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \ + CMD curl -f http://localhost:5050/ || exit 1 # 暴露端口 EXPOSE 5050 diff --git a/fastapi_server.py b/fastapi_server.py index 82af2df..65e6063 100644 --- a/fastapi_server.py +++ b/fastapi_server.py @@ -87,46 +87,20 @@ 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") - + yapi: Optional[str] = Field(None, description="YAPI定义文件路径", exclude=True) + swagger: Optional[str] = Field(None, description="Swagger/OpenAPI定义文件路径", exclude=True) + dms: Optional[str] = Field("./assets/doc/dms/domain.json", description="DMS服务发现的domain mapping文件路径", example="./assets/doc/dms/domain.json") # 基本配置 - base_url: str = Field(..., description="API基础URL", example="https://api.example.com") + base_url: str = Field("https://www.dev.ideas.cnpc/", description="API基础URL", example="https://www.dev.ideas.cnpc/") # 分页配置 - page_size: int = Field(1000, description="DMS API分页大小，默认1000。较小的值可以减少内存使用", ge=1, le=10000) + page_size: int = Field(10, description="DMS API分页大小，默认10。较小的值可以减少内存使用", ge=1, le=10000) page_no: int = Field(1, description="起始页码，从1开始。可用于断点续传或跳过前面的页面", ge=1) - fetch_all_pages: bool = Field(True, description="是否获取所有页面。True=获取所有数据，False=只获取指定页面") + fetch_all_pages: bool = Field(False, 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): @@ -239,7 +213,7 @@ def run_tests_logic(config: dict): 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()}") - + print(f"config{config}") # Initialize the orchestrator orchestrator = APITestOrchestrator( base_url=config['base_url'], @@ -291,9 +265,7 @@ def run_tests_logic(config: dict): 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) + orchestrator.run_stages_from_spec(parsed_spec, test_summary) if test_summary: # Save main summary @@ -344,29 +316,29 @@ def save_api_call_details_to_markdown(api_call_details: List[APICallDetail], out 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") + 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("请求头:\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("请求体:\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") + f.write(f"响应状态码: `{detail.response_status_code}`\n\n") if detail.response_headers: - f.write("**响应头**:\n```json\n") + 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("响应体:\n```json\n") f.write(json.dumps(detail.response_body, indent=2, ensure_ascii=False)) f.write("\n```\n\n") @@ -383,16 +355,16 @@ def save_api_call_details_to_markdown(api_call_details: List[APICallDetail], out 执行API合规性测试的主要端点。 支持三种API定义源： - - **YAPI**: 基于YAPI定义文件 - - **Swagger/OpenAPI**: 基于OpenAPI规范文件 - - **DMS**: 动态发现DMS服务的API + - YAPI: 基于YAPI定义文件 + - Swagger/OpenAPI: 基于OpenAPI规范文件 + - DMS: 动态发现DMS服务的API - ### 分页支持 + 分页支持 对于DMS测试，支持分页获取API列表，避免内存溢出： - `page_size`: 每页获取的API数量（默认1000） - 返回详细的分页统计信息 - ### LLM集成 + LLM集成 可选择使用大语言模型生成测试数据： - 智能生成请求体、路径参数、查询参数等 - 提高测试覆盖率和数据多样性 @@ -407,17 +379,36 @@ async def run_api_tests(config: TestConfig): """ 执行API合规性测试 - - **config**: 测试配置，包含API定义源、测试参数等 - - **returns**: 测试结果，包含摘要信息和分页信息（如适用） + - 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()} + + # Add hidden parameters with default values + hidden_defaults = { + "categories": [], + "tags": [], + "ignore_ssl": True, + "output": "./test_reports", + "generate_pdf": True, + "custom_test_cases_dir": "./custom_testcases", + "stages_dir": "./custom_stages", + "llm_api_key": "sk-lbGrsUPL1iby86h554FaE536C343435dAa9bA65967A840B2", + "llm_base_url": "https://aiproxy.petrotech.cnpc/v1", + "llm_model_name": "deepseek-v3", + "use_llm_for_request_body": False, + "use_llm_for_path_params": False, + "use_llm_for_query_params": False, + "use_llm_for_headers": False, + "verbose": False + } + + # Merge hidden defaults with config + config_dict.update(hidden_defaults) result = run_tests_logic(config_dict) @@ -456,8 +447,8 @@ async def download_report(report_id: str, file_type: str = "summary.json"): """ 下载测试报告文件 - - **report_id**: 报告ID（通常是时间戳） - - **file_type**: 文件类型，可选值：summary.json, api_call_details.md + - report_id: 报告ID（通常是时间戳） + - file_type: 文件类型，可选值：summary.json, api_call_details.md """ try: report_dir = Path("./test_reports") / report_id diff --git a/test-directory-fix.sh b/test-directory-fix.sh deleted file mode 100644 index 25143a1..0000000 --- a/test-directory-fix.sh +++ /dev/null @@ -1,179 +0,0 @@ -#!/bin/bash - -# 测试目录创建和复制逻辑 - -set -e - -echo "=== 测试目录创建和复制逻辑 ===" - -# 模拟变量 -SELECTED_SERVICE_ARCH="dual" -TARGET_PLATFORM="linux/arm64" -MULTI_PLATFORM=false - -# 更新导出目录名称以包含平台信息 -if [[ "$MULTI_PLATFORM" == "true" ]]; then - EXPORT_DIR="dms-compliance-${SELECTED_SERVICE_ARCH}-multiplatform-$(date +%Y%m%d-%H%M%S)" -else - platform_suffix=$(echo "$TARGET_PLATFORM" | sed 's/linux\///g' | sed 's/\//-/g') - EXPORT_DIR="dms-compliance-${SELECTED_SERVICE_ARCH}-${platform_suffix}-$(date +%Y%m%d-%H%M%S)" -fi -ARCHIVE_NAME="$EXPORT_DIR.tar.gz" - -echo "[信息] 最终输出目录: $EXPORT_DIR" - -# 创建最终导出目录 -rm -rf "$EXPORT_DIR" -mkdir -p "$EXPORT_DIR" - -echo "[步骤 1/4] 复制项目文件..." - -# 创建临时构建目录 -TEMP_BUILD_DIR=$(mktemp -d) -trap "rm -rf $TEMP_BUILD_DIR" EXIT - -# 复制核心目录（排除缓存和临时文件） -echo "[信息] 复制核心目录..." -mkdir -p "$TEMP_BUILD_DIR"/{ddms_compliance_suite,custom_stages,custom_testcases,templates,static,assets} - -# 创建测试文件 -echo "# Test content" > "$TEMP_BUILD_DIR/requirements.txt" -echo "# Test API server" > "$TEMP_BUILD_DIR/api_server.py" -echo "# Test history viewer" > "$TEMP_BUILD_DIR/history_viewer.py" -echo "# Test ddms suite" > "$TEMP_BUILD_DIR/ddms_compliance_suite/test.py" - -echo "[步骤 2/4] 创建 Dockerfile..." -cd "$TEMP_BUILD_DIR" - -# 创建测试Dockerfile -cat > "Dockerfile" << 'EOF' -FROM python:3.11-alpine -WORKDIR /app -COPY . . -EXPOSE 5050 5051 -CMD ["echo", "Test dockerfile"] -EOF - -# 创建supervisor配置 -cat > "supervisord.conf" << 'EOF' -[supervisord] -nodaemon=true - -[program:api_server] -command=python api_server.py - -[program:history_viewer] -command=python history_viewer.py -EOF - -cd .. - -echo "[步骤 3/4] 复制构建文件..." -# 确保目标目录存在 -mkdir -p "$EXPORT_DIR" -cp -r "$TEMP_BUILD_DIR"/* "$EXPORT_DIR/" - -echo "[步骤 4/4] 创建配置文件..." - -# 创建Docker Compose文件 -cat > "$EXPORT_DIR/docker-compose.yml" << EOF -version: '3.8' - -services: - dms-compliance: - build: - context: . - dockerfile: Dockerfile - platforms: - - $TARGET_PLATFORM - image: compliance-dms-multiplatform:latest - container_name: dms-compliance-tool - ports: - - "5050:5050" # API服务器端口 - - "5051:5051" # 历史查看器端口 - environment: - - PYTHONPATH=/app - - TZ=Asia/Shanghai - - FLASK_ENV=production - - PYTHONUNBUFFERED=1 - volumes: - - ./uploads:/app/uploads - - ./logs:/app/logs - - ./test_reports:/app/test_reports - - ./config:/app/config:ro - restart: unless-stopped - healthcheck: - test: ["CMD", "curl", "-f", "http://localhost:5050/"] - interval: 30s - timeout: 10s - retries: 3 - start_period: 40s - networks: - - dms-network - -networks: - dms-network: - driver: bridge - -volumes: - uploads: - logs: - test_reports: - config: -EOF - -# 创建启动脚本 -cat > "$EXPORT_DIR/start.sh" << 'EOF' -#!/bin/bash -echo "=== DMS合规性测试工具启动脚本 ===" -echo "这是一个测试脚本" -EOF - -chmod +x "$EXPORT_DIR/start.sh" - -# 创建README -cat > "$EXPORT_DIR/README.md" << EOF -# DMS合规性测试工具 - 测试部署包 - -## 系统信息 - -- **架构**: 双服务架构 - API服务器(5050) + 历史查看器(5051) -- **端口**: 5050,5051 -- **目标平台**: ARM64 (aarch64) - Apple M1/M2, ARM 64位 -- **构建时间**: $(date '+%Y-%m-%d %H:%M:%S') - -## 测试成功 - -目录创建和文件复制逻辑正常工作。 -EOF - -# 验证文件结构 -echo "" -echo "=== 验证文件结构 ===" -echo "导出目录内容:" -ls -la "$EXPORT_DIR" -echo "" -echo "核心文件检查:" -for file in Dockerfile docker-compose.yml start.sh README.md requirements.txt api_server.py history_viewer.py supervisord.conf; do - if [[ -f "$EXPORT_DIR/$file" ]]; then - echo "✅ $file" - else - echo "❌ $file (缺失)" - fi -done - -# 创建压缩包 -echo "" -echo "[信息] 创建压缩包..." -tar -czf "$ARCHIVE_NAME" "$EXPORT_DIR" - -# 清理临时目录 -rm -rf "$EXPORT_DIR" - -# 显示结果 -echo "" -echo "=== 测试完成 ===" -echo "[成功] 测试部署包已创建: $ARCHIVE_NAME" -echo "[信息] 文件大小: $(du -h "$ARCHIVE_NAME" | cut -f1)" -echo "" -echo "目录创建和复制逻辑测试成功！" diff --git a/test-multiplatform-script.sh b/test-multiplatform-script.sh deleted file mode 100755 index 8ff4ee2..0000000 --- a/test-multiplatform-script.sh +++ /dev/null @@ -1,253 +0,0 @@ -#!/bin/bash - -# 测试多平台脚本的基本功能（不构建Docker） - -set -e - -# 配置变量 -EXPORT_DIR="dms-compliance-multiplatform-test-$(date +%Y%m%d-%H%M%S)" -IMAGE_NAME="compliance-dms-multiplatform" -ARCHIVE_NAME="$EXPORT_DIR.tar.gz" - -# 支持的平台列表 - 使用函数替代关联数组以兼容旧版bash -get_platform() { - case "$1" in - 1) echo "linux/amd64" ;; - 2) echo "linux/arm64" ;; - 3) echo "linux/arm/v7" ;; - 4) echo "linux/arm/v6" ;; - 5) echo "linux/386" ;; - 6) echo "linux/ppc64le" ;; - 7) echo "linux/s390x" ;; - 8) echo "linux/riscv64" ;; - *) echo "" ;; - esac -} - -get_platform_name() { - case "$1" in - "linux/amd64") echo "AMD64 (x86_64) - Intel/AMD 64位" ;; - "linux/arm64") echo "ARM64 (aarch64) - Apple M1/M2, ARM 64位" ;; - "linux/arm/v7") echo "ARMv7 - 树莓派 3/4, ARM 32位" ;; - "linux/arm/v6") echo "ARMv6 - 树莓派 1/Zero, ARM 32位" ;; - "linux/386") echo "i386 - Intel/AMD 32位" ;; - "linux/ppc64le") echo "PowerPC 64位小端" ;; - "linux/s390x") echo "IBM System z" ;; - "linux/riscv64") echo "RISC-V 64位" ;; - *) echo "未知平台" ;; - esac -} - -# 服务架构选择 -get_service_arch() { - case "$1" in - 1) echo "dual" ;; - 2) echo "fastapi" ;; - 3) echo "flask" ;; - *) echo "" ;; - esac -} - -get_service_arch_name() { - case "$1" in - "dual") echo "双服务架构 - API服务器(5050) + 历史查看器(5051)" ;; - "fastapi") echo "FastAPI单服务 - 现代异步框架，自动生成API文档(5051)" ;; - "flask") echo "Flask单服务 - 轻量级传统框架(5050)" ;; - *) echo "未知架构" ;; - esac -} - -get_service_ports() { - case "$1" in - "dual") echo "5050,5051" ;; - "fastapi") echo "5051" ;; - "flask") echo "5050" ;; - *) echo "5050" ;; - esac -} - -echo "=== DMS合规性测试工具 跨平台脚本测试 ===" -echo "" - -# 检测当前平台 -CURRENT_ARCH=$(uname -m) -case "$CURRENT_ARCH" in - x86_64|amd64) CURRENT_PLATFORM="linux/amd64" ;; - aarch64|arm64) CURRENT_PLATFORM="linux/arm64" ;; - armv7l) CURRENT_PLATFORM="linux/arm/v7" ;; - armv6l) CURRENT_PLATFORM="linux/arm/v6" ;; - i386|i686) CURRENT_PLATFORM="linux/386" ;; - *) CURRENT_PLATFORM="linux/amd64" ;; -esac - -echo "[信息] 当前平台: $(get_platform_name "$CURRENT_PLATFORM")" -echo "" - -# 选择服务架构 -echo "请选择服务架构:" -echo " 1) $(get_service_arch_name "dual")" -echo " 2) $(get_service_arch_name "fastapi")" -echo " 3) $(get_service_arch_name "flask")" -echo "" -read -p "请输入选择 (1-3) [默认: 1]: " service_choice -service_choice=${service_choice:-1} - -SELECTED_SERVICE_ARCH=$(get_service_arch "$service_choice") -if [[ -z "$SELECTED_SERVICE_ARCH" ]]; then - echo "[错误] 无效的服务架构选择" - exit 1 -fi - -SELECTED_PORTS=$(get_service_ports "$SELECTED_SERVICE_ARCH") - -echo "[信息] 选择的架构: $(get_service_arch_name "$SELECTED_SERVICE_ARCH")" -echo "[信息] 服务端口: $SELECTED_PORTS" -echo "" - -# 选择目标平台 -echo "请选择目标平台架构:" -for key in 1 2 3 4 5 6 7 8; do - platform=$(get_platform "$key") - name=$(get_platform_name "$platform") - if [[ "$platform" == "$CURRENT_PLATFORM" ]]; then - echo " $key) $name [当前平台]" - else - echo " $key) $name" - fi -done -echo " 9) 多平台构建 (同时构建多个平台)" -echo " 0) 自动检测当前平台" -echo "" -read -p "请输入选择 (0-9) [默认: 0]: " platform_choice -platform_choice=${platform_choice:-0} - -if [[ "$platform_choice" == "0" ]]; then - TARGET_PLATFORM="$CURRENT_PLATFORM" - TARGET_PLATFORM_NAME="$(get_platform_name "$CURRENT_PLATFORM") [自动检测]" - MULTI_PLATFORM=false -elif [[ "$platform_choice" == "9" ]]; then - TARGET_PLATFORM="linux/amd64,linux/arm64" - TARGET_PLATFORM_NAME="常用平台 (AMD64 + ARM64)" - MULTI_PLATFORM=true -else - TARGET_PLATFORM=$(get_platform "$platform_choice") - if [[ -z "$TARGET_PLATFORM" ]]; then - echo "[错误] 无效的平台选择" - exit 1 - fi - TARGET_PLATFORM_NAME=$(get_platform_name "$TARGET_PLATFORM") - MULTI_PLATFORM=false -fi - -echo "[信息] 目标平台: $TARGET_PLATFORM_NAME" -echo "[信息] 多平台构建: $MULTI_PLATFORM" -echo "" - -# 确认构建 -echo "构建配置确认:" -echo " 架构: $(get_service_arch_name "$SELECTED_SERVICE_ARCH")" -echo " 端口: $SELECTED_PORTS" -echo " 平台: $TARGET_PLATFORM_NAME" -echo "" -read -p "确认开始测试? (y/N): " confirm -if [[ ! "$confirm" =~ ^[Yy]$ ]]; then - echo "[信息] 测试已取消" - exit 0 -fi - -# 更新导出目录名称以包含平台信息 -if [[ "$MULTI_PLATFORM" == "true" ]]; then - EXPORT_DIR="dms-compliance-${SELECTED_SERVICE_ARCH}-multiplatform-$(date +%Y%m%d-%H%M%S)" -else - platform_suffix=$(echo "$TARGET_PLATFORM" | sed 's/linux\///g' | sed 's/\//-/g') - EXPORT_DIR="dms-compliance-${SELECTED_SERVICE_ARCH}-${platform_suffix}-$(date +%Y%m%d-%H%M%S)" -fi -ARCHIVE_NAME="$EXPORT_DIR.tar.gz" - -echo "[信息] 最终输出目录: $EXPORT_DIR" - -# 创建最终导出目录 -rm -rf "$EXPORT_DIR" -mkdir -p "$EXPORT_DIR" - -echo "" -echo "[步骤 1/3] 创建测试文件..." - -# 创建临时构建目录 -TEMP_BUILD_DIR=$(mktemp -d) -trap "rm -rf $TEMP_BUILD_DIR" EXIT - -# 创建测试文件 -mkdir -p "$TEMP_BUILD_DIR"/{ddms_compliance_suite,custom_stages,templates,static} -echo "# Test requirements" > "$TEMP_BUILD_DIR/requirements.txt" -echo "# Test API server" > "$TEMP_BUILD_DIR/api_server.py" -echo "# Test history viewer" > "$TEMP_BUILD_DIR/history_viewer.py" - -echo "[步骤 2/3] 创建配置文件..." - -# 创建测试Dockerfile -if [[ "$SELECTED_SERVICE_ARCH" == "dual" ]]; then - cat > "$TEMP_BUILD_DIR/Dockerfile" << 'EOF' -FROM python:3.11-alpine -WORKDIR /app -COPY . . -EXPOSE 5050 5051 -CMD ["echo", "Dual service architecture"] -EOF - - cat > "$TEMP_BUILD_DIR/supervisord.conf" << 'EOF' -[supervisord] -nodaemon=true - -[program:api_server] -command=python api_server.py - -[program:history_viewer] -command=python history_viewer.py -EOF - -else - cat > "$TEMP_BUILD_DIR/Dockerfile" << 'EOF' -FROM python:3.11-alpine -WORKDIR /app -COPY . . -EXPOSE 5050 -CMD ["echo", "Single service architecture"] -EOF -fi - -echo "[步骤 3/3] 复制文件到最终目录..." - -# 复制构建文件到最终目录 -cp -r "$TEMP_BUILD_DIR"/* "$EXPORT_DIR/" - -# 创建README -cat > "$EXPORT_DIR/README.md" << EOF -# DMS合规性测试工具 - 测试部署包 - -## 配置信息 - -- **架构**: $(get_service_arch_name "$SELECTED_SERVICE_ARCH") -- **端口**: $SELECTED_PORTS -- **目标平台**: $TARGET_PLATFORM_NAME -- **构建时间**: $(date '+%Y-%m-%d %H:%M:%S') - -## 测试成功 - -这是一个测试部署包，验证了脚本的基本功能。 -EOF - -# 创建压缩包 -echo "[信息] 创建压缩包..." -tar -czf "$ARCHIVE_NAME" "$EXPORT_DIR" - -# 显示结果 -echo "" -echo "=== 测试完成 ===" -echo "[成功] 测试部署包已创建: $ARCHIVE_NAME" -echo "[信息] 架构: $(get_service_arch_name "$SELECTED_SERVICE_ARCH")" -echo "[信息] 端口: $SELECTED_PORTS" -echo "[信息] 平台: $TARGET_PLATFORM_NAME" -echo "[信息] 文件大小: $(du -h "$ARCHIVE_NAME" | cut -f1)" -echo "" -echo "测试成功！脚本功能正常。" From 402b4662042b9bc57654d8c6f5e3f4f51923ad4c Mon Sep 17 00:00:00 2001 From: gongwenxin Date: Wed, 27 Aug 2025 16:39:08 +0800 Subject: [PATCH 2/2] fix:fastapi --- fastapi_server.py | 638 +++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 602 insertions(+), 36 deletions(-) diff --git a/fastapi_server.py b/fastapi_server.py index 65e6063..de3ea32 100644 --- a/fastapi_server.py +++ b/fastapi_server.py @@ -62,11 +62,11 @@ app = FastAPI( 主要特性 - - 🚀 高性能: 基于FastAPI，支持异步处理 - - 📊 分页支持: 解决大量API节点的内存问题 - - 📝 自动文档: 自动生成交互式API文档 - - 🔧 灵活配置: 支持多种测试配置选项 - - 📈 详细报告: 生成PDF和JSON格式的测试报告 + 🚀 高性能: 基于FastAPI，支持异步处理 + 📊 分页支持: 解决大量API节点的内存问题 + 📝 自动文档: 自动生成交互式API文档 + 🔧 灵活配置: 支持多种测试配置选项 + 📈 详细报告: 生成PDF和JSON格式的测试报告 """, version="1.0.0", docs_url="/docs", # Swagger UI @@ -275,12 +275,17 @@ def run_tests_logic(config: dict): # Save API call details api_calls_filename = "api_call_details.md" - save_api_call_details_to_markdown( + save_api_call_details_to_file( orchestrator.get_api_call_details(), str(output_directory), filename=api_calls_filename ) + # Generate PDF report if reportlab is available + if reportlab_available and config.get('generate_pdf', True): + pdf_report_path = output_directory / "report_cn.pdf" + save_pdf_report(test_summary.to_dict(), pdf_report_path, config.get('strictness_level', 'CRITICAL')) + 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) @@ -307,47 +312,608 @@ def run_tests_logic(config: dict): "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""" +def save_api_call_details_to_file(api_call_details: List[APICallDetail], output_dir_path: str, filename: str = "api_call_details.md"): + """ + 将API调用详情列表保存到指定目录下的 Markdown 文件中。 + 同时，额外生成一个纯文本文件 (.txt)，每行包含一个 cURL 命令。 + """ + if not api_call_details: + logger.info("没有API调用详情可供保存。") + return + + output_dir = Path(output_dir_path) try: - output_path = Path(output_dir) / filename + output_dir.mkdir(parents=True, exist_ok=True) + except OSError as e: + logger.error(f"创建API调用详情输出目录 {output_dir} 失败: {e}") + return - with open(output_path, 'w', encoding='utf-8') as f: - f.write("# API调用详情\n\n") + # 主文件是 Markdown 文件 + md_output_file = output_dir / filename + # 确保它是 .md，尽管 main 函数应该已经处理了 + if md_output_file.suffix.lower() not in ['.md', '.markdown']: + md_output_file = md_output_file.with_suffix('.md') - 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") + markdown_content = [] - 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") + for detail in api_call_details: + + # Request URL with params (if any) + url_to_display = detail.request_url + if detail.request_params: + try: + # Ensure urllib is available for this formatting step + import urllib.parse + query_string = urllib.parse.urlencode(detail.request_params) + url_to_display = f"{detail.request_url}?{query_string}" + except Exception as e: + logger.warning(f"Error formatting URL with params for display: {e}") + # Fallback to just the base URL if params formatting fails + + markdown_content.append(f"## `{detail.request_method} {url_to_display}`") + markdown_content.append("**cURL Command:**") + markdown_content.append("```sh") + markdown_content.append(detail.curl_command) + markdown_content.append("```") + markdown_content.append("### Request Details") + markdown_content.append(f"- **Method:** `{detail.request_method}`") + markdown_content.append(f"- **Full URL:** `{url_to_display}`") + + markdown_content.append("- **Headers:**") + markdown_content.append("```json") + markdown_content.append(json.dumps(detail.request_headers, indent=2, ensure_ascii=False)) + markdown_content.append("```") - 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") + if detail.request_params: + markdown_content.append("- **Query Parameters:**") + markdown_content.append("```json") + markdown_content.append(json.dumps(detail.request_params, indent=2, ensure_ascii=False)) + markdown_content.append("```") - f.write(f"响应状态码: `{detail.response_status_code}`\n\n") + if detail.request_body is not None: + markdown_content.append("- **Body:**") + body_lang = "text" + formatted_body = str(detail.request_body) + try: + # Try to parse as JSON for pretty printing + if isinstance(detail.request_body, str): + try: + parsed_json = json.loads(detail.request_body) + formatted_body = json.dumps(parsed_json, indent=2, ensure_ascii=False) + body_lang = "json" + except json.JSONDecodeError: + pass # Keep as text + elif isinstance(detail.request_body, (dict, list)): + formatted_body = json.dumps(detail.request_body, indent=2, ensure_ascii=False) + body_lang = "json" + except Exception as e: + logger.warning(f"Error formatting request body for Markdown: {e}") + + markdown_content.append(f"```{body_lang}") + markdown_content.append(formatted_body) + markdown_content.append("```") - 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") + markdown_content.append("### Response Details") + markdown_content.append(f"- **Status Code:** `{detail.response_status_code}`") + markdown_content.append(f"- **Elapsed Time:** `{detail.response_elapsed_time:.4f}s`") + + markdown_content.append("- **Headers:**") + markdown_content.append("```json") + markdown_content.append(json.dumps(detail.response_headers, indent=2, ensure_ascii=False)) + markdown_content.append("```") - 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") + if detail.response_body is not None: + markdown_content.append("- **Body:**") + resp_body_lang = "text" + formatted_resp_body = str(detail.response_body) + try: + # Try to parse as JSON for pretty printing + if isinstance(detail.response_body, str): + try: + # If it's already a string that might be JSON, try parsing and re-dumping + parsed_json_resp = json.loads(detail.response_body) + formatted_resp_body = json.dumps(parsed_json_resp, indent=2, ensure_ascii=False) + resp_body_lang = "json" + except json.JSONDecodeError: + # It's a string, but not valid JSON, keep as text + pass + elif isinstance(detail.response_body, (dict, list)): + # It's already a dict/list, dump it as JSON + formatted_resp_body = json.dumps(detail.response_body, indent=2, ensure_ascii=False) + resp_body_lang = "json" + # If it's neither string nor dict/list (e.g. int, bool from parsed json), str() is fine. + except Exception as e: + logger.warning(f"Error formatting response body for Markdown: {e}") - f.write("---\n\n") - - logger.info(f"API call details saved to: {output_path}") + markdown_content.append(f"```{resp_body_lang}") + markdown_content.append(formatted_resp_body) + markdown_content.append("```") + markdown_content.append("") # Add a blank line for spacing before next --- or EOF + markdown_content.append("---") # Separator + try: + with open(md_output_file, 'w', encoding='utf-8') as f_md: + f_md.write("\n".join(markdown_content)) + logger.info(f"API调用详情已保存为 Markdown: {md_output_file}") except Exception as e: - logger.error(f"Error saving API call details: {e}") + logger.error(f"保存API调用详情到 Markdown 文件 {md_output_file} 失败: {e}", exc_info=True) + +def save_pdf_report(summary_data, output_path: Path, strictness_level: str = 'CRITICAL'): + """将测试摘要保存为格式化的PDF文件""" + logger.info(f"开始生成PDF报告: {output_path}") + output_path.parent.mkdir(parents=True, exist_ok=True) + + try: + # --- 统一的字体管理和注册 --- + font_name = 'SimSun' # 使用一个简单清晰的注册名 + font_path = 'assets/fonts/STHeiti-Medium-4.ttc' + + if not Path(font_path).exists(): + logger.error(f"字体文件未找到: {Path(font_path).resolve()}") + return + + # 关键修复: 对于 .ttc (TrueType Collection) 文件, 必须指定 subfontIndex + pdfmetrics.registerFont(TTFont(font_name, font_path, subfontIndex=0)) + # 将注册的字体关联到 'SimSun' 字体族 + pdfmetrics.registerFontFamily(font_name, normal=font_name, bold=font_name, italic=font_name, boldItalic=font_name) + + doc = SimpleDocTemplate(str(output_path), pagesize=A4, title="API测试报告") + elements = [] + + # --- 统一样式定义, 全部使用注册的字体名 --- + styles = getSampleStyleSheet() + title_style = ParagraphStyle('ChineseTitle', parent=styles['Title'], fontName=font_name, fontSize=22, leading=28) + heading_style = ParagraphStyle('ChineseHeading', parent=styles['Heading1'], fontName=font_name, fontSize=16, leading=20, spaceAfter=8) + normal_style = ParagraphStyle('ChineseNormal', parent=styles['Normal'], fontName=font_name, fontSize=10, leading=14) + small_style = ParagraphStyle('ChineseSmall', parent=styles['Normal'], fontName=font_name, fontSize=9, leading=12) + + def to_para(text, style=normal_style, escape=True): + """ + 根据用户建议移除 textwrap 以进行诊断。 + 此版本只包含净化和基本的换行符替换。 + """ + if text is None: + content = "" + else: + content = str(text) + + if escape: + content = html.escape(content) + + # 依然保留Unicode控制字符的净化 + content = "".join(ch for ch in content if unicodedata.category(ch)[0] != 'C') + + if not content.strip(): + # 对于完全空白或None的输入，返回一个安全的非换行空格 + return Paragraph(' ', style) + + # 只使用基本的换行符替换 + content = content.replace('\n', '
') + + return Paragraph(content, style) + + # 3. 填充PDF内容 - 优化后的报告格式 + + # 生成报告编码（基于时间戳） + import time + report_code = f"DMS-TEST-{int(time.time())}" + + # 报告标题 + elements.append(to_para("数据管理服务测试分析报告", title_style, escape=False)) + elements.append(Spacer(1, 15)) + + # 报告基本信息表格 + basic_info_data = [ + [to_para("报告编码", escape=False), to_para(report_code)], + [to_para("报告名称", escape=False), to_para("DMS领域数据服务测试分析报告")], + [to_para("申请日期", escape=False), to_para(datetime.datetime.now().strftime('%Y年%m月%d日'))], + [to_para("申请人", escape=False), to_para("系统管理员")], + [to_para("服务供应商名称", escape=False), to_para("数据管理系统(DMS)")], + ] + basic_info_table = Table(basic_info_data, colWidths=[120, '*']) + basic_info_table.setStyle(TableStyle([ + ('GRID', (0,0), (-1,-1), 1, colors.grey), + ('VALIGN', (0,0), (-1,-1), 'MIDDLE'), + ('BACKGROUND', (0,0), (0,-1), colors.lightgrey) + ])) + elements.append(basic_info_table) + elements.append(Spacer(1, 20)) + + # 摘要部分 + elements.append(to_para("摘要", heading_style, escape=False)) + overall = summary_data.get('overall_summary', {}) + + # 从JSON提取并格式化时间 + try: + start_time_str = summary_data.get('start_time', 'N/A') + end_time_str = summary_data.get('end_time', 'N/A') + duration = summary_data.get('duration_seconds', summary_data.get('duration', 0.0)) + + start_time_formatted = datetime.datetime.fromisoformat(start_time_str).strftime('%Y-%m-%d %H:%M:%S') if start_time_str != 'N/A' else 'N/A' + end_time_formatted = datetime.datetime.fromisoformat(end_time_str).strftime('%Y-%m-%d %H:%M:%S') if end_time_str != 'N/A' else 'N/A' + except: + start_time_formatted = start_time_str + end_time_formatted = end_time_str + + # 摘要内容 - 安全计算跳过的数量 + def safe_subtract(total, passed, failed): + """安全地计算跳过的数量""" + try: + if isinstance(total, (int, float)) and isinstance(passed, (int, float)) and isinstance(failed, (int, float)): + return max(0, total - passed - failed) + else: + return 0 + except: + return 0 + + endpoints_tested = overall.get('endpoints_tested', 0) + endpoints_passed = overall.get('endpoints_passed', 0) + endpoints_failed = overall.get('endpoints_failed', 0) + endpoints_skipped = safe_subtract(endpoints_tested, endpoints_passed, endpoints_failed) + + test_cases_executed = overall.get('total_test_cases_executed', 0) + test_cases_passed = overall.get('test_cases_passed', 0) + test_cases_failed = overall.get('test_cases_failed', 0) + test_cases_skipped = safe_subtract(test_cases_executed, test_cases_passed, test_cases_failed) + + stages_executed = overall.get('total_stages_executed', 0) + stages_passed = overall.get('stages_passed', 0) + stages_failed = overall.get('stages_failed', 0) + stages_skipped = safe_subtract(stages_executed, stages_passed, stages_failed) + + summary_text = f"""本次测试针对DMS（数据管理系统）领域数据服务进行全面的合规性验证。 +测试时间：{start_time_formatted} 至 {end_time_formatted}，总耗时 {float(duration):.2f} 秒。 +共测试 {endpoints_tested} 个API端点，其中 {endpoints_passed} 个通过，{endpoints_failed} 个失败，{endpoints_skipped} 个跳过，端点成功率为 {overall.get('endpoint_success_rate', 'N/A')}。 +执行 {test_cases_executed} 个测试用例，其中 {test_cases_passed} 个通过，{test_cases_failed} 个失败，{test_cases_skipped} 个跳过，测试用例成功率为 {overall.get('test_case_success_rate', 'N/A')}。 +执行 {stages_executed} 个流程测试，其中 {stages_passed} 个通过，{stages_failed} 个失败，{stages_skipped} 个跳过，流程测试成功率为 {overall.get('stage_success_rate', 'N/A')}。""" + + elements.append(to_para(summary_text, normal_style)) + elements.append(Spacer(1, 20)) + + # 测试内容包括 - API列表表格 + elements.append(to_para("测试内容包括", heading_style, escape=False)) + + # 从测试结果中提取API信息 + endpoint_results = summary_data.get('endpoint_results', []) + api_list_data = [ + [to_para("序号", escape=False), to_para("服务名称", escape=False), + to_para("服务功能描述", escape=False), to_para("服务参数描述", escape=False), + to_para("服务返回值描述", escape=False)] + ] + + for i, endpoint in enumerate(endpoint_results[:10], 1): # 限制显示前10个API + endpoint_name = endpoint.get('endpoint_name', 'N/A') + + # 简化的功能描述 + if 'Create' in endpoint_name: + func_desc = "提供数据创建服务" + elif 'List' in endpoint_name or 'Query' in endpoint_name: + func_desc = "提供数据查询和列表服务" + elif 'Read' in endpoint_name: + func_desc = "提供单条数据读取服务" + elif 'Update' in endpoint_name: + func_desc = "提供数据更新服务" + elif 'Delete' in endpoint_name: + func_desc = "提供数据删除服务" + else: + func_desc = "提供数据管理服务" + + api_list_data.append([ + to_para(str(i), small_style), + to_para(endpoint_name, small_style), + to_para(func_desc, small_style), + to_para("标准DMS参数格式", small_style), + to_para("标准DMS响应格式", small_style) + ]) + + api_list_table = Table(api_list_data, colWidths=[30, 80, 120, 80, 80]) + api_list_table.setStyle(TableStyle([ + ('GRID', (0,0), (-1,-1), 1, colors.grey), + ('BACKGROUND', (0,0), (-1,0), colors.lightgrey), + ('ALIGN', (0,0), (-1,-1), 'CENTER'), + ('VALIGN', (0,0), (-1,-1), 'MIDDLE'), + ('FONTSIZE', (0,0), (-1,-1), 8) + ])) + elements.append(api_list_table) + elements.append(Spacer(1, 20)) + + # 测试用例列表 - 根据严格等级分为必须和非必须 + elements.append(to_para("测试用例列表", heading_style, escape=False)) + + # 定义严重性等级的数值映射 + severity_levels = { + 'CRITICAL': 5, + 'HIGH': 4, + 'MEDIUM': 3, + 'LOW': 2, + 'INFO': 1 + } + + strictness_value = severity_levels.get(strictness_level, 5) # 默认为CRITICAL + + # 收集所有测试用例（包括endpoint用例和stage用例） + all_test_cases = [] + failed_test_cases = [] # 专门收集失败的测试用例 + + # 1. 收集endpoint测试用例 + for endpoint_result in endpoint_results: + test_cases = endpoint_result.get('executed_test_cases', []) + for tc in test_cases: + tc_severity = tc.get('test_case_severity', 'MEDIUM') + tc_severity_value = severity_levels.get(tc_severity, 3) + tc_status = tc.get('status', 'N/A') + tc_message = tc.get('message', '') + + test_case_info = { + 'type': 'Endpoint', + 'endpoint': endpoint_result.get('endpoint_name', 'N/A'), + 'endpoint_id': endpoint_result.get('endpoint_id', 'N/A'), + 'case_name': tc.get('test_case_name', 'N/A'), + 'case_id': tc.get('test_case_id', 'N/A'), + 'status': tc_status, + 'message': tc_message, + 'severity': tc_severity, + 'severity_value': tc_severity_value, + 'is_required': tc_severity_value >= strictness_value, + 'duration': tc.get('duration_seconds', 0), + 'timestamp': tc.get('timestamp', '') + } + + all_test_cases.append(test_case_info) + + # 收集失败的测试用例 + if tc_status in ['失败', 'FAILED', '错误', 'ERROR']: + failed_test_cases.append(test_case_info) + + # 2. 收集stage测试用例 + stage_results = summary_data.get('stage_results', []) + for stage_result in stage_results: + stage_name = stage_result.get('stage_name', 'N/A') + stage_status = stage_result.get('overall_status', 'N/A') + stage_message = stage_result.get('message', stage_result.get('error_message', '')) + stage_severity = 'HIGH' # Stage用例通常是高优先级 + stage_severity_value = severity_levels.get(stage_severity, 4) + + # 将stage作为一个测试用例添加 + stage_case_info = { + 'type': 'Stage', + 'endpoint': f"Stage: {stage_name}", + 'endpoint_id': f"STAGE_{stage_name}", + 'case_name': stage_result.get('description', stage_name), + 'case_id': f"STAGE_{stage_name}", + 'status': stage_status, + 'message': stage_message, + 'severity': stage_severity, + 'severity_value': stage_severity_value, + 'is_required': stage_severity_value >= strictness_value, + 'duration': stage_result.get('duration_seconds', 0), + 'timestamp': stage_result.get('start_time', '') + } + + all_test_cases.append(stage_case_info) + + # 收集失败的stage用例 + if stage_status in ['失败', 'FAILED', '错误', 'ERROR']: + failed_test_cases.append(stage_case_info) + + # 分离必须和非必须的测试用例 + required_cases = [case for case in all_test_cases if case['is_required']] + optional_cases = [case for case in all_test_cases if not case['is_required']] + + # 创建分离的测试用例表格 + if all_test_cases: + # 添加严格等级说明 + strictness_text = f"当前严格等级：{strictness_level}。根据此等级，测试用例被分为必须执行和非必须执行两部分。" + elements.append(to_para(strictness_text, small_style)) + elements.append(Spacer(1, 10)) + + # 1. 必须的测试用例表格 + if required_cases: + elements.append(to_para("必须的测试用例（影响测试结果）", heading_style, escape=False)) + + required_table_data = [ + [to_para("序号", escape=False), to_para("类型", escape=False), + to_para("测试用例名称", escape=False), to_para("所属端点/阶段", escape=False), + to_para("优先级", escape=False), to_para("执行结果", escape=False)] + ] + + for i, case in enumerate(required_cases, 1): + status_display = "通过" if case['status'] == "通过" else "失败" if case['status'] == "失败" else case['status'] + required_table_data.append([ + to_para(str(i), small_style), + to_para(case['type'], small_style), + to_para(case['case_name'], small_style), + to_para(case['endpoint'], small_style), + to_para(case['severity'], small_style), + to_para(status_display, small_style) + ]) + + required_table = Table(required_table_data, colWidths=[25, 35, 110, 90, 45, 45]) + required_table.setStyle(TableStyle([ + ('GRID', (0,0), (-1,-1), 1, colors.grey), + ('BACKGROUND', (0,0), (-1,0), colors.lightblue), # 使用浅蓝色突出必须用例 + ('ALIGN', (0,0), (-1,-1), 'CENTER'), + ('VALIGN', (0,0), (-1,-1), 'MIDDLE'), + ('FONTSIZE', (0,0), (-1,-1), 8) + ])) + elements.append(required_table) + elements.append(Spacer(1, 15)) + + # 2. 非必须的测试用例表格 + if optional_cases: + elements.append(to_para("非必须的测试用例（不影响测试结果）", heading_style, escape=False)) + + optional_table_data = [ + [to_para("序号", escape=False), to_para("类型", escape=False), + to_para("测试用例名称", escape=False), to_para("所属端点/阶段", escape=False), + to_para("优先级", escape=False), to_para("执行结果", escape=False)] + ] + + for i, case in enumerate(optional_cases, 1): + status_display = "通过" if case['status'] == "通过" else "失败" if case['status'] == "失败" else case['status'] + optional_table_data.append([ + to_para(str(i), small_style), + to_para(case['type'], small_style), + to_para(case['case_name'], small_style), + to_para(case['endpoint'], small_style), + to_para(case['severity'], small_style), + to_para(status_display, small_style) + ]) + + optional_table = Table(optional_table_data, colWidths=[25, 35, 110, 90, 45, 45]) + optional_table.setStyle(TableStyle([ + ('GRID', (0,0), (-1,-1), 1, colors.grey), + ('BACKGROUND', (0,0), (-1,0), colors.lightgrey), # 使用浅灰色表示非必须用例 + ('ALIGN', (0,0), (-1,-1), 'CENTER'), + ('VALIGN', (0,0), (-1,-1), 'MIDDLE'), + ('FONTSIZE', (0,0), (-1,-1), 8) + ])) + elements.append(optional_table) + elements.append(Spacer(1, 10)) + + # 添加用例统计信息 + total_cases = len(all_test_cases) + endpoint_cases = len([c for c in all_test_cases if c['type'] == 'Endpoint']) + stage_cases = len([c for c in all_test_cases if c['type'] == 'Stage']) + required_count = len(required_cases) + optional_count = len(optional_cases) + + stats_text = f"""测试用例统计： +总计 {total_cases} 个用例，其中端点用例 {endpoint_cases} 个，阶段用例 {stage_cases} 个。 +必须用例 {required_count} 个，非必须用例 {optional_count} 个。 +严格等级：{strictness_level}（{severity_levels.get(strictness_level, 5)}级及以上为必须）。""" + + elements.append(to_para(stats_text, small_style)) + else: + elements.append(to_para("无测试用例执行记录。", normal_style)) + + elements.append(Spacer(1, 20)) + + # 失败用例详情部分 + if failed_test_cases: + elements.append(to_para("失败用例详情分析", heading_style, escape=False)) + elements.append(Spacer(1, 10)) + + # 按严重性分组失败用例 + critical_failures = [tc for tc in failed_test_cases if tc['severity'] == 'CRITICAL'] + high_failures = [tc for tc in failed_test_cases if tc['severity'] == 'HIGH'] + medium_failures = [tc for tc in failed_test_cases if tc['severity'] == 'MEDIUM'] + low_failures = [tc for tc in failed_test_cases if tc['severity'] == 'LOW'] + + failure_summary = f"""失败用例统计： +总计 {len(failed_test_cases)} 个失败用例，其中： +• 严重级别：{len(critical_failures)} 个 +• 高级别：{len(high_failures)} 个 +• 中级别：{len(medium_failures)} 个 +• 低级别：{len(low_failures)} 个 + +以下是详细的失败原因分析：""" + + elements.append(to_para(failure_summary, normal_style)) + elements.append(Spacer(1, 15)) + + # 详细失败用例列表 + for i, failed_case in enumerate(failed_test_cases, 1): + # 用例标题 + case_title = f"{i}. {failed_case['case_name']}" + elements.append(to_para(case_title, ParagraphStyle('case_title', parent=normal_style, fontSize=11, textColor=colors.darkred, spaceAfter=5))) + + # 用例基本信息 + case_info = f"""• 用例ID：{failed_case['case_id']} +• 所属端点：{failed_case['endpoint']} +• 严重级别：{failed_case['severity']} +• 执行状态：{failed_case['status']}""" + + elements.append(to_para(case_info, ParagraphStyle('case_info', parent=small_style, leftIndent=15, spaceAfter=5))) + + # 失败原因 + failure_reason = failed_case.get('message', '无详细错误信息') + if failure_reason: + elements.append(to_para("失败原因：", ParagraphStyle('failure_label', parent=normal_style, fontSize=10, textColor=colors.darkblue, leftIndent=15))) + + # 处理长文本，确保在PDF中正确显示 + if len(failure_reason) > 200: + # 对于很长的错误信息，进行适当的分段 + failure_reason = failure_reason[:200] + "..." + + elements.append(to_para(failure_reason, ParagraphStyle('failure_reason', parent=small_style, leftIndent=30, rightIndent=20, spaceAfter=10, textColor=colors.red))) + + # 添加分隔线 + if i < len(failed_test_cases): + elements.append(HRFlowable(width="80%", thickness=0.5, color=colors.lightgrey)) + elements.append(Spacer(1, 10)) + + elements.append(Spacer(1, 20)) + + elements.append(Spacer(1, 20)) + + # 测试情况说明 + elements.append(to_para("测试情况说明", heading_style, escape=False)) + + test_situation_text = f"""本次测试是对DMS领域数据管理服务V1.0版本下的{overall.get('endpoints_tested', 'N/A')}个API进行验证测试。 +测试：累计发现缺陷{overall.get('test_cases_failed', 0)}个。 +测试执行时间：{start_time_formatted} 至 {end_time_formatted} +测试环境：开发测试环境 +测试方法：自动化API合规性测试""" + + elements.append(to_para(test_situation_text, normal_style)) + elements.append(Spacer(1, 20)) + + # 测试结论 + elements.append(to_para("测试结论", heading_style, escape=False)) + + # 根据测试结果生成结论 + success_rate = overall.get('test_case_success_rate', '0%') + success_rate_num = float(success_rate.replace('%', '')) if success_rate != 'N/A' else 0 + + if success_rate_num >= 90: + conclusion_status = "通过" + conclusion_text = f"""本套领域数据服务已通过环境验证，系统可以正常运行。验收测试通过标准关于用例执行、DMS业务流相关文档等两个方面分析，该项目通过验收测试。 +测试用例成功率达到{success_rate}，符合验收标准。""" + elif success_rate_num >= 70: + conclusion_status = "基本通过" + conclusion_text = f"""本套领域数据服务基本满足验收要求，但存在部分问题需要修复。测试用例成功率为{success_rate}，建议修复失败用例后重新测试。""" + else: + conclusion_status = "不通过" + conclusion_text = f"""本套领域数据服务未达到验收标准，存在较多问题需要修复。测试用例成功率仅为{success_rate}，需要全面检查和修复后重新测试。""" + + elements.append(to_para(conclusion_text, normal_style)) + elements.append(Spacer(1, 20)) + + # 检测依据 + elements.append(to_para("检测依据", heading_style, escape=False)) + + detection_basis_text = """集成开发应用支撑系统开放数据生态数据共享要求和评价第1部分：关于DMS领域数据服务的接口要求和测试细则。 +参考标准： +1. DMS数据管理系统API规范V1.0 +2. RESTful API设计规范 +3. 数据安全和隐私保护要求 +4. 系统集成测试标准""" + + elements.append(to_para(detection_basis_text, normal_style)) + elements.append(Spacer(1, 20)) + + # 报告生成信息 + elements.append(to_para("报告生成信息", heading_style, escape=False)) + generation_info_data = [ + [to_para("生成时间", escape=False), to_para(datetime.datetime.now().strftime('%Y年%m月%d日 %H:%M:%S'))], + [to_para("生成工具", escape=False), to_para("DMS合规性测试工具")], + [to_para("工具版本", escape=False), to_para("V1.0.0")], + [to_para("测试结论", escape=False), to_para(f"{conclusion_status}", escape=False)], + ] + generation_info_table = Table(generation_info_data, colWidths=[120, '*']) + generation_info_table.setStyle(TableStyle([ + ('GRID', (0,0), (-1,-1), 1, colors.grey), + ('VALIGN', (0,0), (-1,-1), 'MIDDLE'), + ('BACKGROUND', (0,0), (0,-1), colors.lightgrey) + ])) + elements.append(generation_info_table) + + # 构建PDF + doc.build(elements) + logger.info(f"PDF报告已成功生成: {output_path}") + except Exception as e: + logger.error(f"构建PDF文档时出错: {e}", exc_info=True) @app.post("/run", summary="执行API合规性测试",