diff --git a/tests/unit/test_preview_generation_fixes.py b/tests/unit/test_preview_generation_fixes.py new file mode 100644 index 000000000..23781db72 --- /dev/null +++ b/tests/unit/test_preview_generation_fixes.py @@ -0,0 +1,410 @@ +"""Tests for preview generation fixes: URL signing, duration capping, resolution.""" + +from __future__ import annotations + +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + + +# ── Fix 1: URL 签名 ────────────────────────────────────────────────────────── + + +class TestSignVideoUrl: + """_sign_video_url 单元测试。""" + + def test_empty_url_returns_empty(self): + """空 URL 直接返回空字符串。""" + from app.api.routes.generation_preview import _sign_video_url + + assert _sign_video_url("") == "" + + def test_signs_oss_url(self): + """OSS URL 应被签名。""" + from app.api.routes.generation_preview import _sign_video_url + + mock_storage = MagicMock() + mock_storage.get_download_url.return_value = "https://signed-url.example.com/video.mp4?sig=abc" + + with patch("app.api.routes.generation_preview.get_storage_service", return_value=mock_storage): + result = _sign_video_url("https://bucket.oss-cn-hangzhou.aliyuncs.com/generated/video.mp4") + + assert result == "https://signed-url.example.com/video.mp4?sig=abc" + mock_storage.get_download_url.assert_called_once_with( + "https://bucket.oss-cn-hangzhou.aliyuncs.com/generated/video.mp4", + expires_seconds=7200, + ) + + def test_fallback_on_sign_failure(self): + """签名失败时降级返回原始 URL。""" + from app.api.routes.generation_preview import _sign_video_url + + mock_storage = MagicMock() + mock_storage.get_download_url.side_effect = Exception("OSS not configured") + + with patch("app.api.routes.generation_preview.get_storage_service", return_value=mock_storage): + result = _sign_video_url("https://bucket.oss.example.com/video.mp4") + + assert result == "https://bucket.oss.example.com/video.mp4" + + def test_fallback_on_storage_error(self): + """get_storage_service 抛异常时降级返回原始 URL。""" + from app.api.routes.generation_preview import _sign_video_url + + with patch("app.api.routes.generation_preview.get_storage_service", side_effect=RuntimeError("no storage")): + result = _sign_video_url("https://bucket.oss.example.com/video.mp4") + + assert result == "https://bucket.oss.example.com/video.mp4" + + def test_sign_returns_none_fallback(self): + """get_download_url 返回 None 时降级返回原始 URL。""" + from app.api.routes.generation_preview import _sign_video_url + + mock_storage = MagicMock() + mock_storage.get_download_url.return_value = None + + with patch("app.api.routes.generation_preview.get_storage_service", return_value=mock_storage): + result = _sign_video_url("https://bucket.oss.example.com/video.mp4") + + assert result == "https://bucket.oss.example.com/video.mp4" + + +# ── Fix 2: 模板 segment 时长约束 ───────────────────────────────────────────── + + +class TestLoadTemplateSegmentDurations: + """_load_template_segment_durations 单元测试。""" + + def test_empty_template_id_returns_empty(self): + """空 template_id 返回空列表。""" + import sys + import os + + sys.path.insert(0, os.path.join(os.path.dirname(__file__), "apps", "worker")) + from worker_app.tasks.generation import _load_template_segment_durations + + assert _load_template_segment_durations("") == [] + + def test_loads_durations_ordered(self): + """按 segment_order 排序返回 duration_max 列表。""" + from worker_app.tasks.generation import _load_template_segment_durations + + mock_segment1 = MagicMock() + mock_segment1.duration_max = 5.0 + mock_segment2 = MagicMock() + mock_segment2.duration_max = 8.0 + mock_segment3 = MagicMock() + mock_segment3.duration_max = 3.0 + + mock_query = MagicMock() + mock_query.filter.return_value.order_by.return_value.all.return_value = [ + mock_segment1, + mock_segment2, + mock_segment3, + ] + + mock_session = MagicMock() + mock_session.query.return_value = mock_query + + mock_session_local = MagicMock(return_value=mock_session) + + with patch("worker_app.tasks.generation.SessionLocal", mock_session_local): + result = _load_template_segment_durations("tpl_123") + + assert result == [5.0, 8.0, 3.0] + + def test_filters_zero_durations(self): + """duration_max <= 0 的 segment 被过滤。""" + from worker_app.tasks.generation import _load_template_segment_durations + + mock_seg_valid = MagicMock() + mock_seg_valid.duration_max = 5.0 + mock_seg_zero = MagicMock() + mock_seg_zero.duration_max = 0.0 + mock_seg_neg = MagicMock() + mock_seg_neg.duration_max = -1.0 + + mock_query = MagicMock() + mock_query.filter.return_value.order_by.return_value.all.return_value = [ + mock_seg_valid, + mock_seg_zero, + mock_seg_neg, + ] + + mock_session = MagicMock() + mock_session.query.return_value = mock_query + + with patch("worker_app.tasks.generation.SessionLocal", MagicMock(return_value=mock_session)): + result = _load_template_segment_durations("tpl_123") + + assert result == [5.0] + + def test_db_error_returns_empty(self): + """数据库异常返回空列表,不抛出。""" + from worker_app.tasks.generation import _load_template_segment_durations + + with patch("worker_app.tasks.generation.SessionLocal", side_effect=Exception("DB down")): + result = _load_template_segment_durations("tpl_123") + + assert result == [] + + +class TestDurationCappingInBuildPlan: + """_build_plan_and_clips_from_task 中时长约束的集成测试。""" + + def test_clips_capped_by_segment_max(self): + """clip 时长超过 segment duration_max 时应被截断。""" + from worker_app.tasks.generation import _build_plan_and_clips_from_task + + # 创建临时假视频文件 + import tempfile + + with tempfile.TemporaryDirectory() as tmpdir: + paths = [] + for i in range(3): + p = Path(tmpdir) / f"video_{i}.mp4" + p.write_bytes(b"\x00" * 100) # 假文件 + paths.append(p) + + # Mock probe_duration 返回很长的时长 + with patch("worker_app.tasks.generation.probe_duration", return_value=30.0): + # Mock segment durations: 5s, 4s, 3s + with patch( + "worker_app.tasks.generation._load_template_segment_durations", + return_value=[5.0, 4.0, 3.0], + ): + # Mock _load_template_clip_configs 返回空(跳过效果层) + with patch( + "worker_app.tasks.generation._load_template_clip_configs", + return_value=[], + ): + plan, clips, asset_map = _build_plan_and_clips_from_task( + task_id="test_task_123", + downloaded_paths=paths, + mode="one_take", + template_id="tpl_test", + ) + + # 每个 clip 的时长应被截断到对应 segment 的 duration_max + 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 时不截断。""" + from worker_app.tasks.generation import _build_plan_and_clips_from_task + + import tempfile + + with tempfile.TemporaryDirectory() as tmpdir: + paths = [Path(tmpdir) / "video_0.mp4"] + paths[0].write_bytes(b"\x00" * 100) + + 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=[], + ): + plan, clips, asset_map = _build_plan_and_clips_from_task( + task_id="test_task_456", + downloaded_paths=paths, + mode="one_take", + template_id="tpl_test", + ) + + # 3.0 < 5.0, 不应截断 + assert clips[0].duration == 3.0 + + def test_no_capping_without_template(self): + """无 template_id 时不截断。""" + from worker_app.tasks.generation import _build_plan_and_clips_from_task + + import tempfile + + with tempfile.TemporaryDirectory() as tmpdir: + paths = [Path(tmpdir) / "video_0.mp4"] + paths[0].write_bytes(b"\x00" * 100) + + with patch("worker_app.tasks.generation.probe_duration", return_value=30.0): + plan, clips, asset_map = _build_plan_and_clips_from_task( + task_id="test_task_789", + downloaded_paths=paths, + mode="one_take", + template_id="", + ) + + # 无模板,使用素材完整时长 + assert clips[0].duration == 30.0 + + +# ── Fix 3: 预览分辨率 ──────────────────────────────────────────────────────── + + +class TestPreviewResolution: + """_render_video 预览分辨率逻辑测试。""" + + def test_preview_uses_passed_resolution(self): + """is_preview=True 且有 resolution 参数时,使用传入的分辨率。""" + from worker_app.tasks.generation import _render_video + + # 验证逻辑:检查 _render_video 在 is_preview + resolution 时的行为 + # 由于 _render_video 内部会调用 RenderAdapter,这里只验证分辨率配置逻辑 + # 通过 mock 掉渲染部分,检查 export_cfg + + import tempfile + + with tempfile.TemporaryDirectory() as tmpdir: + temp_path = Path(tmpdir) + video_path = temp_path / "input.mp4" + video_path.write_bytes(b"\x00" * 100) + + # Mock probe_duration + with patch("worker_app.tasks.generation.probe_duration", return_value=5.0): + with patch("worker_app.tasks.generation._build_plan_and_clips_from_task") as mock_build: + mock_clip = MagicMock() + mock_clip.duration = 5.0 + mock_clip.id = "vc_000" + mock_clip.plan_id = "test" + mock_clip.clip_type = "main" + mock_clip.order = 0 + mock_clip.asset_id = "asset_0" + mock_clip.config = {} + + mock_plan = MagicMock() + mock_plan.config = {} + mock_plan.id = "test" + mock_plan.name = "test" + + mock_build.return_value = (mock_plan, [mock_clip], {"asset_0": video_path}) + + # Mock RenderAdapter + with patch("worker_app.tasks.generation.RenderAdapter") as MockAdapter: + mock_result = MagicMock() + mock_result.success = True + mock_result.output_path = temp_path / "output.mp4" + mock_result.output_path.write_bytes(b"\x00" * 100) + mock_result.duration = 5.0 + + mock_adapter_instance = MagicMock() + mock_adapter_instance.render_from_memory.return_value = mock_result + MockAdapter.return_value = mock_adapter_instance + + with patch("worker_app.tasks.generation.SessionLocal"): + try: + _render_video( + task_id="test_resolution", + downloaded_videos=[video_path], + voice_path=None, + editing_mode=MagicMock(value="one_take"), + project_id="", + template_id="", + user_id="", + temp_path=temp_path, + output_name="output.mp4", + resolution="480x854", + is_preview=True, + ) + except Exception: + pass # 可能会在其他地方失败,但我们只关心分辨率配置 + + # 检查传给 RenderAdapter 的 plan.config 中的分辨率 + if mock_adapter_instance.render_from_memory.called: + call_args = mock_adapter_instance.render_from_memory.call_args + plan_arg = call_args.kwargs.get("plan") or call_args[1].get("plan") + if plan_arg and hasattr(plan_arg, "config"): + export = (plan_arg.config or {}).get("export", {}) + assert export.get("resolution") == "480x854" + + def test_preview_defaults_to_landscape_when_no_resolution(self): + """is_preview=True 且无 resolution 参数时,默认 854x480。""" + from worker_app.tasks.generation import _render_video + + import tempfile + + with tempfile.TemporaryDirectory() as tmpdir: + temp_path = Path(tmpdir) + video_path = temp_path / "input.mp4" + video_path.write_bytes(b"\x00" * 100) + + with patch("worker_app.tasks.generation.probe_duration", return_value=5.0): + with patch("worker_app.tasks.generation._build_plan_and_clips_from_task") as mock_build: + mock_clip = MagicMock() + mock_clip.duration = 5.0 + mock_clip.config = {} + + mock_plan = MagicMock() + mock_plan.config = {} + mock_plan.id = "test" + mock_plan.name = "test" + + mock_build.return_value = (mock_plan, [mock_clip], {"": video_path}) + + with patch("worker_app.tasks.generation.RenderAdapter") as MockAdapter: + mock_result = MagicMock() + mock_result.success = True + mock_result.output_path = temp_path / "output.mp4" + mock_result.output_path.write_bytes(b"\x00" * 100) + mock_result.duration = 5.0 + + mock_adapter_instance = MagicMock() + mock_adapter_instance.render_from_memory.return_value = mock_result + MockAdapter.return_value = mock_adapter_instance + + with patch("worker_app.tasks.generation.SessionLocal"): + try: + _render_video( + task_id="test_resolution_default", + downloaded_videos=[video_path], + voice_path=None, + editing_mode=MagicMock(value="one_take"), + project_id="", + template_id="", + user_id="", + temp_path=temp_path, + output_name="output.mp4", + resolution="", + is_preview=True, + ) + except Exception: + pass + + if mock_adapter_instance.render_from_memory.called: + call_args = mock_adapter_instance.render_from_memory.call_args + plan_arg = call_args.kwargs.get("plan") or call_args[1].get("plan") + if plan_arg and hasattr(plan_arg, "config"): + export = (plan_arg.config or {}).get("export", {}) + assert export.get("resolution") == "854x480" + + +class TestCalcPreviewResolution: + """_calc_preview_resolution 单元测试。""" + + def test_portrait_9_16(self): + from app.api.routes.generation_preview import _calc_preview_resolution + + assert _calc_preview_resolution("9:16") == "480x854" + + def test_landscape_16_9(self): + from app.api.routes.generation_preview import _calc_preview_resolution + + assert _calc_preview_resolution("16:9") == "854x480" + + def test_square_1_1(self): + from app.api.routes.generation_preview import _calc_preview_resolution + + assert _calc_preview_resolution("1:1") == "480x480" + + def test_unknown_defaults_to_landscape(self): + from app.api.routes.generation_preview import _calc_preview_resolution + + assert _calc_preview_resolution("unknown") == "854x480" + + def test_empty_defaults_to_landscape(self): + from app.api.routes.generation_preview import _calc_preview_resolution + + assert _calc_preview_resolution("") == "854x480"