xiaoxia-saas/tests/unit/test_unified_render_reuse.py

"""统一渲染路径 — 编辑器预览产物复用逻辑单元测试。

覆盖：
- _find_reusable_preview_task: 查找可复用的预览任务
- _get_task_output_url: 获取任务输出 URL
- 编辑器 generate 接口复用预览产物路径
"""

from __future__ import annotations

import os
import sys
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from typing import Any, Optional
from unittest.mock import MagicMock, patch

os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")

import pytest

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api"))

from packages.domain.generation_task import GenerationTask, GenerationTaskStatus

# ── Stub Repository ──────────────────────────────────────────────────────────


class StubGenTaskRepo:
    def __init__(self):
        self._store = {}

    def create(self, task):
        self._store[task.id] = task
        return task

    def get(self, task_id):
        return self._store.get(task_id)

    def update(self, task):
        self._store[task.id] = task
        return task

    def list_by_source_edit_plan(self, plan_id):
        return [t for t in self._store.values() if (t.source_edit_plan_id or "") == plan_id]


def _make_task(**kwargs):
    defaults = dict(
        id="task-001",
        project_id="proj-1",
        asset_library_id="lib-1",
        strategy_id="one_take",
        voice_library_id="",
        template_id="tmpl-1",
        asset_ids=["a1"],
        title_ids=[],
        voice_ids=[],
        status=GenerationTaskStatus.COMPLETED,
        progress=100.0,
        result_count=1,
        error_message="",
        created_by_user_id="user-1",
        source_edit_plan_id="plan-1",
        asset_select_mode="all",
        is_preview=True,
        source_task_id="",
        output_width=1920,
        output_height=1080,
        cover_url="",
        custom_title="",
        video_title="",
        resolution="",
        bgm_config={},
        completed_at=datetime.now(timezone.utc),
    )
    defaults.update(kwargs)
    return GenerationTask(**defaults)


def _make_plan(updated_at=None):
    plan = MagicMock()
    plan.id = "plan-1"
    plan.updated_at = updated_at or datetime.now(timezone.utc)
    plan.status = MagicMock()
    plan.status.value = "editing"
    plan.config = {"clips": [{"id": "c1"}, {"id": "c2"}]}
    return plan


# ── _find_reusable_preview_task ──────────────────────────────────────────────


class TestFindReusablePreviewTask:
    def test_returns_completed_preview_task(self):
        from app.api.routes.templates_editor.generation import _find_reusable_preview_task

        repo = StubGenTaskRepo()
        now = datetime.now(timezone.utc)
        task = _make_task(
            source_edit_plan_id="plan-1",
            is_preview=True,
            status=GenerationTaskStatus.COMPLETED,
            completed_at=now - timedelta(minutes=5),
        )
        repo.create(task)

        plan = _make_plan(updated_at=now - timedelta(minutes=10))
        result = _find_reusable_preview_task(repo, "plan-1", plan)

        assert result is not None
        assert result.id == "task-001"

    def test_returns_none_when_no_tasks(self):
        from app.api.routes.templates_editor.generation import _find_reusable_preview_task

        repo = StubGenTaskRepo()
        plan = _make_plan()
        result = _find_reusable_preview_task(repo, "plan-1", plan)
        assert result is None

    def test_returns_none_when_preview_not_completed(self):
        from app.api.routes.templates_editor.generation import _find_reusable_preview_task

        repo = StubGenTaskRepo()
        task = _make_task(
            source_edit_plan_id="plan-1",
            is_preview=True,
            status=GenerationTaskStatus.RUNNING,
            completed_at=None,
        )
        repo.create(task)

        plan = _make_plan()
        result = _find_reusable_preview_task(repo, "plan-1", plan)
        assert result is None

    def test_returns_none_when_plan_modified_after_preview(self):
        from app.api.routes.templates_editor.generation import _find_reusable_preview_task

        repo = StubGenTaskRepo()
        now = datetime.now(timezone.utc)
        task = _make_task(
            source_edit_plan_id="plan-1",
            is_preview=True,
            status=GenerationTaskStatus.COMPLETED,
            completed_at=now - timedelta(minutes=10),
        )
        repo.create(task)

        # Plan was updated AFTER preview completed
        plan = _make_plan(updated_at=now)
        result = _find_reusable_preview_task(repo, "plan-1", plan)
        assert result is None

    def test_skips_non_preview_tasks(self):
        from app.api.routes.templates_editor.generation import _find_reusable_preview_task

        repo = StubGenTaskRepo()
        task = _make_task(
            source_edit_plan_id="plan-1",
            is_preview=False,  # not a preview task
            status=GenerationTaskStatus.COMPLETED,
        )
        repo.create(task)

        plan = _make_plan()
        result = _find_reusable_preview_task(repo, "plan-1", plan)
        assert result is None

    def test_handles_repo_exception(self):
        from app.api.routes.templates_editor.generation import _find_reusable_preview_task

        repo = MagicMock()
        repo.list_by_source_edit_plan.side_effect = Exception("db error")
        plan = _make_plan()
        result = _find_reusable_preview_task(repo, "plan-1", plan)
        assert result is None


