diff --git a/tests/unit/test_preview_generation_fixes.py b/tests/unit/test_preview_generation_fixes.py deleted file mode 100644 index 9705a8edf..000000000 --- a/tests/unit/test_preview_generation_fixes.py +++ /dev/null @@ -1,409 +0,0 @@ -"""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 os - import sys - - 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 时应被截断。""" - # 创建临时假视频文件 - import tempfile - - from worker_app.tasks.generation import _build_plan_and_clips_from_task - - 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 时不截断。""" - import tempfile - - from worker_app.tasks.generation import _build_plan_and_clips_from_task - - 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 时不截断。""" - import tempfile - - from worker_app.tasks.generation import _build_plan_and_clips_from_task - - 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 参数时，使用传入的分辨率。""" - import tempfile - - from worker_app.tasks.generation import _render_video - - # 验证逻辑：检查 _render_video 在 is_preview + resolution 时的行为 - # 由于 _render_video 内部会调用 RenderAdapter，这里只验证分辨率配置逻辑 - # 通过 mock 掉渲染部分，检查 export_cfg - - 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。""" - import tempfile - - from worker_app.tasks.generation import _render_video - - 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"