feat: CI集成测试引入 pytest-xdist 并行执行

2026-07-22 12:01:46 +08:00
parent 50250536a9
commit d24caefbca
1 changed files with 51 additions and 20 deletions
@@ -1,6 +1,7 @@
 #!/bin/bash
 # CI Integration Tests Job 主脚本
 # 包含：依赖安装、ffmpeg安装、Redis启动、PG启动、迁移、测试、清理、覆盖率
+# 支持 pytest-xdist 并行执行：每个 worker 使用独立数据库，预期加速 2-4 倍
 set -eu

 echo "=== CI Integration Tests 开始 ==="
@@ -28,12 +29,13 @@ for i in 1 2 3; do
  sleep 5
 done
 for i in 1 2 3; do
-  python3 -m pip install -q pytest-rerunfailures && break
-  echo "pip install pytest-rerunfailures 失败，重试 $i/3..."
+  python3 -m pip install -q pytest-rerunfailures pytest-xdist && break
+  echo "pip install pytest-rerunfailures/pytest-xdist 失败，重试 $i/3..."
  [ $i -eq 3 ] && exit 1
  sleep 5
 done
 pytest --version
+echo "pytest-xdist: $(python3 -c "import xdist; print(xdist.__version__)" 2>/dev/null || echo 'not installed')"

 # --- 安装 ffmpeg ---
 echo ""
@@ -187,8 +189,8 @@ if [ "$USE_SHARED_PG" = "true" ]; then
  echo "等待共享PG连接就绪..."
  wait_tcp_ready "$SHARED_PG_HOST" "$SHARED_PG_PORT" 5

-  # 创建独立数据库
-  echo "创建测试数据库: $CI_DB_NAME"
+  # 创建主数据库（xdist 模式下各 worker 会创建自己的数据库，主库作为 fallback）
+  echo "创建主测试数据库: $CI_DB_NAME"
  PGPASSWORD="$SHARED_PG_PASSWORD" python3 -c "
 import psycopg2
 conn = psycopg2.connect(host='$SHARED_PG_HOST', port=$SHARED_PG_PORT, user='$SHARED_PG_USER', password='$SHARED_PG_PASSWORD', dbname='postgres')
@@ -238,30 +240,44 @@ else
  echo "TCP connectivity to PostgreSQL confirmed on port $PG_PORT"
 fi

-# --- 执行迁移 ---
+# --- 执行迁移（主数据库，xdist worker 会各自创建自己的库并迁移） ---
 echo ""
-echo "=== 执行 Alembic 迁移 ==="
+echo "=== 执行 Alembic 迁移（主数据库） ==="
 PYTHONPATH="$PWD/apps/api:$PWD" python3 -m alembic upgrade head
 echo "✅ 迁移完成"

-# --- 运行集成测试 ---
+# --- 运行集成测试（pytest-xdist 并行） ---
 echo ""
-echo "=== 运行集成测试 ==="
-PYTHONPATH="$PWD/apps/api:$PWD" python3 -m coverage run \
-  --source=apps/api/app,packages \
-  --omit="*/migrations/*,*/tests/*,*/test_*.py,*/site-packages/*" \
-  --branch \
-  -m pytest tests/integration -q --timeout=60 -x --reruns 2 --reruns-delay 1 -m "not performance"
-python3 -m coverage report --show-missing
-python3 -m coverage xml -o coverage.xml
-python3 -m coverage report --fail-under=40 > /dev/null
+echo "=== 运行集成测试（pytest-xdist 并行模式） ==="
+echo "CPU 核数: $(nproc 2>/dev/null || echo 'unknown')"
+
+# 使用 pytest-cov + pytest-xdist，多进程下 coverage 自动合并
+# -n auto: 自动使用 CPU 核数
+# --dist loadfile: 同一测试文件分配到同一 worker（共享 fixture 更高效）
+# --maxfail=1: 遇到失败停止调度新测试（并行模式下等价于 -x）
+PYTHONPATH="$PWD/apps/api:$PWD" python3 -m pytest tests/integration \
+  -q --timeout=60 --maxfail=1 --reruns 2 --reruns-delay 1 \
+  -m "not performance" \
+  -n auto --dist loadfile \
+  --cov=apps/api/app \
+  --cov=packages \
+  --cov-branch \
+  --cov-omit="*/migrations/*" \
+  --cov-omit="*/tests/*" \
+  --cov-omit="*/test_*.py" \
+  --cov-omit="*/site-packages/*" \
+  --cov-report=term-missing \
+  --cov-report=xml:coverage.xml \
+  --cov-fail-under=40
+
 echo "✅ 集成测试通过"

-# --- API 性能基线测试（仅告警） ---
+# --- API 性能基线测试（仅告警，串行执行） ---
 echo ""
 echo "=== API 性能基线测试（仅告警） ==="
 set +e
 PERF_OUTPUT=$(mktemp)
+# 性能测试单独串行运行（不参与并行，避免资源竞争影响测量结果）
 PYTHONPATH="$PWD/apps/api:$PWD" python3 -m pytest tests/integration/test_api_performance.py \
  -v --timeout=120 -p no:cacheprovider 2>&1 | tee "$PERF_OUTPUT"
 echo ""
@@ -284,14 +300,29 @@ set -e
 echo ""
 echo "=== 清理 ==="
 if [ "$USE_SHARED_PG" = "true" ]; then
-  # 清理共享PG上的测试数据库
-  echo "清理共享PG测试数据库: $CI_DB_NAME"
+  # 清理共享PG上的测试数据库（主库 + 可能残留的 worker 库）
+  echo "清理共享PG测试数据库..."
+
+  # 清理所有以 CI_DB_NAME 开头的数据库（主库 + worker 库）
  PGPASSWORD="${SHARED_PG_PASSWORD}" python3 -c "
 import psycopg2
 conn = psycopg2.connect(host='${SHARED_PG_HOST}', port=${SHARED_PG_PORT}, user='${SHARED_PG_USER}', password='${SHARED_PG_PASSWORD}', dbname='postgres')
 conn.autocommit = True
 cur = conn.cursor()
-cur.execute(f'DROP DATABASE IF EXISTS \"$CI_DB_NAME\" WITH (FORCE)')
+
+# 查找所有需要清理的数据库（主库 + worker 库）
+cur.execute(\"SELECT datname FROM pg_database WHERE datname LIKE '$CI_DB_NAME%'\")
+dbs = [row[0] for row in cur.fetchall()]
+
+for db in dbs:
+    try:
+        # 强制断开所有连接
+        cur.execute(f\"SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = '{db}' AND pid <> pg_backend_pid()\")
+        cur.execute(f'DROP DATABASE IF EXISTS \"{db}\" WITH (FORCE)')
+        print(f'  已清理: {db}')
+    except Exception as e:
+        print(f'  警告: 清理 {db} 失败: {e}')
+
 cur.close()
 conn.close()
 " 2>/dev/null || echo "WARN: 数据库清理失败（可能已被清理）"