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