test(unit): 第87波 - worker层speed_engine+reverse_engine纯逻辑单测 (+85)

- test_speed_engine_pure.py (57个): SpeedConfig解析/钳制/isOriginal, SpeedEngine视频滤镜, atempo多级拆分算法, 时长计算, resolve_clip_speed
- test_reverse_engine_pure.py (28个): ReverseConfig.from_dict容错解析, 视频/音频滤镜构建, 120s安全时长限制, 组合场景验证

tests/unit/test_reverse_engine_pure.py

@@ -0,0 +1,228 @@
+"""ReverseEngine 纯逻辑单测 — 配置解析 + 滤镜构建 + 安全限制.
+
+全纯函数测试，不依赖 FFmpeg。
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pytest
+
+from video_processing.reverse_engine import ReverseConfig, ReverseEngine
+
+# ── ReverseConfig 解析 ────────────────────────────────────────────────────
+
+
+class TestReverseConfigFromDict:
+    """ReverseConfig.from_dict 配置解析测试."""
+
+    def test_none_returns_disabled(self):
+        """None 输入返回 disabled 默认配置."""
+        config = ReverseConfig.from_dict(None)
+        assert config.enabled is False
+        assert config.reverse_video is True
+        assert config.reverse_audio is True
+
+    def test_empty_dict_returns_disabled(self):
+        """空 dict 返回 disabled."""
+        config = ReverseConfig.from_dict({})
+        assert config.enabled is False
+
+    def test_enabled_false_returns_disabled(self):
+        """显式 enabled=False."""
+        config = ReverseConfig.from_dict({"enabled": False})
+        assert config.enabled is False
+
+    def test_enabled_default_flags_default_video_audio(self):
+        """只启用时默认视频音频都倒放."""
+        config = ReverseConfig.from_dict({"enabled": True})
+        assert config.enabled is True
+        assert config.reverse_video is True
+        assert config.reverse_audio is True
+
+    def test_video_only(self):
+        """只倒放视频."""
+        config = ReverseConfig.from_dict(
+            {
+                "enabled": True,
+                "reverse_video": True,
+                "reverse_audio": False,
+            }
+        )
+        assert config.enabled is True
+        assert config.reverse_video is True
+        assert config.reverse_audio is False
+
+    def test_audio_only(self):
+        """只倒放音频."""
+        config = ReverseConfig.from_dict(
+            {
+                "enabled": True,
+                "reverse_video": False,
+                "reverse_audio": True,
+            }
+        )
+        assert config.enabled is True
+        assert config.reverse_video is False
+        assert config.reverse_audio is True
+
+    def test_both_disabled_but_enabled_flag_true(self):
+        """enabled=True 但两个子选项都关了（边缘情况）."""
+        config = ReverseConfig.from_dict(
+            {
+                "enabled": True,
+                "reverse_video": False,
+                "reverse_audio": False,
+            }
+        )
+        assert config.enabled is True
+        assert config.reverse_video is False
+        assert config.reverse_audio is False
+
+    def test_invalid_type_falls_back(self):
+        """非 dict 类型回退到默认 disabled."""
+        config = ReverseConfig.from_dict("reverse=true")
+        assert config.enabled is False
+
+    def test_attribute_error_falls_back(self):
+        """属性错误时回退到默认."""
+
+        class WeirdObj:
+            def get(self, key, default=None):
+                raise AttributeError("nope")
+
+        config = ReverseConfig.from_dict(WeirdObj())
+        assert config.enabled is False
+
+    def test_type_error_falls_back(self):
+        """类型错误时回退到默认."""
+        config = ReverseConfig.from_dict([1, 2, 3])
+        assert config.enabled is False
+
+
+# ── ReverseEngine 视频滤镜 ──────────────────────────────────────────────────
+
+
+class TestReverseEngineBuildVideoFilter:
+    """ReverseEngine.build_video_filter 视频滤镜测试."""
+
+    def test_disabled_returns_empty(self):
+        """disabled 返回空字符串."""
+        config = ReverseConfig(enabled=False)
+        assert ReverseEngine.build_video_filter(config) == ""
+
+    def test_enabled_returns_reverse(self):
+        """启用返回 reverse 滤镜."""
+        config = ReverseConfig(enabled=True, reverse_video=True)
+        assert ReverseEngine.build_video_filter(config) == "reverse"
+
+    def test_video_disabled_returns_empty(self):
+        """reverse_video=False 返回空."""
+        config = ReverseConfig(enabled=True, reverse_video=False, reverse_audio=True)
+        assert ReverseEngine.build_video_filter(config) == ""
+
+    def test_short_duration_ok(self):
+        """短时长正常返回 reverse."""
+        config = ReverseConfig(enabled=True)
+        result = ReverseEngine.build_video_filter(config, duration=10.0)
+        assert result == "reverse"
+
+    def test_exactly_max_duration_ok(self):
+        """恰好等于安全上限正常."""
+        config = ReverseConfig(enabled=True)
+        result = ReverseEngine.build_video_filter(config, duration=ReverseEngine.MAX_SAFE_DURATION)
+        assert result == "reverse"
+
+    def test_over_max_duration_skipped(self):
+        """超过安全时长跳过倒放."""
+        config = ReverseConfig(enabled=True)
+        result = ReverseEngine.build_video_filter(config, duration=200.0)
+        assert result == ""
+
+    def test_zero_duration_ok(self):
+        """零时长正常倒放."""
+        config = ReverseConfig(enabled=True)
+        result = ReverseEngine.build_video_filter(config, duration=0.0)
+        assert result == "reverse"
+
+
+# ── ReverseEngine 音频滤镜 ──────────────────────────────────────────────
+
+
+class TestReverseEngineBuildAudioFilter:
+    """ReverseEngine.build_audio_filter 音频滤镜测试."""
+
+    def test_disabled_returns_empty(self):
+        """disabled 返回空字符串."""
+        config = ReverseConfig(enabled=False)
+        assert ReverseEngine.build_audio_filter(config) == ""
+
+    def test_enabled_returns_areverse(self):
+        """启用返回 areverse 滤镜."""
+        config = ReverseConfig(enabled=True, reverse_audio=True)
+        assert ReverseEngine.build_audio_filter(config) == "areverse"
+
+    def test_audio_disabled_returns_empty(self):
+        """reverse_audio=False 返回空."""
+        config = ReverseConfig(enabled=True, reverse_video=True, reverse_audio=False)
+        assert ReverseEngine.build_audio_filter(config) == ""
+
+    def test_short_duration_ok(self):
+        """短时长正常返回 areverse."""
+        config = ReverseConfig(enabled=True)
+        result = ReverseEngine.build_audio_filter(config, duration=30.0)
+        assert result == "areverse"
+
+    def test_over_max_duration_skipped(self):
+        """超过安全时长跳过音频倒放."""
+        config = ReverseConfig(enabled=True)
+        result = ReverseEngine.build_audio_filter(config, duration=150.0)
+        assert result == ""
+
+    def test_exactly_max_duration_ok(self):
+        """恰好等于安全上限正常."""
+        config = ReverseConfig(enabled=True)
+        result = ReverseEngine.build_audio_filter(config, duration=ReverseEngine.MAX_SAFE_DURATION)
+        assert result == "areverse"
+
+
+# ── 组合场景 ────────────────────────────────────────────────────────────
+
+
+class TestReverseEngineCombined:
+    """组合场景测试."""
+
+    def test_both_video_audio_reverse(self):
+        """视频音频都倒放."""
+        config = ReverseConfig(enabled=True)
+        vf = ReverseEngine.build_video_filter(config)
+        af = ReverseEngine.build_audio_filter(config)
+        assert vf == "reverse"
+        assert af == "areverse"
+
+    def test_neither_video_nor_audio(self):
+        """都不倒放."""
+        config = ReverseConfig(enabled=True, reverse_video=False, reverse_audio=False)
+        assert ReverseEngine.build_video_filter(config) == ""
+        assert ReverseEngine.build_audio_filter(config) == ""
+
+    def test_long_video_both_skipped(self):
+        """超长视频两个都跳过."""
+        config = ReverseConfig(enabled=True)
+        duration = ReverseEngine.MAX_SAFE_DURATION + 1
+        assert ReverseEngine.build_video_filter(config, duration=duration) == ""
+        assert ReverseEngine.build_audio_filter(config, duration=duration) == ""
+
+    def test_from_dict_full_config_flow(self):
+        """从 dict 解析到滤镜构建的完整流程."""
+        data = {"enabled": True, "reverse_video": True, "reverse_audio": False}
+        config = ReverseConfig.from_dict(data)
+        assert ReverseEngine.build_video_filter(config, duration=10) == "reverse"
+        assert ReverseEngine.build_audio_filter(config, duration=10) == ""
+
+    def test_from_dict_disabled_flow(self):
+        """disabled 配置完整流程."""
+        config = ReverseConfig.from_dict({"enabled": False})
+        assert ReverseEngine.build_video_filter(config) == ""
+        assert ReverseEngine.build_audio_filter(config) == ""

tests/unit/test_speed_engine_pure.py

@@ -0,0 +1,404 @@
+"""SpeedEngine 纯逻辑单测 — 配置解析 + 调速滤镜 + 时长计算.
+
+全纯函数测试，不依赖 FFmpeg 或外部服务。
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from video_processing.speed_engine import (
+    MAX_SPEED,
+    MIN_SPEED,
+    SpeedConfig,
+    SpeedEngine,
+)
+
+# ── SpeedConfig 解析 ──────────────────────────────────────────────────────
+
+
+class TestSpeedConfigParse:
+    """SpeedConfig.from_dict / parse 解析测试."""
+
+    def test_none_data_returns_default(self):
+        """None 输入返回默认配置."""
+        config = SpeedConfig.parse(None)
+        assert config.speed == 1.0
+        assert config.pitch_correct is True
+
+    def test_empty_dict_returns_default(self):
+        """空 dict 返回默认配置."""
+        config = SpeedConfig.parse({})
+        assert config.speed == 1.0
+        assert config.pitch_correct is True
+
+    def test_valid_speed_and_pitch(self):
+        """正常速度和音调配置."""
+        config = SpeedConfig.parse({"speed": 2.0, "pitch_correct": False})
+        assert config.speed == 2.0
+        assert config.pitch_correct is False
+
+    def test_speed_as_int(self):
+        """整数 speed 自动转 float."""
+        config = SpeedConfig.parse({"speed": 2})
+        assert config.speed == 2.0
+        assert isinstance(config.speed, float)
+
+    def test_invalid_speed_type_falls_back(self):
+        """speed 类型错误回退到默认."""
+        config = SpeedConfig.parse({"speed": "fast"})
+        assert config.speed == 1.0
+
+    def test_invalid_pitch_correct_type_falls_back(self):
+        """pitch_correct 非 bool 回退到 True."""
+        config = SpeedConfig.parse({"pitch_correct": "yes"})
+        assert config.pitch_correct is True
+
+    def test_non_dict_input_falls_back(self):
+        """非 dict 输入回退到默认."""
+        config = SpeedConfig.parse("speed=2x")
+        assert config.speed == 1.0
+        assert config.pitch_correct is True
+
+
+class TestSpeedConfigClamp:
+    """SpeedConfig.clamp 钳制测试."""
+
+    def test_zero_speed_clamps_to_default(self):
+        """speed=0 钳制到默认 1.0."""
+        config = SpeedConfig(speed=0.0)
+        config.clamp()
+        assert config.speed == 1.0
+
+    def test_negative_speed_clamps_to_default(self):
+        """负速度钳制到默认 1.0."""
+        config = SpeedConfig(speed=-1.0)
+        config.clamp()
+        assert config.speed == 1.0
+
+    def test_below_min_clamps_to_min(self):
+        """低于最小速度钳制到 MIN_SPEED."""
+        config = SpeedConfig(speed=0.1)
+        config.clamp()
+        assert config.speed == MIN_SPEED
+
+    def test_at_min_stays(self):
+        """恰好在最小值保持不变."""
+        config = SpeedConfig(speed=MIN_SPEED)
+        config.clamp()
+        assert config.speed == MIN_SPEED
+
+    def test_above_max_clamps_to_max(self):
+        """超过最大速度钳制到 MAX_SPEED."""
+        config = SpeedConfig(speed=5.0)
+        config.clamp()
+        assert config.speed == MAX_SPEED
+
+    def test_at_max_stays(self):
+        """恰好在最大值保持不变."""
+        config = SpeedConfig(speed=MAX_SPEED)
+        config.clamp()
+        assert config.speed == MAX_SPEED
+
+    def test_normal_speed_stays(self):
+        """正常范围内速度保持不变."""
+        config = SpeedConfig(speed=1.5)
+        config.clamp()
+        assert config.speed == 1.5
+
+    def test_parse_auto_clamps(self):
+        """parse 自动执行 clamp."""
+        config = SpeedConfig.parse({"speed": 10.0})
+        assert config.speed == MAX_SPEED
+
+
+class TestSpeedConfigIsOriginal:
+    """SpeedConfig.is_original 属性测试."""
+
+    def test_default_is_original(self):
+        """默认配置为原速."""
+        assert SpeedConfig().is_original is True
+
+    def test_exactly_one_is_original(self):
+        """speed=1.0 为原速."""
+        assert SpeedConfig(speed=1.0).is_original is True
+
+    def test_very_close_is_original(self):
+        """浮点精度接近 1.0 视为原速."""
+        assert SpeedConfig(speed=1.0000001).is_original is True
+
+    def test_different_speed_not_original(self):
+        """非 1.0 速度不是原速."""
+        assert SpeedConfig(speed=2.0).is_original is False
+        assert SpeedConfig(speed=0.5).is_original is False
+
+
+# ── SpeedEngine 滤镜构建 ──────────────────────────────────────────────────
+
+
+class TestSpeedEngineBuildVideoFilter:
+    """SpeedEngine.build_video_filter 视频滤镜测试."""
+
+    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):
+        """2倍速 setpts=PTS/2."""
+        config = SpeedConfig(speed=2.0)
+        result = self.engine.build_video_filter(config)
+        assert result == "setpts=PTS/2.0000"
+
+    def test_half_speed(self):
+        """0.5倍速 setpts=PTS/0.5."""
+        config = SpeedConfig(speed=0.5)
+        result = self.engine.build_video_filter(config)
+        assert result == "setpts=PTS/0.5000"
+
+    def test_quarter_speed(self):
+        """0.25倍速."""
+        config = SpeedConfig(speed=0.25)
+        result = self.engine.build_video_filter(config)
+        assert "setpts=PTS/0.25" in result
+
+    def test_quad_speed(self):
+        """4倍速."""
+        config = SpeedConfig(speed=4.0)
+        result = self.engine.build_video_filter(config)
+        assert "setpts=PTS/4.0" in result
+
+    def test_custom_speed_precision(self):
+        """自定义速度保留4位小数."""
+        config = SpeedConfig(speed=1.333)
+        result = self.engine.build_video_filter(config)
+        assert result == "setpts=PTS/1.3330"
+
+
+class TestSpeedEngineBuildAudioFilter:
+    """SpeedEngine.build_audio_filter 音频滤镜测试."""
+
+    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_within_single_stage_range(self):
+        """单级 atempo 范围内返回一级."""
+        config = SpeedConfig(speed=1.5)
+        result = self.engine.build_audio_filter(config)
+        assert result == "atempo=1.5000"
+
+    def test_at_max_single_stage(self):
+        """恰好 2.0 单级."""
+        config = SpeedConfig(speed=2.0)
+        result = self.engine.build_audio_filter(config)
+        assert result == "atempo=2.0000"
+
+    def test_at_min_single_stage(self):
+        """恰好 0.5 单级."""
+        config = SpeedConfig(speed=0.5)
+        result = self.engine.build_audio_filter(config)
+        assert result == "atempo=0.5000"
+
+    def test_quad_speed_two_stages(self):
+        """4倍速 = atempo=2.0,atempo=2.0."""
+        config = SpeedConfig(speed=4.0)
+        result = self.engine.build_audio_filter(config)
+        stages = result.split(",")
+        assert len(stages) == 2
+        assert stages[0] == "atempo=2.0000"
+        assert stages[1] == "atempo=2.0000"
+
+    def test_quarter_speed_two_stages(self):
+        """0.25倍速 = atempo=0.5,atempo=0.5."""
+        config = SpeedConfig(speed=0.25)
+        result = self.engine.build_audio_filter(config)
+        stages = result.split(",")
+        assert len(stages) == 2
+        assert stages[0] == "atempo=0.5000"
+        assert stages[1] == "atempo=0.5000"
+
+    def test_triple_speed_two_stages(self):
+        """3倍速 = atempo=2.0,atempo=1.5 (2.0 * 1.5 = 3.0)."""
+        config = SpeedConfig(speed=3.0)
+        result = self.engine.build_audio_filter(config)
+        stages = result.split(",")
+        assert len(stages) == 2
+        # 乘积应为 3.0
+        values = [float(s.split("=")[1]) for s in stages]
+        product = 1.0
+        for v in values:
+            product *= v
+        assert abs(product - 3.0) < 0.01
+
+    def test_low_speed_two_stages(self):
+        """0.3倍速多级串联，乘积为 0.3."""
+        config = SpeedConfig(speed=0.3)
+        result = self.engine.build_audio_filter(config)
+        stages = result.split(",")
+        assert len(stages) >= 2
+        values = [float(s.split("=")[1]) for s in stages]
+        product = 1.0
+        for v in values:
+            product *= v
+        assert abs(product - 0.3) < 0.01
+
+    def test_all_stages_within_valid_range(self):
+        """所有 atempo 级都在 [0.5, 2.0] 范围内."""
+        for speed in [0.25, 0.3, 0.5, 0.8, 1.5, 2.0, 3.0, 4.0]:
+            config = SpeedConfig(speed=speed)
+            result = self.engine.build_audio_filter(config)
+            if not result:
+                continue
+            stages = result.split(",")
+            for stage in stages:
+                val = float(stage.split("=")[1])
+                assert 0.5 <= val <= 2.0, f"speed={speed}, stage={val} out of range"
+
+
+class TestSpeedEngineSplitAtempoStages:
+    """SpeedEngine._split_atempo_stages 拆分算法测试."""
+
+    def test_single_stage_within_range(self):
+        """范围内单级."""
+        stages = SpeedEngine._split_atempo_stages(1.5)
+        assert stages == [1.5]
+
+    def test_exactly_max_single_stage(self):
+        """恰好 2.0 单级."""
+        stages = SpeedEngine._split_atempo_stages(2.0)
+        assert stages == [2.0]
+
+    def test_exactly_min_single_stage(self):
+        """恰好 0.5 单级."""
+        stages = SpeedEngine._split_atempo_stages(0.5)
+        assert stages == [0.5]
+
+    def test_four_x_two_stages(self):
+        """4.0 拆为两级 2.0."""
+        stages = SpeedEngine._split_atempo_stages(4.0)
+        assert stages == [2.0, 2.0]
+
+    def test_quarter_x_two_stages(self):
+        """0.25 拆为两级 0.5."""
+        stages = SpeedEngine._split_atempo_stages(0.25)
+        assert stages == [0.5, 0.5]
+
+    def test_product_matches_original_speed(self):
+        """拆分后乘积应等于原速度."""
+        for speed in [0.25, 0.3, 0.5, 0.8, 1.0, 1.5, 2.0, 2.5, 3.0, 4.0]:
+            stages = SpeedEngine._split_atempo_stages(speed)
+            product = 1.0
+            for s in stages:
+                product *= s
+            assert abs(product - speed) < 0.001, f"speed={speed}, product={product}"
+
+
+class TestSpeedEngineAdjustDuration:
+    """SpeedEngine.adjust_duration 时长计算测试."""
+
+    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):
+        """2倍速时长减半."""
+        config = SpeedConfig(speed=2.0)
+        assert self.engine.adjust_duration(10.0, config) == 5.0
+
+    def test_half_speed_double_duration(self):
+        """0.5倍速时长加倍."""
+        config = SpeedConfig(speed=0.5)
+        assert self.engine.adjust_duration(10.0, config) == 20.0
+
+    def test_zero_duration_stays_zero(self):
+        """零时长保持零."""
+        config = SpeedConfig(speed=2.0)
+        assert self.engine.adjust_duration(0.0, config) == 0.0
+
+    def test_negative_duration_stays(self):
+        """负时长直接返回（不做调速）."""
+        config = SpeedConfig(speed=2.0)
+        assert self.engine.adjust_duration(-5.0, config) == -5.0
+
+    def test_quad_speed_quarter_duration(self):
+        """4倍速时长为1/4."""
+        config = SpeedConfig(speed=4.0)
+        assert self.engine.adjust_duration(20.0, config) == 5.0
+
+
+class TestSpeedEngineBuildClipSpeedFilter:
+    """SpeedEngine.build_clip_speed_filter 便捷方法测试."""
+
+    def setup_method(self):
+        self.engine = SpeedEngine()
+
+    def test_default_speed_returns_empty_filters(self):
+        """默认速度返回空滤镜."""
+        vf, af, config = self.engine.build_clip_speed_filter(1.0)
+        assert vf == ""
+        assert af == ""
+        assert config.is_original is True
+
+    def test_double_speed_filters(self):
+        """2倍速返回对应滤镜."""
+        vf, af, config = self.engine.build_clip_speed_filter(2.0)
+        assert vf == "setpts=PTS/2.0000"
+        assert af == "atempo=2.0000"
+        assert config.speed == 2.0
+
+    def test_speed_gets_clamped(self):
+        """超范围速度自动钳制."""
+        vf, af, config = self.engine.build_clip_speed_filter(10.0)
+        assert config.speed == MAX_SPEED
+        assert "setpts" in vf
+
+    def test_pitch_correct_false_still_has_audio_filter(self):
+        """pitch_correct=False 也返回音频滤镜（只是方式不同，当前实现仍用atempo）."""
+        vf, af, config = self.engine.build_clip_speed_filter(2.0, pitch_correct=False)
+        assert config.pitch_correct is False
+        # 当前实现 pitch_correct 不影响滤镜输出（atempo 本身保持音调）
+        assert "atempo" in af
+
+
+class TestSpeedEngineResolveClipSpeed:
+    """SpeedEngine.resolve_clip_speed 速度解析测试."""
+
+    def test_no_clip_config_uses_global(self):
+        """无 clip config 使用全局速度."""
+        assert SpeedEngine.resolve_clip_speed(None, 2.0) == 2.0
+
+    def test_empty_config_uses_global(self):
+        """空 config 使用全局速度."""
+        assert SpeedEngine.resolve_clip_speed({}, 1.5) == 1.5
+
+    def test_zero_speed_uses_global(self):
+        """playback_speed=0 使用全局."""
+        assert SpeedEngine.resolve_clip_speed({"playback_speed": 0}, 2.0) == 2.0
+
+    def test_valid_clip_speed(self):
+        """有效 clip 速度优先."""
+        assert SpeedEngine.resolve_clip_speed({"playback_speed": 1.5}, 1.0) == 1.5
+
+    def test_invalid_speed_type_uses_global(self):
+        """速度类型错误回退到全局."""
+        assert SpeedEngine.resolve_clip_speed({"playback_speed": "fast"}, 2.0) == 2.0
+
+    def test_negative_speed_uses_global(self):
+        """负速度回退到全局."""
+        assert SpeedEngine.resolve_clip_speed({"playback_speed": -1}, 1.0) == 1.0
+
+    def test_default_global_is_one(self):
+        """默认全局速度为 1.0."""
+        assert SpeedEngine.resolve_clip_speed({}) == 1.0