xiaoxia-saas/tests/unit/test_templates_editor_api.py

"""
templates_editor.py 模板编辑器 API 端点单元测试

覆盖核心端点（25个测试用例）：
- 草稿：GET/PUT/发布
- 片段：list/create/get/update/delete/split/merge
- BGM：GET/PUT
- 时间线：GET
- 生成状态查询
- 预设：BGM预设
"""

from __future__ import annotations

import os
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch

os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))

from fastapi import FastAPI
from fastapi.testclient import TestClient

# ---------------------------------------------------------------------------
# 测试常量
# ---------------------------------------------------------------------------

TEST_TEMPLATE_ID = "tmpl-test-001"
TEST_PLAN_ID = "plan-draft-001"
TEST_USER_ID = "user-001"


def _make_auth_user():
    """构造一个认证用户 stub（MagicMock 兼容不同属性名）"""
    auth = MagicMock()
    auth.user.id = TEST_USER_ID
    auth.user.email = "test@example.com"
    auth.user.display_name = "测试用户"
    auth.user_id = TEST_USER_ID
    return auth


def _make_mock_clip(clip_id="clip-001", order=0, duration=10.0, clip_type="video"):
    """构造一个 mock 片段"""
    clip = MagicMock()
    clip.id = clip_id
    clip.plan_id = TEST_PLAN_ID
    clip.clip_type = clip_type
    clip.order = order
    clip.duration = duration
    clip.start_time = 0.0
    clip.text_content = ""
    clip.transition_effect = "none"
    clip.playback_speed = 1.0
    clip.config = {}
    clip.asset_id = "asset-001"
    clip.status = "ready"
    return clip


def _make_mock_plan(status="editing", config=None):
    """构造一个 mock 剪辑计划"""
    plan = MagicMock()
    plan.id = TEST_PLAN_ID
    plan.status = status
    plan.config = config or {"is_template_draft": True, "asset_ids": []}
    plan.template_id = TEST_TEMPLATE_ID
    plan.project_id = "proj-001"
    plan.name = "测试草稿"
    plan.total_duration = 30.0
    plan.generation_task_id = None
    return plan


# ---------------------------------------------------------------------------
# Test App Setup
# ---------------------------------------------------------------------------


def _create_test_app():
    """创建带 mock 注入的模板编辑器测试应用"""
    from app.api.routes import templates_editor as editor_module

    mock_plan = _make_mock_plan()
    mock_clip_1 = _make_mock_clip("clip-001", 0, 10.0)
    mock_clip_2 = _make_mock_clip("clip-002", 1, 20.0)

    mock_template_svc = MagicMock()
    mock_template_svc.publish_template_from_draft.return_value = MagicMock(
        id=TEST_TEMPLATE_ID,
        name="发布后的模板",
        status="published",
    )

    mock_plan_svc = MagicMock()
    mock_plan_svc.get_plan_or_raise.return_value = mock_plan
    mock_plan_svc.update_plan.return_value = mock_plan
    mock_plan_svc.update_plan_config.return_value = mock_plan
    mock_plan_svc.list_clips.return_value = [mock_clip_1, mock_clip_2]
    mock_plan_svc.count_clips.return_value = 2
    mock_plan_svc.get_clip.return_value = mock_clip_1
    mock_plan_svc.create_clip.return_value = _make_mock_clip("clip-new", order=2)
    mock_plan_svc.update_clip.return_value = _make_mock_clip("clip-001", duration=15.0)
    mock_plan_svc.delete_clip.return_value = True
    mock_plan_svc.split_clip.return_value = {
        "left_clip": _make_mock_clip("clip-left", 0, 5.0),
        "right_clip": _make_mock_clip("clip-right", 1, 5.0),
    }
    mock_plan_svc.merge_clips.return_value = _make_mock_clip("clip-merged", 0, 20.0)
    mock_plan_svc.can_generate.return_value = (True, None)
    mock_plan_svc.mark_clips_ready.return_value = 2
    mock_plan_svc.transition_status.return_value = mock_plan

    # 生成任务 mock
    mock_gen_task = MagicMock()
    mock_gen_task.id = "task-001"
    mock_gen_task.status = "pending"
    mock_plan_svc.create_generation_task.return_value = mock_gen_task
    # 生成状态返回结构（需要 plan + clips）
    mock_plan_svc.get_generation_status.return_value = {
        "plan": mock_plan,
        "clips": [mock_clip_1, mock_clip_2],
        "task_id": "task-001",
        "generation_task_id": "task-001",
        "status": "processing",
    }
    mock_plan_svc.list_generation_tasks.return_value = {"items": [], "total": 0}

    # mock get_editor_services 依赖
    def _mock_get_editor_services():
        return mock_template_svc, mock_plan_svc

    app = FastAPI()
    app.include_router(
        editor_module.router,
        prefix="/api/v1/templates/{template_id}/editor",
    )

    # 覆盖依赖
    app.dependency_overrides[editor_module.get_current_user] = _make_auth_user
    app.dependency_overrides[editor_module.get_db_session] = lambda: MagicMock()
    app.dependency_overrides[editor_module.get_draft_plan_id] = lambda: TEST_PLAN_ID
    app.dependency_overrides[editor_module.get_editor_services] = _mock_get_editor_services

    return app, mock_template_svc, mock_plan_svc


