xiaoxia-saas/tests/unit/domain/test_speed_config.py

"""speed_config 调速配置领域模型单测."""

import pytest
from domain.speed_config import (
    DEFAULT_SPEED,
    MAX_SPEED,
    MIN_SPEED,
    SpeedConfig,
    adjust_duration,
    build_audio_filter,
    build_clip_speed_filter,
    build_video_filter,
    resolve_clip_speed,
)

# ── 常量测试 ─────────────────────────────────────────────────────────────────


class TestConstants:
    """模块常量"""

    def test_speed_limits(self):
        assert MIN_SPEED == 0.25
        assert MAX_SPEED == 4.0
        assert DEFAULT_SPEED == 1.0


# ── SpeedConfig 默认值与基础 ────────────────────────────────────────────────


class TestSpeedConfigDefaults:
    """SpeedConfig 默认值"""

    def test_default_values(self):
        c = SpeedConfig()
        assert c.speed == 1.0
        assert c.pitch_correct is True

    def test_custom_values(self):
        c = SpeedConfig(speed=2.0, pitch_correct=False)
        assert c.speed == 2.0
        assert c.pitch_correct is False


# ── SpeedConfig.parse ───────────────────────────────────────────────────────


class TestSpeedConfigParse:
    """SpeedConfig.parse 工厂方法"""

    def test_none_returns_default(self):
        c = SpeedConfig.parse(None)
        assert c.speed == 1.0
        assert c.pitch_correct is True

    def test_empty_dict_returns_default(self):
        c = SpeedConfig.parse({})
        assert c.speed == 1.0

    def test_not_dict_returns_default(self):
        c = SpeedConfig.parse("not a dict")
        assert c.speed == 1.0

    def test_valid_speed(self):
        c = SpeedConfig.parse({"speed": 2.0})
        assert c.speed == 2.0

    def test_valid_speed_int(self):
        c = SpeedConfig.parse({"speed": 2})
        assert c.speed == 2.0
        assert isinstance(c.speed, float)

    def test_pitch_correct_false(self):
        c = SpeedConfig.parse({"pitch_correct": False})
        assert c.pitch_correct is False

    def test_pitch_correct_non_bool_falls_back(self):
        c = SpeedConfig.parse({"pitch_correct": "true"})
        assert c.pitch_correct is True

    def test_invalid_speed_string_falls_back(self):
        c = SpeedConfig.parse({"speed": "fast"})
        assert c.speed == 1.0

    def test_speed_below_min_clamped(self):
        c = SpeedConfig.parse({"speed": 0.1})
        assert c.speed == MIN_SPEED

    def test_speed_above_max_clamped(self):
        c = SpeedConfig.parse({"speed": 10.0})
        assert c.speed == MAX_SPEED

    def test_zero_speed_falls_back_to_default(self):
        c = SpeedConfig.parse({"speed": 0})
        assert c.speed == DEFAULT_SPEED

    def test_negative_speed_falls_back(self):
        c = SpeedConfig.parse({"speed": -1.0})
        assert c.speed == DEFAULT_SPEED

    def test_min_speed_boundary(self):
        c = SpeedConfig.parse({"speed": 0.25})
        assert c.speed == 0.25

    def test_max_speed_boundary(self):
        c = SpeedConfig.parse({"speed": 4.0})
        assert c.speed == 4.0


# ── SpeedConfig.clamp ───────────────────────────────────────────────────────


class TestSpeedConfigClamp:
    """SpeedConfig.clamp 方法"""

    def test_normal_speed_no_change(self):
        c = SpeedConfig(speed=1.5)
        c.clamp()
        assert c.speed == 1.5

    def test_zero_speed_reset_default(self):
        c = SpeedConfig(speed=0.0)
        c.clamp()
        assert c.speed == DEFAULT_SPEED

    def test_negative_speed_reset_default(self):
        c = SpeedConfig(speed=-0.5)
        c.clamp()
        assert c.speed == DEFAULT_SPEED

    def test_below_min_clamped(self):
        c = SpeedConfig(speed=0.1)
        c.clamp()
        assert c.speed == MIN_SPEED

    def test_above_max_clamped(self):
        c = SpeedConfig(speed=5.0)
        c.clamp()
        assert c.speed == MAX_SPEED

    def test_exact_min_unchanged(self):
        c = SpeedConfig(speed=MIN_SPEED)
        c.clamp()
        assert c.speed == MIN_SPEED

    def test_exact_max_unchanged(self):
        c = SpeedConfig(speed=MAX_SPEED)
        c.clamp()
        assert c.speed == MAX_SPEED


# ── SpeedConfig 属性方法 ────────────────────────────────────────────────────


