diff --git a/tests/unit/test_1294_preview_voice_injection.py b/tests/unit/test_1294_preview_voice_injection.py index 7158ae985..5d8c560f3 100644 --- a/tests/unit/test_1294_preview_voice_injection.py +++ b/tests/unit/test_1294_preview_voice_injection.py @@ -1,75 +1,93 @@ -"""测试 #1294 修复:预览视频配音注入。 +"""测试 #1294 修复:预览视频配音注入(#1749 后重写)。 -验证: -1. _load_task_info 正确加载 voice_ids -3. voice_ids 正确注入到 plan config 中(实际执行代码路径,diff-cover 可达) +#1749 起冗余 voice_ids 字段已从 task_info 移除(DB 列保留只读), +配音一律以 voice_library_id 为准(批量独立配音每变体各自绑定)。 +本测试断言: +1. _load_task_info 正确加载 voice_library_id; +2. task_info 不再包含 voice_ids 键; +3. 配音解析 effective_voice_id 直接取 voice_library_id,不再有 [0] 兜底。 """ -from pathlib import Path from unittest.mock import MagicMock, patch -import pytest + +def _mock_task(voice_library_id: str = "voice_lib_1"): + mock_task = MagicMock() + mock_task.project_id = "proj_1" + mock_task.asset_library_id = "lib_1" + mock_task.voice_library_id = voice_library_id + mock_task.template_id = "tmpl_1" + mock_task.strategy_id = "one_take" + mock_task.asset_ids = ["a1", "a2"] + mock_task.batch_id = "batch_1" + mock_task.created_by_user_id = "user_1" + mock_task.video_title = "test" + mock_task.resolution = "854x480" + mock_task.bgm_config = {} + mock_task.is_preview = True + mock_task.voice_ids = ["voice_1", "voice_2"] # DB 列仍存在但不再读 + mock_task.source_task_id = "" + mock_task.output_width = 854 + mock_task.output_height = 480 + mock_task.cover_url = "" + mock_task.title_config = {} + mock_task.source_edit_plan_id = "plan_1" + return mock_task -class TestLoadTaskInfoVoiceIds: - """验证 _load_task_info 包含 voice_ids""" +def test_voice_library_id_loaded_from_task(): + with patch( + "packages.adapters.sqlalchemy_impl.generation_task_repository.SQLAlchemyGenerationTaskRepository" + ) as MockRepo: + mock_repo = MagicMock() + mock_repo.get.return_value = _mock_task("voice_lib_abc") + MockRepo.return_value = mock_repo - def test_voice_ids_loaded_from_task(self): - """voice_ids 从 gen_task 正确加载""" - mock_task = MagicMock() - mock_task.project_id = "proj_1" - mock_task.asset_library_id = "lib_1" - mock_task.voice_library_id = "voice_lib_1" - mock_task.template_id = "tmpl_1" - mock_task.strategy_id = "one_take" - mock_task.asset_ids = ["a1", "a2"] - mock_task.batch_id = "batch_1" - mock_task.created_by_user_id = "user_1" - mock_task.video_title = "test" - mock_task.resolution = "854x480" - mock_task.bgm_config = {} - mock_task.is_preview = True - mock_task.voice_ids = ["voice_1", "voice_2"] + from worker_app.tasks.generation import _load_task_info - with patch( - "packages.adapters.sqlalchemy_impl.generation_task_repository.SQLAlchemyGenerationTaskRepository" - ) as MockRepo: - mock_repo = MagicMock() - mock_repo.get.return_value = mock_task - MockRepo.return_value = mock_repo + result = _load_task_info("test_task_id") - from worker_app.tasks.generation import _load_task_info + assert result is not None + assert result["voice_library_id"] == "voice_lib_abc" - result = _load_task_info("test_task_id") - assert result is not None - assert result["voice_ids"] == ["voice_1", "voice_2"] +def test_task_info_has_no_voice_ids_key(): + """#1749:冗余 voice_ids 已从 task_info 移除。""" + with patch( + "packages.adapters.sqlalchemy_impl.generation_task_repository.SQLAlchemyGenerationTaskRepository" + ) as MockRepo: + mock_repo = MagicMock() + mock_repo.get.return_value = _mock_task() + MockRepo.return_value = mock_repo - def test_voice_ids_empty_when_none(self): - """voice_ids 为 None 时返回空列表""" - mock_task = MagicMock() - mock_task.project_id = "proj_1" - mock_task.asset_library_id = "lib_1" - mock_task.voice_library_id = "" - mock_task.template_id = "tmpl_1" - mock_task.strategy_id = "one_take" - mock_task.asset_ids = ["a1"] - mock_task.batch_id = "" - mock_task.created_by_user_id = "user_1" - mock_task.video_title = "" - mock_task.resolution = "" - mock_task.bgm_config = {} - mock_task.is_preview = False - mock_task.voice_ids = None + from worker_app.tasks.generation import _load_task_info - with patch( - "packages.adapters.sqlalchemy_impl.generation_task_repository.SQLAlchemyGenerationTaskRepository" - ) as MockRepo: - mock_repo = MagicMock() - mock_repo.get.return_value = mock_task - MockRepo.return_value = mock_repo + result = _load_task_info("test_task_id") + assert "voice_ids" not in result - from worker_app.tasks.generation import _load_task_info - result = _load_task_info("test_task_id") - assert result["voice_ids"] == [] +def test_effective_voice_uses_voice_library_id_only(): + """配音解析:voice_library_id 即最终配音 ID,无 voice_ids[0] 兜底。""" + # 模拟 _sync_task_config_to_plan 中的解析逻辑 + task_info = {"voice_library_id": "ANDT_voice"} + voice_library_id = task_info.get("voice_library_id", "") + effective_voice_id = voice_library_id or "" + assert effective_voice_id == "ANDT_voice" + + # 空配音 + task_info2 = {"voice_library_id": ""} + assert (task_info2.get("voice_library_id", "") or "") == "" + + +def test_empty_voice_library_id(): + with patch( + "packages.adapters.sqlalchemy_impl.generation_task_repository.SQLAlchemyGenerationTaskRepository" + ) as MockRepo: + mock_repo = MagicMock() + mock_repo.get.return_value = _mock_task("") + MockRepo.return_value = mock_repo + + from worker_app.tasks.generation import _load_task_info + + result = _load_task_info("test_task_id") + assert result["voice_library_id"] == "" diff --git a/tests/unit/test_1677_batch_variants.py b/tests/unit/test_1677_batch_variants.py index c7c71f884..a792f3e0d 100644 --- a/tests/unit/test_1677_batch_variants.py +++ b/tests/unit/test_1677_batch_variants.py @@ -97,7 +97,7 @@ class TestVariantArrayValidation: """voice_library_ids 长度非法 → 报错""" from pydantic import ValidationError - with pytest.raises(ValidationError, match="voice_library_ids"): + with pytest.raises(ValidationError, match="配音|视频数量"): _make_preview_request(preview_count=4, voice_library_ids=["v1", "v2"]) def test_preview_empty_arrays_ok(self): @@ -202,17 +202,19 @@ class TestBatchPreviewRoute: with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc: reselect_results = [MagicMock(id=pid) for pid in reselect_plan_ids] MockPlanSvc.return_value.reselect_plan_for_variant.side_effect = reselect_results + MockPlanSvc.return_value.clone_plan_for_variant.return_value = MagicMock(id="clone_v0") + MockPlanSvc.return_value.get_asset_durations.return_value = {} create_preview_generation_task( _make_preview_request(preview_count=3), authenticated_user=_make_user(), generation_task_repository=repo, db=MagicMock(), ) - # 变体 1..N-1 各独立选片一次(共 2 次);count>1 不再走 clone + # #1749:变体0 clone(不污染源 plan);变体 1..N-1 各独立选片一次(共 2 次) assert MockPlanSvc.return_value.reselect_plan_for_variant.call_count == 2 - MockPlanSvc.return_value.clone_plan_for_variant.assert_not_called() - # 变体0保留源 plan;变体1/2 关联各自独立选出的 plan - assert tasks[0].source_edit_plan_id == "source_plan" + MockPlanSvc.return_value.clone_plan_for_variant.assert_called_once() + # 变体0 关联 clone plan;变体1/2 关联各自独立选出的 plan + assert tasks[0].source_edit_plan_id == "clone_v0" assert tasks[1].source_edit_plan_id == "reselect_1" assert tasks[2].source_edit_plan_id == "reselect_2" @@ -487,6 +489,8 @@ class TestBatchGenerationVariantConfig: MagicMock(id="reselect_1"), MagicMock(id="reselect_2"), ] + MockPlanSvc.return_value.clone_plan_for_variant.return_value = MagicMock(id="clone_v0") + MockPlanSvc.return_value.get_asset_durations.return_value = {} req = CreateGenerationTaskRequest( template_id="tpl_1", asset_ids=["a1"], @@ -498,7 +502,9 @@ class TestBatchGenerationVariantConfig: cover_urls=["http://c1", "http://c2", "http://c3"], ) resp = self._call_create_tasks(req) + # #1749:变体0 clone;变体1/2 reselect assert MockPlanSvc.return_value.reselect_plan_for_variant.call_count == 2 + MockPlanSvc.return_value.clone_plan_for_variant.assert_called_once() assert resp.total == 3 assert [c.title_config["text"] for c in captured] == ["成片标题1", "成片标题2", "成片标题3"] assert [c.voice_library_id for c in captured] == ["v1", "v2", "v3"] @@ -557,12 +563,14 @@ class TestBatchGenerationVariantConfig: MagicMock(id="reselect_1"), MagicMock(id="reselect_2"), ] + MockPlanSvc.return_value.clone_plan_for_variant.return_value = MagicMock(id="clone_v0") + MockPlanSvc.return_value.get_asset_durations.return_value = {} req = CreateGenerationTaskRequest( template_id="tpl_1", asset_ids=["a1"], count=3, source_edit_plan_id="source_plan", - voice_library_ids=["shared_voice"], + voice_library_id="shared_voice", cover_urls=["http://shared"], ) self._call_create_tasks(req) @@ -595,10 +603,12 @@ class TestBatchGenerationVariantConfig: MagicMock(id="reselect_1"), MagicMock(id="reselect_2"), ] + MockPlanSvc.return_value.clone_plan_for_variant.return_value = MagicMock(id="clone_v0") + MockPlanSvc.return_value.get_asset_durations.return_value = {} self._call_create_tasks(req, db_latest_plan=latest) - # 兜底 plan 被用作源;变体0关联兜底 plan,变体1/2关联 reselect plan - assert _execute.caps[0].source_edit_plan_id == "fallback_plan_id" + # #1749:变体0 关联 clone plan(源为兜底 plan),变体1/2关联 reselect plan + assert _execute.caps[0].source_edit_plan_id == "clone_v0" assert _execute.caps[1].source_edit_plan_id == "reselect_1" assert _execute.caps[2].source_edit_plan_id == "reselect_2" diff --git a/tests/unit/test_variant_plan_selector_1749.py b/tests/unit/test_variant_plan_selector_1749.py new file mode 100644 index 000000000..b5c0e7973 --- /dev/null +++ b/tests/unit/test_variant_plan_selector_1749.py @@ -0,0 +1,135 @@ +"""#1749 问题 C:跨视频素材级去重 + target_durations 测试。""" + +import random + +import pytest + +from packages.domain.variant_plan_selector import ( + BATCH_CLIP_OVERLAP_LIMIT, + _clip_overlap_ratio, + reselect_clips_for_variant, +) + + +def _src_clips(n=3, dur=5.0): + return [ + { + "order": i, + "asset_id": f"src{i}", + "start_time": 0.0, + "duration": dur, + "clip_type": "main", + "transition_effect": "cut", + "transition_duration": 0.0, + } + for i in range(n) + ] + + +def test_three_variants_three_clips_eleven_assets_no_cross_reuse(): + """3 变体 × 3 片段 / 11 素材:fresh 优先 → 跨变体零重复。""" + pool = [f"a{i}" for i in range(11)] + durs = {a: 30.0 for a in pool} + batch: dict = {} + used_per_variant = [] + for v in range(3): + clips = reselect_clips_for_variant( + _src_clips(3), pool, asset_durations=durs, batch_segments=batch, rng=random.Random(100 + v) + ) + assert len(clips) == 3 + used_per_variant.append({c["asset_id"] for c in clips}) + # 两两交集为空(9 个素材位置,11 素材足够 fresh 分配) + assert used_per_variant[0] & used_per_variant[1] == set() + assert used_per_variant[0] & used_per_variant[2] == set() + assert used_per_variant[1] & used_per_variant[2] == set() + + +def test_reuse_overlap_under_limit_and_no_full_overlap(): + """素材池不足被迫复用时:重叠 ≤20% 且不得完全重叠。""" + # 2 个长素材、3 变体 × 3 片段 → 必然复用 + pool = ["x", "y"] + durs = {"x": 60.0, "y": 60.0} + batch: dict = {} + for v in range(3): + reselect_clips_for_variant( + _src_clips(3, 5.0), pool, asset_durations=durs, batch_segments=batch, rng=random.Random(7 + v) + ) + # 校验 batch_segments 中同素材任意两区间重叠占比 + for asset, segs in batch.items(): + for i in range(len(segs)): + for j in range(i + 1, len(segs)): + s1, e1 = segs[i] + s2, e2 = segs[j] + ov = max(0.0, min(e1, e2) - max(s1, s2)) + seg_dur = min(e1 - s1, e2 - s2) + ratio = ov / seg_dur if seg_dur > 0 else 0.0 + assert ratio <= BATCH_CLIP_OVERLAP_LIMIT + 0.01, f"{asset} overlap {ratio}" + # 不得完全重叠 + assert not (abs(s1 - s2) < 0.01 and abs(e1 - e2) < 0.01), f"{asset} 完全重叠" + + +def test_short_asset_cannot_be_reused_across_variants(): + """短素材(时长 < 段长 80%)数学上无法错开 → 禁跨变体复用。""" + pool = ["short", "long1", "long2"] + durs = {"short": 6.2, "long1": 40.0, "long2": 40.0} + batch = {"short": [(0.0, 6.2)]} # 短素材已被变体0使用 + clips = reselect_clips_for_variant( + _src_clips(1, 10.0), pool, asset_durations=durs, batch_segments=batch, rng=random.Random(1) + ) + assert clips[0]["asset_id"] != "short" + + +def test_target_durations_applied_to_clips(): + """target_durations 落库到片段 duration。""" + pool = ["a", "b", "c"] + durs = {"a": 30.0, "b": 30.0, "c": 30.0} + clips = reselect_clips_for_variant( + _src_clips(3, 5.0), + pool, + asset_durations=durs, + batch_segments={}, + target_durations=[7.333, 7.333, 7.334], + rng=random.Random(3), + ) + durs_out = sorted(c["duration"] for c in clips) + assert durs_out == pytest.approx([7.333, 7.333, 7.334], abs=0.01) + + +def test_short_asset_freeze_start_zero(): + """素材短于目标段长:起点为 0(末帧冻结由渲染侧铺满),不报错。""" + pool = ["short6s"] + durs = {"short6s": 6.0} + clips = reselect_clips_for_variant( + _src_clips(1, 10.0), + pool, + asset_durations=durs, + batch_segments={}, + target_durations=[10.0], + rng=random.Random(5), + ) + assert clips[0]["asset_id"] == "short6s" + assert clips[0]["start_time"] == 0.0 + assert clips[0]["duration"] == pytest.approx(10.0) + + +def test_empty_pool_raises(): + with pytest.raises(ValueError): + reselect_clips_for_variant(_src_clips(2), [], asset_durations={}, rng=random.Random(1)) + + +def test_empty_source_raises(): + with pytest.raises(ValueError): + reselect_clips_for_variant([], ["a"], asset_durations={"a": 10.0}, rng=random.Random(1)) + + +def test_zero_duration_assets_raises(): + with pytest.raises(ValueError): + reselect_clips_for_variant(_src_clips(2), ["a", "b"], asset_durations={"a": 0.0, "b": 0.0}, rng=random.Random(1)) + + +def test_overlap_ratio_helper(): + batch = {"a": [(0.0, 5.0)]} + assert _clip_overlap_ratio("a", 0.0, 5.0, batch) == pytest.approx(1.0) + assert _clip_overlap_ratio("a", 5.0, 5.0, batch) == pytest.approx(0.0) + assert _clip_overlap_ratio("a", 4.0, 5.0, batch) == pytest.approx(0.2) + assert _clip_overlap_ratio("b", 0.0, 5.0, batch) == 0.0 diff --git a/tests/unit/test_variant_voice_resolver_1749.py b/tests/unit/test_variant_voice_resolver_1749.py new file mode 100644 index 000000000..4ad0bad2f --- /dev/null +++ b/tests/unit/test_variant_voice_resolver_1749.py @@ -0,0 +1,51 @@ +"""#1749 问题 A:变体配音严格守卫测试。""" + +import pytest + +from packages.domain.variant_voice_resolver import VariantVoiceError, resolve_variant_voice_ids + + +def test_unified_voice_repeated(): + assert resolve_variant_voice_ids(count=3, voice_library_id="V1") == ["V1", "V1", "V1"] + + +def test_independent_voices(): + assert resolve_variant_voice_ids(count=3, voice_library_ids=["a", "b", "c"]) == ["a", "b", "c"] + + +def test_no_voice_returns_empty_strings(): + assert resolve_variant_voice_ids(count=2) == ["", ""] + + +def test_independent_length_mismatch_raises(): + with pytest.raises(VariantVoiceError): + resolve_variant_voice_ids(count=3, voice_library_ids=["a", "b"]) + + +def test_independent_missing_value_raises(): + with pytest.raises(VariantVoiceError): + resolve_variant_voice_ids(count=3, voice_library_ids=["a", "", "c"]) + + +def test_independent_whitespace_missing_raises(): + with pytest.raises(VariantVoiceError): + resolve_variant_voice_ids(count=2, voice_library_ids=["a", " "]) + + +def test_count_zero_raises(): + with pytest.raises(VariantVoiceError): + resolve_variant_voice_ids(count=0) + + +def test_explicit_multi_does_not_fallback_to_single(): + """传了 voice_library_ids 但长度错 → 400,禁止静默 fallback 到单值。""" + with pytest.raises(VariantVoiceError): + resolve_variant_voice_ids(count=3, voice_library_id="SINGLE", voice_library_ids=["x", "y"]) + + +def test_single_voice_with_empty_multi_list(): + assert resolve_variant_voice_ids(count=2, voice_library_id="S", voice_library_ids=[]) == ["S", "S"] + + +def test_count_one_independent(): + assert resolve_variant_voice_ids(count=1, voice_library_ids=["solo"]) == ["solo"] diff --git a/tests/unit/test_voice_duration_alignment.py b/tests/unit/test_voice_duration_alignment.py index 11f3a5036..8606301e6 100644 --- a/tests/unit/test_voice_duration_alignment.py +++ b/tests/unit/test_voice_duration_alignment.py @@ -1,6 +1,12 @@ -"""Tests for voice duration alignment feature. +"""#1749 配音对齐行为测试(重写)。 -Tests the _align_clips_to_voice_duration method in UnifiedRenderService. +旧行为(已删除):渲染期全局等比裁剪片段 / 全局慢放补偿配音时长。 +新行为(定稿): +- 段长由 voice_duration_planner 在选片阶段精确分配,成片总时长=配音; +- 渲染期 _align_clips_to_voice_duration 仅保留 ±5% 守卫日志,**不修改任何 clip** + (不裁剪、不慢放); +- 素材短于段长 → freeze(config['_freeze_seconds'])+ 目标段长始终为准, + 由 tpad/apad 铺满。 """ from __future__ import annotations @@ -8,198 +14,111 @@ from __future__ import annotations from pathlib import Path from unittest.mock import MagicMock, patch -import pytest from video_processing.unified_render_service import RenderLayer, ResolvedClip, UnifiedRenderService -class TestAlignClipsToVoiceDuration: - """Test clip duration alignment to voice audio.""" - - def _make_clip( - self, - clip_id: str, - duration: float, - actual_duration: float = 0.0, - playback_speed: float = 1.0, - ) -> ResolvedClip: - """Helper to create a ResolvedClip for testing.""" - return ResolvedClip( - clip_id=clip_id, - asset_id=f"asset_{clip_id}", - local_path=Path(f"/tmp/{clip_id}.mp4"), - clip_type="main", - order=0, - duration=duration, - actual_duration=actual_duration or duration, - playback_speed=playback_speed, - ) - - def _make_layer(self, role: str, clips: list[ResolvedClip]) -> RenderLayer: - """Helper to create a RenderLayer for testing.""" - return RenderLayer(role=role, clips=clips, z_index=0) - - def _make_service(self, voiceover_path: str | None = None) -> UnifiedRenderService: - """Helper to create a mock UnifiedRenderService.""" - plan = MagicMock() - plan.id = "test_plan" - plan.config = {} - - with patch.object(UnifiedRenderService, "__init__", lambda self, **kwargs: None): - service = UnifiedRenderService.__new__(UnifiedRenderService) - service.plan = plan - service.voiceover_audio_path = voiceover_path - service.transition_duration = 0.0 - return service - - def test_no_voice_audio_no_adjustment(self): - """No voice audio → no adjustment.""" - service = self._make_service(voiceover_path=None) - clips = [self._make_clip("c1", 10.0), self._make_clip("c2", 10.0)] - layers = [self._make_layer("main", clips)] - - service._align_clips_to_voice_duration(layers, voice_duration=0.0) - - # No change - assert clips[0].duration == 10.0 - assert clips[1].duration == 10.0 - - def test_ratio_within_5_percent_no_adjustment(self): - """Ratio within ±5% → no adjustment.""" - service = self._make_service() - clips = [self._make_clip("c1", 10.0)] - layers = [self._make_layer("main", clips)] - - # Total clips = 10s, voice = 10.3s → ratio = 1.03 (within 5%) - service._align_clips_to_voice_duration(layers, voice_duration=10.3) - - assert clips[0].duration == 10.0 # Unchanged - - def test_ratio_less_than_1_trim_clips(self): - """Ratio < 1 (clips too long) → trim clips proportionally.""" - service = self._make_service() - clips = [self._make_clip("c1", 10.0), self._make_clip("c2", 10.0)] - layers = [self._make_layer("main", clips)] - - # Total clips = 20s, voice = 15s → ratio = 0.75 - service._align_clips_to_voice_duration(layers, voice_duration=15.0) - - # Each clip should be trimmed to 75% - assert abs(clips[0].duration - 7.5) < 0.01 - assert abs(clips[1].duration - 7.5) < 0.01 - - def test_ratio_greater_than_1_slowdown_clips(self): - """Ratio > 1 (clips too short) → slow down clips.""" - service = self._make_service() - clips = [self._make_clip("c1", 10.0), self._make_clip("c2", 10.0)] - layers = [self._make_layer("main", clips)] - - # Total clips = 20s, voice = 25s → ratio = 1.25 - service._align_clips_to_voice_duration(layers, voice_duration=25.0) - - # Each clip's speed should be reduced: 1.0 / 1.25 = 0.8 - assert abs(clips[0].playback_speed - 0.8) < 0.01 - assert abs(clips[1].playback_speed - 0.8) < 0.01 - - def test_speed_lower_bound_025(self): - """Playback speed should not go below 0.25x.""" - service = self._make_service() - clips = [self._make_clip("c1", 5.0)] - layers = [self._make_layer("main", clips)] - - # Total clips = 5s, voice = 50s → ratio = 10.0 - # Speed would be 1.0 / 10 = 0.1, but should be clamped to 0.25 - service._align_clips_to_voice_duration(layers, voice_duration=50.0) - - assert clips[0].playback_speed == 0.25 - - def test_only_video_layers_adjusted(self): - """Only main/broll/background layers are adjusted, not audio.""" - service = self._make_service() - - video_clips = [self._make_clip("v1", 10.0)] - audio_clips = [self._make_clip("a1", 10.0)] - - layers = [ - self._make_layer("main", video_clips), - self._make_layer("audio", audio_clips), - ] - - # ratio = 0.5 → should trim video but not audio - service._align_clips_to_voice_duration(layers, voice_duration=5.0) - - assert abs(video_clips[0].duration - 5.0) < 0.01 # Trimmed - assert audio_clips[0].duration == 10.0 # Unchanged - - def test_multiple_video_layers_all_adjusted(self): - """All video layers (main, broll, background) are adjusted.""" - service = self._make_service() - - main_clips = [self._make_clip("m1", 10.0)] - broll_clips = [self._make_clip("b1", 10.0)] - bg_clips = [self._make_clip("bg1", 10.0)] - - layers = [ - self._make_layer("main", main_clips), - self._make_layer("broll", broll_clips), - self._make_layer("background", bg_clips), - ] - - # Total video = 30s, voice = 15s → ratio = 0.5 - service._align_clips_to_voice_duration(layers, voice_duration=15.0) - - # All should be trimmed to 50% - assert abs(main_clips[0].duration - 5.0) < 0.01 - assert abs(broll_clips[0].duration - 5.0) < 0.01 - assert abs(bg_clips[0].duration - 5.0) < 0.01 - - def test_trim_config_also_adjusted(self): - """When clip has trim_config, it should also be adjusted.""" - from video_processing.trim_engine import TrimConfig - - service = self._make_service() - - clip = self._make_clip("c1", 10.0) - clip.trim_config = TrimConfig(start_time=0.0, duration=10.0) - - layers = [self._make_layer("main", [clip])] - - # ratio = 0.5 - service._align_clips_to_voice_duration(layers, voice_duration=5.0) - - assert abs(clip.duration - 5.0) < 0.01 - assert clip.trim_config is not None - assert abs(clip.trim_config.duration - 5.0) < 0.01 +def _make_clip(clip_id: str, duration: float, actual_duration: float = 0.0, playback_speed: float = 1.0, + config: dict | None = None, transition_effect: str = "cut") -> ResolvedClip: + return ResolvedClip( + clip_id=clip_id, + asset_id=f"asset_{clip_id}", + local_path=Path(f"/tmp/{clip_id}.mp4"), + clip_type="main", + order=0, + duration=duration, + actual_duration=actual_duration or duration, + playback_speed=playback_speed, + config=config or {}, + transition_effect=transition_effect, + transition_duration=0.0, + ) -class TestGetVoiceAudioDuration: - """Test voice audio duration probing.""" +def _make_layer(role: str, clips: list[ResolvedClip]) -> RenderLayer: + return RenderLayer(role=role, clips=clips, z_index=0) - def test_no_voiceover_path_returns_zero(self): - """No voiceover path → return 0.""" - with patch.object(UnifiedRenderService, "__init__", lambda self, **kwargs: None): - service = UnifiedRenderService.__new__(UnifiedRenderService) - service.voiceover_audio_path = None - assert service._get_voice_audio_duration() == 0.0 +def _make_service() -> UnifiedRenderService: + plan = MagicMock() + plan.id = "test_plan" + plan.config = {} + with patch.object(UnifiedRenderService, "__init__", lambda self, **kwargs: None): + service = UnifiedRenderService.__new__(UnifiedRenderService) + service.plan = plan + service.voiceover_audio_path = None + service.transition_duration = 0.0 + return service - def test_nonexistent_file_returns_zero(self): - """Nonexistent file → return 0.""" - with patch.object(UnifiedRenderService, "__init__", lambda self, **kwargs: None): - service = UnifiedRenderService.__new__(UnifiedRenderService) - service.voiceover_audio_path = "/nonexistent/path.mp3" - assert service._get_voice_audio_duration() == 0.0 +def test_align_does_not_trim_clips_when_clips_longer(): + """片段合计比配音长:旧逻辑裁剪,新逻辑不动。""" + service = _make_service() + clips = [_make_clip("c1", 10.0), _make_clip("c2", 10.0)] + layers = [_make_layer("main", clips)] + service._align_clips_to_voice_duration(layers, voice_duration=10.0) + assert clips[0].duration == 10.0 + assert clips[1].duration == 10.0 - @patch("video_processing.unified_render_service.probe_duration") - @patch("video_processing.unified_render_service.Path.exists", return_value=True) - @patch("video_processing.unified_render_service.Path.stat") - def test_probes_duration_from_file(self, mock_stat, mock_exists, mock_probe): - """Valid file → probe duration.""" - mock_stat.return_value.st_size = 1000 # Non-empty file - mock_probe.return_value = 42.5 - with patch.object(UnifiedRenderService, "__init__", lambda self, **kwargs: None): - service = UnifiedRenderService.__new__(UnifiedRenderService) - service.voiceover_audio_path = "/tmp/voice.mp3" +def test_align_does_not_slow_down_when_clips_shorter(): + """片段合计比配音短:旧逻辑慢放,新逻辑不动。""" + service = _make_service() + clips = [_make_clip("c1", 5.0, actual_duration=5.0), _make_clip("c2", 5.0, actual_duration=5.0)] + layers = [_make_layer("main", clips)] + service._align_clips_to_voice_duration(layers, voice_duration=20.0) + assert clips[0].playback_speed == 1.0 + assert clips[1].playback_speed == 1.0 + assert clips[0].duration == 5.0 - assert service._get_voice_audio_duration() == 42.5 + +def test_align_zero_voice_noop(): + service = _make_service() + clips = [_make_clip("c1", 10.0)] + layers = [_make_layer("main", clips)] + service._align_clips_to_voice_duration(layers, voice_duration=0.0) + assert clips[0].duration == 10.0 + + +def test_effective_duration_uses_target_duration_not_actual(): + """素材短于目标段长:effective_duration = 目标段长(冻结铺满),不被 actual 钳制。""" + clip = _make_clip("c1", duration=10.0, actual_duration=6.2) + eff = UnifiedRenderService._clip_effective_duration(clip) + assert eff == 10.0 + + +def test_freeze_marked_in_config_respects_target_duration(): + """freeze 秒数 = 目标段长 − 素材内可用时长;config 标记供 tpad/apad 读取。""" + # 模拟 _resolve_clips 单段路径的 freeze 计算 + target = 10.0 + actual = 6.2 + start = 0.0 + avail = max(0.0, actual - start) + freeze = round(target - avail, 3) if avail < target - 0.05 else 0.0 + assert freeze == 3.8 + clip = _make_clip("c1", duration=target, actual_duration=actual, config={"_freeze_seconds": freeze}) + assert clip.config["_freeze_seconds"] == 3.8 + + +def test_estimate_total_duration_matches_planner_cut(): + """cut 零重叠:总时长 = Σ段长。""" + service = _make_service() + clips = [_make_clip("c1", 7.333, actual_duration=30.0, transition_effect="cut"), + _make_clip("c2", 7.333, actual_duration=30.0, transition_effect="cut"), + _make_clip("c3", 7.334, actual_duration=30.0, transition_effect="cut")] + clips[0].order, clips[1].order, clips[2].order = 0, 1, 2 + total = service._estimate_total_duration([_make_layer("main", clips)]) + assert abs(total - 22.0) < 0.1 + + +def test_estimate_total_duration_deducts_xfade_overlap(): + """xfade 转场:逐处扣减重叠(与 planner 同口径)。""" + service = _make_service() + clips = [ + _make_clip("c1", 4.333, actual_duration=30.0, transition_effect="cut"), + _make_clip("c2", 4.333, actual_duration=30.0, transition_effect="xfade"), + _make_clip("c3", 4.334, actual_duration=30.0, transition_effect="xfade"), + ] + clips[0].transition_duration, clips[1].transition_duration, clips[2].transition_duration = 0.0, 0.5, 0.5 + clips[0].order, clips[1].order, clips[2].order = 0, 1, 2 + total = service._estimate_total_duration([_make_layer("main", clips)]) + assert abs(total - 12.0) < 0.1 diff --git a/tests/unit/test_voice_duration_planner_1749.py b/tests/unit/test_voice_duration_planner_1749.py new file mode 100644 index 000000000..096a788a8 --- /dev/null +++ b/tests/unit/test_voice_duration_planner_1749.py @@ -0,0 +1,83 @@ +"""#1749 配音时长分配纯函数测试。""" + +import pytest + +from packages.domain.voice_duration_planner import ( + MIN_CLIP_DURATION, + plan_clip_durations, + total_output_duration, + transition_overlap_seconds, +) + + +def test_cut_22s_3clips_even_split(): + d = plan_clip_durations(3, 22.0) + assert len(d) == 3 + assert d == pytest.approx([7.333, 7.333, 7.334], abs=0.01) + assert total_output_duration(d) == pytest.approx(22.0, abs=0.05) + + +def test_xfade_12s_3clips_two_transitions(): + effects = ["cut", "xfade", "xfade"] + tdurs = [0.0, 0.5, 0.5] + d = plan_clip_durations(3, 12.0, effects, tdurs) + # Σ段长 − 2×0.5 = 12 → Σ段长 = 13 + assert sum(d) == pytest.approx(13.0, abs=0.01) + assert total_output_duration(d, effects, tdurs) == pytest.approx(12.0, abs=0.05) + + +def test_cut_zero_overlap(): + assert transition_overlap_seconds("cut", 0.5) == 0.0 + assert transition_overlap_seconds(None, 0.5) == 0.0 + assert transition_overlap_seconds("none", 0.5) == 0.0 + assert transition_overlap_seconds("xfade", 0.5) == 0.5 + assert transition_overlap_seconds("fade", 0.0) == 0.0 + + +def test_zero_voice_returns_empty(): + assert plan_clip_durations(3, 0.0) == [] + assert plan_clip_durations(3, -1.0) == [] + assert plan_clip_durations(0, 10.0) == [] + + +def test_invalid_voice_returns_empty(): + assert plan_clip_durations(3, None) == [] # type: ignore[arg-type] + assert plan_clip_durations(3, "abc") == [] # type: ignore[arg-type] + + +def test_short_voice_floors_min_clip(): + d = plan_clip_durations(5, 2.0) + assert len(d) == 5 + assert all(x >= MIN_CLIP_DURATION - 0.001 for x in d) + + +def test_last_segment_absorbs_rounding(): + d = plan_clip_durations(3, 10.0) + assert total_output_duration(d) == pytest.approx(10.0, abs=0.05) + d2 = plan_clip_durations(7, 23.7) + assert total_output_duration(d2) == pytest.approx(23.7, abs=0.05) + + +def test_all_effects_cut_total_equals_voice(): + effects = ["cut"] * 4 + d = plan_clip_durations(4, 30.0, effects, [0.0] * 4) + assert total_output_duration(d, effects, [0.0] * 4) == pytest.approx(30.0, abs=0.05) + + +def test_mixed_transitions(): + effects = ["cut", "cut", "xfade", "slide"] + tdurs = [0.0, 0.0, 0.4, 0.6] + d = plan_clip_durations(4, 20.0, effects, tdurs) + # 重叠 0.4 + 0.6 = 1.0 → Σ段长 = 21 + assert sum(d) == pytest.approx(21.0, abs=0.01) + assert total_output_duration(d, effects, tdurs) == pytest.approx(20.0, abs=0.05) + + +def test_single_clip_no_transition(): + d = plan_clip_durations(1, 8.0) + assert len(d) == 1 + assert d[0] == pytest.approx(8.0, abs=0.05) + + +def test_total_output_duration_empty(): + assert total_output_duration([]) == 0.0