diff --git a/tests/unit/test_unified_render_reuse.py b/tests/unit/test_unified_render_reuse.py new file mode 100644 index 000000000..3d6bd1aee --- /dev/null +++ b/tests/unit/test_unified_render_reuse.py @@ -0,0 +1,261 @@ +"""统一渲染路径 — 编辑器预览产物复用逻辑单元测试。 + +覆盖: +- _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() + + mock_video = MagicMock() + mock_video.file_url = "https://oss.example.com/video.mp4" + + with patch( + "app.api.routes.templates_editor.generation.get_generated_videos_for_task", + return_value=[mock_video], + ): + result = _get_task_output_url(task, repo) + + 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() + + with patch( + "app.api.routes.templates_editor.generation.get_generated_videos_for_task", + return_value=[], + ): + result = _get_task_output_url(task, repo) + + 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() + + with patch( + "app.api.routes.templates_editor.generation.get_generated_videos_for_task", + side_effect=Exception("db error"), + ): + result = _get_task_output_url(task, repo) + + 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