test: 补充编辑器预览产物复用逻辑单测
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 27s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m27s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m28s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m40s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m40s
AI Code Review / AI Code Review (pull_request) Failing after 2m26s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m21s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 2m54s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 2m54s
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 5m14s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 1m56s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 7m13s
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / CI Gate (pull_request) Failing after 5s
CI/CD Pipeline / ACR Image Cleanup (pull_request) Failing after 13m39s
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 27s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m27s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m28s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m40s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m40s
AI Code Review / AI Code Review (pull_request) Failing after 2m26s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m21s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 2m54s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 2m54s
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 5m14s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 1m56s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 7m13s
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / CI Gate (pull_request) Failing after 5s
CI/CD Pipeline / ACR Image Cleanup (pull_request) Failing after 13m39s
- _find_reusable_preview_task: 6 个用例(正常复用/无任务/未完成/plan已修改/非预览/异常) - _get_task_output_url: 3 个用例(正常/空视频/异常) - mark_confirmed: 5 个用例(is_preview/cover/title/dimensions)
This commit is contained in:
@@ -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
|
||||
Reference in New Issue
Block a user