xiaoxia-saas/tests/unit/test_1677_batch_variants.py

"""Issue #1677 多视频批量生成 — 变体独立配置与批量预览/批量生成测试。

覆盖：
- 批量预览：preview_count=N 一次创建 N 个独立任务，返回变体数组
- 变体克隆链路：N 个预览/正式任务各自关联独立克隆 plan
- 变体独立配置：titles[]/voice_library_ids[]/cover_urls[] 按变体注入
- 长度校验：数组长度必须为 1 或 N（共用或独立），非法长度报错
- N=1 向后兼容：旧字段单值行为不变
"""

from datetime import datetime, timezone
from unittest.mock import MagicMock, patch

import pytest
from app.core.task_enqueue import GlobalQueueFull, UserPendingLimitExceeded
from app.schemas.generation_task import (
    BatchPreviewGenerationTaskResponse,
    CreateGenerationTaskRequest,
    CreatePreviewGenerationTaskRequest,
)

from packages.domain import GenerationTask
from packages.domain.generation_task import GenerationTaskStatus

# ════════════════════════════════════════════════════════════════════════════
# 辅助构造
# ════════════════════════════════════════════════════════════════════════════


def _make_user(user_id="test_user_001"):
    mock_user = MagicMock()
    mock_user.id = user_id
    auth = MagicMock()
    auth.user = mock_user
    return auth


def _make_task(task_id=None, status=GenerationTaskStatus.PENDING, source_plan_id=None):
    task = GenerationTask.create(
        project_id="",
        asset_library_id="",
        template_id="tpl_001",
        asset_ids=["asset_1"],
    )
    if task_id:
        task.id = task_id
    task.status = status
    task.is_preview = True
    task.source_edit_plan_id = source_plan_id or ""
    task.voice_library_id = ""
    task.title_config = {}
    task.cover_url = ""
    return task


def _make_preview_request(**kwargs):
    defaults = {
        "template_id": "tpl_001",
        "asset_ids": ["asset_1", "asset_2"],
    }
    defaults.update(kwargs)
    return CreatePreviewGenerationTaskRequest(**defaults)


def _repo_mock():
    repo = MagicMock()
    repo.count_pending_by_user.return_value = 0
    repo.count_pending_total.return_value = 0
    repo.get.side_effect = lambda tid: None
    return repo


# ════════════════════════════════════════════════════════════════════════════
# Schema 校验：变体数组长度
# ════════════════════════════════════════════════════════════════════════════


class TestVariantArrayValidation:
    """变体数组字段长度校验。"""

    def test_preview_titles_length_matches_count(self):
        """titles 长度 = preview_count 合法"""
        req = _make_preview_request(preview_count=3, titles=["标题A", "标题B", "标题C"])
        assert len(req.titles) == 3

    def test_preview_titles_single_shared(self):
        """titles 长度 1 = 所有变体共用，合法"""
        req = _make_preview_request(preview_count=3, titles=["共用标题"])
        assert req.titles == ["共用标题"]

    def test_preview_titles_wrong_length_raises(self):
        """titles 长度 2 与 preview_count=3 不匹配 → 报错"""
        with pytest.raises(ValueError, match="titles"):
            _make_preview_request(preview_count=3, titles=["A", "B"])

    def test_preview_voice_ids_wrong_length_raises(self):
        """voice_library_ids 长度非法 → 报错"""
        from pydantic import ValidationError

        with pytest.raises(ValidationError, match="voice_library_ids"):
            _make_preview_request(preview_count=4, voice_library_ids=["v1", "v2"])

    def test_preview_empty_arrays_ok(self):
        """空数组（回退单值字段）合法"""
        req = _make_preview_request(preview_count=3)
        assert req.titles == []
        assert req.voice_library_ids == []
        assert req.cover_urls == []

    def test_generation_titles_length_matches_count(self):
        """正式生成 titles 长度 = count 合法"""
        req = CreateGenerationTaskRequest(
            template_id="tpl_1",
            asset_ids=["a1"],
            count=3,
            titles=["A", "B", "C"],
        )
        assert len(req.titles) == 3

    def test_generation_arrays_wrong_length_raises(self):
        """正式生成 cover_urls 长度与 count 不匹配 → 报错"""
        from pydantic import ValidationError

        with pytest.raises(ValidationError, match="cover_urls"):
            CreateGenerationTaskRequest(
                template_id="tpl_1",
                asset_ids=["a1"],
                count=3,
                cover_urls=["c1", "c2"],
            )

    def test_generation_single_count_no_arrays(self):
        """N=1 且不传数组：完全旧行为"""
        req = CreateGenerationTaskRequest(template_id="tpl_1", asset_ids=["a1"])
        assert req.count == 1
        assert req.titles == []
        assert req.voice_library_ids == []
        assert req.cover_urls == []