class TestSpeedConfigProperties:
    """SpeedConfig 属性方法"""

    def test_is_original_true(self):
        c = SpeedConfig(speed=1.0)
        assert c.is_original is True

    def test_is_original_very_close(self):
        c = SpeedConfig(speed=1.0 + 1e-7)
        assert c.is_original is True

    def test_is_original_false_fast(self):
        c = SpeedConfig(speed=1.5)
        assert c.is_original is False

    def test_is_original_false_slow(self):
        c = SpeedConfig(speed=0.8)
        assert c.is_original is False

    def test_is_fast_true(self):
        c = SpeedConfig(speed=2.0)
        assert c.is_fast is True

    def test_is_fast_false(self):
        c = SpeedConfig(speed=0.5)
        assert c.is_fast is False

    def test_is_fast_at_one(self):
        c = SpeedConfig(speed=1.0)
        assert c.is_fast is False

    def test_is_slow_true(self):
        c = SpeedConfig(speed=0.5)
        assert c.is_slow is True

    def test_is_slow_false(self):
        c = SpeedConfig(speed=2.0)
        assert c.is_slow is False

    def test_is_slow_at_one(self):
        c = SpeedConfig(speed=1.0)
        assert c.is_slow is False


# ── build_video_filter ───────────────────────────────────────────────────────


class TestBuildVideoFilter:
    """build_video_filter 视频滤镜构建"""

    def test_original_speed_empty(self):
        c = SpeedConfig(speed=1.0)
        assert build_video_filter(c) == ""

    def test_double_speed(self):
        c = SpeedConfig(speed=2.0)
        result = build_video_filter(c)
        assert "setpts=PTS/2.0" in result

    def test_half_speed(self):
        c = SpeedConfig(speed=0.5)
        result = build_video_filter(c)
        assert "setpts=PTS/0.5" in result

    def test_format_precision(self):
        c = SpeedConfig(speed=1.5)
        result = build_video_filter(c)
        # 应该是 4 位小数
        assert "1.5000" in result

    def test_min_speed(self):
        c = SpeedConfig(speed=0.25)
        result = build_video_filter(c)
        assert result.startswith("setpts=PTS/")

    def test_max_speed(self):
        c = SpeedConfig(speed=4.0)
        result = build_video_filter(c)
        assert "4.0000" in result


# ── build_audio_filter / atempo 拆分 ────────────────────────────────────────


class TestBuildAudioFilter:
    """build_audio_filter 音频滤镜构建"""

    def test_original_speed_empty(self):
        c = SpeedConfig(speed=1.0)
        assert build_audio_filter(c) == ""

    def test_within_range_single_stage(self):
        c = SpeedConfig(speed=1.5)
        result = build_audio_filter(c)
        assert result == "atempo=1.5000"

    def test_05_speed_single_stage(self):
        c = SpeedConfig(speed=0.5)
        result = build_audio_filter(c)
        assert result == "atempo=0.5000"

    def test_20_speed_single_stage(self):
        c = SpeedConfig(speed=2.0)
        result = build_audio_filter(c)
        assert result == "atempo=2.0000"

    def test_4x_speed_two_stages(self):
        c = SpeedConfig(speed=4.0)
        result = build_audio_filter(c)
        # 2.0 * 2.0 = 4.0
        assert result == "atempo=2.0000,atempo=2.0000"

    def test_025_speed_two_stages(self):
        c = SpeedConfig(speed=0.25)
        result = build_audio_filter(c)
        # 0.5 * 0.5 = 0.25
        assert result == "atempo=0.5000,atempo=0.5000"

    def test_3x_speed_two_stages(self):
        c = SpeedConfig(speed=3.0)
        result = build_audio_filter(c)
        # 2.0 * 1.5 = 3.0
        stages = result.split(",")
        assert len(stages) == 2
        assert "atempo=2.0000" in stages[0]
        assert "atempo=1.5000" in stages[1]

    def test_03_speed_two_stages(self):
        c = SpeedConfig(speed=0.3)
        result = build_audio_filter(c)
        stages = result.split(",")
        assert len(stages) == 2
        # 0.5 * 0.6 = 0.3
        assert "atempo=0.5000" in stages[0]

    def test_format_each_stage(self):
        c = SpeedConfig(speed=1.2345)
        result = build_audio_filter(c)
        assert "atempo=1.2345" in result


