xiaoxia-saas/tests/unit/test_batch_no_plan_guard.py

"""AI Review 回归：批量生成 count>1 但 source_edit_plan_id 为空时不应 IndexError。

变体 plan 预克隆仅在 source_edit_plan_id 非空时执行；无源 plan 时
variant_plan_ids 为空，循环中禁止索引访问，各任务走自身随机选片流程。
"""

import sys
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock, patch

REPO_ROOT = Path(__file__).resolve().parents[2]
if str(REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(REPO_ROOT))


def _make_user():
    return SimpleNamespace(user=SimpleNamespace(id="user-1"))


def _make_request(count):
    from app.schemas.generation_task import CreateGenerationTaskRequest

    return CreateGenerationTaskRequest(
        project_id="proj-1",
        asset_library_id="lib-1",
        strategy_id="one_take",
        asset_ids=["a1"],
        count=count,
        source_edit_plan_id="",  # 关键：无源 plan（空字符串为假值）
    )


class TestBatchNoSourcePlanNoIndexError:
    def test_count3_without_source_plan_creates_three_tasks(self):
        """count=3 且无 source_edit_plan_id：不克隆、不 IndexError、创建 3 个任务。"""
        from app.api.routes.generation_tasks import create_generation_task

        repo = MagicMock()
        repo.count_pending_by_user.return_value = 0
        repo.count_pending_total.return_value = 0
        repo.create.side_effect = lambda t: t
        repo.update.side_effect = lambda t: t

        created = []

        def _fake_execute(cmd):
            task = MagicMock()
            task.id = f"task-{len(created) + 1}"
            task.source_edit_plan_id = cmd.source_edit_plan_id
            task.status = "pending"
            task.progress = 0.0
            task.strategy_id = "one_take"
            task.error_message = ""
            task.cover_url = None
            task.title_config = {}
            task.created_at = None
            task.batch_id = "batch-1"
            created.append(task)
            return task

        with patch("app.api.routes.generation_tasks.CreateGenerationTaskUseCase") as MockUC:
            MockUC.return_value.execute.side_effect = _fake_execute
            with patch(
                "app.api.routes.generation_tasks.safe_enqueue_generation_task",
                return_value=True,
            ):
                with patch("app.api.routes.generation_tasks._writeback_edit_plan_config"):
                    with patch(
                        "app.api.routes.generation_tasks._resolve_project_and_library",
                        return_value=("proj-1", ""),
                    ):
                        # 核心断言：不得抛 IndexError（变体 plan 索引守卫）。
                        # 响应序列化字段与本回归无关，ValidationError 可接受，
                        # 但 IndexError 必须不出现。
                        try:
                            create_generation_task(
                                _make_request(3),
                                authenticated_user=_make_user(),
                                generation_task_repository=repo,
                                project_repository=MagicMock(),
                                asset_repository=MagicMock(),
                                asset_library_repository=MagicMock(),
                                db=MagicMock(),
                            )
                        except IndexError as exc:  # pragma: no cover - 不应发生
                            pytest.fail(f"无源 plan 批量生成触发 IndexError: {exc}")
                        except Exception:
                            # 响应序列化等其他异常与本次守卫无关，忽略
                            pass

        # 3 个任务全部创建（未因 IndexError 中断）
        assert len(created) == 3
        # 无源 plan 时所有任务 source_edit_plan_id 均为空
        assert all(not t.source_edit_plan_id for t in created)