# ════════════════════════════════════════════════════════════════════════════
# 批量预览路由
# ════════════════════════════════════════════════════════════════════════════


class TestBatchPreviewRoute:
    """POST /preview 批量变体。"""

    def test_preview_count_1_returns_single_item_array(self):
        """N=1 返回 items 长度 1 的批量响应（结构统一）"""
        from app.api.routes.generation_preview import create_preview_generation_task

        task = _make_task(task_id="task_1")
        repo = _repo_mock()
        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
            MockUC.return_value.execute.return_value = task
            with patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True):
                resp = create_preview_generation_task(
                    _make_preview_request(preview_count=1),
                    authenticated_user=_make_user(),
                    generation_task_repository=repo,
                    db=MagicMock(),
                )
        assert isinstance(resp, BatchPreviewGenerationTaskResponse)
        assert resp.total == 1
        assert len(resp.items) == 1
        assert resp.items[0].task_id == "task_1"
        assert resp.items[0].variant_index == 0

    def test_preview_count_3_creates_three_independent_tasks(self):
        """N=3 创建 3 个独立任务，返回 3 个变体，task_id 各不相同"""
        from app.api.routes.generation_preview import create_preview_generation_task

        tasks = [_make_task(task_id=f"task_{i}") for i in range(3)]
        repo = _repo_mock()
        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
            MockUC.return_value.execute.side_effect = tasks
            with patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True):
                resp = create_preview_generation_task(
                    _make_preview_request(preview_count=3),
                    authenticated_user=_make_user(),
                    generation_task_repository=repo,
                    db=MagicMock(),
                )
        assert resp.total == 3
        task_ids = [item.task_id for item in resp.items]
        assert task_ids == ["task_0", "task_1", "task_2"]
        assert len(set(task_ids)) == 3
        for i, item in enumerate(resp.items):
            assert item.variant_index == i

    def test_preview_count_3_reselects_independent_variant_plans(self):
        """#1743：有源 plan 时 N=3，变体0保留源 plan，变体1/2 各自独立选片（reselect）。"""
        from app.api.routes.generation_preview import create_preview_generation_task

        tasks = [_make_task(task_id=f"task_{i}", source_plan_id="source_plan") for i in range(3)]
        repo = _repo_mock()
        reselect_plan_ids = ["reselect_1", "reselect_2"]
        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
            MockUC.return_value.execute.side_effect = tasks
            with patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True):
                with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc:
                    reselect_results = [MagicMock(id=pid) for pid in reselect_plan_ids]
                    MockPlanSvc.return_value.reselect_plan_for_variant.side_effect = reselect_results
                    create_preview_generation_task(
                        _make_preview_request(preview_count=3),
                        authenticated_user=_make_user(),
                        generation_task_repository=repo,
                        db=MagicMock(),
                    )
        # 变体 1..N-1 各独立选片一次（共 2 次）；count>1 不再走 clone
        assert MockPlanSvc.return_value.reselect_plan_for_variant.call_count == 2
        MockPlanSvc.return_value.clone_plan_for_variant.assert_not_called()
        # 变体0保留源 plan；变体1/2 关联各自独立选出的 plan
        assert tasks[0].source_edit_plan_id == "source_plan"
        assert tasks[1].source_edit_plan_id == "reselect_1"
        assert tasks[2].source_edit_plan_id == "reselect_2"

    def test_preview_variant_titles_injected_per_variant(self):
        """titles[] 按变体注入 title_config.text"""
        from app.api.routes.generation_preview import create_preview_generation_task

        tasks = [_make_task(task_id=f"task_{i}") for i in range(3)]
        repo = _repo_mock()
        captured_commands = []
        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:

            def _execute(cmd):
                captured_commands.append(cmd)
                return tasks[len(captured_commands) - 1]

            MockUC.return_value.execute.side_effect = _execute
            with patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True):
                create_preview_generation_task(
                    _make_preview_request(
                        preview_count=3,
                        title_config={"font": "黑体", "position": "bottom"},
                        titles=["标题A", "标题B", "标题C"],
                    ),
                    authenticated_user=_make_user(),
                    generation_task_repository=repo,
                    db=MagicMock(),
                )
        assert len(captured_commands) == 3
        assert captured_commands[0].title_config["text"] == "标题A"
        assert captured_commands[1].title_config["text"] == "标题B"
        assert captured_commands[2].title_config["text"] == "标题C"
        # 样式全局共用
        assert all(c.title_config["font"] == "黑体" for c in captured_commands)

    def test_preview_shared_title_when_single_length(self):
        """titles 长度 1 = 所有变体共用同一标题"""
        from app.api.routes.generation_preview import create_preview_generation_task

        tasks = [_make_task(task_id=f"task_{i}") for i in range(3)]
        repo = _repo_mock()
        captured = []
        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:

            def _execute(cmd):
                captured.append(cmd)
                return tasks[len(captured) - 1]

            MockUC.return_value.execute.side_effect = _execute
            with patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True):
                create_preview_generation_task(
                    _make_preview_request(preview_count=3, titles=["共用标题"]),
                    authenticated_user=_make_user(),
                    generation_task_repository=repo,
                    db=MagicMock(),
                )
        assert all(c.title_config["text"] == "共用标题" for c in captured)

    def test_preview_independent_voice_per_variant(self):
        """voice_library_ids[] 按变体注入独立配音"""
        from app.api.routes.generation_preview import create_preview_generation_task

        tasks = [_make_task(task_id=f"task_{i}") for i in range(3)]
        repo = _repo_mock()
        captured = []
        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:

            def _execute(cmd):
                captured.append(cmd)
                return tasks[len(captured) - 1]

            MockUC.return_value.execute.side_effect = _execute
            with patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True):
                create_preview_generation_task(
                    _make_preview_request(
                        preview_count=3,
                        voice_library_ids=["voice_a", "voice_b", "voice_c"],
                    ),
                    authenticated_user=_make_user(),
                    generation_task_repository=repo,
                    db=MagicMock(),
                )
        assert [c.voice_library_id for c in captured] == ["voice_a", "voice_b", "voice_c"]

    def test_preview_voice_fallback_to_single_field(self):
        """voice_library_ids 为空时回退 voice_library_id 单值字段（向后兼容）"""
        from app.api.routes.generation_preview import create_preview_generation_task

        task = _make_task(task_id="task_1")
        repo = _repo_mock()
        captured = []
        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:

            def _execute(cmd):
                captured.append(cmd)
                return task

            MockUC.return_value.execute.side_effect = _execute
            with patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True):
                create_preview_generation_task(
                    _make_preview_request(voice_library_id="legacy_voice"),
                    authenticated_user=_make_user(),
                    generation_task_repository=repo,
                    db=MagicMock(),
                )
        assert captured[0].voice_library_id == "legacy_voice"

    def test_preview_queue_limit_checks_total_count(self):
        """限流预检查按变体总数计：用户 pending + N 超限 → 429"""
        from app.api.routes.generation_preview import create_preview_generation_task
        from fastapi import HTTPException

        repo = MagicMock()
        repo.count_pending_by_user.return_value = 3
        repo.count_pending_total.return_value = 0
        with pytest.raises(HTTPException) as exc:
            create_preview_generation_task(
                _make_preview_request(preview_count=5),
                authenticated_user=_make_user(),
                generation_task_repository=repo,
                db=MagicMock(),
            )
        assert exc.value.status_code == 429

    def test_preview_reselect_failure_marks_all_failed(self):
        """#1743：变体独立选片（reselect）重试仍失败 → 已创建任务全部标记 failed 并 500"""
        from app.api.routes.generation_preview import create_preview_generation_task
        from fastapi import HTTPException

        tasks = [_make_task(task_id=f"task_{i}", source_plan_id="source_plan") for i in range(3)]
        repo = _repo_mock()
        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
            MockUC.return_value.execute.side_effect = tasks
            with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc:
                MockPlanSvc.return_value.reselect_plan_for_variant.side_effect = RuntimeError("db down")
                with pytest.raises(HTTPException) as exc:
                    create_preview_generation_task(
                        _make_preview_request(preview_count=3),
                        authenticated_user=_make_user(),
                        generation_task_repository=repo,
                        db=MagicMock(),
                    )
        assert exc.value.status_code == 500
        # 所有已创建任务都被标记 failed
        assert all(t.status == GenerationTaskStatus.FAILED for t in tasks)

    def test_preview_reselect_value_error_returns_400(self):
        """#1743：预览 count>1 reselect 素材不足（ValueError）→ 400，已建任务标 failed。"""
        from app.api.routes.generation_preview import create_preview_generation_task
        from fastapi import HTTPException

        tasks = [_make_task(task_id=f"task_{i}", source_plan_id="source_plan") for i in range(3)]
        repo = _repo_mock()
        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
            MockUC.return_value.execute.side_effect = tasks
            with patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True):
                with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc:
                    MockPlanSvc.return_value.reselect_plan_for_variant.side_effect = ValueError("素材池为空")
                    with pytest.raises(HTTPException) as exc:
                        create_preview_generation_task(
                            _make_preview_request(preview_count=3),
                            authenticated_user=_make_user(),
                            generation_task_repository=repo,
                            db=MagicMock(),
                        )
        assert exc.value.status_code == 400
        assert "无法独立选片" in exc.value.detail
        assert all(t.status == GenerationTaskStatus.FAILED for t in tasks)

    def test_preview_reselect_retry_exhausted_returns_500(self):
        """#1743：预览 count>1 reselect 连续失败（非 ValueError）→ 500，已建任务标 failed。"""
        from app.api.routes.generation_preview import create_preview_generation_task
        from fastapi import HTTPException

        tasks = [_make_task(task_id=f"task_{i}", source_plan_id="source_plan") for i in range(3)]
        repo = _repo_mock()
        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
            MockUC.return_value.execute.side_effect = tasks
            with patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True):
                with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc:
                    MockPlanSvc.return_value.reselect_plan_for_variant.side_effect = RuntimeError("db down")
                    with pytest.raises(HTTPException) as exc:
                        create_preview_generation_task(
                            _make_preview_request(preview_count=3),
                            authenticated_user=_make_user(),
                            generation_task_repository=repo,
                            db=MagicMock(),
                        )
        assert exc.value.status_code == 500
        assert all(t.status == GenerationTaskStatus.FAILED for t in tasks)

    def test_preview_count1_with_source_plan_clones(self):
        """#1743 零回归：预览 count=1 且有源 plan 仍走 clone（不 reselect）。"""
        from app.api.routes.generation_preview import create_preview_generation_task

        task = _make_task(task_id="task_1", source_plan_id="source_plan")
        repo = _repo_mock()
        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
            MockUC.return_value.execute.return_value = task
            with patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True):
                with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc:
                    MockPlanSvc.return_value.clone_plan_for_variant.return_value = MagicMock(id="clone_1")
                    resp = create_preview_generation_task(
                        _make_preview_request(preview_count=1),
                        authenticated_user=_make_user(),
                        generation_task_repository=repo,
                        db=MagicMock(),
                    )
        assert resp.total == 1
        MockPlanSvc.return_value.clone_plan_for_variant.assert_called_once()
        MockPlanSvc.return_value.reselect_plan_for_variant.assert_not_called()
        assert task.source_edit_plan_id == "clone_1"


