xiaoxia-saas/tests/unit/test_bgm_mixer_pure.py

"""bgm_mixer_pure 单元测试."""

from apps.worker.video_processing.bgm_mixer_pure import (
    BGMPureConfig,
    build_bgm_filter_chain,
    build_sidechain_mix_filter,
    build_simple_mix_filter,
    calculate_fade_out_start,
    calculate_loop_count,
    calculate_sidechain_ratio,
    estimate_bgm_processing_duration,
    normalize_bgm_config,
    should_loop_bgm,
    validate_bgm_config,
)

# ── BGMPureConfig ────────────────────────────────────────────────────────────


class TestBGMPureConfig:
    def test_default_values(self):
        cfg = BGMPureConfig()
        assert cfg.volume == 0.3
        assert cfg.fade_in == 0.0
        assert cfg.fade_out == 0.0
        assert cfg.loop_enabled is True
        assert cfg.sidechain_enabled is False
        assert cfg.sidechain_ratio == 0.3
        assert cfg.sidechain_attack == 0.02
        assert cfg.sidechain_release == 0.5
        assert cfg.sidechain_threshold == -25.0

    def test_custom_values(self):
        cfg = BGMPureConfig(
            volume=0.5,
            fade_in=1.0,
            fade_out=2.0,
            loop_enabled=False,
            sidechain_enabled=True,
            sidechain_ratio=0.5,
        )
        assert cfg.volume == 0.5
        assert cfg.loop_enabled is False
        assert cfg.sidechain_enabled is True
        assert cfg.sidechain_ratio == 0.5


# ── should_loop_bgm ─────────────────────────────────────────────────────────


class TestShouldLoopBgm:
    def test_loop_enabled_much_shorter(self):
        # BGM 10秒，目标60秒 → 需要循环
        assert should_loop_bgm(10, 60) is True

    def test_loop_disabled(self):
        assert should_loop_bgm(10, 60, loop_enabled=False) is False

    def test_bgm_longer_than_target(self):
        # BGM 100秒，目标60秒 → 不需要循环
        assert should_loop_bgm(100, 60) is False

    def test_bgm_slightly_shorter_no_loop(self):
        # BGM 58秒，目标60秒 → 58 > 60*0.9=54，不需要循环
        assert should_loop_bgm(58, 60) is False

    def test_bgm_significantly_shorter_loops(self):
        # BGM 50秒，目标60秒 → 50 < 54，需要循环
        assert should_loop_bgm(50, 60) is True

    def test_zero_bgm_duration(self):
        assert should_loop_bgm(0, 60) is False

    def test_negative_bgm_duration(self):
        assert should_loop_bgm(-1, 60) is False

    def test_zero_target_duration(self):
        assert should_loop_bgm(10, 0) is False

    def test_negative_target_duration(self):
        assert should_loop_bgm(10, -1) is False

    def test_exact_90_percent_no_loop(self):
        # 边界：bgm == target * 0.9 → 不小于，不循环
        assert should_loop_bgm(54, 60) is False


# ── calculate_loop_count ────────────────────────────────────────────────────


class TestCalculateLoopCount:
    def test_exact_fit_returns_1(self):
        assert calculate_loop_count(60, 60) == 1

    def test_bgm_longer_returns_1(self):
        assert calculate_loop_count(100, 60) == 1

    def test_needs_3_loops_plus_2_margin(self):
        # 60/20 = 3 + 2 = 5
        assert calculate_loop_count(20, 60) == 5

    def test_needs_2_loops_plus_2_margin(self):
        # 60/30 = 2 + 2 = 4
        assert calculate_loop_count(30, 60) == 4

    def test_zero_bgm_duration(self):
        assert calculate_loop_count(0, 60) == 1

    def test_negative_bgm_duration(self):
        assert calculate_loop_count(-1, 60) == 1

    def test_zero_target_duration(self):
        assert calculate_loop_count(10, 0) == 1

    def test_negative_target_duration(self):
        assert calculate_loop_count(10, -1) == 1

    def test_minimum_is_1(self):
        assert calculate_loop_count(10, 5) == 1


# ── build_bgm_filter_chain ──────────────────────────────────────────────────


