[Do] refactor: 统一渲染链路，预览即所得 #1479
@@ -5,7 +5,6 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
|
||||
from app.auth import AuthenticatedUser, get_current_user
|
||||
@@ -268,19 +267,7 @@ def create_preview_generation_task(
|
||||
# 从模板读取 editing_mode / mode 作为 strategy_id（渲染 pipeline 的 mode 参数）
|
||||
strategy_id = _resolve_strategy_id_from_template(request.template_id, db, user_id)
|
||||
|
||||
# 处理标题配置：如果有标题文本，序列化到 custom_title 字段传递给 worker
|
||||
title_config = request.title_config or {}
|
||||
title_text = (title_config.get("text") or "").strip()
|
||||
custom_title_value = ""
|
||||
if title_text:
|
||||
# 将标题文本和样式配置序列化为 JSON 存入 custom_title
|
||||
# Worker 端会解析 JSON 获取完整标题配置
|
||||
custom_title_value = json.dumps(title_config, ensure_ascii=False)
|
||||
logger.info(
|
||||
"[预览生成] 标题配置: text=%s, config_keys=%s",
|
||||
title_text[:30],
|
||||
list(title_config.keys()),
|
||||
)
|
||||
|
||||
use_case = CreateGenerationTaskUseCase(generation_task_repository)
|
||||
|
||||
@@ -305,7 +292,7 @@ def create_preview_generation_task(
|
||||
auto_retry_enabled=False,
|
||||
auto_retry_max=0,
|
||||
is_preview=True,
|
||||
custom_title=custom_title_value,
|
||||
title_config=title_config,
|
||||
)
|
||||
)
|
||||
except ValueError as e:
|
||||
|
||||
@@ -70,7 +70,6 @@ def _to_generation_task_response(task) -> GenerationTaskResponse:
|
||||
output_width=getattr(task, "output_width", 1280),
|
||||
output_height=getattr(task, "output_height", 720),
|
||||
cover_url=getattr(task, "cover_url", ""),
|
||||
custom_title=getattr(task, "custom_title", ""),
|
||||
title_config=getattr(task, "title_config", {}) or {},
|
||||
logs=getattr(task, "logs", "[]"),
|
||||
status=task.status,
|
||||
@@ -355,7 +354,6 @@ def create_generation_task(
|
||||
output_width=request.output_width,
|
||||
output_height=request.output_height,
|
||||
cover_url=request.cover_url,
|
||||
custom_title=request.custom_title,
|
||||
title_config=request.title_config or {},
|
||||
)
|
||||
)
|
||||
@@ -473,7 +471,6 @@ def confirm_generation(
|
||||
if resolution_match:
|
||||
source_task.mark_confirmed(
|
||||
cover_url=request.cover_url,
|
||||
custom_title=request.custom_title,
|
||||
output_width=request.output_width,
|
||||
output_height=request.output_height,
|
||||
)
|
||||
@@ -519,7 +516,6 @@ def confirm_generation(
|
||||
output_width=request.output_width,
|
||||
output_height=request.output_height,
|
||||
cover_url=request.cover_url,
|
||||
custom_title=request.custom_title,
|
||||
)
|
||||
)
|
||||
|
||||
@@ -652,7 +648,6 @@ def retry_generation_task(
|
||||
output_width=getattr(task, "output_width", 1280),
|
||||
output_height=getattr(task, "output_height", 720),
|
||||
cover_url=getattr(task, "cover_url", ""),
|
||||
custom_title=getattr(task, "custom_title", ""),
|
||||
)
|
||||
)
|
||||
try:
|
||||
|
||||
@@ -10,7 +10,6 @@ class ConfirmGenerationRequest(BaseModel):
|
||||
output_width: int = Field(default=1080, ge=100, description="输出视频宽度")
|
||||
output_height: int = Field(default=1920, ge=100, description="输出视频高度")
|
||||
cover_url: str = Field(default="", description="自定义封面图片 URL")
|
||||
custom_title: str = Field(default="", description="自定义视频标题")
|
||||
|
||||
|
||||
class CreateGenerationTaskRequest(BaseModel):
|
||||
@@ -33,7 +32,7 @@ class CreateGenerationTaskRequest(BaseModel):
|
||||
voice_ids: list[str] = Field(default_factory=list)
|
||||
# ── 来源剪辑计划 ──
|
||||
source_edit_plan_id: str = ""
|
||||
# ── 标题配置（结构化，优先于 custom_title 纯文本）──
|
||||
# ── 标题配置（结构化）──
|
||||
title_config: dict | None = Field(
|
||||
default=None,
|
||||
description="标题样式对象，包含 text/font/font_size/font_color/position/bold/stroke/shadow 等。为空时不影响现有行为。",
|
||||
@@ -77,7 +76,6 @@ class CreateGenerationTaskRequest(BaseModel):
|
||||
output_width: int = Field(default=1280, description="输出视频宽度")
|
||||
output_height: int = Field(default=720, description="输出视频高度")
|
||||
cover_url: str = Field(default="", description="封面图片 URL")
|
||||
custom_title: str = Field(default="", description="自定义视频标题")
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _check_at_least_one_mode(self) -> "CreateGenerationTaskRequest":
|
||||
@@ -113,7 +111,6 @@ class GenerationTaskResponse(BaseModel):
|
||||
output_width: int = 1280
|
||||
output_height: int = 720
|
||||
cover_url: str = ""
|
||||
custom_title: str = ""
|
||||
title_config: dict = Field(default_factory=dict)
|
||||
status: str
|
||||
progress: float
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -41,7 +41,6 @@ def _to_domain(model: GenerationTaskModel) -> GenerationTask:
|
||||
output_width=getattr(model, "output_width", 1280) or 1280,
|
||||
output_height=getattr(model, "output_height", 720) or 720,
|
||||
cover_url=getattr(model, "cover_url", "") or "",
|
||||
custom_title=getattr(model, "custom_title", "") or "",
|
||||
title_config=dict(getattr(model, "title_config", {}) or {}),
|
||||
logs=model.logs or "[]",
|
||||
created_at=model.created_at,
|
||||
@@ -86,7 +85,6 @@ class SQLAlchemyGenerationTaskRepository:
|
||||
output_width=task.output_width,
|
||||
output_height=task.output_height,
|
||||
cover_url=task.cover_url or "",
|
||||
custom_title=task.custom_title or "",
|
||||
title_config=dict(task.title_config) if task.title_config else {},
|
||||
logs=task.logs,
|
||||
created_at=task.created_at,
|
||||
@@ -274,7 +272,6 @@ class SQLAlchemyGenerationTaskRepository:
|
||||
model.output_width = task.output_width
|
||||
model.output_height = task.output_height
|
||||
model.cover_url = task.cover_url or ""
|
||||
model.custom_title = task.custom_title or ""
|
||||
model.title_config = dict(task.title_config) if task.title_config else {}
|
||||
model.logs = task.logs
|
||||
self.session.commit()
|
||||
|
||||
@@ -31,7 +31,6 @@ class CreateGenerationTaskCommand:
|
||||
output_width: int = 1280
|
||||
output_height: int = 720
|
||||
cover_url: str = ""
|
||||
custom_title: str = ""
|
||||
title_config: dict = field(default_factory=dict)
|
||||
|
||||
|
||||
@@ -68,7 +67,6 @@ class CreateGenerationTaskUseCase:
|
||||
output_width=command.output_width,
|
||||
output_height=command.output_height,
|
||||
cover_url=command.cover_url,
|
||||
custom_title=command.custom_title,
|
||||
title_config=command.title_config,
|
||||
)
|
||||
return self.generation_task_repository.create(task)
|
||||
|
||||
@@ -120,7 +120,6 @@ class GenerationTask:
|
||||
output_width: int = 1280
|
||||
output_height: int = 720
|
||||
cover_url: str = ""
|
||||
custom_title: str = ""
|
||||
title_config: dict = field(default_factory=dict)
|
||||
extra_meta: dict = field(default_factory=dict)
|
||||
logs: str = "[]"
|
||||
@@ -153,7 +152,6 @@ class GenerationTask:
|
||||
output_width: int = 1280,
|
||||
output_height: int = 720,
|
||||
cover_url: str = "",
|
||||
custom_title: str = "",
|
||||
title_config: dict | None = None,
|
||||
extra_meta: dict | None = None,
|
||||
) -> "GenerationTask":
|
||||
@@ -185,7 +183,6 @@ class GenerationTask:
|
||||
output_width=output_width,
|
||||
output_height=output_height,
|
||||
cover_url=cover_url,
|
||||
custom_title=custom_title,
|
||||
title_config=dict(title_config) if title_config else {},
|
||||
extra_meta=dict(extra_meta) if extra_meta else {},
|
||||
)
|
||||
@@ -306,7 +303,6 @@ class GenerationTask:
|
||||
self,
|
||||
*,
|
||||
cover_url: str = "",
|
||||
custom_title: str = "",
|
||||
extra_meta: dict | None = None,
|
||||
output_width: int = 0,
|
||||
output_height: int = 0,
|
||||
@@ -319,8 +315,6 @@ class GenerationTask:
|
||||
self.is_preview = False
|
||||
if cover_url:
|
||||
self.cover_url = cover_url
|
||||
if custom_title:
|
||||
self.custom_title = custom_title
|
||||
if output_width > 0:
|
||||
self.output_width = output_width
|
||||
if output_height > 0:
|
||||
|
||||
@@ -1,234 +0,0 @@
|
||||
"""剪辑模式渲染集成测试.
|
||||
|
||||
验证剪辑模式（ONE_TAKE / VOICE_OVER）通过
|
||||
_build_plan_and_clips_from_task + UnifiedRenderService 的完整渲染流程。
|
||||
注：PIP / VOICE_PIP 已下线，统一映射为 ONE_TAKE。
|
||||
|
||||
需要 ffmpeg 可用；CI 无 ffmpeg 时自动跳过。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from video_processing.unified_render_service import (
|
||||
RenderResult,
|
||||
UnifiedRenderService,
|
||||
_resolve_layer_role,
|
||||
)
|
||||
from worker_app.tasks.generation import _build_plan_and_clips_from_task
|
||||
|
||||
pytestmark = pytest.mark.skipif(
|
||||
not shutil.which("ffmpeg"),
|
||||
reason="ffmpeg not available",
|
||||
)
|
||||
|
||||
|
||||
# ── 辅助函数 ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _generate_test_video(path: Path, duration: float = 3.0, color: str = "red") -> None:
|
||||
"""生成一个纯色测试视频。"""
|
||||
cmd = [
|
||||
"ffmpeg",
|
||||
"-y",
|
||||
"-f",
|
||||
"lavfi",
|
||||
"-i",
|
||||
f"color=c={color}:s=640x360:d={duration}:r=25",
|
||||
"-c:v",
|
||||
"libx264",
|
||||
"-pix_fmt",
|
||||
"yuv420p",
|
||||
"-movflags",
|
||||
"+faststart",
|
||||
str(path),
|
||||
]
|
||||
subprocess.run(cmd, check=True, capture_output=True, timeout=30)
|
||||
|
||||
|
||||
def _render_with_mode(
|
||||
mode: str,
|
||||
num_clips: int = 3,
|
||||
duration: float = 2.0,
|
||||
) -> tuple[RenderResult, Path]:
|
||||
"""用指定模式生成测试视频并渲染，返回 (result, work_dir)。
|
||||
|
||||
调用方负责清理 work_dir。
|
||||
"""
|
||||
work_dir = Path(tempfile.mkdtemp(prefix="test_4mode_"))
|
||||
|
||||
# 生成测试视频素材
|
||||
colors = ["red", "green", "blue", "yellow", "purple"]
|
||||
downloaded_paths: list[Path] = []
|
||||
for i in range(num_clips):
|
||||
p = work_dir / f"test_{i:03d}.mp4"
|
||||
_generate_test_video(p, duration=duration, color=colors[i % len(colors)])
|
||||
downloaded_paths.append(p)
|
||||
|
||||
# 构建虚拟 plan + clips
|
||||
task_id = f"test_task_{mode}"
|
||||
plan, clips, asset_path_map = _build_plan_and_clips_from_task(
|
||||
task_id=task_id,
|
||||
downloaded_paths=downloaded_paths,
|
||||
mode=mode,
|
||||
)
|
||||
|
||||
# 渲染
|
||||
service = UnifiedRenderService(
|
||||
plan=plan,
|
||||
clips=clips,
|
||||
asset_path_map=asset_path_map,
|
||||
work_dir=work_dir,
|
||||
output_width=640,
|
||||
output_height=360,
|
||||
output_fps=25,
|
||||
)
|
||||
result = service.render()
|
||||
return result, work_dir
|
||||
|
||||
|
||||
# ── 测试 _build_plan_and_clips_from_task ──────────────────────────────────────
|
||||
|
||||
|
||||
class TestBuildPlanAndClips:
|
||||
"""测试 4 种模式的虚拟 plan 构建。"""
|
||||
|
||||
def _make_paths(self, n: int) -> list[Path]:
|
||||
return [Path(f"/tmp/test_{i}.mp4") for i in range(n)]
|
||||
|
||||
def test_one_take_mode(self):
|
||||
paths = self._make_paths(3)
|
||||
plan, clips, asset_map = _build_plan_and_clips_from_task("t1", paths, "one_take")
|
||||
|
||||
assert plan.id == "t1"
|
||||
assert len(clips) == 3
|
||||
assert all(c.clip_type == "main" for c in clips)
|
||||
assert len(asset_map) == 3
|
||||
|
||||
def test_pip_mode_maps_to_one_take(self):
|
||||
"""PIP 已下线，映射为 one_take → 全部 main clips。"""
|
||||
paths = self._make_paths(3)
|
||||
plan, clips, asset_map = _build_plan_and_clips_from_task("t2", paths, "pip")
|
||||
|
||||
assert len(clips) == 3
|
||||
assert all(c.clip_type == "main" for c in clips)
|
||||
|
||||
def test_voice_over_mode(self):
|
||||
paths = self._make_paths(3)
|
||||
plan, clips, asset_map = _build_plan_and_clips_from_task("t3", paths, "voice_over")
|
||||
|
||||
assert len(clips) == 3
|
||||
assert all(c.clip_type == "main" for c in clips)
|
||||
assert all(c.config.get("role") == "b_roll" for c in clips)
|
||||
|
||||
def test_voice_pip_mode_maps_to_one_take(self):
|
||||
"""VOICE_PIP 已下线，映射为 one_take → 全部 main clips。"""
|
||||
paths = self._make_paths(4)
|
||||
plan, clips, asset_map = _build_plan_and_clips_from_task("t4", paths, "voice_pip")
|
||||
|
||||
assert len(clips) == 4
|
||||
assert all(c.clip_type == "main" for c in clips)
|
||||
|
||||
def test_unknown_mode_defaults_to_one_take(self):
|
||||
paths = self._make_paths(2)
|
||||
plan, clips, asset_map = _build_plan_and_clips_from_task("t5", paths, "unknown_mode")
|
||||
|
||||
assert len(clips) == 2
|
||||
assert all(c.clip_type == "main" for c in clips)
|
||||
|
||||
def test_asset_path_map_keys_match_clip_asset_ids(self):
|
||||
paths = self._make_paths(3)
|
||||
_, clips, asset_map = _build_plan_and_clips_from_task("t6", paths, "one_take")
|
||||
|
||||
clip_asset_ids = {c.asset_id for c in clips}
|
||||
map_keys = set(asset_map.keys())
|
||||
assert clip_asset_ids == map_keys
|
||||
|
||||
|
||||
# ── 测试图层分组（4 模式） ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestFourModeLayerGrouping:
|
||||
"""验证 4 种模式的 clip_type 分布经 _resolve_layer_role 后产生正确的图层。"""
|
||||
|
||||
def test_one_take_layers(self):
|
||||
"""ONE_TAKE: 3 main → 1 main layer。"""
|
||||
paths = [Path(f"/tmp/ot_{i}.mp4") for i in range(3)]
|
||||
_, clips, _ = _build_plan_and_clips_from_task("ot", paths, "one_take")
|
||||
|
||||
roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips}
|
||||
assert roles == {"main"}
|
||||
|
||||
def test_pip_layers_now_one_take(self):
|
||||
"""PIP 已下线 → one_take: 3 main → 1 main layer。"""
|
||||
paths = [Path(f"/tmp/pip_{i}.mp4") for i in range(3)]
|
||||
_, clips, _ = _build_plan_and_clips_from_task("pip", paths, "pip")
|
||||
|
||||
roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips}
|
||||
assert roles == {"main"}
|
||||
|
||||
def test_voice_over_layers(self):
|
||||
"""VOICE_OVER: 3 main(b_roll) → broll。"""
|
||||
paths = [Path(f"/tmp/vo_{i}.mp4") for i in range(3)]
|
||||
_, clips, _ = _build_plan_and_clips_from_task("vo", paths, "voice_over")
|
||||
|
||||
roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips}
|
||||
assert roles == {"broll"}
|
||||
|
||||
def test_voice_pip_layers_now_one_take(self):
|
||||
"""VOICE_PIP 已下线 → one_take: 4 main → main layer。"""
|
||||
paths = [Path(f"/tmp/vpip_{i}.mp4") for i in range(4)]
|
||||
_, clips, _ = _build_plan_and_clips_from_task("vpip", paths, "voice_pip")
|
||||
|
||||
roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips}
|
||||
assert roles == {"main"}
|
||||
|
||||
|
||||
# ── 端到端渲染测试（需要 ffmpeg） ─────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestEndToEndRendering:
|
||||
"""4 种模式的完整渲染测试，验证输出文件存在且时长合理。"""
|
||||
|
||||
def test_one_take_render(self):
|
||||
result, work_dir = _render_with_mode("one_take", num_clips=2, duration=2.0)
|
||||
try:
|
||||
assert result.output_path.exists()
|
||||
assert result.file_size > 0
|
||||
assert result.duration > 0
|
||||
assert result.width == 640
|
||||
assert result.height == 360
|
||||
finally:
|
||||
shutil.rmtree(work_dir, ignore_errors=True)
|
||||
|
||||
def test_pip_render(self):
|
||||
result, work_dir = _render_with_mode("pip", num_clips=2, duration=2.0)
|
||||
try:
|
||||
assert result.output_path.exists()
|
||||
assert result.file_size > 0
|
||||
assert result.duration > 0
|
||||
finally:
|
||||
shutil.rmtree(work_dir, ignore_errors=True)
|
||||
|
||||
def test_voice_over_render(self):
|
||||
result, work_dir = _render_with_mode("voice_over", num_clips=2, duration=2.0)
|
||||
try:
|
||||
assert result.output_path.exists()
|
||||
assert result.file_size > 0
|
||||
assert result.duration > 0
|
||||
finally:
|
||||
shutil.rmtree(work_dir, ignore_errors=True)
|
||||
|
||||
def test_voice_pip_render(self):
|
||||
result, work_dir = _render_with_mode("voice_pip", num_clips=3, duration=2.0)
|
||||
try:
|
||||
assert result.output_path.exists()
|
||||
assert result.file_size > 0
|
||||
assert result.duration > 0
|
||||
finally:
|
||||
shutil.rmtree(work_dir, ignore_errors=True)
|
||||
@@ -1,189 +0,0 @@
|
||||
"""全链路集成测试.
|
||||
|
||||
验证 PlanGeneratorService → UnifiedRenderService → 查重 的端到端流程。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from video_processing.unified_render_service import (
|
||||
UnifiedRenderService,
|
||||
)
|
||||
from worker_app.tasks.generation import _build_plan_and_clips_from_task
|
||||
|
||||
pytestmark = pytest.mark.skipif(
|
||||
not shutil.which("ffmpeg"),
|
||||
reason="ffmpeg not available",
|
||||
)
|
||||
|
||||
|
||||
def _generate_test_video(path: Path, duration: float = 3.0) -> None:
|
||||
"""生成一个测试视频。"""
|
||||
cmd = [
|
||||
"ffmpeg",
|
||||
"-y",
|
||||
"-f",
|
||||
"lavfi",
|
||||
"-i",
|
||||
f"color=c=blue:s=640x360:d={duration}:r=25",
|
||||
"-c:v",
|
||||
"libx264",
|
||||
"-pix_fmt",
|
||||
"yuv420p",
|
||||
str(path),
|
||||
]
|
||||
subprocess.run(cmd, check=True, capture_output=True, timeout=30)
|
||||
|
||||
|
||||
def _generate_test_audio(path: Path, duration: float = 5.0) -> None:
|
||||
"""生成一个测试音频文件。"""
|
||||
cmd = [
|
||||
"ffmpeg",
|
||||
"-y",
|
||||
"-f",
|
||||
"lavfi",
|
||||
"-i",
|
||||
f"sine=frequency=440:duration={duration}",
|
||||
"-c:a",
|
||||
"aac",
|
||||
"-b:a",
|
||||
"128k",
|
||||
str(path),
|
||||
]
|
||||
subprocess.run(cmd, check=True, capture_output=True, timeout=30)
|
||||
|
||||
|
||||
class TestFullPipeline:
|
||||
"""验证从虚拟 plan 构建到渲染输出的完整流程。"""
|
||||
|
||||
def test_one_take_pipeline(self):
|
||||
"""ONE_TAKE 模式完整流程。"""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
work_dir = Path(tmpdir)
|
||||
|
||||
# 生成测试素材
|
||||
paths = []
|
||||
for i in range(3):
|
||||
p = work_dir / f"clip_{i}.mp4"
|
||||
_generate_test_video(p, duration=2.0)
|
||||
paths.append(p)
|
||||
|
||||
# 构建虚拟 plan
|
||||
plan, clips, asset_map = _build_plan_and_clips_from_task("pipeline_test", paths, "one_take")
|
||||
|
||||
# 渲染
|
||||
service = UnifiedRenderService(
|
||||
plan=plan,
|
||||
clips=clips,
|
||||
asset_path_map=asset_map,
|
||||
work_dir=work_dir,
|
||||
output_width=640,
|
||||
output_height=360,
|
||||
)
|
||||
result = service.render()
|
||||
|
||||
assert result.output_path.exists()
|
||||
assert result.duration > 0
|
||||
assert result.file_size > 0
|
||||
assert result.width == 640
|
||||
assert result.height == 360
|
||||
|
||||
def test_pipeline_with_audio_mux(self):
|
||||
"""渲染 + 混音后处理。"""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
work_dir = Path(tmpdir)
|
||||
|
||||
# 生成测试素材
|
||||
video_path = work_dir / "clip_0.mp4"
|
||||
_generate_test_video(video_path, duration=3.0)
|
||||
|
||||
# 构建虚拟 plan
|
||||
plan, clips, asset_map = _build_plan_and_clips_from_task("audio_test", [video_path], "one_take")
|
||||
|
||||
# 渲染
|
||||
service = UnifiedRenderService(
|
||||
plan=plan,
|
||||
clips=clips,
|
||||
asset_path_map=asset_map,
|
||||
work_dir=work_dir,
|
||||
output_width=640,
|
||||
output_height=360,
|
||||
)
|
||||
render_result = service.render()
|
||||
|
||||
# 混音 - 直接用 ffmpeg（_mux_audio_track 已被清理）
|
||||
audio_path = work_dir / "voice.aac"
|
||||
_generate_test_audio(audio_path, duration=5.0)
|
||||
|
||||
final_path = work_dir / "final.mp4"
|
||||
mux_cmd = [
|
||||
"ffmpeg",
|
||||
"-y",
|
||||
"-i",
|
||||
str(render_result.output_path),
|
||||
"-i",
|
||||
str(audio_path),
|
||||
"-c:v",
|
||||
"copy",
|
||||
"-c:a",
|
||||
"aac",
|
||||
"-shortest",
|
||||
str(final_path),
|
||||
]
|
||||
subprocess.run(mux_cmd, check=True, capture_output=True, timeout=30)
|
||||
|
||||
assert final_path.exists()
|
||||
assert final_path.stat().st_size > 0
|
||||
|
||||
def test_single_clip_pipeline(self):
|
||||
"""单 clip 渲染（无转场）。"""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
work_dir = Path(tmpdir)
|
||||
|
||||
video_path = work_dir / "single.mp4"
|
||||
_generate_test_video(video_path, duration=5.0)
|
||||
|
||||
plan, clips, asset_map = _build_plan_and_clips_from_task("single_test", [video_path], "one_take")
|
||||
|
||||
service = UnifiedRenderService(
|
||||
plan=plan,
|
||||
clips=clips,
|
||||
asset_path_map=asset_map,
|
||||
work_dir=work_dir,
|
||||
output_width=640,
|
||||
output_height=360,
|
||||
)
|
||||
result = service.render()
|
||||
|
||||
assert result.output_path.exists()
|
||||
assert result.duration > 0
|
||||
|
||||
def test_dedup_helper_integration(self):
|
||||
"""验证 dedup_helpers.create_video_record_and_dedup 的导入和签名。"""
|
||||
# 只验证函数存在且签名正确（不实际调用，需要数据库）
|
||||
import inspect
|
||||
|
||||
from video_processing.dedup_helpers import create_video_record_and_dedup
|
||||
|
||||
sig = inspect.signature(create_video_record_and_dedup)
|
||||
params = set(sig.parameters.keys())
|
||||
expected = {
|
||||
"generation_task_id",
|
||||
"project_id",
|
||||
"batch_id",
|
||||
"file_url",
|
||||
"file_size",
|
||||
"duration",
|
||||
"video_path",
|
||||
"mode",
|
||||
"session",
|
||||
"width",
|
||||
"height",
|
||||
"fps",
|
||||
}
|
||||
assert expected.issubset(params), f"Missing params: {expected - params}"
|
||||
@@ -149,14 +149,6 @@ class TestWorkerGenerationNoPreviewOverride:
|
||||
assert 'resolution = "854x480"' not in source, "Should not override resolution to 480p in preview mode"
|
||||
assert 'bitrate = "1M"' not in source, "Should not override bitrate to 1M in preview mode"
|
||||
|
||||
def test_parallel_download_still_works(self):
|
||||
"""并行下载逻辑保留。"""
|
||||
with open("apps/worker/worker_app/tasks/generation.py") as f:
|
||||
source = f.read()
|
||||
|
||||
assert "ThreadPoolExecutor" in source, "Should use ThreadPoolExecutor for parallel downloads"
|
||||
assert "as_completed" in source, "Should use as_completed for result collection"
|
||||
|
||||
|
||||
# ── 5. generation_preview.py 不再有 PREVIEW_RESOLUTION ──
|
||||
|
||||
|
||||
@@ -2,7 +2,6 @@
|
||||
|
||||
验证：
|
||||
1. _load_task_info 正确加载 voice_ids
|
||||
2. _render_video 接受 voice_ids 参数
|
||||
3. voice_ids 正确注入到 plan config 中（实际执行代码路径，diff-cover 可达）
|
||||
"""
|
||||
|
||||
@@ -74,185 +73,3 @@ class TestLoadTaskInfoVoiceIds:
|
||||
|
||||
result = _load_task_info("test_task_id")
|
||||
assert result["voice_ids"] == []
|
||||
|
||||
|
||||
class TestRenderVideoVoiceInjection:
|
||||
"""验证 _render_video 正确注入 voice_id 到 plan config（实际执行代码路径）"""
|
||||
|
||||
def test_render_video_accepts_voice_ids(self):
|
||||
"""_render_video 签名包含 voice_ids 参数"""
|
||||
import inspect
|
||||
|
||||
from worker_app.tasks.generation import _render_video
|
||||
|
||||
sig = inspect.signature(_render_video)
|
||||
assert "voice_ids" in sig.parameters
|
||||
|
||||
def test_voice_ids_default_none(self):
|
||||
"""voice_ids 参数默认为 None"""
|
||||
import inspect
|
||||
|
||||
from worker_app.tasks.generation import _render_video
|
||||
|
||||
sig = inspect.signature(_render_video)
|
||||
param = sig.parameters["voice_ids"]
|
||||
assert param.default is None
|
||||
|
||||
def test_voice_ids_injected_into_plan_config(self):
|
||||
"""voice_ids 非空时，voice_id 和 subtitle.auto_generated 被注入到 plan config。
|
||||
|
||||
此测试实际执行 _render_video 的配音注入代码路径，确保 diff-cover 覆盖新增行。
|
||||
"""
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
@dataclass
|
||||
class MockClip:
|
||||
"""模拟 VirtualClip，至少需要 duration 属性。"""
|
||||
|
||||
id: str = "clip_1"
|
||||
duration: float = 5.0
|
||||
config: dict = field(default_factory=dict)
|
||||
|
||||
@dataclass
|
||||
class MockPlan:
|
||||
"""模拟 VirtualPlan，至少需要 config 属性。"""
|
||||
|
||||
id: str = "test_plan"
|
||||
name: str = "test"
|
||||
config: dict = field(default_factory=dict)
|
||||
|
||||
mock_plan = MockPlan(config={"some_key": "some_value"})
|
||||
mock_clips = [MockClip(duration=5.0), MockClip(duration=3.0)]
|
||||
mock_asset_path_map = {"asset_1": Path("/tmp/video1.mp4")}
|
||||
|
||||
# Mock RenderAdapter 和 render 结果
|
||||
mock_render_result = MagicMock()
|
||||
mock_render_result.success = True
|
||||
mock_render_result.output_path = Path("/tmp/output.mp4")
|
||||
mock_render_result.duration = 8.0
|
||||
|
||||
mock_adapter_cls = MagicMock(return_value=MagicMock())
|
||||
mock_adapter_cls.return_value.render_from_memory.return_value = mock_render_result
|
||||
|
||||
mock_db = MagicMock()
|
||||
|
||||
with (
|
||||
patch(
|
||||
"worker_app.tasks.generation._build_plan_and_clips_from_task",
|
||||
return_value=(mock_plan, mock_clips, mock_asset_path_map),
|
||||
),
|
||||
patch(
|
||||
"worker_app.tasks.generation._load_template_plan_config",
|
||||
return_value=None,
|
||||
),
|
||||
patch(
|
||||
"video_processing.render_adapter.RenderAdapter",
|
||||
mock_adapter_cls,
|
||||
),
|
||||
patch(
|
||||
"worker_app.tasks.generation.SessionLocal",
|
||||
return_value=mock_db,
|
||||
),
|
||||
):
|
||||
from worker_app.tasks.generation import _render_video
|
||||
|
||||
from packages.domain import EditingMode
|
||||
|
||||
output_path, render_duration, cover_candidates = _render_video(
|
||||
task_id="test_task_123",
|
||||
downloaded_videos=[Path("/tmp/video1.mp4")],
|
||||
voice_path=None,
|
||||
editing_mode=EditingMode.ONE_TAKE,
|
||||
project_id="proj_1",
|
||||
template_id="tmpl_1",
|
||||
user_id="user_1",
|
||||
temp_path=Path("/tmp"),
|
||||
output_name="test_output.mp4",
|
||||
resolution="854x480",
|
||||
voice_ids=["voice_abc"],
|
||||
)
|
||||
|
||||
# 验证 voice_id 被注入到 plan config（覆盖新增代码行）
|
||||
assert mock_plan.config.get("voice_id") == "voice_abc"
|
||||
# 验证 subtitle.auto_generated 被设置为 True
|
||||
assert mock_plan.config.get("subtitle", {}).get("auto_generated") is True
|
||||
# 验证 RenderAdapter 被调用
|
||||
mock_adapter_cls.return_value.render_from_memory.assert_called_once()
|
||||
# 验证返回值
|
||||
assert output_path == Path("/tmp/output.mp4")
|
||||
assert render_duration == 8.0
|
||||
|
||||
def test_voice_ids_empty_skips_injection(self):
|
||||
"""voice_ids 为空时，不注入 voice_id 到 plan config"""
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
@dataclass
|
||||
class MockClip:
|
||||
id: str = "clip_1"
|
||||
duration: float = 5.0
|
||||
|
||||
@dataclass
|
||||
class MockPlan:
|
||||
id: str = "test_plan"
|
||||
name: str = "test"
|
||||
config: dict = field(default_factory=dict)
|
||||
|
||||
mock_plan = MockPlan(config={"export": {"resolution": "854x480"}})
|
||||
mock_clips = [MockClip(duration=5.0)]
|
||||
|
||||
mock_render_result = MagicMock()
|
||||
mock_render_result.success = True
|
||||
mock_render_result.output_path = Path("/tmp/output.mp4")
|
||||
mock_render_result.duration = 5.0
|
||||
|
||||
mock_adapter_cls = MagicMock(return_value=MagicMock())
|
||||
mock_adapter_cls.return_value.render_from_memory.return_value = mock_render_result
|
||||
|
||||
with (
|
||||
patch(
|
||||
"worker_app.tasks.generation._build_plan_and_clips_from_task",
|
||||
return_value=(mock_plan, mock_clips, {}),
|
||||
),
|
||||
patch(
|
||||
"worker_app.tasks.generation._load_template_plan_config",
|
||||
return_value=None,
|
||||
),
|
||||
patch(
|
||||
"video_processing.render_adapter.RenderAdapter",
|
||||
mock_adapter_cls,
|
||||
),
|
||||
patch(
|
||||
"worker_app.tasks.generation.SessionLocal",
|
||||
return_value=MagicMock(),
|
||||
),
|
||||
):
|
||||
from worker_app.tasks.generation import _render_video
|
||||
|
||||
from packages.domain import EditingMode
|
||||
|
||||
_render_video(
|
||||
task_id="test_task_456",
|
||||
downloaded_videos=[Path("/tmp/video1.mp4")],
|
||||
voice_path=None,
|
||||
editing_mode=EditingMode.ONE_TAKE,
|
||||
project_id="proj_1",
|
||||
template_id="",
|
||||
user_id="user_1",
|
||||
temp_path=Path("/tmp"),
|
||||
output_name="test_output.mp4",
|
||||
voice_ids=[],
|
||||
)
|
||||
|
||||
# 验证 voice_id 没有被注入
|
||||
assert "voice_id" not in mock_plan.config
|
||||
|
||||
|
||||
class TestGenerateVideoPassesVoiceIds:
|
||||
"""验证 generate_video 调用 _render_video 时传递 voice_ids"""
|
||||
|
||||
def test_generate_video_passes_voice_ids(self):
|
||||
"""generate_video 中 _render_video 调用包含 voice_ids 参数"""
|
||||
with open("apps/worker/worker_app/tasks/generation.py", "r") as f:
|
||||
content = f.read()
|
||||
|
||||
assert 'voice_ids=task_info.get("voice_ids", [])' in content
|
||||
|
||||
@@ -1,130 +0,0 @@
|
||||
"""
|
||||
测试：模板 config 字段存储了非 dict 值（如 True / False / str）时，
|
||||
渲染链路不会崩溃（'bool' object has no attribute 'get'）。
|
||||
|
||||
覆盖两个关键文件：
|
||||
1. generation.py — _load_template_plan_config 旧系统路径
|
||||
2. unified_render_service.py — _maybe_generate_ass
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
# Add worker app to path
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
|
||||
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret")
|
||||
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
|
||||
|
||||
|
||||
class TestLoadTemplatePlanConfigBoolDefense:
|
||||
"""_load_template_plan_config 旧系统路径对非 dict 值的防护。"""
|
||||
|
||||
def _call_old_path(self, title_cfg, subtitle_cfg, bgm_cfg):
|
||||
"""通过 mock 新模板系统返回 None，强制走旧模板系统 fallback 路径。"""
|
||||
from worker_app.tasks.generation import _load_template_plan_config
|
||||
|
||||
mock_old_template = MagicMock()
|
||||
mock_old_template.title_config = title_cfg
|
||||
mock_old_template.subtitle_config = subtitle_cfg
|
||||
mock_old_template.bgm_config = bgm_cfg
|
||||
|
||||
mock_session = MagicMock()
|
||||
# 旧系统 query 返回 mock template
|
||||
mock_session.query.return_value.filter.return_value.first.return_value = mock_old_template
|
||||
|
||||
# Mock 新模板系统 repo.get() 返回 None（强制走 fallback）
|
||||
mock_repo_cls = MagicMock()
|
||||
mock_repo_cls.return_value.get.return_value = None
|
||||
|
||||
with (
|
||||
patch("worker_app.tasks.generation.SessionLocal", return_value=mock_session),
|
||||
patch("packages.adapters.sqlalchemy_impl.SQLAlchemyEditTemplateRepository", mock_repo_cls),
|
||||
patch("packages.adapters.sqlalchemy_impl.SQLAlchemyTemplateClipConfigRepository", MagicMock()),
|
||||
):
|
||||
return _load_template_plan_config("fake-id")
|
||||
|
||||
def test_bool_values_return_empty(self):
|
||||
"""title_config=True / subtitle_config=False / bgm_config='str' → 全部过滤掉"""
|
||||
result = self._call_old_path(True, False, "not_a_dict")
|
||||
assert isinstance(result, dict)
|
||||
assert "title" not in result
|
||||
assert "subtitle" not in result
|
||||
assert "bgm" not in result
|
||||
|
||||
def test_valid_dict_passes_through(self):
|
||||
"""正常 dict 正常传递"""
|
||||
result = self._call_old_path(
|
||||
{"text": "标题", "enabled": True},
|
||||
{"text": "副标题"},
|
||||
{"enabled": True, "source": "test.mp3"},
|
||||
)
|
||||
assert result["title"] == {"text": "标题", "enabled": True}
|
||||
assert result["subtitle"] == {"text": "副标题"}
|
||||
assert result["bgm"] == {"enabled": True, "source": "test.mp3"}
|
||||
|
||||
def test_none_returns_empty(self):
|
||||
"""None → 空 dict"""
|
||||
result = self._call_old_path(None, None, None)
|
||||
assert result == {}
|
||||
|
||||
def test_mixed_valid_and_invalid(self):
|
||||
"""部分有效、部分无效时只保留有效的"""
|
||||
result = self._call_old_path({"text": "OK"}, True, None)
|
||||
assert "title" in result
|
||||
assert "subtitle" not in result
|
||||
assert "bgm" not in result
|
||||
|
||||
def test_int_and_list_also_filtered(self):
|
||||
"""int / list 类型也被过滤"""
|
||||
result = self._call_old_path(42, [1, 2, 3], 0)
|
||||
assert result == {}
|
||||
|
||||
|
||||
class TestUnifiedRenderBoolConfigDefense:
|
||||
"""_maybe_generate_ass 对 plan.config 中非 dict title/subtitle 的防护。"""
|
||||
|
||||
def _make_service(self, config):
|
||||
from video_processing.unified_render_service import UnifiedRenderService
|
||||
|
||||
service = UnifiedRenderService.__new__(UnifiedRenderService)
|
||||
mock_plan = MagicMock()
|
||||
mock_plan.config = config
|
||||
service.plan = mock_plan
|
||||
service.task_id = "test-task"
|
||||
return service
|
||||
|
||||
def test_bool_title_does_not_crash(self):
|
||||
"""config['title']=True → 不崩溃，返回 None"""
|
||||
service = self._make_service({"title": True, "subtitle": {}})
|
||||
result = service._maybe_generate_ass(video_duration=10.0)
|
||||
assert result is None
|
||||
|
||||
def test_bool_subtitle_does_not_crash(self):
|
||||
"""config['subtitle']=False → 不崩溃，返回 None"""
|
||||
service = self._make_service({"title": {}, "subtitle": False})
|
||||
result = service._maybe_generate_ass(video_duration=10.0)
|
||||
assert result is None
|
||||
|
||||
def test_str_title_does_not_crash(self):
|
||||
"""config['title']='plain string' → 不崩溃"""
|
||||
service = self._make_service({"title": "plain string", "subtitle": {}})
|
||||
result = service._maybe_generate_ass(video_duration=10.0)
|
||||
assert result is None
|
||||
|
||||
def test_none_config_does_not_crash(self):
|
||||
"""config=None → 不崩溃"""
|
||||
service = self._make_service(None)
|
||||
result = service._maybe_generate_ass(video_duration=10.0)
|
||||
assert result is None
|
||||
|
||||
def test_int_title_does_not_crash(self):
|
||||
"""config['title']=42 → 不崩溃"""
|
||||
service = self._make_service({"title": 42, "subtitle": 0})
|
||||
result = service._maybe_generate_ass(video_duration=10.0)
|
||||
assert result is None
|
||||
@@ -5,7 +5,7 @@
|
||||
- 预览任务未完成 → 创建新任务走渲染流程
|
||||
- 预览任务不存在 → 404
|
||||
- 权限不足 → 403
|
||||
- cover_url 和 custom_title 正确传递
|
||||
- cover_url 正确传递
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -201,7 +201,6 @@ def _make_preview_task(**kwargs: Any) -> GenerationTask:
|
||||
output_width=1080,
|
||||
output_height=1920,
|
||||
cover_url="",
|
||||
custom_title="",
|
||||
video_title="",
|
||||
resolution="",
|
||||
bgm_config={},
|
||||
@@ -233,7 +232,6 @@ class TestConfirmGenerationReuse:
|
||||
"output_width": 1080,
|
||||
"output_height": 1920,
|
||||
"cover_url": "https://example.com/cover.jpg",
|
||||
"custom_title": "我的视频",
|
||||
},
|
||||
)
|
||||
|
||||
@@ -251,7 +249,6 @@ class TestConfirmGenerationReuse:
|
||||
assert item["output_height"] == 1920
|
||||
# 封面和标题更新
|
||||
assert item["cover_url"] == "https://example.com/cover.jpg"
|
||||
assert item["custom_title"] == "我的视频"
|
||||
|
||||
# 没有创建新任务
|
||||
assert len(gen_task_repo._store) == initial_count
|
||||
@@ -277,7 +274,6 @@ class TestConfirmGenerationReuse:
|
||||
assert updated is not None
|
||||
assert updated.is_preview is False
|
||||
assert updated.cover_url == "https://cdn.example.com/cover.png"
|
||||
assert updated.custom_title == "测试标题"
|
||||
|
||||
def test_confirm_creates_new_task_when_preview_not_completed(
|
||||
self,
|
||||
@@ -356,7 +352,6 @@ class TestConfirmGenerationErrors:
|
||||
assert item["output_height"] == 1080
|
||||
# 默认封面和标题为空
|
||||
assert item["cover_url"] == ""
|
||||
assert item["custom_title"] == ""
|
||||
# 配置保留
|
||||
assert item["voice_library_id"] == "voice-001"
|
||||
assert item["template_id"] == "tmpl-001"
|
||||
@@ -368,7 +363,7 @@ class TestConfirmGenerationErrors:
|
||||
client: TestClient,
|
||||
gen_task_repo: StubGenerationTaskRepository,
|
||||
) -> None:
|
||||
"""cover_url 和 custom_title 正确传递"""
|
||||
"""cover_url 正确传递"""
|
||||
preview = _make_preview_task()
|
||||
gen_task_repo.create(preview)
|
||||
|
||||
@@ -378,14 +373,12 @@ class TestConfirmGenerationErrors:
|
||||
"output_width": 1080,
|
||||
"output_height": 1920,
|
||||
"cover_url": "https://cdn.example.com/my-cover.png",
|
||||
"custom_title": "测试视频标题",
|
||||
},
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
item = resp.json()["items"][0]
|
||||
assert item["cover_url"] == "https://cdn.example.com/my-cover.png"
|
||||
assert item["custom_title"] == "测试视频标题"
|
||||
|
||||
def test_confirm_default_resolution(
|
||||
self,
|
||||
|
||||
@@ -1,389 +0,0 @@
|
||||
"""P3 优化单元测试 — generation.py 三项优化.
|
||||
|
||||
覆盖：
|
||||
P3-1: _download_library_assets strict 模式
|
||||
P3-2: 归属校验合并到同一 DB session
|
||||
P3-3: _verify_url_accessible HEAD 重试
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
|
||||
|
||||
# ── 预注入 mock 模块，防止 worker_app.db 触发真实数据库连接 ──
|
||||
# worker_app.db 在模块级别调用 ensure_database_exists() 尝试连接 PostgreSQL，
|
||||
# 增量测试单独跑这些文件时会失败。与 test_voice_clone_task.py 同理。
|
||||
_mock_db_module = MagicMock()
|
||||
_mock_db_module.SessionLocal = MagicMock()
|
||||
sys.modules.setdefault("worker_app.db", _mock_db_module)
|
||||
if "worker_app" in sys.modules:
|
||||
sys.modules["worker_app"].db = _mock_db_module
|
||||
|
||||
|
||||
# ── P3-3: _verify_url_accessible 重试 ───────────────────────────────────────
|
||||
|
||||
|
||||
class TestVerifyUrlAccessibleRetry:
|
||||
"""_verify_url_accessible 重试逻辑."""
|
||||
|
||||
@patch("time.sleep")
|
||||
@patch("urllib.request.OpenerDirector.open")
|
||||
def test_first_attempt_success(self, mock_open, mock_sleep):
|
||||
"""首次成功，不重试."""
|
||||
from worker_app.tasks.generation import _verify_url_accessible
|
||||
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.status = 200
|
||||
mock_resp.__enter__ = MagicMock(return_value=mock_resp)
|
||||
mock_resp.__exit__ = MagicMock(return_value=False)
|
||||
mock_open.return_value = mock_resp
|
||||
|
||||
assert _verify_url_accessible("https://example.com/file.mp4") is True
|
||||
assert mock_open.call_count == 1
|
||||
mock_sleep.assert_not_called()
|
||||
|
||||
@patch("time.sleep")
|
||||
@patch("urllib.request.OpenerDirector.open")
|
||||
def test_retry_then_success(self, mock_open, mock_sleep):
|
||||
"""首次失败，重试后成功."""
|
||||
from worker_app.tasks.generation import _verify_url_accessible
|
||||
|
||||
# 第一次失败（网络异常），第二次成功
|
||||
mock_resp_ok = MagicMock()
|
||||
mock_resp_ok.status = 200
|
||||
mock_resp_ok.__enter__ = MagicMock(return_value=mock_resp_ok)
|
||||
mock_resp_ok.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
mock_open.side_effect = [
|
||||
OSError("connection reset"),
|
||||
mock_resp_ok,
|
||||
]
|
||||
|
||||
assert _verify_url_accessible("https://example.com/file.mp4") is True
|
||||
assert mock_open.call_count == 2
|
||||
mock_sleep.assert_called_once_with(1)
|
||||
|
||||
@patch("time.sleep")
|
||||
@patch("urllib.request.OpenerDirector.open")
|
||||
def test_all_retries_exhausted(self, mock_open, mock_sleep):
|
||||
"""全部重试耗尽，返回 False."""
|
||||
from worker_app.tasks.generation import _verify_url_accessible
|
||||
|
||||
mock_open.side_effect = OSError("connection refused")
|
||||
|
||||
assert _verify_url_accessible("https://example.com/file.mp4") is False
|
||||
# 1 首次 + 2 重试 = 3 次
|
||||
assert mock_open.call_count == 3
|
||||
assert mock_sleep.call_count == 2
|
||||
|
||||
@patch("time.sleep")
|
||||
@patch("urllib.request.OpenerDirector.open")
|
||||
def test_http_500_then_success(self, mock_open, mock_sleep):
|
||||
"""HTTP 500 后重试成功."""
|
||||
from worker_app.tasks.generation import _verify_url_accessible
|
||||
|
||||
mock_resp_500 = MagicMock()
|
||||
mock_resp_500.status = 500
|
||||
mock_resp_500.__enter__ = MagicMock(return_value=mock_resp_500)
|
||||
mock_resp_500.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
mock_resp_200 = MagicMock()
|
||||
mock_resp_200.status = 200
|
||||
mock_resp_200.__enter__ = MagicMock(return_value=mock_resp_200)
|
||||
mock_resp_200.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
mock_open.side_effect = [mock_resp_500, mock_resp_200]
|
||||
|
||||
assert _verify_url_accessible("https://example.com/file.mp4") is True
|
||||
assert mock_open.call_count == 2
|
||||
|
||||
@patch("time.sleep")
|
||||
@patch("urllib.request.OpenerDirector.open")
|
||||
def test_custom_retries_zero(self, mock_open, mock_sleep):
|
||||
"""retries=0 时不重试."""
|
||||
from worker_app.tasks.generation import _verify_url_accessible
|
||||
|
||||
mock_open.side_effect = OSError("timeout")
|
||||
|
||||
assert _verify_url_accessible("https://example.com/file.mp4", retries=0) is False
|
||||
assert mock_open.call_count == 1
|
||||
mock_sleep.assert_not_called()
|
||||
|
||||
|
||||
# ── P3-1: _download_library_assets strict 模式 ──────────────────────────────
|
||||
|
||||
|
||||
def _make_mock_asset(
|
||||
asset_id: str,
|
||||
name: str,
|
||||
file_url: str | None,
|
||||
asset_library_id: str = "lib-1",
|
||||
project_id: str = "",
|
||||
):
|
||||
"""构造 mock AssetModel 实例."""
|
||||
return SimpleNamespace(
|
||||
id=asset_id,
|
||||
name=name,
|
||||
file_url=file_url,
|
||||
asset_library_id=asset_library_id,
|
||||
project_id=project_id,
|
||||
status="ready",
|
||||
file_type="video",
|
||||
created_at="2026-01-01",
|
||||
)
|
||||
|
||||
|
||||
def _setup_mock_session(assets):
|
||||
"""构造 mock session，返回 (mock_session, mock_query_chain)."""
|
||||
mock_session = MagicMock()
|
||||
mock_query = MagicMock()
|
||||
|
||||
# chain: session.query().filter().filter().order_by().all()
|
||||
mock_session.query.return_value = mock_query
|
||||
mock_query.filter.return_value = mock_query
|
||||
mock_query.order_by.return_value = mock_query
|
||||
mock_query.all.return_value = assets
|
||||
|
||||
return mock_session
|
||||
|
||||
|
||||
class TestDownloadLibraryAssetsStrictMode:
|
||||
"""_download_library_assets strict 模式."""
|
||||
|
||||
@patch("worker_app.tasks.generation.SessionLocal")
|
||||
@patch("worker_app.tasks.generation.download_asset")
|
||||
def test_strict_mode_raises_on_download_failure(self, mock_download, mock_session_factory):
|
||||
"""strict=True 时，单个素材下载失败立即抛 RuntimeError."""
|
||||
from worker_app.tasks.generation import _download_library_assets
|
||||
|
||||
assets = [
|
||||
_make_mock_asset("a1", "video1.mp4", "uploads/video1.mp4"),
|
||||
_make_mock_asset("a2", "video2.mp4", "uploads/video2.mp4"),
|
||||
]
|
||||
mock_session = _setup_mock_session(assets)
|
||||
mock_session_factory.return_value = mock_session
|
||||
|
||||
# 第一个成功，第二个失败
|
||||
mock_download.side_effect = [True, False]
|
||||
|
||||
with pytest.raises(RuntimeError, match="素材下载失败"):
|
||||
_download_library_assets(
|
||||
Path("/tmp"),
|
||||
asset_library_id="lib-1",
|
||||
asset_ids=["a1", "a2"],
|
||||
strict=True,
|
||||
)
|
||||
|
||||
@patch("worker_app.tasks.generation.SessionLocal")
|
||||
@patch("worker_app.tasks.generation.download_asset")
|
||||
def test_non_strict_mode_returns_partial_results(self, mock_download, mock_session_factory):
|
||||
"""strict=False 时，跳过失败素材，返回成功列表."""
|
||||
from worker_app.tasks.generation import _download_library_assets
|
||||
|
||||
assets = [
|
||||
_make_mock_asset("a1", "video1.mp4", "uploads/video1.mp4"),
|
||||
_make_mock_asset("a2", "video2.mp4", "uploads/video2.mp4"),
|
||||
_make_mock_asset("a3", "video3.mp4", "uploads/video3.mp4"),
|
||||
]
|
||||
mock_session = _setup_mock_session(assets)
|
||||
mock_session_factory.return_value = mock_session
|
||||
|
||||
# 第一个成功，第二个失败，第三个成功
|
||||
mock_download.side_effect = [True, False, True]
|
||||
|
||||
result = _download_library_assets(
|
||||
Path("/tmp"),
|
||||
asset_library_id="lib-1",
|
||||
asset_ids=["a1", "a2", "a3"],
|
||||
strict=False,
|
||||
)
|
||||
|
||||
assert len(result) == 2
|
||||
|
||||
@patch("worker_app.tasks.generation.SessionLocal")
|
||||
@patch("worker_app.tasks.generation.download_asset")
|
||||
def test_non_strict_all_fail_raises(self, mock_download, mock_session_factory):
|
||||
"""strict=False 但全部失败时仍抛 RuntimeError."""
|
||||
from worker_app.tasks.generation import _download_library_assets
|
||||
|
||||
assets = [
|
||||
_make_mock_asset("a1", "video1.mp4", "uploads/video1.mp4"),
|
||||
]
|
||||
mock_session = _setup_mock_session(assets)
|
||||
mock_session_factory.return_value = mock_session
|
||||
mock_download.return_value = False
|
||||
|
||||
with pytest.raises(RuntimeError, match="全部下载失败"):
|
||||
_download_library_assets(
|
||||
Path("/tmp"),
|
||||
asset_library_id="lib-1",
|
||||
asset_ids=["a1"],
|
||||
strict=False,
|
||||
)
|
||||
|
||||
@patch("worker_app.tasks.generation.SessionLocal")
|
||||
@patch("worker_app.tasks.generation.download_asset")
|
||||
def test_strict_mode_raises_on_missing_file_url(self, mock_download, mock_session_factory):
|
||||
"""strict=True 时，素材缺少 file_url 立即抛异常."""
|
||||
from worker_app.tasks.generation import _download_library_assets
|
||||
|
||||
assets = [
|
||||
_make_mock_asset("a1", "video1.mp4", None), # file_url 为空
|
||||
]
|
||||
mock_session = _setup_mock_session(assets)
|
||||
mock_session_factory.return_value = mock_session
|
||||
|
||||
with pytest.raises(RuntimeError, match="素材缺少 file_url"):
|
||||
_download_library_assets(
|
||||
Path("/tmp"),
|
||||
asset_library_id="lib-1",
|
||||
strict=True,
|
||||
)
|
||||
|
||||
# download_asset 不应被调用
|
||||
mock_download.assert_not_called()
|
||||
|
||||
@patch("worker_app.tasks.generation.SessionLocal")
|
||||
@patch("worker_app.tasks.generation.download_asset")
|
||||
def test_default_is_strict(self, mock_download, mock_session_factory):
|
||||
"""默认 strict=True."""
|
||||
from worker_app.tasks.generation import _download_library_assets
|
||||
|
||||
assets = [
|
||||
_make_mock_asset("a1", "video1.mp4", "uploads/video1.mp4"),
|
||||
]
|
||||
mock_session = _setup_mock_session(assets)
|
||||
mock_session_factory.return_value = mock_session
|
||||
mock_download.return_value = False
|
||||
|
||||
# 不传 strict 参数，默认严格模式
|
||||
with pytest.raises(RuntimeError, match="素材下载失败"):
|
||||
_download_library_assets(
|
||||
Path("/tmp"),
|
||||
asset_library_id="lib-1",
|
||||
)
|
||||
|
||||
|
||||
# ── P3-2: 归属校验合并到同一 session ────────────────────────────────────────
|
||||
|
||||
|
||||
class TestDownloadLibraryAssetsOwnershipValidation:
|
||||
"""归属校验合并到 _download_library_assets 同一 session."""
|
||||
|
||||
@patch("worker_app.tasks.generation.SessionLocal")
|
||||
@patch("worker_app.tasks.generation.download_asset")
|
||||
def test_ownership_mismatch_raises_value_error(self, mock_download, mock_session_factory):
|
||||
"""asset_ids 不属于指定素材库时抛 ValueError."""
|
||||
from worker_app.tasks.generation import _download_library_assets
|
||||
|
||||
# asset 属于 lib-2，但请求的是 lib-1
|
||||
assets = [
|
||||
_make_mock_asset("a1", "video1.mp4", "uploads/video1.mp4", asset_library_id="lib-2"),
|
||||
]
|
||||
mock_session = _setup_mock_session(assets)
|
||||
mock_session_factory.return_value = mock_session
|
||||
|
||||
with pytest.raises(ValueError, match="素材不属于指定素材库"):
|
||||
_download_library_assets(
|
||||
Path("/tmp"),
|
||||
asset_library_id="lib-1",
|
||||
asset_ids=["a1"],
|
||||
)
|
||||
|
||||
# 不应调用 download_asset（校验在下载前）
|
||||
mock_download.assert_not_called()
|
||||
|
||||
@patch("worker_app.tasks.generation.SessionLocal")
|
||||
@patch("worker_app.tasks.generation.download_asset")
|
||||
def test_missing_asset_ids_raises_value_error(self, mock_download, mock_session_factory):
|
||||
"""指定的 asset_ids 不存在时抛 ValueError."""
|
||||
from worker_app.tasks.generation import _download_library_assets
|
||||
|
||||
# DB 返回空（asset_ids 不存在，query 过滤后无结果）
|
||||
mock_session = _setup_mock_session([])
|
||||
mock_session_factory.return_value = mock_session
|
||||
|
||||
with pytest.raises(RuntimeError, match="未找到视频素材"):
|
||||
_download_library_assets(
|
||||
Path("/tmp"),
|
||||
asset_library_id="lib-1",
|
||||
asset_ids=["nonexistent-id"],
|
||||
)
|
||||
|
||||
@patch("worker_app.tasks.generation.SessionLocal")
|
||||
@patch("worker_app.tasks.generation.download_asset")
|
||||
def test_project_ownership_mismatch_raises(self, mock_download, mock_session_factory):
|
||||
"""项目级模式下归属不匹配抛 ValueError."""
|
||||
from worker_app.tasks.generation import _download_library_assets
|
||||
|
||||
assets = [
|
||||
_make_mock_asset(
|
||||
"a1",
|
||||
"video1.mp4",
|
||||
"uploads/video1.mp4",
|
||||
asset_library_id="",
|
||||
project_id="proj-2",
|
||||
),
|
||||
]
|
||||
mock_session = _setup_mock_session(assets)
|
||||
mock_session_factory.return_value = mock_session
|
||||
|
||||
with pytest.raises(ValueError, match="素材不属于指定项目"):
|
||||
_download_library_assets(
|
||||
Path("/tmp"),
|
||||
project_id="proj-1",
|
||||
asset_ids=["a1"],
|
||||
)
|
||||
|
||||
mock_download.assert_not_called()
|
||||
|
||||
@patch("worker_app.tasks.generation.SessionLocal")
|
||||
@patch("worker_app.tasks.generation.download_asset")
|
||||
def test_ownership_pass_then_download(self, mock_download, mock_session_factory):
|
||||
"""归属校验通过后正常下载."""
|
||||
from worker_app.tasks.generation import _download_library_assets
|
||||
|
||||
assets = [
|
||||
_make_mock_asset("a1", "video1.mp4", "uploads/video1.mp4", asset_library_id="lib-1"),
|
||||
]
|
||||
mock_session = _setup_mock_session(assets)
|
||||
mock_session_factory.return_value = mock_session
|
||||
mock_download.return_value = True
|
||||
|
||||
result = _download_library_assets(
|
||||
Path("/tmp"),
|
||||
asset_library_id="lib-1",
|
||||
asset_ids=["a1"],
|
||||
)
|
||||
|
||||
assert len(result) == 1
|
||||
mock_download.assert_called_once()
|
||||
|
||||
@patch("worker_app.tasks.generation.SessionLocal")
|
||||
@patch("worker_app.tasks.generation.download_asset")
|
||||
def test_single_session_used(self, mock_download, mock_session_factory):
|
||||
"""验证只创建了一个 DB session（P3-2 核心）."""
|
||||
from worker_app.tasks.generation import _download_library_assets
|
||||
|
||||
assets = [
|
||||
_make_mock_asset("a1", "video1.mp4", "uploads/video1.mp4", asset_library_id="lib-1"),
|
||||
]
|
||||
mock_session = _setup_mock_session(assets)
|
||||
mock_session_factory.return_value = mock_session
|
||||
mock_download.return_value = True
|
||||
|
||||
_download_library_assets(
|
||||
Path("/tmp"),
|
||||
asset_library_id="lib-1",
|
||||
asset_ids=["a1"],
|
||||
)
|
||||
|
||||
# SessionLocal 只调用一次（合并前会调用两次：校验 + 下载）
|
||||
assert mock_session_factory.call_count == 1
|
||||
@@ -1,228 +0,0 @@
|
||||
"""Tests for generation.py worker-side fixes: segment durations + preview resolution."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
# Add worker app to path
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
|
||||
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret")
|
||||
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
|
||||
|
||||
|
||||
def _patch_session_local(mock_session):
|
||||
"""Patch worker_app.db.SessionLocal robustly even when other tests
|
||||
have pre-registered a MagicMock for worker_app.db in sys.modules.
|
||||
Uses patch.dict to inject a clean module so that
|
||||
'from worker_app.db import SessionLocal' resolves correctly."""
|
||||
from types import ModuleType
|
||||
|
||||
_fresh_db = ModuleType("worker_app.db")
|
||||
_fresh_db.SessionLocal = lambda *a, **kw: mock_session
|
||||
return patch.dict(sys.modules, {"worker_app.db": _fresh_db})
|
||||
|
||||
|
||||
class TestLoadTemplateSegmentDurations:
|
||||
"""_load_template_segment_durations 单元测试 (covers lines 198-226)."""
|
||||
|
||||
def test_empty_template_id(self):
|
||||
"""空 template_id 直接返回空列表。"""
|
||||
from worker_app.tasks.generation import _load_template_segment_durations
|
||||
|
||||
result = _load_template_segment_durations("")
|
||||
assert result == []
|
||||
|
||||
def test_loads_durations_ordered(self):
|
||||
"""按 segment_order 排序返回 duration_max 列表。"""
|
||||
from worker_app.tasks.generation import _load_template_segment_durations
|
||||
|
||||
mock_seg1 = MagicMock(duration_max=5.0)
|
||||
mock_seg2 = MagicMock(duration_max=8.0)
|
||||
mock_seg3 = MagicMock(duration_max=3.0)
|
||||
|
||||
mock_query = MagicMock()
|
||||
mock_query.filter.return_value.order_by.return_value.all.return_value = [
|
||||
mock_seg1,
|
||||
mock_seg2,
|
||||
mock_seg3,
|
||||
]
|
||||
mock_session = MagicMock()
|
||||
mock_session.query.return_value = mock_query
|
||||
|
||||
with _patch_session_local(mock_session):
|
||||
result = _load_template_segment_durations("tpl_123")
|
||||
|
||||
assert result == [5.0, 8.0, 3.0]
|
||||
|
||||
def test_filters_zero_and_negative(self):
|
||||
"""duration_max <= 0 的 segment 被过滤。"""
|
||||
from worker_app.tasks.generation import _load_template_segment_durations
|
||||
|
||||
mock_seg_valid = MagicMock(duration_max=5.0)
|
||||
mock_seg_zero = MagicMock(duration_max=0.0)
|
||||
mock_seg_none = MagicMock(duration_max=None)
|
||||
|
||||
mock_query = MagicMock()
|
||||
mock_query.filter.return_value.order_by.return_value.all.return_value = [
|
||||
mock_seg_valid,
|
||||
mock_seg_zero,
|
||||
mock_seg_none,
|
||||
]
|
||||
mock_session = MagicMock()
|
||||
mock_session.query.return_value = mock_query
|
||||
|
||||
with _patch_session_local(mock_session):
|
||||
result = _load_template_segment_durations("tpl_456")
|
||||
|
||||
assert result == [5.0]
|
||||
|
||||
def test_db_error_returns_empty(self):
|
||||
"""数据库异常返回空列表，不抛出。"""
|
||||
from types import ModuleType
|
||||
|
||||
from worker_app.tasks.generation import _load_template_segment_durations
|
||||
|
||||
_err_db = ModuleType("worker_app.db")
|
||||
|
||||
def _raise(*a, **kw):
|
||||
raise Exception("DB down")
|
||||
|
||||
_err_db.SessionLocal = _raise
|
||||
with patch.dict(sys.modules, {"worker_app.db": _err_db}):
|
||||
result = _load_template_segment_durations("tpl_789")
|
||||
|
||||
assert result == []
|
||||
|
||||
def test_empty_segments_returns_empty(self):
|
||||
"""没有 segment 时返回空列表。"""
|
||||
from worker_app.tasks.generation import _load_template_segment_durations
|
||||
|
||||
mock_query = MagicMock()
|
||||
mock_query.filter.return_value.order_by.return_value.all.return_value = []
|
||||
mock_session = MagicMock()
|
||||
mock_session.query.return_value = mock_query
|
||||
|
||||
with _patch_session_local(mock_session):
|
||||
result = _load_template_segment_durations("tpl_empty")
|
||||
|
||||
assert result == []
|
||||
|
||||
|
||||
class TestDurationCappingInBuildPlan:
|
||||
"""_build_plan_and_clips_from_task 时长约束测试 (covers lines 322-333)."""
|
||||
|
||||
def _make_temp_video(self, tmpdir: Path, name: str = "v.mp4") -> Path:
|
||||
p = tmpdir / name
|
||||
p.write_bytes(b"\x00" * 100)
|
||||
return p
|
||||
|
||||
def test_clips_capped_by_segment_max(self):
|
||||
"""clip 时长超过 segment duration_max 时截断。"""
|
||||
import tempfile
|
||||
|
||||
from worker_app.tasks.generation import _build_plan_and_clips_from_task
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tmpdir = Path(tmpdir)
|
||||
paths = [self._make_temp_video(tmpdir, f"v{i}.mp4") for i in range(3)]
|
||||
|
||||
with patch("worker_app.tasks.generation.probe_duration", return_value=30.0):
|
||||
with patch(
|
||||
"worker_app.tasks.generation._load_template_segment_durations",
|
||||
return_value=[5.0, 4.0, 3.0],
|
||||
):
|
||||
with patch(
|
||||
"worker_app.tasks.generation._load_template_clip_configs",
|
||||
return_value=[],
|
||||
):
|
||||
_, clips, _ = _build_plan_and_clips_from_task(
|
||||
task_id="test_cap",
|
||||
downloaded_paths=paths,
|
||||
mode="one_take",
|
||||
template_id="tpl_test",
|
||||
)
|
||||
|
||||
assert clips[0].duration == 5.0
|
||||
assert clips[1].duration == 4.0
|
||||
assert clips[2].duration == 3.0
|
||||
|
||||
def test_clips_not_capped_when_under_max(self):
|
||||
"""clip 时长小于 segment duration_max 时不截断。"""
|
||||
import tempfile
|
||||
|
||||
from worker_app.tasks.generation import _build_plan_and_clips_from_task
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tmpdir = Path(tmpdir)
|
||||
paths = [self._make_temp_video(tmpdir)]
|
||||
|
||||
with patch("worker_app.tasks.generation.probe_duration", return_value=3.0):
|
||||
with patch(
|
||||
"worker_app.tasks.generation._load_template_segment_durations",
|
||||
return_value=[5.0],
|
||||
):
|
||||
with patch(
|
||||
"worker_app.tasks.generation._load_template_clip_configs",
|
||||
return_value=[],
|
||||
):
|
||||
_, clips, _ = _build_plan_and_clips_from_task(
|
||||
task_id="test_no_cap",
|
||||
downloaded_paths=paths,
|
||||
mode="one_take",
|
||||
template_id="tpl_test",
|
||||
)
|
||||
|
||||
assert clips[0].duration == 3.0
|
||||
|
||||
def test_no_capping_without_template(self):
|
||||
"""无 template_id 时不截断。"""
|
||||
import tempfile
|
||||
|
||||
from worker_app.tasks.generation import _build_plan_and_clips_from_task
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tmpdir = Path(tmpdir)
|
||||
paths = [self._make_temp_video(tmpdir)]
|
||||
|
||||
with patch("worker_app.tasks.generation.probe_duration", return_value=30.0):
|
||||
_, clips, _ = _build_plan_and_clips_from_task(
|
||||
task_id="test_no_tpl",
|
||||
downloaded_paths=paths,
|
||||
mode="one_take",
|
||||
template_id="",
|
||||
)
|
||||
|
||||
assert clips[0].duration == 30.0
|
||||
|
||||
def test_partial_segments_only_caps_matching(self):
|
||||
"""segment 数量少于 clip 时，只截断有对应 segment 的 clip。"""
|
||||
import tempfile
|
||||
|
||||
from worker_app.tasks.generation import _build_plan_and_clips_from_task
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tmpdir = Path(tmpdir)
|
||||
paths = [self._make_temp_video(tmpdir, f"v{i}.mp4") for i in range(3)]
|
||||
|
||||
with patch("worker_app.tasks.generation.probe_duration", return_value=20.0):
|
||||
with patch(
|
||||
"worker_app.tasks.generation._load_template_segment_durations",
|
||||
return_value=[5.0], # only 1 segment for 3 clips
|
||||
):
|
||||
with patch(
|
||||
"worker_app.tasks.generation._load_template_clip_configs",
|
||||
return_value=[],
|
||||
):
|
||||
_, clips, _ = _build_plan_and_clips_from_task(
|
||||
task_id="test_partial",
|
||||
downloaded_paths=paths,
|
||||
mode="one_take",
|
||||
template_id="tpl_test",
|
||||
)
|
||||
|
||||
assert clips[0].duration == 5.0 # capped
|
||||
assert clips[1].duration == 20.0 # not capped (no matching segment)
|
||||
assert clips[2].duration == 20.0 # not capped
|
||||
@@ -1,7 +1,6 @@
|
||||
"""P0/P1 修复单元测试 — 一键生成 P0 问题 + P1 校验.
|
||||
|
||||
覆盖：
|
||||
P0-1: _download_library_assets 双模式查询（asset_library_id / project_id）
|
||||
P0-2: OSS 上传失败抛异常 + URL 可访问性校验
|
||||
P0-3: FFmpeg 失败时完整 stderr 日志
|
||||
P1: template_id 存在性校验 + asset_ids 归属校验
|
||||
@@ -26,130 +25,6 @@ if str(_WORKER_ROOT) not in sys.path:
|
||||
# ── P0-1: _download_library_assets ────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestDownloadLibraryAssets:
|
||||
"""P0-1: 素材下载双模式 + 错误处理."""
|
||||
|
||||
def _make_asset(self, id_: str, file_url: str, project_id: str = "p1", library_id: str = "lib1"):
|
||||
mock = MagicMock()
|
||||
mock.id = id_
|
||||
mock.file_url = file_url
|
||||
mock.name = f"asset_{id_}"
|
||||
mock.project_id = project_id
|
||||
mock.asset_library_id = library_id
|
||||
return mock
|
||||
|
||||
@patch("worker_app.tasks.generation.SessionLocal")
|
||||
@patch("worker_app.tasks.generation.download_asset")
|
||||
def test_asset_library_mode(self, mock_download, mock_session_factory):
|
||||
"""素材库模式：按 asset_library_id 查询."""
|
||||
from worker_app.tasks.generation import _download_library_assets
|
||||
|
||||
session = MagicMock()
|
||||
mock_session_factory.return_value = session
|
||||
query = MagicMock()
|
||||
session.query.return_value = query
|
||||
filter_result = MagicMock()
|
||||
query.filter.return_value = filter_result
|
||||
in_filter = MagicMock()
|
||||
filter_result.filter.return_value = in_filter
|
||||
assets = [self._make_asset("a1", "video/a1.mp4")]
|
||||
in_filter.order_by.return_value.all.return_value = assets
|
||||
|
||||
mock_download.return_value = True
|
||||
|
||||
with patch("worker_app.tasks.generation.AssetModel", create=True):
|
||||
result = _download_library_assets(
|
||||
Path("/tmp/test"),
|
||||
asset_library_id="lib1",
|
||||
)
|
||||
|
||||
assert len(result) == 1
|
||||
mock_download.assert_called_once()
|
||||
|
||||
@patch("worker_app.tasks.generation.SessionLocal")
|
||||
@patch("worker_app.tasks.generation.download_asset")
|
||||
def test_project_mode(self, mock_download, mock_session_factory):
|
||||
"""项目级模式：asset_library_id 为空时按 project_id 查询."""
|
||||
from worker_app.tasks.generation import _download_library_assets
|
||||
|
||||
session = MagicMock()
|
||||
mock_session_factory.return_value = session
|
||||
query = MagicMock()
|
||||
session.query.return_value = query
|
||||
filter_result = MagicMock()
|
||||
query.filter.return_value = filter_result
|
||||
proj_filter = MagicMock()
|
||||
filter_result.filter.return_value = proj_filter
|
||||
assets = [self._make_asset("a1", "video/a1.mp4", project_id="proj1")]
|
||||
proj_filter.order_by.return_value.all.return_value = assets
|
||||
|
||||
mock_download.return_value = True
|
||||
|
||||
result = _download_library_assets(
|
||||
Path("/tmp/test"),
|
||||
project_id="proj1",
|
||||
)
|
||||
|
||||
assert len(result) == 1
|
||||
|
||||
def test_both_empty_raises(self):
|
||||
"""asset_library_id 和 project_id 都为空时抛 ValueError."""
|
||||
from worker_app.tasks.generation import _download_library_assets
|
||||
|
||||
with pytest.raises(ValueError, match="至少需要提供一个"):
|
||||
_download_library_assets(Path("/tmp/test"))
|
||||
|
||||
@patch("worker_app.tasks.generation.SessionLocal")
|
||||
def test_no_assets_found_raises(self, mock_session_factory):
|
||||
"""查不到素材时抛 RuntimeError."""
|
||||
from worker_app.tasks.generation import _download_library_assets
|
||||
|
||||
session = MagicMock()
|
||||
mock_session_factory.return_value = session
|
||||
query = MagicMock()
|
||||
session.query.return_value = query
|
||||
filter_result = MagicMock()
|
||||
query.filter.return_value = filter_result
|
||||
in_filter = MagicMock()
|
||||
filter_result.filter.return_value = in_filter
|
||||
in_filter.order_by.return_value.all.return_value = []
|
||||
|
||||
with pytest.raises(RuntimeError, match="未找到视频素材"):
|
||||
_download_library_assets(
|
||||
Path("/tmp/test"),
|
||||
asset_library_id="lib1",
|
||||
)
|
||||
|
||||
@patch("worker_app.tasks.generation.SessionLocal")
|
||||
@patch("worker_app.tasks.generation.download_asset")
|
||||
def test_all_asset_ids_fail_raises(self, mock_download, mock_session_factory):
|
||||
"""指定 asset_ids 但全部下载失败时抛 RuntimeError."""
|
||||
from worker_app.tasks.generation import _download_library_assets
|
||||
|
||||
session = MagicMock()
|
||||
mock_session_factory.return_value = session
|
||||
query = MagicMock()
|
||||
session.query.return_value = query
|
||||
filter_result = MagicMock()
|
||||
query.filter.return_value = filter_result
|
||||
id_filter = MagicMock()
|
||||
filter_result.filter.return_value = id_filter
|
||||
assets = [self._make_asset("a1", "video/a1.mp4")]
|
||||
id_filter.order_by.return_value.all.return_value = assets
|
||||
|
||||
mock_download.return_value = False # 全部下载失败
|
||||
|
||||
with pytest.raises(RuntimeError, match="素材下载失败"):
|
||||
_download_library_assets(
|
||||
Path("/tmp/test"),
|
||||
asset_library_id="lib1",
|
||||
asset_ids=["a1"],
|
||||
)
|
||||
|
||||
|
||||
# ── P0-2: OSS 上传 + URL 校验 ────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestOSSUploadAndVerify:
|
||||
"""P0-2: OSS 上传失败抛异常 + URL 可访问性校验."""
|
||||
|
||||
@@ -290,318 +165,3 @@ class TestP1Validations:
|
||||
|
||||
|
||||
# ── P1: 一键生成 clip 级效果层映射 ───────────────────────────────────────────
|
||||
|
||||
|
||||
class TestTemplateClipEffectMapping:
|
||||
"""P1: 模板 clip 级效果层映射到一键生成素材 clips."""
|
||||
|
||||
def _make_virtual_clip(self, idx: int, clip_type: str = "main", config: dict | None = None):
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
_clip_type_val = clip_type
|
||||
|
||||
@dataclass
|
||||
class FakeClip:
|
||||
id: str = f"vc_{idx:03d}"
|
||||
plan_id: str = "task_001"
|
||||
clip_type: str = _clip_type_val
|
||||
order: int = idx
|
||||
asset_id: str = f"asset_{idx}"
|
||||
duration: float = 5.0
|
||||
transition_effect: str = "cut"
|
||||
transition_duration: float = 0.0
|
||||
playback_speed: float = 1.0
|
||||
config: dict = field(default_factory=dict)
|
||||
|
||||
return FakeClip(config=config or {})
|
||||
|
||||
def _make_template_clip_config(self, clip_type: str = "main", transition: str = "cut", config: dict | None = None):
|
||||
mock = MagicMock()
|
||||
mock.clip_type = clip_type
|
||||
mock.transition_effect = transition
|
||||
mock.config = config or {}
|
||||
mock.default_duration = 3.0
|
||||
mock.text_template = ""
|
||||
return mock
|
||||
|
||||
def test_transition_effect_mapped(self):
|
||||
"""转场效果正确映射到素材 clips."""
|
||||
from worker_app.tasks.generation import _apply_template_clip_effects
|
||||
|
||||
clips = [self._make_virtual_clip(i) for i in range(3)]
|
||||
clip_configs = [
|
||||
self._make_template_clip_config("main", transition="fade"),
|
||||
self._make_template_clip_config("main", transition="dissolve"),
|
||||
]
|
||||
|
||||
_apply_template_clip_effects(clips, clip_configs, "one_take")
|
||||
|
||||
# 前两个按顺序映射，第三个用最后一个模板配置
|
||||
assert clips[0].transition_effect == "fade"
|
||||
assert clips[1].transition_effect == "dissolve"
|
||||
assert clips[2].transition_effect == "dissolve" # 复用最后一个
|
||||
|
||||
def test_color_grade_mapped(self):
|
||||
"""滤镜配置正确映射到 clip.config.color_grade."""
|
||||
from worker_app.tasks.generation import _apply_template_clip_effects
|
||||
|
||||
clips = [self._make_virtual_clip(i) for i in range(2)]
|
||||
clip_configs = [
|
||||
self._make_template_clip_config(
|
||||
"main", config={"color_grade": {"enabled": True, "filter": "vintage", "brightness": 0.1}}
|
||||
),
|
||||
]
|
||||
|
||||
_apply_template_clip_effects(clips, clip_configs, "one_take")
|
||||
|
||||
assert clips[0].config["color_grade"]["filter"] == "vintage"
|
||||
assert clips[0].config["color_grade"]["brightness"] == 0.1
|
||||
# 第二个素材复用第一个模板配置
|
||||
assert clips[1].config["color_grade"]["filter"] == "vintage"
|
||||
|
||||
def test_existing_config_preserved(self):
|
||||
"""已有 clip.config 内容（如 role）被保留."""
|
||||
from worker_app.tasks.generation import _apply_template_clip_effects
|
||||
|
||||
clips = [self._make_virtual_clip(0, config={"role": "b_roll"})]
|
||||
clip_configs = [
|
||||
self._make_template_clip_config("main", config={"color_grade": {"enabled": True, "filter": "warm"}}),
|
||||
]
|
||||
|
||||
_apply_template_clip_effects(clips, clip_configs, "voice_over")
|
||||
|
||||
assert clips[0].config["role"] == "b_roll" # 保留原有配置
|
||||
assert clips[0].config["color_grade"]["filter"] == "warm" # 新增滤镜配置
|
||||
|
||||
def test_empty_clip_configs_no_change(self):
|
||||
"""空模板配置时 clips 保持不变."""
|
||||
from worker_app.tasks.generation import _apply_template_clip_effects
|
||||
|
||||
clips = [self._make_virtual_clip(i) for i in range(2)]
|
||||
_apply_template_clip_effects(clips, [], "one_take")
|
||||
|
||||
assert clips[0].transition_effect == "cut"
|
||||
assert clips[1].transition_effect == "cut"
|
||||
|
||||
def test_cut_transition_not_overwritten(self):
|
||||
"""模板转场为 cut 时不覆盖（保持默认）."""
|
||||
from worker_app.tasks.generation import _apply_template_clip_effects
|
||||
|
||||
clips = [self._make_virtual_clip(0)]
|
||||
clips[0].transition_effect = "fade" # 已有非默认值
|
||||
clip_configs = [
|
||||
self._make_template_clip_config("main", transition="cut"),
|
||||
]
|
||||
|
||||
_apply_template_clip_effects(clips, clip_configs, "one_take")
|
||||
|
||||
# 模板是 cut 时，保留原有值（避免无意义覆盖）
|
||||
assert clips[0].transition_effect == "fade"
|
||||
|
||||
def test_transition_duration_mapped(self):
|
||||
"""转场时长（transition_duration）从模板 config 正确映射到 clip."""
|
||||
from worker_app.tasks.generation import _apply_template_clip_effects
|
||||
|
||||
clips = [self._make_virtual_clip(i) for i in range(3)]
|
||||
clip_configs = [
|
||||
self._make_template_clip_config("main", transition="fade", config={"transition_duration": 0.8}),
|
||||
self._make_template_clip_config("main", transition="dissolve", config={"transition_duration": 1.2}),
|
||||
]
|
||||
|
||||
_apply_template_clip_effects(clips, clip_configs, "one_take")
|
||||
|
||||
# 前两个按顺序映射，第三个复用最后一个
|
||||
assert clips[0].transition_effect == "fade"
|
||||
assert clips[0].transition_duration == 0.8
|
||||
assert clips[1].transition_effect == "dissolve"
|
||||
assert clips[1].transition_duration == 1.2
|
||||
assert clips[2].transition_effect == "dissolve"
|
||||
assert clips[2].transition_duration == 1.2
|
||||
|
||||
def test_transition_duration_ignored_for_cut(self):
|
||||
"""模板转场为 cut 时，transition_duration 不生效（保持默认0）."""
|
||||
from worker_app.tasks.generation import _apply_template_clip_effects
|
||||
|
||||
clips = [self._make_virtual_clip(0)]
|
||||
clip_configs = [
|
||||
self._make_template_clip_config("main", transition="cut", config={"transition_duration": 0.5}),
|
||||
]
|
||||
|
||||
_apply_template_clip_effects(clips, clip_configs, "one_take")
|
||||
|
||||
# cut 转场不映射，transition_duration 也不应用
|
||||
assert clips[0].transition_duration == 0.0
|
||||
|
||||
def test_transition_duration_invalid_value_skipped(self):
|
||||
"""transition_duration 为无效值时安全跳过."""
|
||||
from worker_app.tasks.generation import _apply_template_clip_effects
|
||||
|
||||
clips = [self._make_virtual_clip(0)]
|
||||
clip_configs = [
|
||||
self._make_template_clip_config("main", transition="fade", config={"transition_duration": "abc"}),
|
||||
]
|
||||
|
||||
_apply_template_clip_effects(clips, clip_configs, "one_take")
|
||||
|
||||
assert clips[0].transition_effect == "fade"
|
||||
assert clips[0].transition_duration == 0.0 # 无效值保持默认
|
||||
|
||||
def test_intro_outro_extracted(self):
|
||||
"""intro/outro 类型 clip_config 正确提取为 plan 级 intro_outro 配置."""
|
||||
from worker_app.tasks.generation import _extract_intro_outro_from_clip_configs
|
||||
|
||||
clip_configs = [
|
||||
self._make_template_clip_config("intro", config={"intro_type": "text", "intro_text_color": "#ffffff"}),
|
||||
self._make_template_clip_config("main"),
|
||||
self._make_template_clip_config("outro", config={"outro_type": "follow", "outro_follow_text": "关注我们"}),
|
||||
]
|
||||
# 设置 intro/outro 的 text_template
|
||||
clip_configs[0].text_template = "精彩视频"
|
||||
clip_configs[0].default_duration = 2.5
|
||||
|
||||
result = _extract_intro_outro_from_clip_configs(clip_configs)
|
||||
|
||||
assert result["has_intro"] is True
|
||||
assert result["intro_type"] == "text"
|
||||
assert result["intro_text"] == "精彩视频"
|
||||
assert result["intro_duration"] == 2.5
|
||||
assert result["intro_text_color"] == "#ffffff"
|
||||
assert result["has_outro"] is True
|
||||
assert result["outro_type"] == "follow"
|
||||
assert result["outro_follow_text"] == "关注我们"
|
||||
|
||||
def test_intro_outro_empty_when_none(self):
|
||||
"""没有 intro/outro 时返回空 dict."""
|
||||
from worker_app.tasks.generation import _extract_intro_outro_from_clip_configs
|
||||
|
||||
clip_configs = [
|
||||
self._make_template_clip_config("main"),
|
||||
self._make_template_clip_config("main"),
|
||||
]
|
||||
|
||||
result = _extract_intro_outro_from_clip_configs(clip_configs)
|
||||
assert result == {}
|
||||
|
||||
|
||||
class TestTemplatePlanConfigLoading:
|
||||
"""验证从模板加载 plan 级配置（BGM、字幕、标题）的逻辑。"""
|
||||
|
||||
def _mock_template(
|
||||
self,
|
||||
title_config=None,
|
||||
subtitle_config=None,
|
||||
bgm_config=None,
|
||||
is_active=True,
|
||||
):
|
||||
template = MagicMock()
|
||||
template.id = "tmpl_001"
|
||||
template.name = "Test Template"
|
||||
template.is_active = is_active
|
||||
template.title_config = title_config or {}
|
||||
template.subtitle_config = subtitle_config or {}
|
||||
template.bgm_config = bgm_config or {}
|
||||
return template
|
||||
|
||||
def _mock_session(self, template):
|
||||
session = MagicMock()
|
||||
|
||||
# EditTemplateModel 查询返回 None（走旧模板系统 fallback）
|
||||
edit_query = MagicMock()
|
||||
edit_filter = MagicMock()
|
||||
edit_query.filter.return_value = edit_filter
|
||||
edit_filter.first.return_value = None
|
||||
|
||||
# TemplateModel 查询返回 template（旧模板系统）
|
||||
old_query = MagicMock()
|
||||
old_filter = MagicMock()
|
||||
old_query.filter.return_value = old_filter
|
||||
old_filter.first.return_value = template
|
||||
|
||||
def _query_side_effect(model):
|
||||
name = getattr(model, "__name__", "")
|
||||
if "EditTemplate" in name:
|
||||
return edit_query
|
||||
return old_query
|
||||
|
||||
session.query.side_effect = _query_side_effect
|
||||
return session
|
||||
|
||||
def test_load_template_config_assembles_three_fields(self):
|
||||
"""模板的三个独立字段正确组装成 plan.config 格式。"""
|
||||
from worker_app.tasks.generation import _load_template_plan_config
|
||||
|
||||
title_cfg = {"enabled": True, "text": "我的标题", "font_size": 36}
|
||||
subtitle_cfg = {"enabled": True, "auto_generated": True, "language": "zh"}
|
||||
bgm_cfg = {"enabled": True, "preset_id": "bgm-001", "volume": 0.5}
|
||||
|
||||
template = self._mock_template(
|
||||
title_config=title_cfg,
|
||||
subtitle_config=subtitle_cfg,
|
||||
bgm_config=bgm_cfg,
|
||||
)
|
||||
session = self._mock_session(template)
|
||||
|
||||
with patch("worker_app.tasks.generation.SessionLocal", return_value=session):
|
||||
result = _load_template_plan_config("tmpl_001")
|
||||
|
||||
assert result["title"] == title_cfg
|
||||
assert result["subtitle"] == subtitle_cfg
|
||||
assert result["bgm"] == bgm_cfg
|
||||
|
||||
def test_load_template_config_empty_template_returns_empty(self):
|
||||
"""模板三个字段都为空时返回空 dict。"""
|
||||
from worker_app.tasks.generation import _load_template_plan_config
|
||||
|
||||
template = self._mock_template()
|
||||
session = self._mock_session(template)
|
||||
|
||||
with patch("worker_app.tasks.generation.SessionLocal", return_value=session):
|
||||
result = _load_template_plan_config("tmpl_001")
|
||||
|
||||
assert result == {}
|
||||
|
||||
def test_load_template_config_only_bgm(self):
|
||||
"""只有 BGM 配置时只返回 bgm 字段。"""
|
||||
from worker_app.tasks.generation import _load_template_plan_config
|
||||
|
||||
bgm_cfg = {"enabled": True, "audio_url": "https://example.com/bgm.mp3"}
|
||||
template = self._mock_template(bgm_config=bgm_cfg)
|
||||
session = self._mock_session(template)
|
||||
|
||||
with patch("worker_app.tasks.generation.SessionLocal", return_value=session):
|
||||
result = _load_template_plan_config("tmpl_001")
|
||||
|
||||
assert "bgm" in result
|
||||
assert result["bgm"] == bgm_cfg
|
||||
assert "title" not in result
|
||||
assert "subtitle" not in result
|
||||
|
||||
def test_load_template_config_empty_template_id(self):
|
||||
"""空 template_id 直接返回空 dict。"""
|
||||
from worker_app.tasks.generation import _load_template_plan_config
|
||||
|
||||
result = _load_template_plan_config("")
|
||||
assert result == {}
|
||||
|
||||
result = _load_template_plan_config(None)
|
||||
assert result == {}
|
||||
|
||||
def test_load_template_config_not_found_returns_empty(self):
|
||||
"""模板不存在时返回空 dict，不抛异常。"""
|
||||
from worker_app.tasks.generation import _load_template_plan_config
|
||||
|
||||
session = MagicMock()
|
||||
|
||||
def _query_side_effect(model):
|
||||
q = MagicMock()
|
||||
f = MagicMock()
|
||||
q.filter.return_value = f
|
||||
f.first.return_value = None
|
||||
return q
|
||||
|
||||
session.query.side_effect = _query_side_effect
|
||||
|
||||
with patch("worker_app.tasks.generation.SessionLocal", return_value=session):
|
||||
result = _load_template_plan_config("tmpl_nonexist")
|
||||
|
||||
assert result == {}
|
||||
|
||||
@@ -1,190 +0,0 @@
|
||||
"""Tests for preview title_config feature.
|
||||
|
||||
验证预览 API 的 title_config 字段和 Worker 的标题配置解析逻辑。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
class TestPreviewTitleConfigSchema:
|
||||
"""测试 CreatePreviewGenerationTaskRequest 的 title_config 字段."""
|
||||
|
||||
def test_title_config_default_empty(self):
|
||||
"""title_config 默认为空 dict."""
|
||||
from app.schemas.generation_task import CreatePreviewGenerationTaskRequest
|
||||
|
||||
req = CreatePreviewGenerationTaskRequest(
|
||||
template_id="test_template",
|
||||
asset_ids=["asset1"],
|
||||
)
|
||||
assert req.title_config == {}
|
||||
|
||||
def test_title_config_with_text(self):
|
||||
"""传入标题文本."""
|
||||
from app.schemas.generation_task import CreatePreviewGenerationTaskRequest
|
||||
|
||||
req = CreatePreviewGenerationTaskRequest(
|
||||
template_id="test_template",
|
||||
asset_ids=["asset1"],
|
||||
title_config={"text": "测试标题"},
|
||||
)
|
||||
assert req.title_config["text"] == "测试标题"
|
||||
|
||||
def test_title_config_with_full_style(self):
|
||||
"""传入完整标题样式配置."""
|
||||
from app.schemas.generation_task import CreatePreviewGenerationTaskRequest
|
||||
|
||||
config = {
|
||||
"text": "我的视频标题",
|
||||
"font": "思源黑体",
|
||||
"font_size": 48,
|
||||
"font_color": "#ffffff",
|
||||
"position": "top",
|
||||
"bold": True,
|
||||
"stroke": 2,
|
||||
"shadow": True,
|
||||
}
|
||||
req = CreatePreviewGenerationTaskRequest(
|
||||
template_id="test_template",
|
||||
asset_ids=["asset1"],
|
||||
title_config=config,
|
||||
)
|
||||
assert req.title_config["text"] == "我的视频标题"
|
||||
assert req.title_config["font_size"] == 48
|
||||
assert req.title_config["position"] == "top"
|
||||
|
||||
|
||||
class TestCommandTitleConfig:
|
||||
"""测试 CreateGenerationTaskCommand 的 title_config 字段."""
|
||||
|
||||
def test_command_has_title_config(self):
|
||||
"""Command 包含 title_config 字段."""
|
||||
from packages.application.generation_tasks import CreateGenerationTaskCommand
|
||||
|
||||
cmd = CreateGenerationTaskCommand(
|
||||
title_config={"text": "hello", "font_size": 32},
|
||||
)
|
||||
assert cmd.title_config["text"] == "hello"
|
||||
assert cmd.title_config["font_size"] == 32
|
||||
|
||||
def test_command_title_config_default_empty(self):
|
||||
"""Command 的 title_config 默认为空 dict."""
|
||||
from packages.application.generation_tasks import CreateGenerationTaskCommand
|
||||
|
||||
cmd = CreateGenerationTaskCommand()
|
||||
assert cmd.title_config == {}
|
||||
|
||||
|
||||
class TestWorkerTitleConfigParsing:
|
||||
"""测试 Worker 渲染时的标题配置解析逻辑."""
|
||||
|
||||
def test_json_format_parsing(self):
|
||||
"""JSON 格式的 custom_title 能正确解析."""
|
||||
config = {"text": "测试标题", "font_size": 48, "font_color": "#ff0000"}
|
||||
custom_title = json.dumps(config, ensure_ascii=False)
|
||||
|
||||
ct_stripped = custom_title.strip()
|
||||
parsed = None
|
||||
if ct_stripped.startswith("{"):
|
||||
try:
|
||||
parsed = json.loads(ct_stripped)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
parsed = None
|
||||
|
||||
assert parsed is not None
|
||||
assert parsed["text"] == "测试标题"
|
||||
assert parsed["font_size"] == 48
|
||||
|
||||
def test_plain_text_fallback(self):
|
||||
"""纯文本的 custom_title 不触发 JSON 解析."""
|
||||
custom_title = "简单的标题文字"
|
||||
|
||||
ct_stripped = custom_title.strip()
|
||||
parsed = None
|
||||
if ct_stripped.startswith("{"):
|
||||
try:
|
||||
parsed = json.loads(ct_stripped)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
parsed = None
|
||||
|
||||
assert parsed is None
|
||||
|
||||
def test_invalid_json_fallback(self):
|
||||
"""无效 JSON 的 custom_title 降级为纯文本."""
|
||||
custom_title = "{invalid json"
|
||||
|
||||
ct_stripped = custom_title.strip()
|
||||
parsed = None
|
||||
if ct_stripped.startswith("{"):
|
||||
try:
|
||||
parsed = json.loads(ct_stripped)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
parsed = None
|
||||
|
||||
assert parsed is None
|
||||
|
||||
def test_json_without_text_skipped(self):
|
||||
"""JSON 格式但缺少 text 字段时，跳过标题注入."""
|
||||
config = {"font_size": 48}
|
||||
custom_title = json.dumps(config, ensure_ascii=False)
|
||||
|
||||
ct_stripped = custom_title.strip()
|
||||
parsed = json.loads(ct_stripped)
|
||||
title_text = (parsed.get("text") or "").strip()
|
||||
|
||||
assert title_text == ""
|
||||
|
||||
def test_style_key_mapping(self):
|
||||
"""前端字段名正确映射到 ASS 字段名."""
|
||||
config = {
|
||||
"text": "标题",
|
||||
"font_size": 48,
|
||||
"font_color": "#ffffff",
|
||||
"font_preset": "思源黑体",
|
||||
}
|
||||
|
||||
style_keys = ["font", "font_size", "font_color", "position", "bold", "stroke", "shadow", "font_preset"]
|
||||
title_cfg = {}
|
||||
for key in style_keys:
|
||||
if key in config and config[key] is not None:
|
||||
mapped_key = {
|
||||
"font_size": "size",
|
||||
"font_color": "color",
|
||||
"font_preset": "font",
|
||||
}.get(key, key)
|
||||
title_cfg[mapped_key] = config[key]
|
||||
|
||||
assert title_cfg["size"] == 48
|
||||
assert title_cfg["color"] == "#ffffff"
|
||||
assert title_cfg["font"] == "思源黑体"
|
||||
|
||||
|
||||
class TestPreviewRouteTitleConfigPassing:
|
||||
"""测试预览路由正确序列化 title_config 到 custom_title."""
|
||||
|
||||
def test_title_config_serialization(self):
|
||||
"""title_config 序列化为 JSON 字符串."""
|
||||
title_config = {
|
||||
"text": "我的标题",
|
||||
"font_size": 32,
|
||||
"font_color": "#d4a843",
|
||||
}
|
||||
serialized = json.dumps(title_config, ensure_ascii=False)
|
||||
|
||||
parsed = json.loads(serialized)
|
||||
assert parsed["text"] == "我的标题"
|
||||
assert parsed["font_size"] == 32
|
||||
|
||||
def test_empty_title_config_produces_empty_string(self):
|
||||
"""空 title_config 时 custom_title 为空字符串."""
|
||||
title_config = {}
|
||||
title_text = (title_config.get("text") or "").strip()
|
||||
custom_title_value = ""
|
||||
if title_text:
|
||||
custom_title_value = json.dumps(title_config, ensure_ascii=False)
|
||||
|
||||
assert custom_title_value == ""
|
||||
@@ -1,256 +0,0 @@
|
||||
"""预览视频标题渲染修复测试 — 覆盖3个断点。
|
||||
|
||||
断点1: generate_video() → _render_video() 传递 custom_title
|
||||
断点2: _render_video() 解析 custom_title 并注入 virtual_plan.config["title"]
|
||||
断点3: generate_ass_from_timeline() ASR路径也渲染标题
|
||||
"""
|
||||
|
||||
import json
|
||||
import time
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, PropertyMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
# ── 断点2: _render_video 标题注入 ─────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestRenderVideoCustomTitleInjection:
|
||||
"""验证 _render_video 正确接收并注入 custom_title 到 virtual_plan.config['title']。"""
|
||||
|
||||
@pytest.fixture
|
||||
def sample_custom_title(self):
|
||||
"""模拟前端发送的 custom_title JSON（含 font_size/font_color）。"""
|
||||
return json.dumps(
|
||||
{
|
||||
"text": "测试标题",
|
||||
"font": "思源黑体",
|
||||
"font_size": 30,
|
||||
"font_color": "#FF0000",
|
||||
"position": "top",
|
||||
"bold": True,
|
||||
"stroke": True,
|
||||
"shadow": False,
|
||||
},
|
||||
ensure_ascii=False,
|
||||
)
|
||||
|
||||
def _call_render_video_with_capture(self, custom_title, template_config=None, tmp_path=None):
|
||||
"""调用 _render_video，在 RenderAdapter 处中断并捕获 virtual_plan.config。"""
|
||||
from worker_app.tasks.generation import _render_video
|
||||
|
||||
captured_config = {}
|
||||
|
||||
class FakePlan:
|
||||
def __init__(self):
|
||||
self.config = {}
|
||||
self.id = "test-plan"
|
||||
|
||||
fake_plan = FakePlan()
|
||||
|
||||
def capture_and_raise(*args, **kwargs):
|
||||
# 此时 title 已注入到 fake_plan.config
|
||||
captured_config.update(fake_plan.config or {})
|
||||
raise RuntimeError("STOP_HERE")
|
||||
|
||||
with (
|
||||
patch("worker_app.tasks.generation._build_plan_and_clips_from_task") as mock_build,
|
||||
patch("worker_app.tasks.generation._load_template_plan_config", return_value=template_config),
|
||||
patch("worker_app.tasks.generation.time.monotonic", side_effect=[0.0, 1.0]),
|
||||
patch("video_processing.render_adapter.RenderAdapter") as mock_adapter_cls,
|
||||
):
|
||||
|
||||
mock_build.return_value = (fake_plan, [], {})
|
||||
mock_adapter_cls.side_effect = capture_and_raise
|
||||
|
||||
with pytest.raises(RuntimeError, match="STOP_HERE"):
|
||||
_render_video(
|
||||
task_id="test-task",
|
||||
downloaded_videos=[tmp_path / "v1.mp4"] if tmp_path else [Path("/tmp/v1.mp4")],
|
||||
voice_path=None,
|
||||
editing_mode=MagicMock(value="one_take"),
|
||||
project_id="proj-1",
|
||||
template_id="tpl-1",
|
||||
user_id="user-1",
|
||||
temp_path=tmp_path or Path("/tmp"),
|
||||
output_name="test_output",
|
||||
resolution="1280x720",
|
||||
bgm_config={},
|
||||
voice_ids=[],
|
||||
custom_title=custom_title,
|
||||
)
|
||||
|
||||
return captured_config
|
||||
|
||||
def test_custom_title_injected_into_plan_config(self, sample_custom_title, tmp_path):
|
||||
"""custom_title JSON 应被解析并注入 virtual_plan.config['title']。"""
|
||||
config = self._call_render_video_with_capture(sample_custom_title, tmp_path=tmp_path)
|
||||
|
||||
assert "title" in config
|
||||
title_cfg = config["title"]
|
||||
assert title_cfg["text"] == "测试标题"
|
||||
# 字段归一化: font_size → size
|
||||
assert title_cfg["size"] == 30
|
||||
# 字段归一化: font_color → color
|
||||
assert title_cfg["color"] == "#FF0000"
|
||||
|
||||
def test_custom_title_overrides_template_title(self, sample_custom_title, tmp_path):
|
||||
"""用户自定义标题应覆盖模板默认标题。"""
|
||||
template_config = {"title": {"text": "模板默认标题", "size": 24}}
|
||||
config = self._call_render_video_with_capture(
|
||||
sample_custom_title, template_config=template_config, tmp_path=tmp_path
|
||||
)
|
||||
|
||||
# 用户标题应覆盖模板标题
|
||||
assert config["title"]["text"] == "测试标题"
|
||||
assert config["title"]["size"] == 30
|
||||
|
||||
def test_empty_custom_title_no_injection(self, tmp_path):
|
||||
"""空 custom_title 不应注入 title 字段。"""
|
||||
config = self._call_render_video_with_capture("", tmp_path=tmp_path)
|
||||
assert "title" not in config
|
||||
|
||||
def test_malformed_custom_title_gracefully_ignored(self, tmp_path):
|
||||
"""非法 JSON 不应崩溃，应跳过注入。"""
|
||||
config = self._call_render_video_with_capture("{invalid json!!!", tmp_path=tmp_path)
|
||||
assert "title" not in config
|
||||
|
||||
|
||||
# ── 断点3: generate_ass_from_timeline ASR路径支持标题 ──────────────────────────
|
||||
|
||||
|
||||
class TestGenerateAssFromTimelineWithTitle:
|
||||
"""验证 generate_ass_from_timeline 在有标题时生成包含 TitleStyle 的 ASS。"""
|
||||
|
||||
def test_title_included_in_ass_output(self, tmp_path):
|
||||
"""有 title_text 时，ASS 输出应包含 TitleStyle 和标题事件。"""
|
||||
from video_processing.subtitle_generator import generate_ass_from_timeline
|
||||
|
||||
from packages.domain.subtitle import SubtitleSegment, SubtitleTimeline
|
||||
|
||||
timeline = SubtitleTimeline(
|
||||
segments=[
|
||||
SubtitleSegment(start=0.0, end=2.0, text="你好世界"),
|
||||
]
|
||||
)
|
||||
|
||||
ass_path = tmp_path / "test.ass"
|
||||
result = generate_ass_from_timeline(
|
||||
ass_path,
|
||||
timeline,
|
||||
video_width=1280,
|
||||
video_height=720,
|
||||
video_duration=10.0,
|
||||
subtitle_config={"font": "思源黑体", "size": 24},
|
||||
title_text="我的标题",
|
||||
title_config={"font": "思源黑体", "size": 36, "color": "#FFFFFF", "position": "top"},
|
||||
)
|
||||
|
||||
content = result.read_text(encoding="utf-8")
|
||||
# 应包含 TitleStyle
|
||||
assert "TitleStyle" in content
|
||||
# 应包含标题文本
|
||||
assert "我的标题" in content
|
||||
# 也应包含 ASR 字幕
|
||||
assert "你好世界" in content
|
||||
|
||||
def test_no_title_no_title_style(self, tmp_path):
|
||||
"""无标题时，ASS 输出不应包含 TitleStyle。"""
|
||||
from video_processing.subtitle_generator import generate_ass_from_timeline
|
||||
|
||||
from packages.domain.subtitle import SubtitleSegment, SubtitleTimeline
|
||||
|
||||
timeline = SubtitleTimeline(
|
||||
segments=[
|
||||
SubtitleSegment(start=0.0, end=2.0, text="只有字幕"),
|
||||
]
|
||||
)
|
||||
|
||||
ass_path = tmp_path / "test.ass"
|
||||
result = generate_ass_from_timeline(
|
||||
ass_path,
|
||||
timeline,
|
||||
video_width=1280,
|
||||
video_height=720,
|
||||
video_duration=10.0,
|
||||
subtitle_config={},
|
||||
title_text="",
|
||||
title_config={},
|
||||
)
|
||||
|
||||
content = result.read_text(encoding="utf-8")
|
||||
assert "TitleStyle" not in content
|
||||
assert "只有字幕" in content
|
||||
|
||||
def test_title_field_normalization_in_ass(self, tmp_path):
|
||||
"""前端字段名 font_size/font_color 应被正确归一化。"""
|
||||
from video_processing.subtitle_generator import generate_ass_from_timeline
|
||||
|
||||
from packages.domain.subtitle import SubtitleSegment, SubtitleTimeline
|
||||
|
||||
timeline = SubtitleTimeline(segments=[SubtitleSegment(start=0.0, end=2.0, text="test")])
|
||||
|
||||
ass_path = tmp_path / "test.ass"
|
||||
result = generate_ass_from_timeline(
|
||||
ass_path,
|
||||
timeline,
|
||||
video_width=1280,
|
||||
video_height=720,
|
||||
video_duration=10.0,
|
||||
subtitle_config={},
|
||||
title_text="归一化测试",
|
||||
title_config={
|
||||
"font_size": 30, # 前端字段名
|
||||
"font_color": "#FF0000", # 前端字段名
|
||||
"position": "top",
|
||||
},
|
||||
)
|
||||
|
||||
content = result.read_text(encoding="utf-8")
|
||||
assert "TitleStyle" in content
|
||||
assert "归一化测试" in content
|
||||
|
||||
def test_title_boolean_stroke_shadow_compat(self, tmp_path):
|
||||
"""boolean stroke/shadow 应被兼容处理。"""
|
||||
from video_processing.subtitle_generator import generate_ass_from_timeline
|
||||
|
||||
from packages.domain.subtitle import SubtitleSegment, SubtitleTimeline
|
||||
|
||||
timeline = SubtitleTimeline(segments=[SubtitleSegment(start=0.0, end=2.0, text="test")])
|
||||
|
||||
ass_path = tmp_path / "test.ass"
|
||||
result = generate_ass_from_timeline(
|
||||
ass_path,
|
||||
timeline,
|
||||
video_width=1280,
|
||||
video_height=720,
|
||||
video_duration=10.0,
|
||||
subtitle_config={},
|
||||
title_text="描边测试",
|
||||
title_config={
|
||||
"size": 36,
|
||||
"stroke": True, # boolean
|
||||
"shadow": False, # boolean
|
||||
},
|
||||
)
|
||||
|
||||
content = result.read_text(encoding="utf-8")
|
||||
assert "TitleStyle" in content
|
||||
assert "描边测试" in content
|
||||
|
||||
|
||||
# ── 断点1: _render_video 签名包含 custom_title ────────────────────────────────
|
||||
|
||||
|
||||
class TestRenderVideoSignature:
|
||||
"""验证 _render_video 函数签名正确。"""
|
||||
|
||||
def test_custom_title_parameter_exists(self):
|
||||
"""_render_video 应有 custom_title 参数，默认空字符串。"""
|
||||
import inspect
|
||||
|
||||
from worker_app.tasks.generation import _render_video
|
||||
|
||||
sig = inspect.signature(_render_video)
|
||||
assert "custom_title" in sig.parameters
|
||||
assert sig.parameters["custom_title"].default == ""
|
||||
@@ -1,151 +0,0 @@
|
||||
"""Tests for voice_ids fallback in _download_all_assets.
|
||||
|
||||
When voice_library_id is empty but voice_ids is non-empty, the Worker
|
||||
should fallback to voice_ids[0] as the audio asset_id.
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
class TestDownloadAllAssetsVoiceIdsFallback:
|
||||
"""_download_all_assets 配音下载 fallback 逻辑测试。"""
|
||||
|
||||
@patch("worker_app.tasks.generation._download_voice_asset")
|
||||
@patch("worker_app.tasks.generation._download_library_assets")
|
||||
def test_voice_library_id_takes_priority(self, mock_download_videos, mock_download_voice, tmp_path):
|
||||
"""voice_library_id 存在时优先使用，不 fallback 到 voice_ids。"""
|
||||
from worker_app.tasks.generation import _download_all_assets
|
||||
|
||||
mock_download_videos.return_value = [tmp_path / "v1.mp4"]
|
||||
mock_download_voice.return_value = True
|
||||
|
||||
videos, audio = _download_all_assets(
|
||||
temp_path=tmp_path,
|
||||
asset_library_id="lib-1",
|
||||
project_id="proj-1",
|
||||
task_asset_ids=["a1"],
|
||||
voice_library_id="voice-lib-123",
|
||||
task_id="task-1",
|
||||
voice_ids=["voice-ids-456"],
|
||||
)
|
||||
|
||||
assert audio is not None
|
||||
mock_download_voice.assert_called_once()
|
||||
call_args = mock_download_voice.call_args
|
||||
assert call_args[0][0] == "voice-lib-123" # first positional arg
|
||||
|
||||
@patch("worker_app.tasks.generation._download_voice_asset")
|
||||
@patch("worker_app.tasks.generation._download_library_assets")
|
||||
def test_fallback_to_voice_ids_when_voice_library_id_empty(
|
||||
self, mock_download_videos, mock_download_voice, tmp_path
|
||||
):
|
||||
"""voice_library_id 为空时 fallback 到 voice_ids[0]。"""
|
||||
from worker_app.tasks.generation import _download_all_assets
|
||||
|
||||
mock_download_videos.return_value = [tmp_path / "v1.mp4"]
|
||||
mock_download_voice.return_value = True
|
||||
|
||||
videos, audio = _download_all_assets(
|
||||
temp_path=tmp_path,
|
||||
asset_library_id="lib-1",
|
||||
project_id="proj-1",
|
||||
task_asset_ids=["a1"],
|
||||
voice_library_id="", # 空字符串
|
||||
task_id="task-2",
|
||||
voice_ids=["voice-asset-789"],
|
||||
)
|
||||
|
||||
assert audio is not None
|
||||
mock_download_voice.assert_called_once()
|
||||
call_args = mock_download_voice.call_args
|
||||
assert call_args[0][0] == "voice-asset-789"
|
||||
|
||||
@patch("worker_app.tasks.generation._download_voice_asset")
|
||||
@patch("worker_app.tasks.generation._download_library_assets")
|
||||
def test_no_audio_when_both_empty(self, mock_download_videos, mock_download_voice, tmp_path):
|
||||
"""voice_library_id 和 voice_ids 都为空时，不下载音频。"""
|
||||
from worker_app.tasks.generation import _download_all_assets
|
||||
|
||||
mock_download_videos.return_value = [tmp_path / "v1.mp4"]
|
||||
|
||||
videos, audio = _download_all_assets(
|
||||
temp_path=tmp_path,
|
||||
asset_library_id="lib-1",
|
||||
project_id="proj-1",
|
||||
task_asset_ids=["a1"],
|
||||
voice_library_id="",
|
||||
task_id="task-3",
|
||||
voice_ids=[],
|
||||
)
|
||||
|
||||
assert audio is None
|
||||
mock_download_voice.assert_not_called()
|
||||
|
||||
@patch("worker_app.tasks.generation._download_voice_asset")
|
||||
@patch("worker_app.tasks.generation._download_library_assets")
|
||||
def test_no_audio_when_voice_ids_none(self, mock_download_videos, mock_download_voice, tmp_path):
|
||||
"""voice_ids 为 None 时，不触发 fallback。"""
|
||||
from worker_app.tasks.generation import _download_all_assets
|
||||
|
||||
mock_download_videos.return_value = [tmp_path / "v1.mp4"]
|
||||
|
||||
videos, audio = _download_all_assets(
|
||||
temp_path=tmp_path,
|
||||
asset_library_id="lib-1",
|
||||
project_id="proj-1",
|
||||
task_asset_ids=["a1"],
|
||||
voice_library_id="",
|
||||
task_id="task-4",
|
||||
voice_ids=None,
|
||||
)
|
||||
|
||||
assert audio is None
|
||||
mock_download_voice.assert_not_called()
|
||||
|
||||
@patch("worker_app.tasks.generation._download_voice_asset")
|
||||
@patch("worker_app.tasks.generation._download_library_assets")
|
||||
def test_voice_library_id_empty_string_fallback(self, mock_download_videos, mock_download_voice, tmp_path):
|
||||
"""voice_library_id 为空字符串且 voice_ids 有多个元素时，取第一个。"""
|
||||
from worker_app.tasks.generation import _download_all_assets
|
||||
|
||||
mock_download_videos.return_value = [tmp_path / "v1.mp4"]
|
||||
mock_download_voice.return_value = True
|
||||
|
||||
videos, audio = _download_all_assets(
|
||||
temp_path=tmp_path,
|
||||
asset_library_id="lib-1",
|
||||
project_id="proj-1",
|
||||
task_asset_ids=["a1"],
|
||||
voice_library_id="",
|
||||
task_id="task-5",
|
||||
voice_ids=["first-id", "second-id", "third-id"],
|
||||
)
|
||||
|
||||
assert audio is not None
|
||||
call_args = mock_download_voice.call_args
|
||||
assert call_args[0][0] == "first-id"
|
||||
|
||||
@patch("worker_app.tasks.generation._download_voice_asset")
|
||||
@patch("worker_app.tasks.generation._download_library_assets")
|
||||
def test_backward_compat_no_voice_ids_param(self, mock_download_videos, mock_download_voice, tmp_path):
|
||||
"""不传 voice_ids 参数时，行为与之前一致（向后兼容）。"""
|
||||
from worker_app.tasks.generation import _download_all_assets
|
||||
|
||||
mock_download_videos.return_value = [tmp_path / "v1.mp4"]
|
||||
mock_download_voice.return_value = True
|
||||
|
||||
# 不传 voice_ids
|
||||
videos, audio = _download_all_assets(
|
||||
temp_path=tmp_path,
|
||||
asset_library_id="lib-1",
|
||||
project_id="proj-1",
|
||||
task_asset_ids=["a1"],
|
||||
voice_library_id="voice-lib-999",
|
||||
task_id="task-6",
|
||||
)
|
||||
|
||||
assert audio is not None
|
||||
mock_download_voice.assert_called_once_with("voice-lib-999", tmp_path / "voice.mp3")
|
||||
@@ -280,7 +280,7 @@ class TestGenerateAssSubtitles:
|
||||
content = output.read_text(encoding="utf-8")
|
||||
assert "有字幕" in content
|
||||
|
||||
def test_custom_title_color(self, tmp_path):
|
||||
def test_title_color_overlay(self, tmp_path):
|
||||
"""自定义标题颜色."""
|
||||
output = tmp_path / "color.ass"
|
||||
result = generate_ass_subtitles(
|
||||
|
||||
@@ -214,7 +214,6 @@ class TestPlanIdFallbackExecution:
|
||||
t.output_width = 1280
|
||||
t.output_height = 720
|
||||
t.cover_url = ""
|
||||
t.custom_title = ""
|
||||
t.title_config = {}
|
||||
t.logs = "[]"
|
||||
t.status = "pending"
|
||||
@@ -255,7 +254,6 @@ class TestPlanIdFallbackExecution:
|
||||
req.output_width = 0
|
||||
req.output_height = 0
|
||||
req.cover_url = ""
|
||||
req.custom_title = ""
|
||||
req.title_config = {}
|
||||
req.project_id = None
|
||||
req.asset_library_id = None
|
||||
@@ -295,7 +293,6 @@ class TestPlanIdFallbackExecution:
|
||||
output_width=1280,
|
||||
output_height=720,
|
||||
cover_url="",
|
||||
custom_title="",
|
||||
title_config={},
|
||||
logs="[]",
|
||||
status="pending",
|
||||
|
||||
@@ -46,13 +46,10 @@ class TestAssetAnalysesDoesNotOverwriteStatus:
|
||||
"Use an independent session to update only extra_meta."
|
||||
)
|
||||
|
||||
def test_asset_analyses_uses_independent_session(self):
|
||||
"""asset_analyses 持久化必须用独立 session 查询最新模型再提交。"""
|
||||
def test_no_stale_repo_update(self):
|
||||
"""旧路径的 _repo.update(gen_task) 模式已随 DEPRECATED 代码一起删除。"""
|
||||
source = _read_source()
|
||||
assert "_meta_session" in source
|
||||
assert "GenerationTaskModel" in source
|
||||
# 必须只更新 extra_meta 字段
|
||||
assert 'existing["asset_analyses"]' in source
|
||||
assert "_repo.update(gen_task)" not in source
|
||||
|
||||
|
||||
class TestGenerationTaskModelOrmAttribute:
|
||||
|
||||
Reference in New Issue
Block a user