@pytest.fixture
def client():
    app, mock_tpl_svc, mock_plan_svc = _create_test_app()
    yield TestClient(app), mock_tpl_svc, mock_plan_svc


BASE = f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor"


# ---------------------------------------------------------------------------
# 草稿端点测试
# ---------------------------------------------------------------------------


class TestDraftEndpoints:
    """草稿查询/更新/发布端点测试"""

    def test_get_draft_success(self, client):
        c, _, _ = client
        resp = c.get(BASE + "/")
        assert resp.status_code == 200
        data = resp.json()
        assert data["plan_id"] == TEST_PLAN_ID
        assert data["template_id"] == TEST_TEMPLATE_ID
        assert "name" in data
        assert "config" in data
        assert "clip_count" in data

    def test_get_draft_returns_is_template_draft(self, client):
        c, _, _ = client
        resp = c.get(BASE + "/")
        data = resp.json()
        assert data["config"]["is_template_draft"] is True

    def test_update_draft_name(self, client):
        c, _, mock_plan_svc = client
        resp = c.put(BASE + "/", json={"name": "新名称"})
        assert resp.status_code == 200
        mock_plan_svc.update_plan.assert_called_once()
        call_kwargs = mock_plan_svc.update_plan.call_args
        assert call_kwargs.kwargs.get("name") == "新名称" or call_kwargs[1].get("name") == "新名称"

    def test_update_draft_empty_body_ok(self, client):
        c, _, _ = client
        resp = c.put(BASE + "/", json={})
        assert resp.status_code == 200

    def test_publish_draft_success(self, client):
        c, mock_tpl_svc, _ = client
        resp = c.post(BASE + "/publish")
        assert resp.status_code == 200
        data = resp.json()
        assert data["status"] == "published"
        assert data["template_id"] == TEST_TEMPLATE_ID
        mock_tpl_svc.publish_template_from_draft.assert_called_once_with(TEST_TEMPLATE_ID, TEST_PLAN_ID)


# ---------------------------------------------------------------------------
# 片段端点测试
# ---------------------------------------------------------------------------


class TestClipEndpoints:
    """片段 CRUD 端点测试"""

    def test_list_clips_success(self, client):
        c, _, mock_plan_svc = client
        resp = c.get(BASE + "/clips")
        assert resp.status_code == 200
        data = resp.json()
        assert "items" in data
        assert "total" in data
        assert data["total"] == 2
        assert len(data["items"]) == 2
        mock_plan_svc.list_clips.assert_called_once()

    def test_list_clips_pagination_params(self, client):
        c, _, mock_plan_svc = client
        resp = c.get(BASE + "/clips?skip=10&limit=20")
        assert resp.status_code == 200
        mock_plan_svc.list_clips.assert_called_once_with(TEST_PLAN_ID, skip=10, limit=20)

    def test_create_clip_success(self, client):
        c, _, mock_plan_svc = client
        resp = c.post(
            BASE + "/clips",
            json={"clip_type": "video", "order": 2, "duration": 5.0},
        )
        assert resp.status_code == 201
        data = resp.json()
        assert data["id"] == "clip-new"
        mock_plan_svc.create_clip.assert_called_once()

    def test_create_clip_missing_type_422(self, client):
        c, _, _ = client
        resp = c.post(BASE + "/clips", json={"order": 0, "duration": 5.0})
        assert resp.status_code == 422

    def test_get_clip_detail_success(self, client):
        c, _, mock_plan_svc = client
        resp = c.get(BASE + "/clips/clip-001")
        assert resp.status_code == 200
        data = resp.json()
        assert data["id"] == "clip-001"
        mock_plan_svc.get_clip.assert_called_once_with("clip-001")

    def test_get_clip_not_found_404(self, client):
        c, _, mock_plan_svc = client
        mock_plan_svc.get_clip.return_value = None
        resp = c.get(BASE + "/clips/nonexistent")
        assert resp.status_code == 404

    def test_update_clip_success(self, client):
        c, _, mock_plan_svc = client
        resp = c.put(
            BASE + "/clips/clip-001",
            json={"duration": 15.0, "playback_speed": 2.0},
        )
        assert resp.status_code == 200
        mock_plan_svc.update_clip.assert_called_once()

    def test_delete_clip_success(self, client):
        c, _, mock_plan_svc = client
        resp = c.delete(BASE + "/clips/clip-001")
        assert resp.status_code == 204
        mock_plan_svc.delete_clip.assert_called_once_with("clip-001")

    def test_delete_clip_not_found_404(self, client):
        c, _, mock_plan_svc = client
        mock_plan_svc.delete_clip.return_value = False
        resp = c.delete(BASE + "/clips/nonexistent")
        assert resp.status_code == 404


