xiaoxia-saas/tests/unit/test_generated_video_creation_logic.py

"""测试：渲染完成后 generated_videos 写表判断逻辑.

验证：project_id 为空时，只要有 generation_task_id 就应该写入 generated_videos。
"""

from __future__ import annotations

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

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from packages.adapters.sqlalchemy_impl.generation_task_repository import (
    SQLAlchemyGenerationTaskRepository,
)
from packages.adapters.sqlalchemy_impl.models import Base
from packages.domain import GenerationTask, GenerationTaskStatus


def _make_task(**kwargs) -> GenerationTask:
    defaults = dict(
        project_id="proj-1",
        asset_library_id="lib-1",
        created_by_user_id="user-1",
        status=GenerationTaskStatus.PENDING,
        progress=0.0,
        source_edit_plan_id="plan-1",
    )
    defaults.update(kwargs)
    return GenerationTask(id="task-" + kwargs.get("project_id", "t1")[:8], **defaults)


class TestCleanupStaleRunning:
    """复用已有的 repo 测试基础设施，验证写表判断逻辑。"""

    def _setup(self):
        engine = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(engine)
        session = sessionmaker(bind=engine)()
        return SQLAlchemyGenerationTaskRepository(session), session

    def test_generation_task_without_project_id_can_be_created(self):
        """验证：project_id 为空的 generation_task 可以正常创建（模拟一键生成场景）。"""
        repo, session = self._setup()
        try:
            task = _make_task(
                project_id="",
                status=GenerationTaskStatus.COMPLETED,
                progress=100.0,
                result_count=1,
                source_edit_plan_id="plan-empty-proj",
            )
            task.id = "task-empty-proj"
            created = repo.create(task)
            assert created.project_id == ""

            fetched = repo.get("task-empty-proj")
            assert fetched is not None
            assert fetched.project_id == ""
            assert fetched.status.value == "completed"
            assert fetched.progress == 100.0
        finally:
            session.close()

    def test_generation_task_with_project_id_can_be_created(self):
        """验证：project_id 有值的 generation_task 正常（回归测试）。"""
        repo, session = self._setup()
        try:
            task = _make_task(
                project_id="proj-123",
                status=GenerationTaskStatus.COMPLETED,
                progress=100.0,
                result_count=1,
            )
            task.id = "task-with-proj"
            created = repo.create(task)
            assert created.project_id == "proj-123"

            fetched = repo.get("task-with-proj")
            assert fetched is not None
            assert fetched.project_id == "proj-123"
        finally:
            session.close()


class TestVideoCreationLogic:
    """验证写表判断逻辑（纯逻辑测试，不依赖 DB）。"""

    def test_should_create_video_with_generation_task_id_and_empty_project(self):
        """有 generation_task_id、无 project_id → 应该创建视频（修复点）。"""
        generation_task_id = "task-123"
        project_id = ""
        # 修复后：只要有 generation_task_id 就创建
        assert bool(generation_task_id) is True
        # 之前的错误逻辑：if generation_task_id and project_id → False
        assert bool(generation_task_id and project_id) is False
        # 修复后的逻辑：if generation_task_id → True
        assert bool(generation_task_id) is True

    def test_should_create_video_with_both_ids(self):
        """有 generation_task_id、有 project_id → 应该创建视频（回归）。"""
        generation_task_id = "task-123"
        project_id = "proj-456"
        assert bool(generation_task_id) is True
        assert bool(generation_task_id and project_id) is True

    def test_should_not_create_video_without_generation_task_id(self):
        """无 generation_task_id → 不创建视频（边界）。"""
        generation_task_id = ""
        project_id = "proj-456"
        assert bool(generation_task_id) is False


class TestVideoNameParameter:
    """验证成片库视频名称逻辑：用户设置标题时用标题，没设置时用默认命名。"""

    def test_name_from_user_title(self):
        """用户设置了标题 → 用标题作为视频名称。"""
        # 验证函数签名包含 name 参数
        import inspect

        from video_processing.dedup_helpers import create_video_record_and_dedup

        sig = inspect.signature(create_video_record_and_dedup)
        assert "name" in sig.parameters, "create_video_record_and_dedup 应支持 name 参数"

    def test_generated_video_requires_name(self):
        """GeneratedVideo.create 要求 name 非空。"""
        from packages.domain import GeneratedVideo

        with pytest.raises(ValueError, match="name cannot be empty"):
            GeneratedVideo.create(
                project_id="proj-1",
                generation_task_id="task-1",
                name="",
                file_url="https://example.com/test.mp4",
            )

    def test_name_fallback_when_empty(self):
        """name 为空或纯空格时，调用方应 fallback 到默认命名。"""
        # 模拟 _finalize_render_success 中的逻辑
        title_text_empty = ""
        title_text_spaces = "   "
        generation_task_id = "task-abcdef123456"

        # 空标题 → fallback
        video_name_1 = title_text_empty.strip() or f"generated-{generation_task_id[:8]}.mp4"
        assert video_name_1 == f"generated-{generation_task_id[:8]}.mp4"

        # 纯空格 → fallback
        video_name_2 = title_text_spaces.strip() or f"generated-{generation_task_id[:8]}.mp4"
        assert video_name_2 == f"generated-{generation_task_id[:8]}.mp4"

        # 有标题 → 用标题
        title_text = "我的旅行vlog"
        video_name_3 = title_text.strip() or f"generated-{generation_task_id[:8]}.mp4"
        assert video_name_3 == "我的旅行vlog"