class TestBuildBgmFilterChain:
    def test_basic_structure(self):
        result = build_bgm_filter_chain(100, 60)
        parts = result.split(",")
        # 至少有 atrim + asetpts
        assert any("atrim=" in p for p in parts)
        assert "asetpts=N/SR/TB" in parts

    def test_volume_filter_applied(self):
        result = build_bgm_filter_chain(100, 60, volume=0.5)
        assert "volume=0.500" in result

    def test_volume_one_omitted(self):
        result = build_bgm_filter_chain(100, 60, volume=1.0)
        assert "volume=" not in result

    def test_volume_clamped(self):
        # volume=2.0钳制到1.0，1.0等于默认值所以被跳过
        result = build_bgm_filter_chain(100, 60, volume=2.0)
        assert "volume=" not in result  # 钳制到1.0后与默认相同，跳过
        # 用0.5验证音量过滤器本身存在
        result2 = build_bgm_filter_chain(100, 60, volume=0.5)
        assert "volume=0.500" in result2

    def test_volume_zero(self):
        result = build_bgm_filter_chain(100, 60, volume=0.0)
        assert "volume=0.000" in result

    def test_fade_in_applied(self):
        result = build_bgm_filter_chain(100, 60, fade_in=1.5)
        assert "afade=t=in:st=0:d=1.500" in result

    def test_fade_in_zero_skipped(self):
        result = build_bgm_filter_chain(100, 60, fade_in=0)
        assert "afade=t=in" not in result

    def test_fade_out_applied(self):
        result = build_bgm_filter_chain(100, 60, fade_out=2.0)
        assert "afade=t=out:st=58.000:d=2.000" in result

    def test_fade_out_longer_than_target_skipped(self):
        result = build_bgm_filter_chain(100, 10, fade_out=20)
        # fade_out >= safe_target，不做淡出
        assert "afade=t=out" not in result

    def test_loop_applied_when_needed(self):
        result = build_bgm_filter_chain(10, 60)
        assert "aloop=loop=" in result

    def test_no_loop_when_bgm_long(self):
        result = build_bgm_filter_chain(100, 60)
        assert "aloop=" not in result

    def test_loop_disabled(self):
        result = build_bgm_filter_chain(10, 60, loop_enabled=False)
        assert "aloop=" not in result

    def test_trim_to_target_duration(self):
        result = build_bgm_filter_chain(100, 60)
        assert "atrim=0:60.000" in result

    def test_zero_target_uses_fallback(self):
        result = build_bgm_filter_chain(100, 0)
        # 兜底5秒
        assert "atrim=0:5.000" in result

    def test_negative_target_uses_fallback(self):
        result = build_bgm_filter_chain(100, -5)
        assert "atrim=0:5.000" in result

    def test_all_features_combined(self):
        result = build_bgm_filter_chain(
            bgm_duration=15,
            target_duration=60,
            volume=0.3,
            fade_in=1.0,
            fade_out=2.0,
            loop_enabled=True,
        )
        assert "aloop=loop=" in result
        assert "volume=0.300" in result
        assert "afade=t=in" in result
        assert "afade=t=out" in result
        assert "atrim=0:60.000" in result
        assert "asetpts=N/SR/TB" in result


# ── calculate_sidechain_ratio ───────────────────────────────────────────────


class TestCalculateSidechainRatio:
    def test_zero_ratio_minimum(self):
        assert calculate_sidechain_ratio(0) == 2.0

    def test_negative_clamped(self):
        assert calculate_sidechain_ratio(-0.5) == 2.0

    def test_one_ratio_maximum(self):
        assert calculate_sidechain_ratio(1.0) == 10.0

    def test_above_one_clamped(self):
        assert calculate_sidechain_ratio(1.5) == 10.0

    def test_mid_value(self):
        # ratio = 1/(1-0.5) = 2.0
        result = calculate_sidechain_ratio(0.5)
        assert abs(result - 2.0) < 0.01

    def test_high_value(self):
        # 1/(1-0.9) = 10 → 钳制到10
        assert calculate_sidechain_ratio(0.9) == 10.0

    def test_03_default(self):
        # 1/(1-0.3) = 1.428... → 钳制到2.0
        result = calculate_sidechain_ratio(0.3)
        assert result >= 2.0


# ── build_simple_mix_filter ─────────────────────────────────────────────────


