"""视频调速引擎单元测试.""" import pytest from video_processing.speed_engine import ( MAX_SPEED, MIN_SPEED, SpeedConfig, SpeedEngine, ) # ─── SpeedConfig 解析与校验 ────────────────────────────────── class TestSpeedConfig: def test_default_values(self): config = SpeedConfig() assert config.speed == 1.0 assert config.pitch_correct is True def test_parse_none(self): config = SpeedConfig.parse(None) assert config.speed == 1.0 assert config.pitch_correct is True def test_parse_empty_dict(self): config = SpeedConfig.parse({}) assert config.speed == 1.0 def test_parse_valid_speed(self): config = SpeedConfig.parse({"speed": 2.0}) assert config.speed == 2.0 def test_parse_pitch_correct_false(self): config = SpeedConfig.parse({"pitch_correct": False}) assert config.pitch_correct is False def test_parse_invalid_speed_type(self): config = SpeedConfig.parse({"speed": "fast"}) assert config.speed == 1.0 def test_parse_invalid_pitch_type(self): config = SpeedConfig.parse({"pitch_correct": "yes"}) assert config.pitch_correct is True def test_clamp_below_min(self): config = SpeedConfig(speed=0.1) config.clamp() assert config.speed == MIN_SPEED def test_clamp_zero(self): config = SpeedConfig(speed=0) config.clamp() assert config.speed == 1.0 def test_clamp_negative(self): config = SpeedConfig(speed=-1.0) config.clamp() assert config.speed == 1.0 def test_clamp_above_max(self): config = SpeedConfig(speed=10.0) config.clamp() assert config.speed == MAX_SPEED def test_clamp_within_range(self): config = SpeedConfig(speed=1.5) config.clamp() assert config.speed == 1.5 def test_is_original_true(self): config = SpeedConfig(speed=1.0) assert config.is_original is True def test_is_original_false(self): config = SpeedConfig(speed=1.5) assert config.is_original is False def test_parse_clamps_automatically(self): """parse 方法应该自动调用 clamp.""" config = SpeedConfig.parse({"speed": 100.0}) assert config.speed == MAX_SPEED # ─── SpeedEngine 视频滤镜 ──────────────────────────────────── class TestSpeedEngineVideoFilter: def setup_method(self): self.engine = SpeedEngine() def test_original_speed_returns_empty(self): config = SpeedConfig(speed=1.0) assert self.engine.build_video_filter(config) == "" def test_double_speed(self): config = SpeedConfig(speed=2.0) result = self.engine.build_video_filter(config) assert "setpts=PTS/2.0" in result def test_half_speed(self): config = SpeedConfig(speed=0.5) result = self.engine.build_video_filter(config) assert "setpts=PTS/0.5" in result def test_quarter_speed(self): config = SpeedConfig(speed=0.25) result = self.engine.build_video_filter(config) assert "setpts=PTS/0.25" in result def test_quad_speed(self): config = SpeedConfig(speed=4.0) result = self.engine.build_video_filter(config) assert "setpts=PTS/4.0" in result # ─── SpeedEngine 音频滤镜（atempo 多级串联） ───────────────── class TestSpeedEngineAudioFilter: def setup_method(self): self.engine = SpeedEngine() def test_original_speed_returns_empty(self): config = SpeedConfig(speed=1.0) assert self.engine.build_audio_filter(config) == "" def test_double_speed_single_stage(self): """2x 在 atempo 单级范围内，只需一个 atempo.""" config = SpeedConfig(speed=2.0) result = self.engine.build_audio_filter(config) assert result == "atempo=2.0000" def test_half_speed_single_stage(self): config = SpeedConfig(speed=0.5) result = self.engine.build_audio_filter(config) assert result == "atempo=0.5000" def test_quad_speed_two_stages(self): """4x 需要两级 atempo: 2.0 * 2.0.""" config = SpeedConfig(speed=4.0) result = self.engine.build_audio_filter(config) assert result == "atempo=2.0000,atempo=2.0000" def test_quarter_speed_two_stages(self): """0.25x 需要两级 atempo: 0.5 * 0.5.""" config = SpeedConfig(speed=0.25) result = self.engine.build_audio_filter(config) assert result == "atempo=0.5000,atempo=0.5000" def test_triple_speed_two_stages(self): """3x: 2.0 * 1.5.""" config = SpeedConfig(speed=3.0) result = self.engine.build_audio_filter(config) parts = result.split(",") assert len(parts) == 2 assert "atempo=2.0000" in parts assert "atempo=1.5000" in parts def test_03_speed_two_stages(self): """0.3x: 0.5 * 0.6.""" config = SpeedConfig(speed=0.3) result = self.engine.build_audio_filter(config) parts = result.split(",") assert len(parts) == 2 assert "atempo=0.5000" in parts assert "atempo=0.6000" in parts def test_split_atempo_inside_range(self): """0.5~2.0 范围内只返回一级.""" stages = SpeedEngine._split_atempo_stages(1.5) assert len(stages) == 1 assert stages[0] == 1.5 def test_split_atempo_boundary_min(self): stages = SpeedEngine._split_atempo_stages(0.5) assert len(stages) == 1 assert stages[0] == 0.5 def test_split_atempo_boundary_max(self): stages = SpeedEngine._split_atempo_stages(2.0) assert len(stages) == 1 assert stages[0] == 2.0 def test_split_atempo_product_equals_speed(self): """所有级联的乘积应该等于原速度.""" test_cases = [0.25, 0.3, 0.5, 0.75, 1.0, 1.5, 2.0, 3.0, 4.0] for speed in test_cases: stages = SpeedEngine._split_atempo_stages(speed) product = 1.0 for s in stages: product *= s assert abs(product - speed) < 1e-6, f"speed={speed}, stages={stages}, product={product}" def test_split_atempo_all_in_range(self): """所有级都应该在 0.5~2.0 范围内.""" test_cases = [0.25, 0.3, 0.5, 0.75, 1.0, 1.5, 2.0, 3.0, 4.0] for speed in test_cases: stages = SpeedEngine._split_atempo_stages(speed) for s in stages: assert 0.5 <= s <= 2.0, f"speed={speed}, stage={s} out of range" # ─── SpeedEngine 时长计算 ──────────────────────────────────── class TestSpeedEngineDuration: def setup_method(self): self.engine = SpeedEngine() def test_original_speed_same_duration(self): config = SpeedConfig(speed=1.0) assert self.engine.adjust_duration(10.0, config) == 10.0 def test_double_speed_half_duration(self): config = SpeedConfig(speed=2.0) assert self.engine.adjust_duration(10.0, config) == 5.0 def test_half_speed_double_duration(self): config = SpeedConfig(speed=0.5) assert self.engine.adjust_duration(10.0, config) == 20.0 def test_quad_speed_quarter_duration(self): config = SpeedConfig(speed=4.0) assert self.engine.adjust_duration(10.0, config) == 2.5 def test_zero_duration(self): config = SpeedConfig(speed=2.0) assert self.engine.adjust_duration(0.0, config) == 0.0 def test_negative_duration(self): config = SpeedConfig(speed=2.0) assert self.engine.adjust_duration(-1.0, config) == -1.0 # ─── SpeedEngine 便捷方法 ──────────────────────────────────── class TestSpeedEngineHelper: def setup_method(self): self.engine = SpeedEngine() def test_build_clip_speed_filter_original(self): v_f, a_f, cfg = self.engine.build_clip_speed_filter(1.0) assert v_f == "" assert a_f == "" assert cfg.speed == 1.0 def test_build_clip_speed_filter_2x(self): v_f, a_f, cfg = self.engine.build_clip_speed_filter(2.0) assert "setpts=PTS/2.0" in v_f assert "atempo=2.0" in a_f assert cfg.speed == 2.0 def test_build_clip_speed_clamped(self): _, _, cfg = self.engine.build_clip_speed_filter(100.0) assert cfg.speed == MAX_SPEED def test_resolve_clip_speed_default(self): assert SpeedEngine.resolve_clip_speed({}) == 1.0 assert SpeedEngine.resolve_clip_speed(None) == 1.0 def test_resolve_clip_speed_zero_uses_global(self): assert SpeedEngine.resolve_clip_speed({"playback_speed": 0}, 1.5) == 1.5 def test_resolve_clip_speed_custom(self): assert SpeedEngine.resolve_clip_speed({"playback_speed": 2.0}) == 2.0 def test_resolve_clip_speed_invalid_type(self): assert SpeedEngine.resolve_clip_speed({"playback_speed": "fast"}) == 1.0