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 "✅ 集成测试通过"

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)