class TestBuildSimpleMixFilter:
    def test_contains_inputs_and_output(self):
        result = build_simple_mix_filter()
        assert "[0:a][1:a]" in result
        assert "amix=inputs=2" in result
        assert "duration=first" in result
        assert "[final]" in result
        assert "volume=2" in result


# ── build_sidechain_mix_filter ──────────────────────────────────────────────


class TestBuildSidechainMixFilter:
    def test_contains_sidechaincompress(self):
        result = build_sidechain_mix_filter()
        assert "sidechaincompress=" in result
        assert "[1:a][0:a]sidechaincompress" in result

    def test_threshold_in_db(self):
        result = build_sidechain_mix_filter(threshold=-30.0)
        assert "threshold=-30.0dB" in result

    def test_attack_and_release(self):
        result = build_sidechain_mix_filter(attack=0.01, release=0.3)
        assert "attack=0.010" in result
        assert "release=0.300" in result

    def test_contains_amix(self):
        result = build_sidechain_mix_filter()
        assert "amix=inputs=2" in result
        assert "duration=first" in result

    def test_contains_volume_compensation(self):
        result = build_sidechain_mix_filter()
        assert "volume=1.5" in result

    def test_output_label(self):
        result = build_sidechain_mix_filter()
        assert "[final]" in result

    def test_bgm_comp_label(self):
        result = build_sidechain_mix_filter()
        assert "[bgm_comp]" in result


# ── normalize_bgm_config ────────────────────────────────────────────────────


class TestNormalizeBgmConfig:
    def test_default_values(self):
        result = normalize_bgm_config({})
        assert result["volume"] == 0.3
        assert result["fade_in"] == 0.0
        assert result["fade_out"] == 0.0
        assert result["loop_enabled"] is True
        assert result["sidechain_enabled"] is False
        assert result["sidechain_ratio"] == 0.3
        assert result["sidechain_attack"] == 0.02
        assert result["sidechain_release"] == 0.5
        assert result["sidechain_threshold"] == -25.0

    def test_volume_clamped(self):
        result = normalize_bgm_config({"volume": 2.0})
        assert result["volume"] == 1.0
        result = normalize_bgm_config({"volume": -1.0})
        assert result["volume"] == 0.0

    def test_fade_in_clamped_to_zero(self):
        result = normalize_bgm_config({"fade_in": -5})
        assert result["fade_in"] == 0.0

    def test_fade_out_clamped_to_zero(self):
        result = normalize_bgm_config({"fade_out": -5})
        assert result["fade_out"] == 0.0

    def test_loop_enabled_bool_conversion(self):
        assert normalize_bgm_config({"loop_enabled": True})["loop_enabled"] is True
        assert normalize_bgm_config({"loop_enabled": False})["loop_enabled"] is False
        assert normalize_bgm_config({"loop_enabled": 1})["loop_enabled"] is True
        assert normalize_bgm_config({"loop_enabled": 0})["loop_enabled"] is False

    def test_sidechain_ratio_clamped(self):
        result = normalize_bgm_config({"sidechain_ratio": 2.0})
        assert result["sidechain_ratio"] == 1.0
        result = normalize_bgm_config({"sidechain_ratio": -1.0})
        assert result["sidechain_ratio"] == 0.0

    def test_sidechain_attack_minimum(self):
        result = normalize_bgm_config({"sidechain_attack": 0})
        assert result["sidechain_attack"] == 0.001

    def test_sidechain_release_minimum(self):
        result = normalize_bgm_config({"sidechain_release": 0})
        assert result["sidechain_release"] == 0.01

    def test_sidechain_threshold_pass_through(self):
        result = normalize_bgm_config({"sidechain_threshold": -40.0})
        assert result["sidechain_threshold"] == -40.0

    def test_string_values_converted(self):
        result = normalize_bgm_config(
            {
                "volume": "0.5",
                "fade_in": "1.0",
                "sidechain_ratio": "0.7",
            }
        )
        assert result["volume"] == 0.5
        assert result["fade_in"] == 1.0
        assert result["sidechain_ratio"] == 0.7


# ── validate_bgm_config ─────────────────────────────────────────────────────