# ---------------------------------------------------------------------------
# 片段分割合并测试
# ---------------------------------------------------------------------------


class TestClipSplitMerge:
    """片段分割与合并端点测试"""

    def test_split_clip_success(self, client):
        c, _, mock_plan_svc = client
        resp = c.post(BASE + "/clips/clip-001/split", json={"split_time": 5.0})
        assert resp.status_code == 200
        data = resp.json()
        assert "left_clip" in data
        assert "right_clip" in data
        mock_plan_svc.split_clip.assert_called_once_with("clip-001", 5.0)

    def test_split_clip_negative_time_422(self, client):
        c, _, _ = client
        resp = c.post(BASE + "/clips/clip-001/split", json={"split_time": -1.0})
        assert resp.status_code == 422

    def test_split_clip_not_found_404(self, client):
        c, _, mock_plan_svc = client
        mock_plan_svc.get_clip.return_value = None
        resp = c.post(BASE + "/clips/nonexistent/split", json={"split_time": 5.0})
        assert resp.status_code == 404

    def test_merge_clips_success(self, client):
        c, _, mock_plan_svc = client
        resp = c.post(BASE + "/clips/merge", json={"clip_ids": ["clip-001", "clip-002"]})
        assert resp.status_code == 200
        data = resp.json()
        assert "id" in data
        mock_plan_svc.merge_clips.assert_called_once_with(["clip-001", "clip-002"])

    def test_merge_clips_single_422(self, client):
        c, _, _ = client
        resp = c.post(BASE + "/clips/merge", json={"clip_ids": ["clip-001"]})
        assert resp.status_code == 422

    def test_merge_clips_not_found_404(self, client):
        c, _, mock_plan_svc = client
        mock_plan_svc.get_clip.return_value = None
        resp = c.post(BASE + "/clips/merge", json={"clip_ids": ["nope", "clip-002"]})
        assert resp.status_code == 404


# ---------------------------------------------------------------------------
# BGM 端点测试
# ---------------------------------------------------------------------------


class TestBGMRoutes:
    """BGM 配置端点测试"""

    def test_get_bgm_success(self, client):
        c, _, mock_plan_svc = client
        resp = c.get(BASE + "/bgm")
        assert resp.status_code == 200
        data = resp.json()
        assert "plan_id" in data
        assert "bgm" in data
        mock_plan_svc.get_plan_or_raise.assert_called()

    def test_update_bgm_success(self, client):
        c, _, mock_plan_svc = client
        resp = c.put(BASE + "/bgm", json={"enabled": True, "asset_id": "asset-001", "volume": 0.6})
        assert resp.status_code == 200
        mock_plan_svc.update_plan_config.assert_called_once()

    def test_get_bgm_presets_success(self, client):
        c, _, _ = client
        resp = c.get(BASE + "/bgm/presets")
        assert resp.status_code == 200
        data = resp.json()
        assert isinstance(data, dict)


# ---------------------------------------------------------------------------
# 时间线端点测试
# ---------------------------------------------------------------------------


class TestTimelineRoute:
    """时间线条端点测试"""

    def test_get_timeline_success(self, client):
        c, _, mock_plan_svc = client
        resp = c.get(BASE + "/timeline")
        assert resp.status_code == 200
        data = resp.json()
        assert "plan_id" in data
        assert "scenes" in data
        mock_plan_svc.list_clips.assert_called()


# ---------------------------------------------------------------------------
# 生成端点测试
# ---------------------------------------------------------------------------


class TestGenerationRoutes:
    """生成端点测试"""

    def test_generation_status_success(self, client):
        c, _, mock_plan_svc = client
        resp = c.get(BASE + "/generation-status")
        assert resp.status_code == 200
        data = resp.json()
        assert "generation_task_id" in data
        assert "clips" in data

    def test_generations_list_success(self, client):
        c, _, mock_plan_svc = client
        resp = c.get(BASE + "/generations")
        assert resp.status_code == 200
        data = resp.json()
        assert "items" in data or "tasks" in data or isinstance(data, dict)