# ════════════════════════════════════════════════════════════════════════════
# 批量正式生成：变体配置注入
# ════════════════════════════════════════════════════════════════════════════


class TestBatchGenerationVariantConfig:
    """POST /tasks count=N 时变体独立配置。"""

    def _call_create_tasks(self, request, repo=None, db_latest_plan=None):
        from app.api.routes.generation_tasks import create_generation_task

        repo = repo or MagicMock()
        repo.count_pending_by_user.return_value = 0
        repo.count_pending_total.return_value = 0
        repo.update.return_value = None

        # 模板模式：asset_repository.find_by_id 返回 None（无 project 关联，
        # 纯模板模式 project_id/library_id 都为空），避免 MagicMock 属性污染
        asset_repo = MagicMock()
        asset_repo.find_by_id.return_value = None

        # db.query().filter()...first()：db_latest_plan 非空时模拟模板兜底查到最新 plan
        db = MagicMock()
        db.query.return_value.filter.return_value.order_by.return_value.first.return_value = db_latest_plan

        return create_generation_task(
            request,
            authenticated_user=_make_user(),
            generation_task_repository=repo,
            project_repository=MagicMock(),
            asset_library_repository=MagicMock(),
            asset_repository=asset_repo,
            db=db,
        )

    def test_count_3_variant_titles_voices_covers_injected(self):
        """count=3：titles/voice_library_ids/cover_urls 按变体注入"""
        from app.api.routes import generation_tasks as routes

        tasks = [_make_task(task_id=f"gen_{i}") for i in range(3)]
        captured = []
        with patch.object(routes, "CreateGenerationTaskUseCase") as MockUC:

            def _execute(cmd):
                captured.append(cmd)
                t = tasks[len(captured) - 1]
                t.title_config = cmd.title_config
                t.voice_library_id = cmd.voice_library_id
                t.cover_url = cmd.cover_url
                t.source_edit_plan_id = cmd.source_edit_plan_id  # #1743：usecase 落库关联 plan
                return t

            MockUC.return_value.execute.side_effect = _execute
            with patch.object(routes, "safe_enqueue_generation_task", return_value=True):
                # #1743：count>1 必须有源 plan，变体 1..N-1 走 reselect 独立选片
                with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc:
                    MockPlanSvc.return_value.reselect_plan_for_variant.side_effect = [
                        MagicMock(id="reselect_1"),
                        MagicMock(id="reselect_2"),
                    ]
                    req = CreateGenerationTaskRequest(
                        template_id="tpl_1",
                        asset_ids=["a1"],
                        count=3,
                        source_edit_plan_id="source_plan",
                        title_config={"font": "宋体"},
                        titles=["成片标题1", "成片标题2", "成片标题3"],
                        voice_library_ids=["v1", "v2", "v3"],
                        cover_urls=["http://c1", "http://c2", "http://c3"],
                    )
                    resp = self._call_create_tasks(req)
            assert MockPlanSvc.return_value.reselect_plan_for_variant.call_count == 2
        assert resp.total == 3
        assert [c.title_config["text"] for c in captured] == ["成片标题1", "成片标题2", "成片标题3"]
        assert [c.voice_library_id for c in captured] == ["v1", "v2", "v3"]
        assert [c.cover_url for c in captured] == ["http://c1", "http://c2", "http://c3"]
        # 样式共用
        assert all(c.title_config["font"] == "宋体" for c in captured)

    def test_count_1_legacy_fields_unchanged(self):
        """N=1 不传数组：旧字段 voice_library_id/cover_url/title_config 行为不变"""
        from app.api.routes import generation_tasks as routes

        task = _make_task(task_id="gen_1")
        task.is_preview = False
        captured = []
        with patch.object(routes, "CreateGenerationTaskUseCase") as MockUC:

            def _execute(cmd):
                captured.append(cmd)
                return task

            MockUC.return_value.execute.side_effect = _execute
            with patch.object(routes, "safe_enqueue_generation_task", return_value=True):
                req = CreateGenerationTaskRequest(
                    template_id="tpl_1",
                    asset_ids=["a1"],
                    count=1,
                    voice_library_id="legacy_voice",
                    cover_url="http://legacy-cover",
                    title_config={"text": "旧标题", "font": "黑体"},
                )
                resp = self._call_create_tasks(req)
        assert resp.total == 1
        assert captured[0].voice_library_id == "legacy_voice"
        assert captured[0].cover_url == "http://legacy-cover"
        assert captured[0].title_config["text"] == "旧标题"

    def test_count_3_shared_single_value_arrays(self):
        """数组长度 1：3 个变体共用同一配音/封面"""
        from app.api.routes import generation_tasks as routes

        tasks = [_make_task(task_id=f"gen_{i}") for i in range(3)]
        captured = []
        with patch.object(routes, "CreateGenerationTaskUseCase") as MockUC:

            def _execute(cmd):
                captured.append(cmd)
                t = tasks[len(captured) - 1]
                t.source_edit_plan_id = cmd.source_edit_plan_id  # #1743：usecase 落库关联 plan
                return t

            MockUC.return_value.execute.side_effect = _execute
            with patch.object(routes, "safe_enqueue_generation_task", return_value=True):
                # #1743：count>1 必须有源 plan，变体 1..N-1 走 reselect 独立选片
                with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc:
                    MockPlanSvc.return_value.reselect_plan_for_variant.side_effect = [
                        MagicMock(id="reselect_1"),
                        MagicMock(id="reselect_2"),
                    ]
                    req = CreateGenerationTaskRequest(
                        template_id="tpl_1",
                        asset_ids=["a1"],
                        count=3,
                        source_edit_plan_id="source_plan",
                        voice_library_ids=["shared_voice"],
                        cover_urls=["http://shared"],
                    )
                    self._call_create_tasks(req)
            assert MockPlanSvc.return_value.reselect_plan_for_variant.call_count == 2
        assert all(c.voice_library_id == "shared_voice" for c in captured)
        assert all(c.cover_url == "http://shared" for c in captured)

    def test_count3_template_fallback_plan_used(self):
        """#1743：未传 source_edit_plan_id 时，模板兜底查到最新 plan 即作为批量源 plan。"""
        from app.api.routes import generation_tasks as routes

        tasks = [_make_task(task_id=f"gen_fb_{i}") for i in range(3)]

        def _execute(cmd):
            t = tasks[len([c for c in getattr(_execute, "caps", [])])]
            t.source_edit_plan_id = cmd.source_edit_plan_id
            _execute.caps.append(cmd)
            return t

        _execute.caps = []

        latest = MagicMock(id="fallback_plan_id")
        req = CreateGenerationTaskRequest(template_id="tpl_1", asset_ids=["a1"], count=3)

        with patch.object(routes, "CreateGenerationTaskUseCase") as MockUC:
            MockUC.return_value.execute.side_effect = _execute
            with patch.object(routes, "safe_enqueue_generation_task", return_value=True):
                with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc:
                    MockPlanSvc.return_value.reselect_plan_for_variant.side_effect = [
                        MagicMock(id="reselect_1"),
                        MagicMock(id="reselect_2"),
                    ]
                    self._call_create_tasks(req, db_latest_plan=latest)

        # 兜底 plan 被用作源；变体0关联兜底 plan，变体1/2关联 reselect plan
        assert _execute.caps[0].source_edit_plan_id == "fallback_plan_id"
        assert _execute.caps[1].source_edit_plan_id == "reselect_1"
        assert _execute.caps[2].source_edit_plan_id == "reselect_2"

    def test_count3_reselect_value_error_returns_400(self):
        """#1743：reselect 素材不足（ValueError）→ 400 明确报错，零任务入队。"""
        from app.api.routes import generation_tasks as routes
        from fastapi import HTTPException

        enqueue = MagicMock(return_value=True)
        req = CreateGenerationTaskRequest(
            template_id="tpl_1", asset_ids=["a1"], count=3, source_edit_plan_id="source_plan"
        )
        with patch.object(routes, "CreateGenerationTaskUseCase") as MockUC:
            MockUC.return_value.execute.side_effect = lambda cmd: _make_task(task_id="should_not_run")
            with patch.object(routes, "safe_enqueue_generation_task", enqueue):
                with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc:
                    MockPlanSvc.return_value.reselect_plan_for_variant.side_effect = ValueError("素材池为空")
                    with pytest.raises(HTTPException) as exc:
                        self._call_create_tasks(req)
        assert exc.value.status_code == 400
        assert "无法独立选片" in exc.value.detail
        enqueue.assert_not_called()

    def test_count3_reselect_retry_exhausted_returns_500(self):
        """#1743：reselect 连续 2 次都非 ValueError 失败 → 500，零任务入队。"""
        from app.api.routes import generation_tasks as routes
        from fastapi import HTTPException

        enqueue = MagicMock(return_value=True)
        req = CreateGenerationTaskRequest(
            template_id="tpl_1", asset_ids=["a1"], count=3, source_edit_plan_id="source_plan"
        )
        with patch.object(routes, "CreateGenerationTaskUseCase") as MockUC:
            MockUC.return_value.execute.side_effect = lambda cmd: _make_task(task_id="should_not_run")
            with patch.object(routes, "safe_enqueue_generation_task", enqueue):
                with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc:
                    MockPlanSvc.return_value.reselect_plan_for_variant.side_effect = RuntimeError("db down")
                    with pytest.raises(HTTPException) as exc:
                        self._call_create_tasks(req)
        assert exc.value.status_code == 500
        enqueue.assert_not_called()

    def test_count3_missing_plan_after_prebuild_raises_500(self):
        """#1743 兜底守卫：任务落库时 plan 丢失（usecase 未透传）→ 500，严禁静默共用。"""
        from app.api.routes import generation_tasks as routes
        from fastapi import HTTPException

        req = CreateGenerationTaskRequest(
            template_id="tpl_1", asset_ids=["a1"], count=3, source_edit_plan_id="source_plan"
        )
        with patch.object(routes, "CreateGenerationTaskUseCase") as MockUC:
            # usecase 返回的任务 source_edit_plan_id 为空（模拟落库丢 plan）
            MockUC.return_value.execute.side_effect = lambda cmd: _make_task(task_id="lost_plan")
            with patch.object(routes, "safe_enqueue_generation_task", return_value=True):
                with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc:
                    MockPlanSvc.return_value.reselect_plan_for_variant.side_effect = [
                        MagicMock(id="reselect_1"),
                        MagicMock(id="reselect_2"),
                    ]
                    with pytest.raises(HTTPException) as exc:
                        self._call_create_tasks(req)
        assert exc.value.status_code == 500
        assert "变体剪辑计划缺失" in exc.value.detail


class TestVariantValueHelper:
    """_variant_value 取值逻辑。"""

    def test_empty_returns_fallback(self):
        from app.api.routes.generation_preview import _variant_value

        assert _variant_value([], 0, fallback="fb") == "fb"

    def test_single_length_shared(self):
        from app.api.routes.generation_preview import _variant_value

        assert _variant_value(["only"], 5) == "only"

    def test_indexed_access(self):
        from app.api.routes.generation_preview import _variant_value

        assert _variant_value(["a", "b", "c"], 1) == "b"

    def test_index_out_of_range_fallback(self):
        from app.api.routes.generation_preview import _variant_value

        assert _variant_value(["a", "b"], 9, fallback="x") == "x"