diff --git a/scripts/ci/run_integration_tests.sh b/scripts/ci/run_integration_tests.sh index 1fca1158f..0e13d585f 100755 --- a/scripts/ci/run_integration_tests.sh +++ b/scripts/ci/run_integration_tests.sh @@ -252,14 +252,29 @@ echo "" echo "=== 运行集成测试(pytest-xdist 并行模式) ===" echo "CPU 核数: $(nproc 2>/dev/null || echo 'unknown')" +# 根据可用内存动态计算 worker 数,防止 OOM +# 每个 worker 约占 200-300MB(含 DB 连接 + FastAPI test client) +# 预留 1GB 给系统 + ffmpeg 等子进程 +AVAIL_MEM_MB=$(($(grep MemAvailable /proc/meminfo 2>/dev/null | awk '{print $2}' || echo 2097152) / 1024)) +RESERVED_MB=1024 +PER_WORKER_MB=256 +MAX_WORKERS=$(( (AVAIL_MEM_MB - RESERVED_MB) / PER_WORKER_MB )) +# 下限 2,上限 8,CPU 核数也作为上限 +CPU_CORES=$(nproc 2>/dev/null || echo 4) +XDIST_WORKERS=$MAX_WORKERS +[ $XDIST_WORKERS -lt 2 ] && XDIST_WORKERS=2 +[ $XDIST_WORKERS -gt 8 ] && XDIST_WORKERS=8 +[ $XDIST_WORKERS -gt $CPU_CORES ] && XDIST_WORKERS=$CPU_CORES +echo "可用内存: ${AVAIL_MEM_MB}MB, CPU核数: ${CPU_CORES}, xdist workers: ${XDIST_WORKERS}" + # 集成测试使用 pytest-xdist 并行加速(coverage 由单元测试负责,并行模式下 coverage 不稳定) -# -n auto: 自动使用 CPU 核数 +# -n N: 并行 worker 数 # --dist loadfile: 同一测试文件分配到同一 worker(共享 fixture 更高效) -# --maxfail=1: 遇到失败停止调度新测试(并行模式下等价于 -x) +# --maxfail=3: 容忍少量失败(避免偶发 OOM 导致全挂) PYTHONPATH="$PWD/apps/api:$PWD" python3 -m pytest tests/integration \ - -q --timeout=60 --maxfail=1 --reruns 2 --reruns-delay 1 \ + -q --timeout=60 --maxfail=3 --reruns 2 --reruns-delay 1 \ -m "not performance" \ - -n auto --dist loadfile \ + -n $XDIST_WORKERS --dist loadfile \ -p no:cacheprovider echo "✅ 集成测试通过" diff --git a/tests/unit/test_auth_service.py b/tests/unit/test_auth_service.py index 42d1c64bc..b0ffa5477 100755 --- a/tests/unit/test_auth_service.py +++ b/tests/unit/test_auth_service.py @@ -456,10 +456,25 @@ class TestJWTService: service.verify_token(token) def test_verify_token_tampered_signature(self, service): - """篡改签名的token无法验证""" + """篡改payload的token无法验证(签名不匹配)""" + import base64 + import json + token = service.create_access_token(user_id="user_123") - # 篡改最后一个字符 - tampered = token[:-1] + ("A" if token[-1] != "A" else "B") + parts = token.split(".") + assert len(parts) == 3 + # 篡改 payload 部分(改用户ID),会导致签名不匹配 + payload_b64 = parts[1] + # 补 padding 以便解码 + padding = 4 - len(payload_b64) % 4 + if padding != 4: + payload_b64 += "=" * padding + payload_bytes = base64.urlsafe_b64decode(payload_b64) + payload = json.loads(payload_bytes) + payload["sub"] = "hacked_user" + new_payload_bytes = json.dumps(payload).encode() + new_payload_b64 = base64.urlsafe_b64encode(new_payload_bytes).rstrip(b"=").decode() + tampered = f"{parts[0]}.{new_payload_b64}.{parts[2]}" with pytest.raises(InvalidTokenError): service.verify_token(tampered)