xiaoxia-saas/tests/integration/test_api_performance.py

"""
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

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):
        # OSS 未配置时跳过此测试
        from app.config import settings

        if not settings.OSS_ACCESS_KEY_ID or not settings.OSS_ACCESS_KEY_SECRET:
            pytest.skip("OSS credentials not configured, skipping upload signature test")

        """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"])