class TestAtempoStages:
    """atempo 多级拆分逻辑验证"""

    def _extract_speeds(self, filter_str: str) -> list[float]:
        """从 atempo 滤镜字符串中提取速度值."""
        import re

        return [float(m) for m in re.findall(r"atempo=([\d.]+)", filter_str)]

    def test_product_equals_speed_fast_3x(self):
        c = SpeedConfig(speed=3.0)
        speeds = self._extract_speeds(build_audio_filter(c))
        product = 1.0
        for s in speeds:
            product *= s
        assert abs(product - 3.0) < 1e-4

    def test_product_equals_speed_4x(self):
        c = SpeedConfig(speed=4.0)
        speeds = self._extract_speeds(build_audio_filter(c))
        product = 1.0
        for s in speeds:
            product *= s
        assert abs(product - 4.0) < 1e-4

    def test_product_equals_speed_slow_025(self):
        c = SpeedConfig(speed=0.25)
        speeds = self._extract_speeds(build_audio_filter(c))
        product = 1.0
        for s in speeds:
            product *= s
        assert abs(product - 0.25) < 1e-4

    def test_product_equals_speed_slow_03(self):
        c = SpeedConfig(speed=0.3)
        speeds = self._extract_speeds(build_audio_filter(c))
        product = 1.0
        for s in speeds:
            product *= s
        assert abs(product - 0.3) < 1e-4

    def test_each_stage_in_range_fast(self):
        c = SpeedConfig(speed=3.5)
        speeds = self._extract_speeds(build_audio_filter(c))
        for s in speeds:
            assert 0.5 <= s <= 2.0

    def test_each_stage_in_range_slow(self):
        c = SpeedConfig(speed=0.35)
        speeds = self._extract_speeds(build_audio_filter(c))
        for s in speeds:
            assert 0.5 <= s <= 2.0


# ── adjust_duration ──────────────────────────────────────────────────────────


class TestAdjustDuration:
    """adjust_duration 时长计算"""

    def test_original_speed_no_change(self):
        assert adjust_duration(10.0, SpeedConfig(speed=1.0)) == 10.0

    def test_double_speed_half_duration(self):
        assert adjust_duration(10.0, SpeedConfig(speed=2.0)) == 5.0

    def test_half_speed_double_duration(self):
        assert adjust_duration(10.0, SpeedConfig(speed=0.5)) == 20.0

    def test_zero_duration_unchanged(self):
        assert adjust_duration(0.0, SpeedConfig(speed=2.0)) == 0.0

    def test_negative_duration_unchanged(self):
        assert adjust_duration(-1.0, SpeedConfig(speed=2.0)) == -1.0

    def test_original_with_zero_duration(self):
        assert adjust_duration(0.0, SpeedConfig(speed=1.0)) == 0.0

    def test_triple_speed(self):
        assert adjust_duration(30.0, SpeedConfig(speed=3.0)) == 10.0

    def test_quarter_speed(self):
        assert adjust_duration(10.0, SpeedConfig(speed=0.25)) == 40.0


# ── build_clip_speed_filter ──────────────────────────────────────────────────


class TestBuildClipSpeedFilter:
    """build_clip_speed_filter 便捷方法"""

    def test_returns_tuple_of_three(self):
        result = build_clip_speed_filter(1.5)
        assert len(result) == 3
        video_filter, audio_filter, config = result
        assert isinstance(video_filter, str)
        assert isinstance(audio_filter, str)
        assert isinstance(config, SpeedConfig)

    def test_normal_speed(self):
        video_filter, audio_filter, config = build_clip_speed_filter(1.0)
        assert video_filter == ""
        assert audio_filter == ""
        assert config.speed == 1.0

    def test_double_speed(self):
        video_filter, audio_filter, config = build_clip_speed_filter(2.0)
        assert "setpts" in video_filter
        assert "atempo" in audio_filter
        assert config.speed == 2.0

    def test_clamps_speed(self):
        _, _, config = build_clip_speed_filter(10.0)
        assert config.speed == MAX_SPEED

    def test_pitch_correct_false(self):
        # pitch_correct=False 时仍然生成滤镜（实际使用中可能换其他算法，但接口返回不变）
        video_filter, audio_filter, config = build_clip_speed_filter(2.0, pitch_correct=False)
        assert config.pitch_correct is False
        assert "setpts" in video_filter


# ── resolve_clip_speed ───────────────────────────────────────────────────────


class TestResolveClipSpeed:
    """resolve_clip_speed 片段速度解析"""

    def test_none_config_uses_global(self):
        assert resolve_clip_speed(None, 1.5) == 1.5

    def test_zero_speed_uses_global(self):
        assert resolve_clip_speed({"playback_speed": 0}, 1.5) == 1.5

    def test_missing_key_uses_global(self):
        assert resolve_clip_speed({}, 2.0) == 2.0

    def test_valid_speed(self):
        assert resolve_clip_speed({"playback_speed": 1.5}, 1.0) == 1.5

    def test_negative_speed_uses_global(self):
        assert resolve_clip_speed({"playback_speed": -1.0}, 1.0) == 1.0

    def test_invalid_type_uses_global(self):
        assert resolve_clip_speed({"playback_speed": "fast"}, 1.0) == 1.0

    def test_default_global_is_one(self):
        assert resolve_clip_speed({"playback_speed": 0}) == 1.0

    def test_int_speed(self):
        result = resolve_clip_speed({"playback_speed": 2})
        assert result == 2.0
        assert isinstance(result, float)

    def test_very_small_positive_uses_it(self):
        # 只要 > 0 就用
        result = resolve_clip_speed({"playback_speed": 0.1})
        assert result == 0.1