diff --git a/.gitea/workflows/ci-cd.yml b/.gitea/workflows/ci-cd.yml index 158cf6f09..5a4a6d98d 100755 --- a/.gitea/workflows/ci-cd.yml +++ b/.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 diff --git a/tests/e2e/api_smoke_test.sh b/tests/e2e/api_smoke_test.sh index 1ccd96ef1..d2aed8952 100755 --- a/tests/e2e/api_smoke_test.sh +++ b/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" diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py new file mode 100644 index 000000000..1ecd127d3 --- /dev/null +++ b/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 diff --git a/tests/integration/test_api_performance.py b/tests/integration/test_api_performance.py new file mode 100644 index 000000000..bc5eb5620 --- /dev/null +++ b/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"])