.gitea/workflows/ci-cd.yml

@@ -211,6 +211,48 @@ jobs:
           pip install -q pytest-rerunfailures
           PYTHONPATH="$PWD/apps/api:$PWD" python3 -m pytest tests/integration -q --timeout=60 -x --reruns 2 --reruns-delay 1
 
+      - name: Run API performance baseline tests
+        shell: sh
+        continue-on-error: true
+        run: |
+          set +e
+          echo "=== API 性能基线测试 ==="
+          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"
+          PERF_EXIT=$?
+          
+          # 提取性能统计
+          echo ""
+          echo "=== 性能测试摘要 ==="
+          grep "PERF_STATS:" "$PERF_OUTPUT" || echo "PERF_STATS: 未找到统计数据"
+          grep "PERF_RESULT:" "$PERF_OUTPUT" || echo "PERF_RESULT: 未找到详细结果"
+          
+          # 统计通过率
+          TOTAL=$(grep -c "PERF_RESULT:" "$PERF_OUTPUT" || echo 0)
+          PASSED=$(grep "PERF_RESULT: PASS" "$PERF_OUTPUT" | wc -l)
+          FAILED=$(grep "PERF_RESULT: FAIL" "$PERF_OUTPUT" | wc -l)
+          
+          echo ""
+          echo "性能测试结果: $PASSED/$TOTAL 通过, $FAILED 未达标"
+          
+          if [ "$FAILED" -gt 0 ]; then
+            echo ""
+            echo "⚠️  警告: $FAILED 个接口性能未达标，请关注以下接口:"
+            grep "PERF_RESULT: FAIL" "$PERF_OUTPUT" | while read line; do
+              echo "   $line"
+            done
+            echo ""
+            echo "性能测试失败不阻塞主流水线，但建议尽快优化。"
+          else
+            echo "✅ 所有接口性能达标！"
+          fi
+          
+          rm -f "$PERF_OUTPUT"
+          # 始终返回 0，不阻塞流水线
+          exit 0
+
+
       - name: Cleanup PostgreSQL
         if: always()
         shell: sh

tests/e2e/api_smoke_test.sh

@@ -33,6 +33,9 @@ CLEANUP_ENABLED="${CLEANUP_ENABLED:-1}"
 CURL_TIMEOUT=30
 CURL_CONNECT_TIMEOUT=15
 CURL_INSECURE="${CURL_INSECURE:-0}"
+PERF_CHECK_ENABLED="${PERF_CHECK_ENABLED:-1}"  # 是否启用响应时间检查
+PERF_WARN_THRESHOLD_MS="${PERF_WARN_THRESHOLD_MS:-3000}"  # 响应时间警告阈值（毫秒）
+PERF_FAIL_THRESHOLD_MS="${PERF_FAIL_THRESHOLD_MS:-10000}"  # 响应时间失败阈值（毫秒）
 
 # 证书不安全的环境（如staging）可设 CURL_INSECURE=1 跳过校验
 if [ "$CURL_INSECURE" = "1" ]; then
@@ -62,6 +65,40 @@ CREATED_TEMPLATES=()
 CREATED_PROJECTS=()
 
 # ===== 工具函数 =====
