xiaoxia/xiaoxia-saas
1
0
Fork 0
Code Issues 14 Pull Requests Actions Packages Projects Releases 7 Wiki Activity

test(#1749): 重建配音分配/守卫/去重测试 + 适配 1677/1294（变体0 clone、voice_ids 移除）

Browse Source
This commit is contained in:
saas-backend-agent
2026-09-07 01:15:40 +08:00
parent eef372073a
commit cbe5b985b4
6 changed files with 468 additions and 252 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_1294_preview_voice_injection.py
+77 -59
View File
@@ -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"] == ""
tests/unit/test_1677_batch_variants.py
+18 -8
View File
@@ -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"
tests/unit/test_variant_plan_selector_1749.py
+135
View File
@@ -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
tests/unit/test_variant_voice_resolver_1749.py
+51
View File
@@ -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"]
tests/unit/test_voice_duration_alignment.py
+104 -185
View File
@@ -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
tests/unit/test_voice_duration_planner_1749.py
+83
View File
@@ -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