xiaoxia-saas/tests/unit/test_finalize_generation.py

"""#2024: 视频生成 finalize 流程单测。

覆盖：
1. GenerationTask 新状态 awaiting_cover 与 mark_awaiting_cover 方法
2. finalize 用例：幂等 / 状态校验 / 正常入库
3. Worker 侧预计算函数 signature 兼容
"""

from __future__ import annotations

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

import pytest

# 使 worker 目录可导入
ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "apps" / "worker"))

from packages.domain.generation_task import (
    TERMINAL_STATUSES,
    GenerationTask,
    GenerationTaskStatus,
)

# ── 1. 状态机 ─────────────────────────────────────────────────────────


class TestAwaitingCoverStatus:
    def test_enum_value(self):
        assert GenerationTaskStatus.AWAITING_COVER == "awaiting_cover"

    def test_not_terminal(self):
        assert GenerationTaskStatus.AWAITING_COVER not in TERMINAL_STATUSES

    def test_is_awaiting_cover_property(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="lib1", asset_ids=["a1"])
        task.mark_processing()
        task.mark_awaiting_cover()
        assert task.is_awaiting_cover
        assert not task.is_completed
        assert not task.is_failed
        assert task.progress == 100.0
        # awaiting_cover 不设置 completed_at
        assert task.completed_at is None

    def test_normal_flow_pending_running_awaiting_completed(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="lib1", asset_ids=["a1"])
        task.mark_processing()
        task.mark_awaiting_cover()
        assert task.status == GenerationTaskStatus.AWAITING_COVER
        task.mark_completed(result_count=1)
        assert task.is_completed
        assert task.completed_at is not None
        assert task.result_count == 1

    def test_awaiting_to_failed_allowed(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="lib1", asset_ids=["a1"])
        task.mark_processing()
        task.mark_awaiting_cover()
        task.mark_failed("test error")
        assert task.is_failed

    def test_awaiting_to_cancelled_allowed(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="lib1", asset_ids=["a1"])
        task.mark_processing()
        task.mark_awaiting_cover()
        task.mark_cancelled()
        assert task.status == GenerationTaskStatus.CANCELLED

    def test_cannot_jump_pending_to_awaiting(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="lib1", asset_ids=["a1"])
        with pytest.raises(ValueError):
            task.mark_awaiting_cover()

    def test_mark_completed_resets_error(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="lib1", asset_ids=["a1"])
        task.mark_processing()
        task.mark_awaiting_cover()
        task.mark_completed()
        assert task.error_message == ""


class TestFinalizeUseCase:
    """finalize_generated_video 用例测试（通过 mock session 避免 DB）。"""

    def _make_task(self, extra_meta=None):
        task = GenerationTask.create(project_id="proj1", asset_library_id="lib1", asset_ids=["a1"])
        task.id = "task-123"
        task.mark_processing()
        task.mark_awaiting_cover()
        task.project_id = "proj1"
        task.created_by_user_id = "user1"
        task.extra_meta = extra_meta or {
            "rendered_output": {
                "file_url": "oss://bucket/v.mp4",
                "file_size": 1024,
                "duration": 12.5,
                "width": 1080,
                "height": 1920,
                "fps": 30.0,
                "name": "demo.mp4",
                "mode": "narrative",
                "batch_id": "",
                "is_duplicate": False,
                "fingerprint_dict": {"md5": "abc"},
            }
        }
        return task

    def test_missing_rendered_output_raises(self):
        """rendered_output.file_url 为空应抛 ValueError。"""
        from packages.application.generated_video_finalize import finalize_generated_video

        task = self._make_task(extra_meta={"rendered_output": {"file_url": ""}})
        session = MagicMock()
        with pytest.raises(ValueError):
            finalize_generated_video(
                task=task,
                session=session,
                effective_cover_url="",
            )

    def test_success_creates_generated_video(self):
        """正常 finalize 创建一条 GeneratedVideo，返回 video_id。"""
        from packages.application.generated_video_finalize import finalize_generated_video

        task = self._make_task()
        session = MagicMock()
        # mock video repo
        with patch(
            "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository"
        ) as mock_repo_cls:
            mock_repo = MagicMock()
            mock_repo_cls.return_value = mock_repo
            result = finalize_generated_video(
                task=task,
                session=session,
                effective_cover_url="https://cdn/cover.jpg",
            )
        assert result["video_id"], "video_id should be non-empty"
        assert mock_repo.create.called, "video_repo.create must be called"
        created_video = mock_repo.create.call_args[0][0]
        assert created_video.generation_task_id == "task-123"
        assert created_video.thumbnail_url == "https://cdn/cover.jpg"
        assert created_video.width == 1080
        assert created_video.height == 1920
        assert created_video.duration == 12.5
        session.commit.assert_called()

    def test_cover_fallback_to_task_cover_url(self):
        """finalize 未传 cover_url 时使用 rendered_output.thumbnail_url。"""
        from packages.application.generated_video_finalize import finalize_generated_video

        task = self._make_task()
        session = MagicMock()
        with patch(
            "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository"
        ) as mock_repo_cls:
            mock_repo = MagicMock()
            mock_repo_cls.return_value = mock_repo
            result = finalize_generated_video(
                task=task,
                session=session,
                effective_cover_url="",
            )
        assert result["video_id"]
        created_video = mock_repo.create.call_args[0][0]
        # rendered_output.thumbnail_url 为空时 thumbnail 为 None
        assert created_video.thumbnail_url is None


class TestRenderedOutputDataclass:
    def test_from_dict_defaults(self):
        from packages.application.generated_video_finalize import RenderedOutput

        ro = RenderedOutput.from_dict({"file_url": "https://x/y.mp4"})
        assert ro.file_url == "https://x/y.mp4"
        assert ro.width == 1280
        assert ro.height == 720
        assert ro.fps == 25.0
        assert ro.is_duplicate is False

    def test_from_dict_full(self):
        from packages.application.generated_video_finalize import RenderedOutput

        ro = RenderedOutput.from_dict(
            {
                "file_url": "https://x/y.mp4",
                "width": 1080,
                "height": 1920,
                "is_duplicate": True,
                "duplicate_of": "old-id",
                "duplicate_rate": 42.5,
            }
        )
        assert ro.width == 1080
        assert ro.is_duplicate is True
        assert ro.duplicate_of == "old-id"
        assert ro.duplicate_rate == 42.5


if __name__ == "__main__":
    sys.exit(pytest.main([__file__, "-v"]))