+# 记录并检查响应时间
+perf_check() {
+    local name="$1"
+    local elapsed_ms="$2"
+    
+    if [ "$PERF_CHECK_ENABLED" != "1" ]; then
+        return 0
+    fi
+    
+    if [ "$elapsed_ms" -ge "$PERF_FAIL_THRESHOLD_MS" ]; then
+        fail "$name 响应时间" "${elapsed_ms}ms > ${PERF_FAIL_THRESHOLD_MS}ms（严重超标）"
+        return 1
+    elif [ "$elapsed_ms" -ge "$PERF_WARN_THRESHOLD_MS" ]; then
+        echo "⚠️  $name 响应时间: ${elapsed_ms}ms（超过警告阈值 ${PERF_WARN_THRESHOLD_MS}ms）"
+        return 0
+    fi
+    return 0
+}
+
+# 带计时的 curl 请求
+curl_timed() {
+    local output_file=$(mktemp)
+    local start_time=$(date +%s%N)
+    curl -s -o "$output_file" -w "%{http_code}" "$@"
+    local code=$?
+    local end_time=$(date +%s%N)
+    local elapsed_ms=$(( (end_time - start_time) / 1000000 ))
+    cat "$output_file"
+    rm -f "$output_file"
+    # 通过 stderr 返回耗时（调用方需重定向）
+    echo "$elapsed_ms" >&2
+    return $code
+}
+
 pass() {
     echo "✅ $1"
     PASSED=$((PASSED + 1))
@@ -199,6 +236,10 @@ setup_auth() {
 test_health() {
     should_run "health" || return 0
     section "1. 基础健康检查"
+    
+    if [ "$PERF_CHECK_ENABLED" = "1" ]; then
+        info "响应时间检查已启用: 警告=${PERF_WARN_THRESHOLD_MS}ms, 失败=${PERF_FAIL_THRESHOLD_MS}ms"
+    fi
 
     local code=$(curl -s -o /dev/null -w "%{http_code}" --max-time 10 "$BASE_URL/health")
     [ "$code" = "200" ] && pass "健康检查 /health" || fail "健康检查" "HTTP $code"

tests/integration/conftest.py

@@ -0,0 +1,293 @@
+"""
+集成测试公共 fixtures
+
+提供性能测试相关的工具、fixture 和 marker。
+"""
+
+from __future__ import annotations
+
+import os
+import time
+from contextlib import contextmanager
+from dataclasses import dataclass, field
+from typing import Callable, Dict, List, Optional
+
+import pytest
+
+# ── 性能阈值配置 ──────────────────────────────────────────────────────────
+PERF_THRESHOLDS: Dict[str, int] = {
+    "core": 500,    # 核心接口：500ms
+    "normal": 1000, # 普通接口：1000ms
+    "heavy": 3000,  # 重操作：3000ms（涉及外部调用或复杂计算）
+}
+
+# 性能测试是否跳过（通过环境变量控制）
+SKIP_PERF_TESTS = os.environ.get("SKIP_PERF_TESTS", "").lower() in ("1", "true", "yes")
+
+# 性能测试容忍度：允许一定比例的请求超标（避免CI偶发波动）
+# 默认：3次请求中允许1次超标（取中位数判断）
+PERF_SAMPLE_COUNT = int(os.environ.get("PERF_SAMPLE_COUNT", "3"))
+PERF_TOLERANCE_RATIO = float(os.environ.get("PERF_TOLERANCE_RATIO", "0.34"))
+
+
+# ── 数据类 ────────────────────────────────────────────────────────────────
+
+@dataclass
+class PerfResult:
+    """单次性能测试结果"""
+    name: str
+    threshold_ms: int
+    times_ms: List[float] = field(default_factory=list)
+    status_code: Optional[int] = None
+
+    @property
+    def median_ms(self) -> float:
+        if not self.times_ms:
+            return 0.0
+        sorted_times = sorted(self.times_ms)
+        n = len(sorted_times)
+        if n % 2 == 0:
+            return (sorted_times[n // 2 - 1] + sorted_times[n // 2]) / 2
+        return sorted_times[n // 2]
+
+    @property
+    def mean_ms(self) -> float:
+        if not self.times_ms:
+            return 0.0
+        return sum(self.times_ms) / len(self.times_ms)
+
+    @property
+    def min_ms(self) -> float:
+        return min(self.times_ms) if self.times_ms else 0.0
+
+    @property
+    def max_ms(self) -> float:
+        return max(self.times_ms) if self.times_ms else 0.0
+
+    @property
+    def passed(self) -> bool:
+        """判断是否通过：基于中位数 + 容忍比例"""
+        if not self.times_ms:
+            return False
+        # 中位数必须在阈值内
+        if self.median_ms > self.threshold_ms:
+            return False
+        # 超标比例不能超过容忍度
+        over_count = sum(1 for t in self.times_ms if t > self.threshold_ms)
+        over_ratio = over_count / len(self.times_ms)
+        return over_ratio <= PERF_TOLERANCE_RATIO
+
+
+# ── 性能断言上下文管理器 ──────────────────────────────────────────────────
+
+class PerfAssert:
+    """
+    性能断言工具。
+
+    使用方式：
+        def test_login_performance(client, perf_assert):
+            with perf_assert("core", name="login") as result:
+                response = client.post("/api/v1/auth/login", json={...})
+                result.status_code = response.status_code
+            # 退出 with 块时自动断言
+    """
+
+    def __init__(self, sample_count: int = PERF_SAMPLE_COUNT):
+        self.sample_count = sample_count
+        self.results: List[PerfResult] = []
+
+    @contextmanager
+    def __call__(self, threshold_level: str = "core", name: str = "", samples: Optional[int] = None):
+        """
+        创建一个性能测试上下文。
+
+        Args:
+            threshold_level: 阈值级别 ("core", "normal", "heavy")
+            name: 测试名称（用于输出报告）
+            samples: 采样次数，默认使用全局配置
+        """
+        if threshold_level not in PERF_THRESHOLDS:
+            raise ValueError(
+                f"未知的阈值级别: {threshold_level}，可选: {list(PERF_THRESHOLDS.keys())}"
+            )
+
+        threshold_ms = PERF_THRESHOLDS[threshold_level]
+        num_samples = samples or self.sample_count
+        result = PerfResult(name=name or threshold_level, threshold_ms=threshold_ms)
+
+        # 预热（第一次请求可能有冷启动开销）
+        yield result
+        # 第一次调用已经记录在 result.times_ms 中（由调用方通过 measure 方法）
+
+    def measure(self, threshold_level: str = "core", name: str = "",
+                samples: Optional[int] = None) -> Callable:
+        """
+        返回一个装饰器/包装器，用于测量函数执行时间。
+
+        使用方式：
+            result = perf_assert.measure("core", "login")(
+                lambda: client.post("/api/v1/auth/login", json={...})
+            )
+        """
+        def wrapper(func):
+            if threshold_level not in PERF_THRESHOLDS:
+                raise ValueError(
+                    f"未知的阈值级别: {threshold_level}，可选: {list(PERF_THRESHOLDS.keys())}"
+                )
+            threshold_ms = PERF_THRESHOLDS[threshold_level]
+            num_samples = samples or self.sample_count
+            result = PerfResult(name=name or threshold_level, threshold_ms=threshold_ms)
+
+            last_response = None
+            for i in range(num_samples):
+                start = time.perf_counter()
+                last_response = func()
+                elapsed = (time.perf_counter() - start) * 1000
+                result.times_ms.append(elapsed)
+
+            if hasattr(last_response, "status_code"):
+                result.status_code = last_response.status_code
+
+            self.results.append(result)
+            return result
+
+        return wrapper
+
+    def assert_all(self):
+        """断言所有性能测试结果都通过"""
+        failed = [r for r in self.results if not r.passed]
+        if failed:
+            lines = []
+            for r in failed:
+                lines.append(
+                    f"  ❌ {r.name}: 中位数 {r.median_ms:.1f}ms "
+                    f"(阈值 {r.threshold_ms}ms) "
+                    f"[min={r.min_ms:.1f}, max={r.max_ms:.1f}, "
+                    f"mean={r.mean_ms:.1f}, samples={len(r.times_ms)}]"
+                )
+            raise AssertionError(
+                f"性能测试失败 ({len(failed)}/{len(self.results)}):\n"
+                + "\n".join(lines)
+            )
+
+    def report(self) -> str:
+        """生成性能报告文本"""
+        lines = ["=" * 60, " 性能测试报告", "=" * 60]
+        for r in self.results:
+            status = "✅" if r.passed else "❌"
+            lines.append(
+                f" {status} {r.name:<40s} "
+                f"median={r.median_ms:>7.1f}ms / {r.threshold_ms:>5d}ms"
+            )
+            lines.append(
+                f"     min={r.min_ms:.1f}ms  max={r.max_ms:.1f}ms  "
+                f"mean={r.mean_ms:.1f}ms  samples={len(r.times_ms)}"
+                f"  status={r.status_code or 'N/A'}"
+            )
+        passed = sum(1 for r in self.results if r.passed)
+        lines.append("=" * 60)
+        lines.append(f" 总计: {passed}/{len(self.results)} 通过")
+        lines.append("=" * 60)
+        return "\n".join(lines)
+
+
+# ── pytest fixtures ──────────────────────────────────────────────────────
+
+def pytest_configure(config):
+    """注册自定义 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）"
+    )
+
+
+def pytest_collection_modifyitems(config, items):
+    """根据环境变量自动跳过性能测试"""
+    if SKIP_PERF_TESTS:
+        skip_perf = pytest.mark.skip(reason="SKIP_PERF_TESTS=1，跳过性能测试")
+        for item in items:
+            if "performance" in item.keywords or "perf_" in item.keywords:
+                item.add_marker(skip_perf)
+
+
+@pytest.fixture
+def perf_assert():
+    """
+    性能断言 fixture。
+
+    使用方式 1（推荐，自动断言）：
+        def test_login(client, perf_assert):
+            @perf_assert.measure("core", "POST /auth/login")
+            def _call():
+                return client.post("/api/v1/auth/login", json={...})
+
+            result = _call()
+            assert result.status_code == 200
+
+    使用方式 2（手动多次调用）：
+        def test_login(client, perf_assert):
+            result = perf_assert.run("core", "POST /auth/login",
+                lambda: client.post("/api/v1/auth/login", json={...})
+            )
+            assert result.status_code == 200
+    """
+    return PerfAssert()
+
+
+@pytest.fixture
+def perf_thresholds():
+    """返回性能阈值配置字典"""
+    return dict(PERF_THRESHOLDS)
+
+
+# ── 辅助函数 ──────────────────────────────────────────────────────────────
+
+def run_perf_test(
+    name: str,
+    threshold_level: str,
+    func: Callable,
+    samples: int = PERF_SAMPLE_COUNT,
+) -> PerfResult:
+    """
+    运行一次性能测试（独立函数，方便在 fixture 外部使用）。
+
+    Args:
+        name: 测试名称
+        threshold_level: 阈值级别
+        func: 要测量的函数（无参数）
+        samples: 采样次数
+
+    Returns:
+        PerfResult 对象
+    """
+    if threshold_level not in PERF_THRESHOLDS:
+        raise ValueError(
+            f"未知的阈值级别: {threshold_level}，可选: {list(PERF_THRESHOLDS.keys())}"
+        )
+
+    threshold_ms = PERF_THRESHOLDS[threshold_level]
+    result = PerfResult(name=name, threshold_ms=threshold_ms)
+
+    last_response = None
+    for i in range(samples):
+        start = time.perf_counter()
+        last_response = func()
+        elapsed = (time.perf_counter() - start) * 1000
+        result.times_ms.append(elapsed)
+
+    if hasattr(last_response, "status_code"):
+        result.status_code = last_response.status_code
+
+    return result

tests/integration/test_api_performance.py

@@ -0,0 +1,626 @@
+"""
+API 性能基线测试
+
+为核心 API 接口添加性能基线测试，确保接口响应时间在合理范围内。
+
+分类：
+- 核心接口（core, 500ms）：登录、获取当前用户、项目列表、素材列表、生成任务列表、订阅信息
+- 普通接口（normal, 1000ms）：创建项目、创建素材、模板列表、剪辑计划列表
+- 重操作接口（heavy, 3000ms）：获取上传签名、创建生成任务、去重上传
+
+运行方式：
+  pytest tests/integration/test_api_performance.py -v
+  SKIP_PERF_TESTS=1 pytest tests/integration/test_api_performance.py -v  # 跳过性能测试
+  pytest tests/integration/test_api_performance.py -m "not performance"  # 同上
+"""
+
+from __future__ import annotations
+
+import os
+import uuid
+from typing import Optional
+
+import pytest
+from fastapi.testclient import TestClient
+
+# 检测是否有可用的 PostgreSQL 数据库
+_HAS_PG = False
+try:
+    if os.environ.get("USE_IN_MEMORY_DB", "").lower() != "true":
+        import psycopg
+
+        conn = psycopg.connect(
+            os.environ.get(
+                "DATABASE_URL",
+                "postgresql+psycopg://postgres:postgres@localhost:5432/xiaoxia_saas",
+            ).replace("postgresql+psycopg://", "postgresql://"),
+            connect_timeout=3,
+        )
+        conn.close()
+        _HAS_PG = True
+except Exception:
+    pass
+
+needs_pg = pytest.mark.skipif(not _HAS_PG, reason="Requires PostgreSQL database")
+skip_perf = os.environ.get("SKIP_PERF_TESTS", "").lower() in ("1", "true", "yes")
+
+from apps.api.main import app
+
+client = TestClient(app)
+
+
+# ── 辅助函数 ──────────────────────────────────────────────────────────────
+
+
+def _register_and_login() -> tuple[str, str, str]:
+    """
+    注册新用户并登录，返回 (access_token, user_id, project_id)。
+    用于需要鉴权的性能测试准备数据。
+    """
+    unique = uuid.uuid4().hex[:8]
+    email = f"perf-{unique}@example.com"
+    username = f"perfuser-{unique}"
+
+    # 注册
+    reg_resp = client.post(
+        "/api/v1/auth/register",
+        json={
+            "email": email,
+            "password": "SecurePass123",
+            "username": username,
+            "display_name": "Perf Test User",
+        },
+    )
+    assert reg_resp.status_code in (200, 201), f"注册失败: {reg_resp.json()}"
+
+    # 登录
+    login_resp = client.post(
+        "/api/v1/auth/login",
+        json={"email": email, "password": "SecurePass123"},
+    )
+    assert login_resp.status_code == 200, f"登录失败: {login_resp.json()}"
+    data = login_resp.json()
+    token = data["access_token"]
+    user_id = data["user_id"]
+
+    # 创建一个项目（用于需要项目的接口）
+    proj_resp = client.post(
+        "/api/v1/projects",
+        headers={"Authorization": f"Bearer {token}"},
+        json={"name": f"perf-project-{unique}"},
+    )
+    assert proj_resp.status_code in (200, 201), f"创建项目失败: {proj_resp.json()}"
+    project_id = proj_resp.json()["id"]
+
+    return token, user_id, project_id
+
+
+# ── Fixtures ──────────────────────────────────────────────────────────────
+
+
+@pytest.fixture(scope="module")
+def perf_test_user():
+    """
+    模块级 fixture：为性能测试准备测试用户。
+
+    由于性能测试关注的是响应时间而非数据正确性，
+    使用同一个用户和同一份数据可以减少 setup 开销，
+    让性能测量更准确。
+    """
+    if skip_perf:
+        pytest.skip("SKIP_PERF_TESTS=1，跳过性能测试")
+    if not _HAS_PG:
+        pytest.skip("Requires PostgreSQL database")
+
+    token, user_id, project_id = _register_and_login()
+    return {
+        "token": token,
+        "user_id": user_id,
+        "project_id": project_id,
+        "headers": {"Authorization": f"Bearer {token}"},
+    }
+
+
+# ══════════════════════════════════════════════════════════════════════════
+# 核心接口性能测试（阈值 500ms）
+# ══════════════════════════════════════════════════════════════════════════
+
+
+@pytest.mark.performance
+@pytest.mark.perf_core
+@needs_pg
+class TestCoreApiPerformance:
+    """
+    核心接口性能测试 —— 阈值 500ms
+
+    这些接口是用户高频使用的功能，必须保证快速响应。
+    """
+
+    def test_login_performance(self, perf_assert):
+        """POST /auth/login 登录接口性能"""
+        # 先注册一个用户
+        unique = uuid.uuid4().hex[:8]
+        email = f"perf-login-{unique}@example.com"
+        client.post(
+            "/api/v1/auth/register",
+            json={
+                "email": email,
+                "password": "SecurePass123",
+                "username": f"perflogin-{unique}",
+                "display_name": "Perf Login Test",
+            },
+        )
+
+        result = perf_assert.measure("core", "POST /auth/login")(
+            lambda: client.post(
+                "/api/v1/auth/login",
+                json={"email": email, "password": "SecurePass123"},
+            )
+        )
+
+        assert result.status_code == 200, (
+            f"登录接口返回状态码 {result.status_code}，预期 200"
+        )
+        assert result.passed, (
+            f"登录接口性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
+            f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
+            f"mean={result.mean_ms:.1f}ms"
+        )
+
+    def test_auth_me_performance(self, perf_test_user, perf_assert):
+        """GET /auth/me 获取当前用户信息性能"""
+        headers = perf_test_user["headers"]
+
+        result = perf_assert.measure("core", "GET /auth/me")(
+            lambda: client.get("/api/v1/auth/me", headers=headers)
+        )
+
+        assert result.status_code == 200, (
+            f"获取当前用户返回状态码 {result.status_code}，预期 200"
+        )
+        assert result.passed, (
+            f"获取当前用户性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
+            f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
+            f"mean={result.mean_ms:.1f}ms"
+        )
+
+    def test_projects_list_performance(self, perf_test_user, perf_assert):
+        """GET /projects 项目列表性能"""
+        headers = perf_test_user["headers"]
+
+        result = perf_assert.measure("core", "GET /projects")(
+            lambda: client.get("/api/v1/projects", headers=headers)
+        )
+
+        assert result.status_code == 200, (
+            f"项目列表返回状态码 {result.status_code}，预期 200"
+        )
+        assert result.passed, (
+            f"项目列表性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
+            f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
+            f"mean={result.mean_ms:.1f}ms"
+        )
+
+    def test_assets_list_performance(self, perf_test_user, perf_assert):
+        """GET /assets 素材列表性能"""
+        headers = perf_test_user["headers"]
+
+        result = perf_assert.measure("core", "GET /assets")(
+            lambda: client.get("/api/v1/assets", headers=headers)
+        )
+
+        assert result.status_code == 200, (
+            f"素材列表返回状态码 {result.status_code}，预期 200"
+        )
+        assert result.passed, (
+            f"素材列表性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
+            f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
+            f"mean={result.mean_ms:.1f}ms"
+        )
+
+    def test_generation_tasks_list_performance(self, perf_test_user, perf_assert):
+        """GET /generation/tasks 生成任务列表性能"""
+        headers = perf_test_user["headers"]
+
+        result = perf_assert.measure("core", "GET /generation/tasks")(
+            lambda: client.get("/api/v1/generation/tasks", headers=headers)
+        )
+
+        assert result.status_code == 200, (
+            f"生成任务列表返回状态码 {result.status_code}，预期 200"
+        )
+        assert result.passed, (
+            f"生成任务列表性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
+            f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
+            f"mean={result.mean_ms:.1f}ms"
+        )
+
+    def test_subscription_current_performance(self, perf_test_user, perf_assert):
+        """GET /subscription/current 订阅信息性能"""
+        headers = perf_test_user["headers"]
+
+        result = perf_assert.measure("core", "GET /subscription/current")(
+            lambda: client.get("/api/v1/subscription/current", headers=headers)
+        )
+
+        assert result.status_code == 200, (
+            f"订阅信息返回状态码 {result.status_code}，预期 200"
+        )
+        assert result.passed, (
+            f"订阅信息性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
+            f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
+            f"mean={result.mean_ms:.1f}ms"
+        )
+
+
+# ══════════════════════════════════════════════════════════════════════════
+# 普通接口性能测试（阈值 1000ms）
+# ══════════════════════════════════════════════════════════════════════════
+
+
+@pytest.mark.performance
+@pytest.mark.perf_normal
+@needs_pg
+class TestNormalApiPerformance:
+    """
+    普通接口性能测试 —— 阈值 1000ms
+
+    这些接口涉及写操作或较多业务逻辑，允许稍长的响应时间。
+    """
+
+    def test_create_project_performance(self, perf_test_user, perf_assert):
+        """POST /projects 创建项目性能"""
+        headers = perf_test_user["headers"]
+        counter = 0
+
+        def _create():
+            nonlocal counter
+            counter += 1
+            return client.post(
+                "/api/v1/projects",
+                headers=headers,
+                json={"name": f"perf-create-{uuid.uuid4().hex[:8]}"},
+            )
+
+        result = perf_assert.measure("normal", "POST /projects")(_create)
+
+        assert result.status_code in (200, 201), (
+            f"创建项目返回状态码 {result.status_code}，预期 200/201"
+        )
+        assert result.passed, (
+            f"创建项目性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
+            f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
+            f"mean={result.mean_ms:.1f}ms"
+        )
+
+    def test_templates_list_performance(self, perf_test_user, perf_assert):
+        """GET /templates 模板列表性能"""
+        headers = perf_test_user["headers"]
+
+        result = perf_assert.measure("normal", "GET /templates")(
+            lambda: client.get("/api/v1/templates", headers=headers)
+        )
+
+        # 模板列表可能返回 200 或空列表，只要不是错误即可
+        assert result.status_code == 200, (
+            f"模板列表返回状态码 {result.status_code}，预期 200"
+        )
+        assert result.passed, (
+            f"模板列表性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
+            f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
+            f"mean={result.mean_ms:.1f}ms"
+        )
+
+    def test_edit_plans_list_performance(self, perf_test_user, perf_assert):
+        """GET /edit-plans 剪辑计划列表性能"""
+        headers = perf_test_user["headers"]
+
+        result = perf_assert.measure("normal", "GET /edit-plans")(
+            lambda: client.get("/api/v1/edit-plans", headers=headers)
+        )
+
+        assert result.status_code == 200, (
+            f"剪辑计划列表返回状态码 {result.status_code}，预期 200"
+        )
+        assert result.passed, (
+            f"剪辑计划列表性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
+            f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
+            f"mean={result.mean_ms:.1f}ms"
+        )
+
+
+# ══════════════════════════════════════════════════════════════════════════
+# 重操作接口性能测试（阈值 3000ms）
+# ══════════════════════════════════════════════════════════════════════════
+
+
+@pytest.mark.performance
+@pytest.mark.perf_heavy
+@needs_pg
+class TestHeavyApiPerformance:
+    """
+    重操作接口性能测试 —— 阈值 3000ms
+
+    这些接口涉及外部服务调用（如 OSS）或复杂业务逻辑，
+    允许较长的响应时间，但仍需有上限。
+    """
+
+    def test_upload_direct_prepare_performance(self, perf_test_user, perf_assert):
+        """POST /upload/direct/prepare 获取上传签名性能"""
+        headers = perf_test_user["headers"]
+        project_id = perf_test_user["project_id"]
+
+        # 获取素材库 ID
+        lib_resp = client.get("/api/v1/asset-libraries", headers=headers)
+        library_id = ""
+        if lib_resp.status_code == 200:
+            items = lib_resp.json().get("items", [])
+            if items:
+                library_id = items[0].get("id", "")
+
+        def _prepare_upload():
+            return client.post(
+                "/api/v1/upload/direct/prepare",
+                headers=headers,
+                json={
+                    "filename": f"perf-test-{uuid.uuid4().hex[:8]}.mp4",
+                    "file_size": 1024 * 1024,  # 1MB
+                    "mime_type": "video/mp4",
+                    "project_id": project_id,
+                    "library_id": library_id,
+                },
+            )
+
+        result = perf_assert.measure("heavy", "POST /upload/direct/prepare")(
+            _prepare_upload
+        )
+
+        # 上传签名接口可能因为 OSS 配置问题返回 503，这是预期的
+        # 只要不超时、不返回 500 即可
+        assert result.status_code in (200, 201, 400, 503), (
+            f"获取上传签名返回状态码 {result.status_code}，预期 200/201/400/503"
+        )
+        assert result.passed, (
+            f"获取上传签名性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
+            f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
+            f"mean={result.mean_ms:.1f}ms"
+        )
+
+    def test_create_generation_task_performance(self, perf_test_user, perf_assert):
+        """POST /generation/tasks 创建生成任务性能"""
+        headers = perf_test_user["headers"]
+        project_id = perf_test_user["project_id"]
+
+        # 获取素材库 ID
+        lib_resp = client.get("/api/v1/asset-libraries", headers=headers)
+        library_id = ""
+        if lib_resp.status_code == 200:
+            items = lib_resp.json().get("items", [])
+            if items:
+                library_id = items[0].get("id", "")
+
+        def _create_task():
+            return client.post(
+                "/api/v1/generation/tasks",
+                headers=headers,
+                json={
+                    "project_id": project_id,
+                    "asset_library_id": library_id,
+                    "template_id": "",
+                    "title_ids": [],
+                    "voice_ids": [],
+                    "asset_ids": [],
+                    "strategy_id": "",
+                },
+            )
+
+        result = perf_assert.measure("heavy", "POST /generation/tasks")(
+            _create_task
+        )
+
+        # 创建生成任务可能因为缺少素材等返回 400，这是预期的
+        # 性能测试关注响应时间，不关注业务成功与否
+        assert result.status_code in (200, 201, 400, 404), (
+            f"创建生成任务返回状态码 {result.status_code}，预期 200/201/400/404"
+        )
+        assert result.passed, (
+            f"创建生成任务性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
+            f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
+            f"mean={result.mean_ms:.1f}ms"
+        )
+
+    def test_duplication_upload_performance(self, perf_test_user, perf_assert):
+        """POST /duplication/upload 去重上传性能"""
+        headers = perf_test_user["headers"]
+
+        # 准备一个小的测试文件（模拟视频文件）
+        test_content = b"fake video content for perf test" * 100
+
+        def _upload():
+            return client.post(
+                "/api/v1/duplication/upload",
+                headers=headers,
+                files={
+                    "file": (
+                        f"perf-dup-{uuid.uuid4().hex[:8]}.mp4",
+                        test_content,
+                        "video/mp4",
+                    )
+                },
+            )
+
+        result = perf_assert.measure("heavy", "POST /duplication/upload")(
+            _upload
+        )
+
+        # 去重上传可能因为 OSS 配置问题返回 503，这是预期的
+        assert result.status_code in (200, 201, 400, 503), (
+            f"去重上传返回状态码 {result.status_code}，预期 200/201/400/503"
+        )
+        assert result.passed, (
+            f"去重上传性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
+            f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
+            f"mean={result.mean_ms:.1f}ms"
+        )
+
+
+# ══════════════════════════════════════════════════════════════════════════
+# 性能测试汇总报告
+# ══════════════════════════════════════════════════════════════════════════
+
+
+@pytest.mark.performance
+@needs_pg
+def test_performance_summary(perf_test_user, perf_assert, capsys):
+    """
+    汇总性能测试结果，输出完整报告。
+
+    这个测试会重新跑一遍所有接口的性能测试，
+    并在最后输出汇总报告，方便在 CI 中查看。
+    """
+    headers = perf_test_user["headers"]
+    project_id = perf_test_user["project_id"]
+
+    # 获取素材库 ID
+    lib_resp = client.get("/api/v1/asset-libraries", headers=headers)
+    library_id = ""
+    if lib_resp.status_code == 200:
+        items = lib_resp.json().get("items", [])
+        if items:
+            library_id = items[0].get("id", "")
+
+    # ── 核心接口 ──
+    # 登录（需要新用户）
+    unique = uuid.uuid4().hex[:8]
+    email = f"perf-summary-{unique}@example.com"
+    client.post(
+        "/api/v1/auth/register",
+        json={
+            "email": email,
+            "password": "SecurePass123",
+            "username": f"perfsummary-{unique}",
+            "display_name": "Perf Summary Test",
+        },
+    )
+
+    perf_assert.measure("core", "POST /auth/login")(
+        lambda: client.post(
+            "/api/v1/auth/login",
+            json={"email": email, "password": "SecurePass123"},
+        )
+    )
+
+    perf_assert.measure("core", "GET /auth/me")(
+        lambda: client.get("/api/v1/auth/me", headers=headers)
+    )
+
+    perf_assert.measure("core", "GET /projects")(
+        lambda: client.get("/api/v1/projects", headers=headers)
+    )
+
+    perf_assert.measure("core", "GET /assets")(
+        lambda: client.get("/api/v1/assets", headers=headers)
+    )
+
+    perf_assert.measure("core", "GET /generation/tasks")(
+        lambda: client.get("/api/v1/generation/tasks", headers=headers)
+    )
+
+    perf_assert.measure("core", "GET /subscription/current")(
+        lambda: client.get("/api/v1/subscription/current", headers=headers)
+    )
+
+    # ── 普通接口 ──
+    perf_assert.measure("normal", "POST /projects")(
+        lambda: client.post(
+            "/api/v1/projects",
+            headers=headers,
+            json={"name": f"perf-summary-{uuid.uuid4().hex[:6]}"},
+        )
+    )
+
+    perf_assert.measure("normal", "GET /templates")(
+        lambda: client.get("/api/v1/templates", headers=headers)
+    )
+
+    perf_assert.measure("normal", "GET /edit-plans")(
+        lambda: client.get("/api/v1/edit-plans", headers=headers)
+    )
+
+    # ── 重操作接口 ──
+    perf_assert.measure("heavy", "POST /upload/direct/prepare")(
+        lambda: client.post(
+            "/api/v1/upload/direct/prepare",
+            headers=headers,
+            json={
+                "filename": f"perf-summary-{uuid.uuid4().hex[:6]}.mp4",
+                "file_size": 1024 * 1024,
+                "mime_type": "video/mp4",
+                "project_id": project_id,
+                "library_id": library_id,
+            },
+        )
+    )
+
+    perf_assert.measure("heavy", "POST /generation/tasks")(
+        lambda: client.post(
+            "/api/v1/generation/tasks",
+            headers=headers,
+            json={
+                "project_id": project_id,
+                "asset_library_id": library_id,
+                "template_id": "",
+                "title_ids": [],
+                "voice_ids": [],
+                "asset_ids": [],
+                "strategy_id": "",
+            },
+        )
+    )
+
+    test_content = b"fake video for summary perf test" * 100
+    perf_assert.measure("heavy", "POST /duplication/upload")(
+        lambda: client.post(
+            "/api/v1/duplication/upload",
+            headers=headers,
+            files={
+                "file": (
+                    f"perf-sum-{uuid.uuid4().hex[:6]}.mp4",
+                    test_content,
+                    "video/mp4",
+                )
+            },
+        )
+    )
+
+    # 输出报告
+    report = perf_assert.report()
+    with capsys.disabled():
+        print("\n" + report)
+
+    # 汇总断言（警告模式：不阻塞，但输出失败信息）
+    # 在 CI 中通过 continue-on-error 控制是否阻塞
+    passed_count = sum(1 for r in perf_assert.results if r.passed)
+    total_count = len(perf_assert.results)
+
+    # 输出统计信息，方便 CI 解析
+    with capsys.disabled():
+        print(f"\nPERF_STATS: total={total_count}, passed={passed_count}, "
+              f"failed={total_count - passed_count}")
+        for r in perf_assert.results:
+            status = "PASS" if r.passed else "FAIL"
+            print(f"PERF_RESULT: {status} | {r.name} | "
+                  f"median={r.median_ms:.1f}ms | threshold={r.threshold_ms}ms | "
+                  f"min={r.min_ms:.1f}ms | max={r.max_ms:.1f}ms | "
+                  f"mean={r.mean_ms:.1f}ms | status_code={r.status_code}")
+
+    # 这里使用宽松断言：只要超过一半通过就不报错
+    # 具体的 CI 阻塞策略由 CI 配置控制（continue-on-error）
+    assert passed_count >= total_count // 2, (
+        f"性能测试通过率过低: {passed_count}/{total_count} "
+        f"({passed_count/total_count*100:.0f}%)，至少需要 50% 通过"
+    )
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v", "--tb=short"])