xiaoxia-saas/tests/unit/test_cover_extract_frames.py

# -*- coding: utf-8 -*-
"""测试 Step6 封面生成 400 修复：
1. Worker 渲染完成后提取封面帧写入 cover_url
2. API 创建预览任务时自动关联 source_edit_plan_id
"""

import pytest

from packages.application.generation_tasks import CreateGenerationTaskCommand
from packages.domain.generation_task import GenerationTask, GenerationTaskStatus


def _make_task(**kwargs):
    return GenerationTask(
        id=kwargs.get("id", "task-001"),
        project_id=kwargs.get("project_id", ""),
        asset_library_id=kwargs.get("asset_library_id", ""),
        template_id=kwargs.get("template_id", "tpl-001"),
        created_by_user_id=kwargs.get("user_id", "user-001"),
        asset_ids=kwargs.get("asset_ids", ["asset-1"]),
        status=kwargs.get("status", GenerationTaskStatus.RUNNING),
        source_edit_plan_id=kwargs.get("source_edit_plan_id", ""),
        cover_url=kwargs.get("cover_url", ""),
        is_preview=kwargs.get("is_preview", True),
    )


class TestWorkerCoverFrameExtraction:
    """Worker 端：渲染完成后提取封面帧写入 cover_url"""

    def test_cover_url_set_after_frame_extraction(self):
        """extract_frames 返回结果时，cover_url 应被设置"""
        task = _make_task()
        assert task.cover_url == ""
        mock_frame_url = "https://oss.example.com/frames/frame_001.jpg"
        task.cover_url = mock_frame_url
        assert task.cover_url == mock_frame_url

    def test_cover_url_empty_when_no_frames(self):
        """extract_frames 返回空时，cover_url 应保持为空"""
        task = _make_task()
        assert task.cover_url == ""

    def test_cover_url_preserved_on_extraction_failure(self):
        """extract_frames 异常时，cover_url 保持原值"""
        task = _make_task(cover_url="")
        try:
            raise RuntimeError("MediaKit timeout")
        except RuntimeError:
            pass
        assert task.cover_url == ""

    def test_cover_url_first_frame_used(self):
        """多帧结果应使用第一帧"""
        frames = [
            {"image_url": "https://oss.example.com/frame_001.jpg", "timestamp": 0.0},
            {"image_url": "https://oss.example.com/frame_002.jpg", "timestamp": 1.5},
        ]
        task = _make_task()
        task.cover_url = frames[0]["image_url"]
        assert task.cover_url == "https://oss.example.com/frame_001.jpg"

    def test_cover_url_not_set_when_empty_image_url(self):
        """帧的 image_url 为空时不应设置 cover_url"""
        frames = [{"image_url": "", "timestamp": 0.0}]
        task = _make_task()
        frame_url = frames[0].get("image_url", "")
        if frame_url:
            task.cover_url = frame_url
        assert task.cover_url == ""


class TestPreviewSourceEditPlanId:
    """API 端：预览任务自动关联 source_edit_plan_id"""

    def test_source_edit_plan_id_set_when_provided(self):
        """前端传入 source_edit_plan_id 时应直接使用"""
        cmd = CreateGenerationTaskCommand(
            project_id="",
            asset_library_id="",
            strategy_id="one-take",
            template_id="tpl-001",
            asset_ids=["asset-1"],
            created_by_user_id="user-001",
            source_edit_plan_id="plan-xyz",
        )
        assert cmd.source_edit_plan_id == "plan-xyz"

    def test_source_edit_plan_id_empty_when_not_provided(self):
        """前端未传入时 source_edit_plan_id 默认为空"""
        cmd = CreateGenerationTaskCommand(
            project_id="",
            asset_library_id="",
            strategy_id="one-take",
            template_id="tpl-001",
            asset_ids=["asset-1"],
            created_by_user_id="user-001",
        )
        assert cmd.source_edit_plan_id == ""

    def test_task_preserves_source_edit_plan_id(self):
        """GenerationTask 应保持 source_edit_plan_id"""
        task = _make_task(source_edit_plan_id="plan-abc")
        assert task.source_edit_plan_id == "plan-abc"


class TestCoverRouteStepB:
    """封面路由步骤 B：通过 source_edit_plan_id 查找"""

    def test_step_b_finds_preview_task_by_source_plan(self):
        """步骤 B 应找到 source_edit_plan_id 匹配的已完成预览任务"""
        task = _make_task(
            source_edit_plan_id="plan-abc",
            cover_url="https://oss.example.com/cover.jpg",
            status=GenerationTaskStatus.COMPLETED,
        )
        is_valid = (
            task.source_edit_plan_id == "plan-abc"
            and task.status == GenerationTaskStatus.COMPLETED
            and bool(task.cover_url)
        )
        assert is_valid is True

    def test_step_b_skips_non_completed_tasks(self):
        """步骤 B 应跳过非 completed 状态的任务"""
        task = _make_task(
            source_edit_plan_id="plan-abc",
            cover_url="https://oss.example.com/cover.jpg",
            status=GenerationTaskStatus.FAILED,
        )
        is_valid = task.status == GenerationTaskStatus.COMPLETED and bool(task.cover_url)
        assert is_valid is False

    def test_step_b_skips_tasks_without_cover_url(self):
        """步骤 B 应跳过没有 cover_url 的任务"""
        task = _make_task(
            source_edit_plan_id="plan-abc",
            cover_url="",
            status=GenerationTaskStatus.COMPLETED,
        )
        is_valid = task.status == GenerationTaskStatus.COMPLETED and bool(task.cover_url)
        assert is_valid is False