# ── _get_task_output_url ─────────────────────────────────────────────────────


class TestGetTaskOutputUrl:
    def test_returns_video_url(self):
        from app.api.routes.templates_editor.generation import _get_task_output_url

        task = _make_task()
        repo = MagicMock()
        db = MagicMock()

        mock_video = MagicMock()
        mock_video.file_url = "https://oss.example.com/video.mp4"

        mock_use_case = MagicMock()
        mock_use_case.execute.return_value = [mock_video]

        with patch(
            "app.api.routes.templates_editor.generation.ListGeneratedVideosByTaskUseCase",
            return_value=mock_use_case,
        ):
            result = _get_task_output_url(task, repo, db)

        assert result == "https://oss.example.com/video.mp4"

    def test_returns_empty_when_no_videos(self):
        from app.api.routes.templates_editor.generation import _get_task_output_url

        task = _make_task()
        repo = MagicMock()
        db = MagicMock()

        mock_use_case = MagicMock()
        mock_use_case.execute.return_value = []

        with patch(
            "app.api.routes.templates_editor.generation.ListGeneratedVideosByTaskUseCase",
            return_value=mock_use_case,
        ):
            result = _get_task_output_url(task, repo, db)

        assert result == ""

    def test_returns_empty_on_exception(self):
        from app.api.routes.templates_editor.generation import _get_task_output_url

        task = _make_task()
        repo = MagicMock()
        db = MagicMock()

        with patch(
            "app.api.routes.templates_editor.generation.ListGeneratedVideosByTaskUseCase",
            side_effect=Exception("db error"),
        ):
            result = _get_task_output_url(task, repo, db)

        assert result == ""


# ── mark_confirmed ───────────────────────────────────────────────────────────


class TestMarkConfirmed:
    def test_sets_is_preview_false(self):
        task = _make_task(is_preview=True)
        task.mark_confirmed()
        assert task.is_preview is False

    def test_sets_cover_url(self):
        task = _make_task()
        task.mark_confirmed(cover_url="https://example.com/cover.jpg")
        assert task.cover_url == "https://example.com/cover.jpg"

    def test_sets_custom_title(self):
        task = _make_task()
        task.mark_confirmed(custom_title="My Video")
        assert task.custom_title == "My Video"

    def test_sets_output_dimensions(self):
        task = _make_task()
        task.mark_confirmed(output_width=1080, output_height=1920)
        assert task.output_width == 1080
        assert task.output_height == 1920

    def test_zero_dimensions_not_applied(self):
        task = _make_task(output_width=1920, output_height=1080)
        task.mark_confirmed(output_width=0, output_height=0)
        assert task.output_width == 1920
        assert task.output_height == 1080

    def test_skips_when_plan_updated_at_is_none(self):
        from app.api.routes.templates_editor.generation import _find_reusable_preview_task

        repo = StubGenTaskRepo()
        now = datetime.now(timezone.utc)
        task = _make_task(
            source_edit_plan_id="plan-1",
            is_preview=True,
            status=GenerationTaskStatus.COMPLETED,
            completed_at=now - timedelta(minutes=5),
        )
        repo.create(task)

        plan = _make_plan(updated_at=None)
        result = _find_reusable_preview_task(repo, "plan-1", plan)
        assert result is None