xiaoxia-saas/tests/unit/test_batch_no_plan_guard.py

"""#1743 批量生成无可用 plan 守卫。

新规则（P0 降重）：count>1 批量生成时必须存在源 plan（前端传入或按模板兜底
解析到最新 plan），为每个变体独立选片；**无任何可用 plan 时直接 4xx 中断、
不创建任务**，严禁 N 个任务兜底共用同一 plan 产出同源成片。
N=1 单视频不受影响（无 plan 时走原有单任务流程）。
"""

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

import pytest
from fastapi import HTTPException

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, **overrides):
    from app.schemas.generation_task import CreateGenerationTaskRequest

    fields = dict(
        project_id="proj-1",
        asset_library_id="lib-1",
        strategy_id="one_take",
        asset_ids=["a1"],
        count=count,
        source_edit_plan_id="",
    )
    fields.update(overrides)
    return CreateGenerationTaskRequest(**fields)


def _common_patches(latest_plan=None):
    """构造通用 patch 上下文（repo/usecase/enqueue 等）。latest_plan 为模板兜底 plan 或 None。"""
    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 = ""
        task.title_config = {}
        task.created_at = None
        task.batch_id = "batch-1"
        created.append(task)
        return task

    db = MagicMock()
    # 模板兜底查最新 plan：返回 latest_plan（None 表示查不到）
    db.query.return_value.filter.return_value.order_by.return_value.first.return_value = latest_plan

    mock_uc = patch("app.api.routes.generation_tasks.CreateGenerationTaskUseCase")
    other_patches = [
        patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True),
        patch("app.api.routes.generation_tasks._writeback_edit_plan_config"),
        patch(
            "app.api.routes.generation_tasks._resolve_project_and_library",
            return_value=("proj-1", ""),
        ),
    ]
    return repo, db, created, mock_uc, other_patches, _fake_execute


class TestBatchNoSourcePlanGuard:
    def test_count3_without_any_plan_rejects_4xx_and_creates_nothing(self):
        """count=3 且无源 plan、模板兜底也查不到 → 400 中断，零任务创建（严禁同源成片）。"""
        from app.api.routes.generation_tasks import create_generation_task

        repo, db, created, mock_uc, other_patches, fake_exec = _common_patches(latest_plan=None)
        MockUC = mock_uc.start()
        MockUC.return_value.execute.side_effect = fake_exec
        for p in other_patches:
            p.start()
        all_patches = [mock_uc] + other_patches
        try:
            with pytest.raises(HTTPException) as exc_info:
                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=db,
                )
            assert exc_info.value.status_code == 400
        finally:
            for p in reversed(all_patches):
                p.stop()
        assert len(created) == 0, "无 plan 批量必须零任务创建"

    def test_count1_without_plan_still_works(self):
        """N=1 单视频无 plan 不触发批量守卫（向后兼容，不 4xx）。"""
        from app.api.routes.generation_tasks import create_generation_task

        repo, db, created, mock_uc, other_patches, fake_exec = _common_patches(latest_plan=None)
        MockUC = mock_uc.start()
        MockUC.return_value.execute.side_effect = fake_exec
        for p in other_patches:
            p.start()
        all_patches = [mock_uc] + other_patches
        try:
            create_generation_task(
                _make_request(1),
                authenticated_user=_make_user(),
                generation_task_repository=repo,
                project_repository=MagicMock(),
                asset_repository=MagicMock(),
                asset_library_repository=MagicMock(),
                db=db,
            )
        except HTTPException as e:
            assert e.status_code != 400, f"N=1 不应被批量守卫拦截: {e.detail}"
        except Exception:
            # MagicMock 任务对象下游响应序列化可能抛 ValidationError 等，与批量守卫无关；
            # 任务已在 usecase.execute 中创建，下方断言 created==1 即证明守卫未拦截。
            pass
        finally:
            for p in reversed(all_patches):
                p.stop()
        assert len(created) == 1