xiaoxia/xiaoxia-saas
1
0
Fork 0
Code Issues 11 Pull Requests 2 Actions Packages Projects Releases 7 Wiki Activity

优化：集成测试引入 pytest-xdist 并行执行 #704

Merged
auto-approve-bot merged 7 commits from feat/ci-pytest-xdist-parallel into develop 2026-07-22 15:32:30 +08:00
Conversation 5 Commits 7 Files Changed 3
+222 -22
3 changed files with 222 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
requirements-dev.txt
+1
View File
@@ -11,4 +11,5 @@ pytest==8.3.3
pytest-asyncio==0.24.0
pytest-cov==6.0.0
pytest-timeout==2.3.1
pytest-xdist==3.6.1
diff-cover==8.0.3
scripts/ci/run_integration_tests.sh
+42 -20
View File
@@ -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,35 @@ 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-xdist 并行加速（coverage 由单元测试负责，并行模式下 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 \
-p no:cacheprovider
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 +291,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: 数据库清理失败（可能已被清理）"
tests/integration/conftest.py
+179 -2
View File
@@ -2,18 +2,167 @@
集成测试公共 fixtures
提供性能测试相关的工具、fixture 和 marker。
支持 pytest-xdist 并行执行：每个 worker 使用独立数据库，数据完全隔离。
"""
from __future__ import annotations
import os
import sys
import time
from contextlib import contextmanager
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable, Dict, List, Optional
import pytest
# ── xdist 并行数据库隔离 ──────────────────────────────────────────────────
# 每个 xdist worker 进程创建独立的数据库并执行迁移，确保测试数据完全隔离
# 通过 PYTEST_XDIST_WORKER 环境变量识别 worker（如 gw0, gw1, ...）
_WORKER_DB_NAME: Optional[str] = None
def _get_worker_id() -> Optional[str]:
"""获取当前 xdist worker ID，非 worker 模式返回 None"""
return os.environ.get("PYTEST_XDIST_WORKER")
def _parse_database_url(url: str) -> Dict[str, str]:
"""
解析 DATABASE_URL，返回各组件。
支持 postgresql+psycopg://user:pass@host:port/dbname 格式
"""
from urllib.parse import urlparse
parsed = urlparse(url)
return {
"driver": parsed.scheme,
"user": parsed.username or "",
"password": parsed.password or "",
"host": parsed.hostname or "",
"port": str(parsed.port or 5432),
"dbname": parsed.path.lstrip("/") or "",
}
def _create_worker_database(worker_id: str) -> str:
"""
为 xdist worker 创建独立数据库并执行迁移。
返回新的 DATABASE_URL。
"""
base_url = os.environ.get(
"DATABASE_URL",
"postgresql+psycopg://postgres:postgres@localhost:5432/xiaoxia_saas",
)
db_info = _parse_database_url(base_url)
# 生成 worker 专属数据库名
base_db = db_info["dbname"]
worker_db = f"{base_db}_{worker_id}"
global _WORKER_DB_NAME
_WORKER_DB_NAME = worker_db
# 使用 psycopg 创建数据库（连接到 postgres 库）
try:
import psycopg
conn_str = (
f"host={db_info['host']} port={db_info['port']} "
f"user={db_info['user']} password={db_info['password']} "
f"dbname=postgres"
)
conn = psycopg.connect(conn_str, autocommit=True)
cur = conn.cursor()
# 先尝试删除（防止残留）
cur.execute(f'DROP DATABASE IF EXISTS "{worker_db}" WITH (FORCE)')
# 创建新数据库
cur.execute(f'CREATE DATABASE "{worker_db}"')
cur.close()
conn.close()
print(f"[xdist {worker_id}] ✅ 创建数据库: {worker_db}")
except ImportError:
print(f"[xdist {worker_id}] ⚠️ psycopg 未安装，跳过数据库创建")
return base_url
except Exception as e:
print(f"[xdist {worker_id}] ⚠️ 创建数据库失败: {e}")
return base_url
# 构建新的 DATABASE_URL
new_url = (
f"{db_info['driver']}://{db_info['user']}:{db_info['password']}"
f"@{db_info['host']}:{db_info['port']}/{worker_db}"
)
# 执行 alembic 迁移
print(f"[xdist {worker_id}] 🔄 执行 Alembic 迁移...")
try:
ROOT = Path(__file__).resolve().parents[2]
api_path = str(ROOT / "apps" / "api")
if api_path not in sys.path:
sys.path.insert(0, api_path)
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from alembic import command as alembic_command
from alembic.config import Config as AlembicConfig
alembic_cfg = AlembicConfig(str(ROOT / "alembic.ini"))
alembic_cfg.set_main_option("sqlalchemy.url", new_url)
# 兼容不同的脚本路径配置
alembic_cfg.set_main_option("script_location", str(ROOT / "alembic"))
# 临时设置环境变量供 alembic env.py 使用
os.environ["DATABASE_URL"] = new_url
alembic_command.upgrade(alembic_cfg, "head")
print(f"[xdist {worker_id}] ✅ 迁移完成")
except Exception as e:
print(f"[xdist {worker_id}] ❌ 迁移失败: {e}")
raise
return new_url
def _cleanup_worker_database(worker_id: str):
"""清理 xdist worker 的数据库"""
global _WORKER_DB_NAME
if not _WORKER_DB_NAME:
return
base_url = os.environ.get(
"DATABASE_URL",
"postgresql+psycopg://postgres:postgres@localhost:5432/xiaoxia_saas",
)
db_info = _parse_database_url(base_url)
try:
import psycopg
conn_str = (
f"host={db_info['host']} port={db_info['port']} "
f"user={db_info['user']} password={db_info['password']} "
f"dbname=postgres"
)
conn = psycopg.connect(conn_str, autocommit=True)
cur = conn.cursor()
# 强制断开所有连接后删除
cur.execute(
f"SELECT pg_terminate_backend(pid) FROM pg_stat_activity "
f"WHERE datname = '{_WORKER_DB_NAME}' AND pid <> pg_backend_pid()"
)
cur.execute(f'DROP DATABASE IF EXISTS "{_WORKER_DB_NAME}" WITH (FORCE)')
cur.close()
conn.close()
print(f"[xdist {worker_id}] 🧹 已清理数据库: {_WORKER_DB_NAME}")
except Exception as e:
print(f"[xdist {worker_id}] ⚠️ 清理数据库失败: {e}")
finally:
_WORKER_DB_NAME = None
# ── 性能阈值配置 ──────────────────────────────────────────────────────────
PERF_THRESHOLDS: Dict[str, int] = {
"core": 500, # 核心接口：500ms
@@ -183,16 +332,41 @@ class PerfAssert:
return "\n".join(lines)
# ── pytest fixtures ──────────────────────────────────────────────────────
# ── pytest hooks ──────────────────────────────────────────────────────────
def pytest_configure(config):
"""注册自定义 marker"""
"""
pytest 配置钩子。
- 注册自定义 marker
- xdist worker 模式下：创建独立数据库 + 执行迁移
"""
# 注册自定义 marker
config.addinivalue_line("markers", "performance: 标记为性能测试（可通过 -m 'not performance' 跳过）")
config.addinivalue_line("markers", "perf_core: 核心接口性能测试（阈值 500ms）")
config.addinivalue_line("markers", "perf_normal: 普通接口性能测试（阈值 1000ms）")
config.addinivalue_line("markers", "perf_heavy: 重操作接口性能测试（阈值 3000ms）")
# xdist worker 模式：创建独立数据库并执行迁移
worker_id = _get_worker_id()
if worker_id:
# 只有当 USE_IN_MEMORY_DB 不为 true 时才创建独立数据库
use_in_memory = os.environ.get("USE_IN_MEMORY_DB", "true").lower() == "true"
if not use_in_memory:
print(f"[xdist {worker_id}] 🚀 worker 启动，准备独立数据库...")
new_db_url = _create_worker_database(worker_id)
os.environ["DATABASE_URL"] = new_db_url
else:
print(f"[xdist {worker_id}] ℹ️ USE_IN_MEMORY_DB=true，跳过 worker 数据库创建")
def pytest_unconfigure(config):
"""pytest 结束钩子：清理 xdist worker 数据库"""
worker_id = _get_worker_id()
if worker_id and _WORKER_DB_NAME:
_cleanup_worker_database(worker_id)
def pytest_collection_modifyitems(config, items):
"""根据环境变量自动跳过性能测试"""
@@ -203,6 +377,9 @@ def pytest_collection_modifyitems(config, items):
item.add_marker(skip_perf)
# ── pytest fixtures ──────────────────────────────────────────────────────
@pytest.fixture
def perf_assert():
"""