class TestValidateBgmConfig:
    def test_valid_config(self):
        valid, errors = validate_bgm_config({"volume": 0.3})
        assert valid is True
        assert errors == []

    def test_invalid_volume_type(self):
        valid, errors = validate_bgm_config({"volume": "abc"})
        assert valid is False
        assert any("volume" in e for e in errors)

    def test_volume_out_of_range(self):
        valid, errors = validate_bgm_config({"volume": -0.1})
        assert valid is False
        assert any("volume" in e for e in errors)
        valid, errors = validate_bgm_config({"volume": 1.1})
        assert valid is False
        assert any("volume" in e for e in errors)

    def test_volume_at_boundaries(self):
        assert validate_bgm_config({"volume": 0})[0] is True
        assert validate_bgm_config({"volume": 1})[0] is True

    def test_invalid_fade_in_type(self):
        valid, errors = validate_bgm_config({"fade_in": "abc"})
        assert valid is False
        assert any("fade_in" in e for e in errors)

    def test_negative_fade_in(self):
        valid, errors = validate_bgm_config({"fade_in": -1})
        assert valid is False
        assert any("fade_in" in e for e in errors)

    def test_invalid_fade_out_type(self):
        valid, errors = validate_bgm_config({"fade_out": "abc"})
        assert valid is False
        assert any("fade_out" in e for e in errors)

    def test_negative_fade_out(self):
        valid, errors = validate_bgm_config({"fade_out": -1})
        assert valid is False
        assert any("fade_out" in e for e in errors)

    def test_invalid_sidechain_ratio_type(self):
        valid, errors = validate_bgm_config({"sidechain_ratio": "abc"})
        assert valid is False
        assert any("sidechain_ratio" in e for e in errors)

    def test_sidechain_ratio_out_of_range(self):
        valid, errors = validate_bgm_config({"sidechain_ratio": -0.1})
        assert valid is False
        assert any("sidechain_ratio" in e for e in errors)
        valid, errors = validate_bgm_config({"sidechain_ratio": 1.1})
        assert valid is False
        assert any("sidechain_ratio" in e for e in errors)

    def test_multiple_errors(self):
        valid, errors = validate_bgm_config(
            {
                "volume": "bad",
                "fade_in": -1,
                "sidechain_ratio": 2.0,
            }
        )
        assert valid is False
        assert len(errors) >= 3


# ── calculate_fade_out_start ────────────────────────────────────────────────


class TestCalculateFadeOutStart:
    def test_normal_case(self):
        assert calculate_fade_out_start(60, 2) == 58.0

    def test_zero_fade_out(self):
        assert calculate_fade_out_start(60, 0) is None

    def test_negative_fade_out(self):
        assert calculate_fade_out_start(60, -1) is None

    def test_zero_target(self):
        assert calculate_fade_out_start(0, 2) is None

    def test_negative_target(self):
        assert calculate_fade_out_start(-5, 2) is None

    def test_fade_longer_than_target(self):
        assert calculate_fade_out_start(10, 20) is None

    def test_fade_equal_to_target(self):
        assert calculate_fade_out_start(10, 10) is None

    def test_float_values(self):
        assert calculate_fade_out_start(60.5, 2.5) == 58.0


# ── estimate_bgm_processing_duration ────────────────────────────────────────


class TestEstimateBgmProcessingDuration:
    def test_bgm_longer_no_loop(self):
        assert estimate_bgm_processing_duration(100, 60, loop_enabled=False) == 60.0

    def test_bgm_longer_with_loop(self):
        # 够长但允许循环，仍然截断到target
        assert estimate_bgm_processing_duration(100, 60, loop_enabled=True) == 60.0

    def test_bgm_shorter_with_loop(self):
        assert estimate_bgm_processing_duration(10, 60, loop_enabled=True) == 60.0

    def test_bgm_shorter_no_loop(self):
        # 需要循环但不允许 → 截断到target
        assert estimate_bgm_processing_duration(10, 60, loop_enabled=False) == 60.0

    def test_zero_target_fallback(self):
        assert estimate_bgm_processing_duration(100, 0) == 5.0

    def test_negative_target_fallback(self):
        assert estimate_bgm_processing_duration(100, -5) == 5.0

    def test_equal_duration(self):
        assert estimate_bgm_processing_duration(60, 60) == 60.0