diff --git a/requirements-dev.txt b/requirements-dev.txt index 041915523..06de6ea0a 100755 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -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 diff --git a/scripts/ci/run_integration_tests.sh b/scripts/ci/run_integration_tests.sh index 73f0e6285..9f7975478 100755 --- a/scripts/ci/run_integration_tests.sh +++ b/scripts/ci/run_integration_tests.sh @@ -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: 数据库清理失败(可能已被清理)" diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 71db53f6b..8dbac13d7 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -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(): """