"""裁剪引擎单元测试.""" import sys import unittest from pathlib import Path # 确保 apps/worker 在路径中 sys.path.insert(0, str(Path(__file__).parent.parent.parent / "apps" / "worker")) from video_processing.trim_engine import ( MIN_TRIM_DURATION, TrimConfig, TrimEngine, TrimSegment, extract_trim_from_clip_config, ) class TestTrimConfig(unittest.TestCase): """TrimConfig 单元测试.""" def test_from_dict_none(self): """空字典返回 None（不裁剪）.""" self.assertIsNone(TrimConfig.from_dict(None)) self.assertIsNone(TrimConfig.from_dict({})) def test_from_dict_with_start(self): """只有 start_time.""" cfg = TrimConfig.from_dict({"start_time": 5.0}) self.assertIsNotNone(cfg) self.assertEqual(cfg.start_time, 5.0) self.assertEqual(cfg.end_time, 0.0) self.assertEqual(cfg.duration, 0.0) def test_from_dict_with_duration(self): """只有 duration.""" cfg = TrimConfig.from_dict({"duration": 10.0}) self.assertIsNotNone(cfg) self.assertEqual(cfg.start_time, 0.0) self.assertEqual(cfg.duration, 10.0) def test_resolve_start_and_end(self): """start + end 推导 duration.""" cfg = TrimConfig(start_time=5.0, end_time=15.0) resolved = cfg.validate_and_resolve(asset_duration=30.0) self.assertEqual(resolved.start_time, 5.0) self.assertEqual(resolved.end_time, 15.0) self.assertAlmostEqual(resolved.duration, 10.0, places=3) self.assertTrue(resolved.is_valid) def test_resolve_start_and_duration(self): """start + duration 推导 end.""" cfg = TrimConfig(start_time=5.0, duration=10.0) resolved = cfg.validate_and_resolve(asset_duration=30.0) self.assertEqual(resolved.start_time, 5.0) self.assertAlmostEqual(resolved.end_time, 15.0, places=3) self.assertEqual(resolved.duration, 10.0) def test_resolve_end_and_duration(self): """end + duration 推导 start.""" cfg = TrimConfig(end_time=20.0, duration=8.0) resolved = cfg.validate_and_resolve(asset_duration=30.0) self.assertAlmostEqual(resolved.start_time, 12.0, places=3) self.assertEqual(resolved.end_time, 20.0) self.assertEqual(resolved.duration, 8.0) def test_resolve_only_start(self): """只有 start → 取到末尾.""" cfg = TrimConfig(start_time=10.0) resolved = cfg.validate_and_resolve(asset_duration=30.0) self.assertEqual(resolved.start_time, 10.0) self.assertEqual(resolved.end_time, 30.0) self.assertAlmostEqual(resolved.duration, 20.0, places=3) def test_resolve_only_duration(self): """只有 duration → 从开头取.""" cfg = TrimConfig(duration=15.0) resolved = cfg.validate_and_resolve(asset_duration=30.0) self.assertEqual(resolved.start_time, 0.0) self.assertAlmostEqual(resolved.end_time, 15.0, places=3) self.assertEqual(resolved.duration, 15.0) def test_boundary_clamp_end(self): """end 超出素材时长 → 钳制.""" cfg = TrimConfig(start_time=5.0, duration=30.0) resolved = cfg.validate_and_resolve(asset_duration=20.0) self.assertEqual(resolved.start_time, 5.0) self.assertEqual(resolved.end_time, 20.0) self.assertAlmostEqual(resolved.duration, 15.0, places=3) def test_boundary_clamp_start_negative(self): """start 为负 → 钳制到 0.""" cfg = TrimConfig(start_time=-5.0, duration=10.0) resolved = cfg.validate_and_resolve(asset_duration=30.0) self.assertEqual(resolved.start_time, 0.0) self.assertAlmostEqual(resolved.end_time, 10.0, places=3) self.assertEqual(resolved.duration, 10.0) def test_boundary_start_past_end(self): """start 超过素材总时长 → 钳制到末尾最小片段.""" cfg = TrimConfig(start_time=50.0, duration=5.0) resolved = cfg.validate_and_resolve(asset_duration=30.0) self.assertTrue(resolved.start_time < 30.0) self.assertEqual(resolved.end_time, 30.0) self.assertTrue(resolved.duration >= MIN_TRIM_DURATION) def test_invalid_end_before_start(self): """end <= start → 无效.""" cfg = TrimConfig(start_time=15.0, end_time=10.0) resolved = cfg.validate_and_resolve(asset_duration=30.0) self.assertFalse(resolved.is_valid) def test_zero_duration_invalid(self): """duration 为 0 → 无效.""" cfg = TrimConfig(start_time=5.0, duration=0.0) resolved = cfg.validate_and_resolve(asset_duration=30.0) # 只有 start 没有 duration → 会被推导为取到末尾 self.assertTrue(resolved.is_valid) self.assertEqual(resolved.end_time, 30.0) def test_is_noop(self): """is_noop 判断.""" noop = TrimConfig(start_time=0.0, end_time=0.0, duration=0.0) self.assertTrue(noop.is_noop) not_noop = TrimConfig(start_time=5.0, duration=10.0) self.assertFalse(not_noop.is_noop) def test_zero_asset_duration(self): """素材时长为 0 → 不裁剪.""" cfg = TrimConfig(start_time=5.0, duration=10.0) resolved = cfg.validate_and_resolve(asset_duration=0.0) self.assertTrue(resolved.is_noop) def test_all_three_params_use_start_duration(self): """三个参数都给了 → 以 start + duration 为准.""" cfg = TrimConfig(start_time=5.0, end_time=20.0, duration=8.0) resolved = cfg.validate_and_resolve(asset_duration=30.0) # validate_and_resolve 中 start+end 优先于 start+duration # 因为先检查的是 start>0 and end>0 self.assertAlmostEqual(resolved.duration, 15.0, places=3) class TestTrimEngine(unittest.TestCase): """TrimEngine 单元测试.""" def test_build_video_trim_with_start_and_duration(self): """视频裁剪：start + duration.""" trim = TrimConfig(start_time=10.0, duration=5.0) result = TrimEngine.build_video_trim_filter("[0:v]", trim, "[v0]") self.assertIn("trim=start=10.000:duration=5.000", result) self.assertIn("setpts=PTS-STARTPTS", result) self.assertTrue(result.startswith("[0:v]")) self.assertTrue(result.endswith("[v0]")) def test_build_video_trim_duration_only(self): """视频裁剪：只有 duration.""" trim = TrimConfig(start_time=0.0, duration=8.0) result = TrimEngine.build_video_trim_filter("[0:v]", trim, "[v0]") self.assertIn("trim=duration=8.000", result) self.assertNotIn("start=", result.split("setpts")[0]) def test_build_audio_trim_with_start(self): """音频裁剪：start + duration.""" trim = TrimConfig(start_time=3.0, duration=7.0) result = TrimEngine.build_audio_trim_filter("[0:a]", trim, "[a0]") self.assertIn("atrim=start=3.000:duration=7.000", result) self.assertIn("asetpts=PTS-STARTPTS", result) def test_build_audio_trim_noop(self): """音频裁剪：noop.""" trim = TrimConfig(start_time=0.0, end_time=0.0, duration=0.0) result = TrimEngine.build_audio_trim_filter("[0:a]", trim, "[a0]") self.assertIn("asetpts=PTS-STARTPTS", result) self.assertNotIn("atrim=", result) def test_resolve_segments(self): """多段裁剪解析.""" segments = [ TrimSegment(segment_id="s1", trim=TrimConfig(start_time=0.0, duration=5.0), order=0), TrimSegment(segment_id="s2", trim=TrimConfig(start_time=10.0, duration=5.0), order=1), TrimSegment(segment_id="s3", trim=TrimConfig(start_time=20.0, duration=5.0), order=2), ] resolved = TrimEngine.resolve_segments(segments, asset_duration=30.0) self.assertEqual(len(resolved), 3) self.assertEqual(resolved[0].segment_id, "s1") self.assertEqual(resolved[0].trim.duration, 5.0) self.assertEqual(resolved[1].segment_id, "s2") self.assertEqual(resolved[1].trim.start_time, 10.0) self.assertEqual(resolved[2].trim.start_time, 20.0) def test_resolve_segments_filter_invalid(self): """多段裁剪：过滤无效段.""" segments = [ TrimSegment(segment_id="good", trim=TrimConfig(start_time=0.0, duration=5.0), order=0), TrimSegment(segment_id="bad", trim=TrimConfig(start_time=10.0, end_time=5.0), order=1), # end < start ] resolved = TrimEngine.resolve_segments(segments, asset_duration=30.0) self.assertEqual(len(resolved), 1) self.assertEqual(resolved[0].segment_id, "good") def test_resolve_segments_boundary_clamp(self): """多段裁剪：边界钳制.""" segments = [ TrimSegment(segment_id="s1", trim=TrimConfig(start_time=25.0, duration=10.0), order=0), ] resolved = TrimEngine.resolve_segments(segments, asset_duration=30.0) self.assertEqual(len(resolved), 1) self.assertEqual(resolved[0].trim.end_time, 30.0) self.assertAlmostEqual(resolved[0].trim.duration, 5.0, places=3) def test_parse_segments_from_list(self): """从 config 解析多段配置.""" config = { "trim_segments": [ {"segment_id": "intro", "start_time": 0, "duration": 3, "order": 0}, {"segment_id": "highlight", "start_time": 10, "duration": 5, "order": 1}, {"segment_id": "outro", "start_time": 50, "duration": 3, "order": 2}, ] } segments = TrimEngine.parse_segments_from_config(config) self.assertEqual(len(segments), 3) self.assertEqual(segments[0].segment_id, "intro") self.assertEqual(segments[1].trim.start_time, 10.0) self.assertEqual(segments[2].trim.duration, 3.0) def test_parse_segments_empty(self): """无裁剪配置 → 空列表.""" self.assertEqual(TrimEngine.parse_segments_from_config(None), []) self.assertEqual(TrimEngine.parse_segments_from_config({}), []) def test_parse_single_trim_legacy(self): """旧格式单段裁剪（trim_start/trim_duration）.""" config = {"trim_start": 5.0, "trim_duration": 10.0} segments = TrimEngine.parse_segments_from_config(config) self.assertEqual(len(segments), 1) self.assertEqual(segments[0].trim.start_time, 5.0) self.assertEqual(segments[0].trim.duration, 10.0) class TestExtractTrimFromClipConfig(unittest.TestCase): """extract_trim_from_clip_config 单元测试.""" def test_trim_subdict(self): """trim 子字典.""" config = {"trim": {"start_time": 5.0, "duration": 10.0}} result = extract_trim_from_clip_config(config) self.assertIsNotNone(result) self.assertEqual(result.start_time, 5.0) self.assertEqual(result.duration, 10.0) def test_flat_fields(self): """扁平字段（trim_start/trim_end/trim_duration）.""" config = {"trim_start": 2.0, "trim_end": 8.0} result = extract_trim_from_clip_config(config) self.assertIsNotNone(result) self.assertEqual(result.start_time, 2.0) self.assertEqual(result.end_time, 8.0) def test_no_trim(self): """无裁剪配置.""" self.assertIsNone(extract_trim_from_clip_config(None)) self.assertIsNone(extract_trim_from_clip_config({})) self.assertIsNone(extract_trim_from_clip_config({"other": "value"})) if __name__ == "__main__": unittest.main()