diff --git a/tests/unit/test_bgm_mixer_pure.py b/tests/unit/test_bgm_mixer_pure.py index 1b4d43ed1..383bea515 100755 --- a/tests/unit/test_bgm_mixer_pure.py +++ b/tests/unit/test_bgm_mixer_pure.py @@ -1,9 +1,6 @@ -"""BGM 混音纯逻辑单元测试.""" +"""bgm_mixer_pure 单元测试.""" -from __future__ import annotations - -import pytest -from video_processing.bgm_mixer_pure import ( +from apps.worker.video_processing.bgm_mixer_pure import ( BGMPureConfig, build_bgm_filter_chain, build_sidechain_mix_filter, @@ -17,408 +14,286 @@ from video_processing.bgm_mixer_pure import ( validate_bgm_config, ) -# ───────────────────────────────────────────────────────────────────────────── -# should_loop_bgm 测试 -# ───────────────────────────────────────────────────────────────────────────── +# ── BGMPureConfig ──────────────────────────────────────────────────────────── -class TestShouldLoopBGM: - """BGM 循环判断测试.""" +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_need_loop_when_much_shorter(self): - """BGM 远短于目标时长，需要循环.""" - assert should_loop_bgm(10, 100, True) is True + 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 - def test_no_loop_when_long_enough(self): - """BGM 够长，不需要循环.""" - assert should_loop_bgm(100, 100, True) is False - def test_no_loop_when_just_slightly_shorter(self): - """BGM 只差一点点（>90%），不循环.""" - assert should_loop_bgm(95, 100, True) is False +# ── should_loop_bgm ───────────────────────────────────────────────────────── - def test_threshold_90_percent(self): - """刚好 90% 阈值，不循环（<90% 才循环）.""" - assert should_loop_bgm(90, 100, True) is False - def test_just_below_threshold(self): - """略低于 90%，需要循环.""" - assert should_loop_bgm(89, 100, True) is True +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): - """禁用循环，即使 BGM 很短也不循环.""" - assert should_loop_bgm(10, 100, False) is False + 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): - """BGM 时长为 0，不循环.""" - assert should_loop_bgm(0, 100, True) is False + assert should_loop_bgm(0, 60) is False def test_negative_bgm_duration(self): - """BGM 时长为负，不循环.""" - assert should_loop_bgm(-5, 100, True) is False + assert should_loop_bgm(-1, 60) is False def test_zero_target_duration(self): - """目标时长为 0，不循环.""" - assert should_loop_bgm(10, 0, True) is False + assert should_loop_bgm(10, 0) is False def test_negative_target_duration(self): - """目标时长为负，不循环.""" - assert should_loop_bgm(10, -10, True) is False + 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 测试 -# ───────────────────────────────────────────────────────────────────────────── +# ── calculate_loop_count ──────────────────────────────────────────────────── class TestCalculateLoopCount: - """循环次数计算测试.""" + def test_exact_fit_returns_1(self): + assert calculate_loop_count(60, 60) == 1 - def test_exact_multiple(self): - """刚好整数倍.""" - # 100/10 = 10, +2 = 12 - assert calculate_loop_count(10, 100) == 12 + def test_bgm_longer_returns_1(self): + assert calculate_loop_count(100, 60) == 1 - def test_not_exact_multiple(self): - """不是整数倍.""" - # 100/30 = 3, +2 = 5 - assert calculate_loop_count(30, 100) == 5 + def test_needs_3_loops_plus_2_margin(self): + # 60/20 = 3 + 2 = 5 + assert calculate_loop_count(20, 60) == 5 - def test_bgm_longer_than_target(self): - """BGM 比目标长，至少 1 次.""" - assert calculate_loop_count(200, 100) == 1 + 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): - """BGM 时长为 0，返回 1.""" - assert calculate_loop_count(0, 100) == 1 + assert calculate_loop_count(0, 60) == 1 def test_negative_bgm_duration(self): - """BGM 时长为负，返回 1.""" - assert calculate_loop_count(-5, 100) == 1 + assert calculate_loop_count(-1, 60) == 1 def test_zero_target_duration(self): - """目标时长为 0，返回 1.""" assert calculate_loop_count(10, 0) == 1 def test_negative_target_duration(self): - """目标时长为负，返回 1.""" - assert calculate_loop_count(10, -10) == 1 + assert calculate_loop_count(10, -1) == 1 - def test_very_short_bgm(self): - """非常短的 BGM，循环次数多.""" - # 100/1 = 100, +2 = 102 - assert calculate_loop_count(1, 100) == 102 + def test_minimum_is_1(self): + assert calculate_loop_count(10, 5) == 1 -# ───────────────────────────────────────────────────────────────────────────── -# build_bgm_filter_chain 测试 -# ───────────────────────────────────────────────────────────────────────────── +# ── build_bgm_filter_chain ────────────────────────────────────────────────── -class TestBuildBGMFilterChain: - """BGM 预处理滤镜链构建测试.""" +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_basic_volume_only(self): - """只有音量调节.""" - result = build_bgm_filter_chain( - bgm_duration=200, - target_duration=100, - volume=0.5, - ) + def test_volume_filter_applied(self): + result = build_bgm_filter_chain(100, 60, volume=0.5) assert "volume=0.500" in result - assert "aloop" not in result - assert "afade=t=in" not in result - assert "afade=t=out" not in result - assert "atrim=0:100.000" in result - assert "asetpts=N/SR/TB" in result - def test_with_loop(self): - """需要循环的情况.""" - result = build_bgm_filter_chain( - bgm_duration=10, - target_duration=100, - volume=0.3, - loop_enabled=True, - ) - assert "aloop=loop=" in result - assert "volume=0.300" 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_no_loop_when_disabled(self): - """禁用循环，即使 BGM 短也不循环.""" - result = build_bgm_filter_chain( - bgm_duration=10, - target_duration=100, - volume=0.3, - loop_enabled=False, - ) - assert "aloop" 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_fade_in_only(self): - """只有淡入.""" - result = build_bgm_filter_chain( - bgm_duration=200, - target_duration=100, - volume=1.0, - fade_in=2.5, - ) - assert "afade=t=in:st=0:d=2.500" in result - assert "afade=t=out" not in result - assert "volume=" not in result # volume=1.0 不加 + def test_volume_zero(self): + result = build_bgm_filter_chain(100, 60, volume=0.0) + assert "volume=0.000" in result - def test_fade_out_only(self): - """只有淡出.""" - result = build_bgm_filter_chain( - bgm_duration=200, - target_duration=100, - volume=1.0, - fade_out=3.0, - ) - assert "afade=t=out:st=97.000:d=3.000" in result - assert "afade=t=in" not in result - - def test_fade_in_and_out(self): - """淡入+淡出.""" - result = build_bgm_filter_chain( - bgm_duration=200, - target_duration=100, - volume=1.0, - fade_in=1.5, - fade_out=2.0, - ) + 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 - assert "afade=t=out:st=98.000:d=2.000" in result - def test_volume_1_0_skipped(self): - """音量为 1.0 时不添加 volume 滤镜.""" - result = build_bgm_filter_chain( - bgm_duration=200, - target_duration=100, - volume=1.0, - ) - assert "volume=" not 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_volume_0(self): - """音量为 0.""" - result = build_bgm_filter_chain( - bgm_duration=200, - target_duration=100, - volume=0.0, - ) - assert "volume=0.000" 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_volume_clamped_high(self): - """音量超过 1.0 被钳制.""" - result = build_bgm_filter_chain( - bgm_duration=200, - target_duration=100, - volume=1.5, - ) - assert "volume=1.000" not in result # 1.0不加 - # 钳制到1.0后和1.0一样，不加volume滤镜 - # 但因为abs(1.0 - 1.0) < 0.001，所以不添加 - assert "volume=" not in result - - def test_volume_clamped_low(self): - """音量为负被钳制到 0.""" - result = build_bgm_filter_chain( - bgm_duration=200, - target_duration=100, - volume=-0.5, - ) - assert "volume=0.000" in result - - def test_fade_out_longer_than_duration(self): - """淡出时长超过总时长，不加淡出.""" - result = build_bgm_filter_chain( - bgm_duration=200, - target_duration=10, - volume=1.0, - fade_out=20.0, - ) + 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_fade_out_equal_to_duration(self): - """淡出时长等于总时长，不加淡出.""" - result = build_bgm_filter_chain( - bgm_duration=200, - target_duration=10, - volume=1.0, - fade_out=10.0, - ) - 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_zero_target_duration_fallback(self): - """目标时长为 0，兜底 5 秒.""" - result = build_bgm_filter_chain( - bgm_duration=3, - target_duration=0, - volume=0.5, - ) + 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_duration_fallback(self): - """目标时长为负，兜底 5 秒.""" - result = build_bgm_filter_chain( - bgm_duration=3, - target_duration=-5, - volume=0.5, - ) + def test_negative_target_uses_fallback(self): + result = build_bgm_filter_chain(100, -5) assert "atrim=0:5.000" in result - def test_full_chain_with_all_effects(self): - """完整滤镜链：循环+音量+淡入淡出+截断+重置.""" + def test_all_features_combined(self): result = build_bgm_filter_chain( - bgm_duration=10, - target_duration=100, - volume=0.4, + bgm_duration=15, + target_duration=60, + volume=0.3, fade_in=1.0, fade_out=2.0, loop_enabled=True, ) - parts = result.split(",") - # 顺序：aloop -> volume -> afade in -> afade out -> atrim -> asetpts - assert len(parts) >= 6 - assert "aloop" in parts[0] - assert "volume" in parts[1] - assert "afade=t=in" in parts[2] - assert "afade=t=out" in parts[3] - assert "atrim" in parts[4] - assert "asetpts" in parts[5] + 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 测试 -# ───────────────────────────────────────────────────────────────────────────── +# ── calculate_sidechain_ratio ─────────────────────────────────────────────── class TestCalculateSidechainRatio: - """Sidechain 压缩比计算测试.""" + def test_zero_ratio_minimum(self): + assert calculate_sidechain_ratio(0) == 2.0 - def test_default_ratio_0_3(self): - """默认 0.3.""" - # 1 / (1 - 0.3) = 1.428... 但下限是 2.0 - assert calculate_sidechain_ratio(0.3) == pytest.approx(2.0, rel=0.01) - - def test_ratio_0_5(self): - """比例 0.5.""" - # 1 / (1 - 0.5) = 2.0 - assert calculate_sidechain_ratio(0.5) == pytest.approx(2.0, rel=0.01) - - def test_ratio_0_8(self): - """比例 0.8.""" - # 1 / (1 - 0.8) = 5.0 - assert calculate_sidechain_ratio(0.8) == pytest.approx(5.0, rel=0.01) - - def test_ratio_0_9(self): - """比例 0.9.""" - # 1 / (1 - 0.9) = 10.0 - assert calculate_sidechain_ratio(0.9) == pytest.approx(10.0, rel=0.01) - - def test_ratio_0(self): - """比例 0，返回下限 2.0.""" - assert calculate_sidechain_ratio(0.0) == 2.0 - - def test_ratio_negative(self): - """比例为负，返回下限 2.0.""" + def test_negative_clamped(self): assert calculate_sidechain_ratio(-0.5) == 2.0 - def test_ratio_1_0(self): - """比例 1.0，返回上限 10.0.""" + def test_one_ratio_maximum(self): assert calculate_sidechain_ratio(1.0) == 10.0 - def test_ratio_greater_than_1(self): - """比例超过 1.0，返回上限 10.0.""" - assert calculate_sidechain_ratio(2.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 测试 -# ───────────────────────────────────────────────────────────────────────────── +# ── build_simple_mix_filter ───────────────────────────────────────────────── class TestBuildSimpleMixFilter: - """普通混音滤镜构建测试.""" - - def test_contains_amix(self): - """包含 amix.""" + 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 - - def test_contains_volume_compensation(self): - """包含 volume=2 补偿.""" - result = build_simple_mix_filter() + assert "duration=first" in result + assert "[final]" in result assert "volume=2" in result - def test_output_label(self): - """输出标签为 [final].""" - result = build_simple_mix_filter() - assert "[final]" in result - def test_duration_first(self): - """duration=first，以主音频时长为准.""" - result = build_simple_mix_filter() - assert "duration=first" in result - - -# ───────────────────────────────────────────────────────────────────────────── -# build_sidechain_mix_filter 测试 -# ───────────────────────────────────────────────────────────────────────────── +# ── build_sidechain_mix_filter ────────────────────────────────────────────── class TestBuildSidechainMixFilter: - """Sidechain 混音滤镜构建测试.""" - def test_contains_sidechaincompress(self): - """包含 sidechaincompress.""" result = build_sidechain_mix_filter() assert "sidechaincompress=" in result + assert "[1:a][0:a]sidechaincompress" in result - def test_threshold_param(self): - """threshold 参数正确.""" + def test_threshold_in_db(self): result = build_sidechain_mix_filter(threshold=-30.0) assert "threshold=-30.0dB" in result - def test_attack_param(self): - """attack 参数正确.""" - result = build_sidechain_mix_filter(attack=0.05) - assert "attack=0.050" in result - - def test_release_param(self): - """release 参数正确.""" - result = build_sidechain_mix_filter(release=0.8) - assert "release=0.800" in result - - def test_knee_param(self): - """knee=6 参数.""" - result = build_sidechain_mix_filter() - assert "knee=6" 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): - """包含 amix 混音.""" result = build_sidechain_mix_filter() assert "amix=inputs=2" in result + assert "duration=first" in result - def test_volume_compensation(self): - """volume=1.5 轻微补偿.""" + def test_contains_volume_compensation(self): result = build_sidechain_mix_filter() assert "volume=1.5" in result - def test_bgmc_comp_label(self): - """包含 [bgm_comp] 中间标签.""" + 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 测试 -# ───────────────────────────────────────────────────────────────────────────── +# ── normalize_bgm_config ──────────────────────────────────────────────────── -class TestNormalizeBGMConfig: - """配置规范化测试.""" - - def test_empty_dict_defaults(self): - """空字典返回默认值.""" +class TestNormalizeBgmConfig: + def test_default_values(self): result = normalize_bgm_config({}) assert result["volume"] == 0.3 assert result["fade_in"] == 0.0 @@ -426,233 +301,184 @@ class TestNormalizeBGMConfig: 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": 1.5}) + result = normalize_bgm_config({"volume": 2.0}) assert result["volume"] == 1.0 - result2 = normalize_bgm_config({"volume": -0.5}) - assert result2["volume"] == 0.0 + result = normalize_bgm_config({"volume": -1.0}) + assert result["volume"] == 0.0 - def test_fade_in_negative(self): - """淡入为负钳制到 0.""" - result = normalize_bgm_config({"fade_in": -1}) + 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_negative(self): - """淡出为负钳制到 0.""" - result = normalize_bgm_config({"fade_out": -1}) + def test_fade_out_clamped_to_zero(self): + result = normalize_bgm_config({"fade_out": -5}) assert result["fade_out"] == 0.0 - def test_sidechain_ratio_clamped(self): - """sidechain_ratio 钳制.""" - result = normalize_bgm_config({"sidechain_ratio": 1.5}) - assert result["sidechain_ratio"] == 1.0 - result2 = normalize_bgm_config({"sidechain_ratio": -0.1}) - assert result2["sidechain_ratio"] == 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_attack_min(self): - """attack 最小值 0.001.""" + 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_min(self): - """release 最小值 0.01.""" + 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": "2.0", + "fade_in": "1.0", + "sidechain_ratio": "0.7", } ) assert result["volume"] == 0.5 - assert result["fade_in"] == 2.0 - - def test_loop_enabled_truthy(self): - """loop_enabled 真值转换.""" - result = normalize_bgm_config({"loop_enabled": 1}) - assert result["loop_enabled"] is True - result2 = normalize_bgm_config({"loop_enabled": 0}) - assert result2["loop_enabled"] is False - - def test_preserves_unknown_keys(self): - """未知 key 不保留.""" - result = normalize_bgm_config({"unknown_key": "value", "volume": 0.5}) - assert "unknown_key" not in result - assert result["volume"] == 0.5 + assert result["fade_in"] == 1.0 + assert result["sidechain_ratio"] == 0.7 -# ───────────────────────────────────────────────────────────────────────────── -# validate_bgm_config 测试 -# ───────────────────────────────────────────────────────────────────────────── +# ── validate_bgm_config ───────────────────────────────────────────────────── -class TestValidateBGMConfig: - """配置验证测试.""" - +class TestValidateBgmConfig: def test_valid_config(self): - """合法配置.""" - ok, errors = validate_bgm_config( - { - "volume": 0.5, - "fade_in": 1.0, - "fade_out": 2.0, - "sidechain_ratio": 0.3, - } - ) - assert ok is True - assert len(errors) == 0 + valid, errors = validate_bgm_config({"volume": 0.3}) + assert valid is True + assert errors == [] - def test_volume_not_number(self): - """volume 不是数字.""" - ok, errors = validate_bgm_config({"volume": "high"}) - assert ok is False + 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): - """volume 超出范围.""" - ok, errors = validate_bgm_config({"volume": 1.5}) - assert ok is False + 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_fade_in_negative(self): - """fade_in 为负.""" - ok, errors = validate_bgm_config({"fade_in": -1}) - assert ok is False + 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_fade_out_negative(self): - """fade_out 为负.""" - ok, errors = validate_bgm_config({"fade_out": -1}) - assert ok is False + 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): - """sidechain_ratio 超出范围.""" - ok, errors = validate_bgm_config({"sidechain_ratio": 2.0}) - assert ok is False + 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): - """多个错误同时报告.""" - ok, errors = validate_bgm_config( + valid, errors = validate_bgm_config( { - "volume": 2.0, + "volume": "bad", "fade_in": -1, - "sidechain_ratio": -0.5, + "sidechain_ratio": 2.0, } ) - assert ok is False + assert valid is False assert len(errors) >= 3 - def test_empty_config_valid(self): - """空配置（全用默认值）视为合法.""" - ok, errors = validate_bgm_config({}) - assert ok is True - assert len(errors) == 0 - -# ───────────────────────────────────────────────────────────────────────────── -# calculate_fade_out_start 测试 -# ───────────────────────────────────────────────────────────────────────────── +# ── calculate_fade_out_start ──────────────────────────────────────────────── class TestCalculateFadeOutStart: - """淡出开始时间计算测试.""" - def test_normal_case(self): - """正常情况.""" - assert calculate_fade_out_start(100, 3) == pytest.approx(97.0) + assert calculate_fade_out_start(60, 2) == 58.0 def test_zero_fade_out(self): - """淡出时长为 0，返回 None.""" - assert calculate_fade_out_start(100, 0) is None + assert calculate_fade_out_start(60, 0) is None def test_negative_fade_out(self): - """淡出时长为负，返回 None.""" - assert calculate_fade_out_start(100, -1) is None - - def test_zero_duration(self): - """总时长为 0，返回 None.""" - assert calculate_fade_out_start(0, 3) is None - - def test_fade_out_longer_than_duration(self): - """淡出超过总时长，返回 None.""" - assert calculate_fade_out_start(10, 20) is None - - def test_fade_out_equal_to_duration(self): - """淡出等于总时长，返回 None.""" - assert calculate_fade_out_start(10, 10) is None - - -# ───────────────────────────────────────────────────────────────────────────── -# estimate_bgm_processing_duration 测试 -# ───────────────────────────────────────────────────────────────────────────── - - -class TestEstimateBGMProcessingDuration: - """BGM 处理时长估算测试.""" - - def test_normal_case_with_loop(self): - """正常循环情况，输出目标时长.""" - assert estimate_bgm_processing_duration(10, 100, True) == 100 - - def test_bgm_longer_no_loop(self): - """BGM 够长，不循环，截断到目标时长.""" - assert estimate_bgm_processing_duration(200, 100, False) == 100 - - def test_bgm_shorter_no_loop(self): - """BGM 短但不循环，仍然截断到目标时长（实际会更短，但 atrim 会截断）.""" - assert estimate_bgm_processing_duration(10, 100, False) == 100 + assert calculate_fade_out_start(60, -1) is None def test_zero_target(self): - """目标时长为 0，兜底 5 秒.""" - assert estimate_bgm_processing_duration(10, 0, True) == 5.0 + assert calculate_fade_out_start(0, 2) is None def test_negative_target(self): - """目标时长为负，兜底 5 秒.""" - assert estimate_bgm_processing_duration(10, -5, True) == 5.0 + 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 -# ───────────────────────────────────────────────────────────────────────────── -# BGMPureConfig 测试 -# ───────────────────────────────────────────────────────────────────────────── +# ── estimate_bgm_processing_duration ──────────────────────────────────────── -class TestBGMPureConfig: - """BGMPureConfig 数据类测试.""" +class TestEstimateBgmProcessingDuration: + def test_bgm_longer_no_loop(self): + assert estimate_bgm_processing_duration(100, 60, loop_enabled=False) == 60.0 - def test_default_values(self): - """默认值正确.""" - config = BGMPureConfig() - assert config.volume == 0.3 - assert config.fade_in == 0.0 - assert config.fade_out == 0.0 - assert config.loop_enabled is True - assert config.sidechain_enabled is False - assert config.sidechain_ratio == 0.3 - assert config.sidechain_attack == 0.02 - assert config.sidechain_release == 0.5 - assert config.sidechain_threshold == -25.0 + def test_bgm_longer_with_loop(self): + # 够长但允许循环，仍然截断到target + assert estimate_bgm_processing_duration(100, 60, loop_enabled=True) == 60.0 - def test_custom_values(self): - """自定义值.""" - config = BGMPureConfig( - volume=0.7, - fade_in=1.0, - fade_out=2.0, - loop_enabled=False, - sidechain_enabled=True, - sidechain_ratio=0.5, - sidechain_attack=0.05, - sidechain_release=0.8, - sidechain_threshold=-30.0, - ) - assert config.volume == 0.7 - assert config.loop_enabled is False - assert config.sidechain_enabled is True - assert config.sidechain_threshold == -30.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 diff --git a/tests/unit/test_multi_track_mixer_pure.py b/tests/unit/test_multi_track_mixer_pure.py index 9d6c2376b..ee63391ae 100755 --- a/tests/unit/test_multi_track_mixer_pure.py +++ b/tests/unit/test_multi_track_mixer_pure.py @@ -1,11 +1,8 @@ -"""多轨混音纯逻辑单元测试.""" - -from __future__ import annotations +"""multi_track_mixer_pure 单元测试.""" import math -import pytest -from video_processing.multi_track_mixer_pure import ( +from apps.worker.video_processing.multi_track_mixer_pure import ( build_amix_filter, build_mix_filter_complex, build_track_filter_chain, @@ -24,615 +21,747 @@ from video_processing.multi_track_mixer_pure import ( validate_mix_config, ) -# ───────────────────────────────────────────────────────────────────────────── -# 时间计算测试 -# ───────────────────────────────────────────────────────────────────────────── +# ── calculate_effective_range ─────────────────────────────────────────────── class TestCalculateEffectiveRange: - """有效时间范围计算测试.""" - - def test_normal_track(self): - """正常轨道.""" - start, dur, trim = calculate_effective_range(5, 10, 30, 60) - assert start == 5.0 - assert dur == 10.0 + def test_simple_inside_target(self): + start, need, trim = calculate_effective_range( + track_start=2.0, + track_duration=5.0, + audio_duration=10.0, + target_duration=20.0, + ) + assert start == 2.0 + assert need == 5.0 assert trim == 0.0 - def test_track_longer_than_audio(self): - """轨道时长超过音频长度.""" - start, dur, trim = calculate_effective_range(0, 100, 30, 60) + def test_zero_duration_uses_full_audio(self): + start, need, trim = calculate_effective_range( + track_start=1.0, + track_duration=0, + audio_duration=8.0, + target_duration=20.0, + ) + assert start == 1.0 + assert need == 8.0 + assert trim == 0.0 + + def test_negative_start_trims_beginning(self): + start, need, trim = calculate_effective_range( + track_start=-2.0, + track_duration=0, + audio_duration=10.0, + target_duration=20.0, + ) assert start == 0.0 - assert dur == 30.0 # 用音频全长 + assert need == 8.0 # 10 - 2 + assert trim == 2.0 - def test_zero_track_duration(self): - """轨道时长为 0（用音频全长）.""" - start, dur, trim = calculate_effective_range(0, 0, 30, 60) + def test_starts_after_target_duration(self): + start, need, trim = calculate_effective_range( + track_start=25.0, + track_duration=5.0, + audio_duration=10.0, + target_duration=20.0, + ) assert start == 0.0 - assert dur == 30.0 + assert need == 0.0 + assert trim == 0.0 - def test_negative_start_time(self): - """负开始时间（从音频中间取）.""" - start, dur, trim = calculate_effective_range(-5, 20, 30, 60) - assert start == 0.0 - assert dur == 15.0 # 20 - 5 = 15 - assert trim == 5.0 + def test_ends_before_zero(self): + start, need, trim = calculate_effective_range( + track_start=-10.0, + track_duration=5.0, + audio_duration=10.0, + target_duration=20.0, + ) + assert need == 0.0 - def test_track_after_target(self): - """轨道完全在目标之后.""" - start, dur, trim = calculate_effective_range(100, 10, 30, 60) - assert dur == 0.0 - - def test_track_before_zero(self): - """轨道完全在 0 之前.""" - start, dur, trim = calculate_effective_range(-50, 10, 30, 60) - assert dur == 0.0 + def test_truncated_at_end(self): + start, need, trim = calculate_effective_range( + track_start=15.0, + track_duration=10.0, + audio_duration=10.0, + target_duration=20.0, + ) + assert start == 15.0 + assert need == 5.0 # 截断到目标时长 + assert trim == 0.0 def test_zero_audio_duration(self): - """音频时长为 0.""" - start, dur, trim = calculate_effective_range(0, 10, 0, 60) - assert dur == 0.0 + start, need, trim = calculate_effective_range( + track_start=0, + track_duration=10, + audio_duration=0, + target_duration=20.0, + ) + assert need == 0.0 - def test_track_extends_beyond_target(self): - """轨道超出目标时长.""" - start, dur, trim = calculate_effective_range(50, 20, 30, 60) - assert start == 50.0 - assert dur == 10.0 # 60 - 50 = 10 + def test_negative_audio_duration(self): + start, need, trim = calculate_effective_range( + track_start=0, + track_duration=10, + audio_duration=-1, + target_duration=20.0, + ) + assert need == 0.0 - def test_full_target_duration(self): - """轨道覆盖整个目标时长.""" - start, dur, trim = calculate_effective_range(0, 0, 100, 60) - assert start == 0.0 - assert dur == 60.0 + def test_track_longer_than_audio(self): + start, need, trim = calculate_effective_range( + track_start=0, + track_duration=20, + audio_duration=10, + target_duration=30, + ) + assert need == 10.0 # 受限于音频长度 + + def test_trim_start_exceeds_audio(self): + start, need, trim = calculate_effective_range( + track_start=-15.0, + track_duration=0, + audio_duration=10.0, + target_duration=20.0, + ) + # 被截掉15秒，但音频只有10秒 → 全没了 + assert need == 0.0 + + +# ── is_track_visible ──────────────────────────────────────────────────────── class TestIsTrackVisible: - """轨道可见性测试.""" - def test_visible_track(self): - """可见轨道.""" - assert is_track_visible(5, 10, 30, 60) is True + assert is_track_visible(2, 5, 10, 20) is True def test_invisible_after_target(self): - """目标之后不可见.""" - assert is_track_visible(100, 10, 30, 60) is False + assert is_track_visible(25, 5, 10, 20) is False - def test_invisible_zero_duration(self): - """零时长不可见.""" - assert is_track_visible(0, 0, 0, 60) is False + def test_invisible_zero_audio(self): + assert is_track_visible(0, 10, 0, 20) is False + + def test_invisible_all_trimmed(self): + assert is_track_visible(-20, 10, 10, 20) is False -# ───────────────────────────────────────────────────────────────────────────── -# 滤镜链构建测试 -# ───────────────────────────────────────────────────────────────────────────── +# ── build_track_filter_chain ──────────────────────────────────────────────── class TestBuildTrackFilterChain: - """单轨滤镜链构建测试.""" - - def test_basic_structure(self): - """基本结构：截断+音量+淡入淡出+延迟+截断.""" - result = build_track_filter_chain( - volume=0.5, - fade_in=1.0, - fade_out=1.0, - effective_start=5.0, - need_duration=10.0, - trim_start=0.0, - target_duration=60.0, - ) - assert "atrim=0.000:10.000" in result - assert "volume=0.500" in result - assert "afade=t=in:st=0:d=1.000" in result - assert "afade=t=out" in result - assert "adelay=5000|5000" in result - assert "atrim=0:60.000" in result - - def test_volume_1_0_skipped(self): - """音量为 1.0 不添加 volume 滤镜.""" + def test_basic_chain_structure(self): result = build_track_filter_chain( volume=1.0, fade_in=0, fade_out=0, effective_start=0, - need_duration=10, + need_duration=5.0, trim_start=0, - target_duration=60, + target_duration=10.0, + ) + parts = result.split(",") + # 至少有: atrim, asetpts, atrim, asetpts + assert any("atrim=" in p for p in parts) + assert parts.count("asetpts=N/SR/TB") == 2 + + def test_volume_filter_applied(self): + result = build_track_filter_chain( + volume=0.5, + fade_in=0, + fade_out=0, + effective_start=0, + need_duration=5.0, + trim_start=0, + target_duration=10.0, + ) + assert "volume=0.500" in result + + def test_volume_one_omitted(self): + result = build_track_filter_chain( + volume=1.0, + fade_in=0, + fade_out=0, + effective_start=0, + need_duration=5.0, + trim_start=0, + target_duration=10.0, ) assert "volume=" not in result - def test_no_fade_in(self): - """无淡入.""" + def test_volume_clamped(self): result = build_track_filter_chain( - volume=1.0, - fade_in=0, - fade_out=2.0, - effective_start=0, - need_duration=10, - trim_start=0, - target_duration=60, - ) - assert "afade=t=in" not in result - assert "afade=t=out" in result - - def test_no_delay(self): - """无延迟（effective_start 很小）.""" - result = build_track_filter_chain( - volume=1.0, + volume=3.0, fade_in=0, fade_out=0, - effective_start=0.001, - need_duration=10, + effective_start=0, + need_duration=5.0, trim_start=0, - target_duration=60, + target_duration=10.0, ) - assert "adelay" not in result + assert "volume=2.000" in result # 钳制到2.0 - def test_with_delay(self): - """有延迟.""" + def test_volume_negative_clamped(self): + result = build_track_filter_chain( + volume=-1.0, + fade_in=0, + fade_out=0, + effective_start=0, + need_duration=5.0, + trim_start=0, + target_duration=10.0, + ) + assert "volume=0.000" in result + + def test_fade_in_applied(self): + result = build_track_filter_chain( + volume=1.0, + fade_in=1.0, + fade_out=0, + effective_start=0, + need_duration=5.0, + trim_start=0, + target_duration=10.0, + ) + assert "afade=t=in:st=0:d=1.000" in result + + def test_fade_in_longer_than_duration_skipped(self): + result = build_track_filter_chain( + volume=1.0, + fade_in=10.0, + fade_out=0, + effective_start=0, + need_duration=5.0, + trim_start=0, + target_duration=10.0, + ) + assert "afade=t=in" not in result + + def test_fade_out_applied(self): + result = build_track_filter_chain( + volume=1.0, + fade_in=0, + fade_out=1.0, + effective_start=0, + need_duration=5.0, + trim_start=0, + target_duration=10.0, + ) + assert "afade=t=out:st=4.000:d=1.000" in result + + def test_fade_out_longer_than_duration_skipped(self): + result = build_track_filter_chain( + volume=1.0, + fade_in=0, + fade_out=10.0, + effective_start=0, + need_duration=5.0, + trim_start=0, + target_duration=10.0, + ) + assert "afade=t=out" not in result + + def test_delay_applied(self): result = build_track_filter_chain( volume=1.0, fade_in=0, fade_out=0, effective_start=2.5, - need_duration=10, + need_duration=5.0, trim_start=0, - target_duration=60, + target_duration=10.0, ) assert "adelay=2500|2500" in result - def test_fade_in_longer_than_duration(self): - """淡入超过总时长，不添加淡入.""" - result = build_track_filter_chain( - volume=1.0, - fade_in=20, - fade_out=0, - effective_start=0, - need_duration=10, - trim_start=0, - target_duration=60, - ) - assert "afade=t=in" not in result - - def test_fade_out_at_start(self): - """淡出从 0 开始（很短的音频）.""" - result = build_track_filter_chain( - volume=1.0, - fade_in=0, - fade_out=15, - effective_start=0, - need_duration=10, - trim_start=0, - target_duration=60, - ) - # fade_out > need_duration，不添加 - assert "afade=t=out" not in result - - def test_trim_start_nonzero(self): - """从音频中间开始截取.""" + def test_zero_delay_skipped(self): result = build_track_filter_chain( volume=1.0, fade_in=0, fade_out=0, effective_start=0, - need_duration=5, - trim_start=3.0, - target_duration=60, + need_duration=5.0, + trim_start=0, + target_duration=10.0, ) - assert "atrim=3.000:8.000" in result # 3.0 to 3.0+5.0 + assert "adelay" not in result + + def test_final_truncation_exists(self): + result = build_track_filter_chain( + volume=1.0, + fade_in=0, + fade_out=0, + effective_start=0, + need_duration=5.0, + trim_start=0, + target_duration=10.0, + ) + assert "atrim=0:10.000" in result # 最终截断 + + def test_full_chain_with_all_features(self): + result = build_track_filter_chain( + volume=0.8, + fade_in=0.5, + fade_out=1.0, + effective_start=2.0, + need_duration=6.0, + trim_start=1.0, + target_duration=10.0, + ) + # 有atrim开头截断 + assert "atrim=1.000:7.000" in result + # 有音量 + assert "volume=0.800" in result + # 有淡入淡出 + assert "afade=t=in" in result + assert "afade=t=out" in result + # 有延迟 + assert "adelay=2000|2000" in result + # 有最终截断 + assert "atrim=0:10.000" in result -# ───────────────────────────────────────────────────────────────────────────── -# amix 滤镜测试 -# ───────────────────────────────────────────────────────────────────────────── +# ── build_amix_filter ─────────────────────────────────────────────────────── class TestBuildAmixFilter: - """amix 滤镜构建测试.""" - - def test_two_inputs(self): - """两路输入.""" - result = build_amix_filter(2) - assert "amix=inputs=2" in result - assert "duration=longest" in result - - def test_five_inputs(self): - """五路输入.""" - result = build_amix_filter(5) - assert "amix=inputs=5" in result - def test_zero_inputs(self): - """零输入.""" assert build_amix_filter(0) == "" - def test_duration_shortest(self): - """shortest 模式.""" - result = build_amix_filter(3, "shortest") + def test_negative_inputs(self): + assert build_amix_filter(-1) == "" + + def test_single_input(self): + result = build_amix_filter(1) + assert "amix=inputs=1:" in result + assert "duration=longest" in result + assert "dropout_transition=0" in result + + def test_multiple_inputs(self): + result = build_amix_filter(5) + assert "amix=inputs=5:" in result + + def test_shortest_mode(self): + result = build_amix_filter(3, duration_mode="shortest") assert "duration=shortest" in result - def test_invalid_duration_mode(self): - """无效模式，默认 longest.""" - result = build_amix_filter(3, "invalid") + def test_first_mode(self): + result = build_amix_filter(3, duration_mode="first") + assert "duration=first" in result + + def test_invalid_mode_falls_back(self): + result = build_amix_filter(3, duration_mode="invalid") assert "duration=longest" in result +# ── calculate_amix_volume_compensation ────────────────────────────────────── + + class TestCalculateAmixVolumeCompensation: - """音量补偿计算测试.""" + def test_zero_inputs(self): + assert calculate_amix_volume_compensation(0) == 1.0 - def test_single_track(self): - """单轨，无需补偿.""" + def test_single_input(self): assert calculate_amix_volume_compensation(1) == 1.0 - def test_two_tracks(self): - """两轨，补偿 2x.""" + def test_two_inputs(self): assert calculate_amix_volume_compensation(2) == 2.0 - def test_five_tracks(self): - """五轨，补偿 5x.""" + def test_five_inputs(self): assert calculate_amix_volume_compensation(5) == 5.0 - def test_zero_tracks(self): - """零轨，返回 1.""" - assert calculate_amix_volume_compensation(0) == 1.0 + def test_negative_inputs(self): + assert calculate_amix_volume_compensation(-1) == 1.0 + + +# ── build_mix_filter_complex ──────────────────────────────────────────────── class TestBuildMixFilterComplex: - """完整混音滤镜测试.""" - - def test_with_main_and_two_tracks(self): - """主音频 + 2 条轨道.""" - result = build_mix_filter_complex(2, has_main=True) - assert "[0:a][1:a][2:a]" in result # 3 路输入 - assert "amix=inputs=3" in result - assert "volume=3" in result # 3x 补偿 - assert "[mixed]" in result - - def test_no_main_three_tracks(self): - """无主音频，3 条轨道.""" - result = build_mix_filter_complex(3, has_main=False) - assert "[0:a][1:a][2:a]" in result - assert "amix=inputs=3" in result - assert "[mixed]" in result - - def test_zero_tracks_no_main(self): - """无轨道无主音频.""" + def test_no_tracks_no_main_empty(self): assert build_mix_filter_complex(0, has_main=False) == "" + def test_main_only(self): + result = build_mix_filter_complex(0, has_main=True) + assert "[0:a]" in result + assert "amix=inputs=1:" in result + assert "[mixed]" in result + # 单路无音量补偿 + assert "volume=" not in result -# ───────────────────────────────────────────────────────────────────────────── -# 音量计算测试 -# ───────────────────────────────────────────────────────────────────────────── + def test_main_plus_tracks(self): + result = build_mix_filter_complex(2, has_main=True) + assert "[0:a][1:a][2:a]" in result + assert "amix=inputs=3:" in result + # 3路有音量补偿 + assert "volume=3.0" in result + + def test_tracks_only_no_main(self): + result = build_mix_filter_complex(3, has_main=False) + assert "[0:a][1:a][2:a]" in result + assert "amix=inputs=3:" in result + assert "volume=3.0" in result + + def test_duration_mode_passed(self): + result = build_mix_filter_complex(2, has_main=True, duration_mode="shortest") + assert "duration=shortest" in result + + def test_output_label(self): + result = build_mix_filter_complex(2, has_main=True) + assert result.endswith("[mixed]") + + +# ── normalize_volume ──────────────────────────────────────────────────────── class TestNormalizeVolume: - """音量规范化测试.""" - - def test_normal_volume(self): - """正常音量.""" - assert normalize_volume(0.5) == 0.5 - - def test_none_default(self): - """None 默认 1.0.""" + def test_none_returns_one(self): assert normalize_volume(None) == 1.0 - def test_below_zero_clamped(self): - """负值钳制到 0.""" - assert normalize_volume(-5) == 0.0 + def test_normal_value(self): + assert normalize_volume(0.5) == 0.5 + + def test_max_value(self): + assert normalize_volume(2.0) == 2.0 def test_above_max_clamped(self): - """超过上限钳制.""" assert normalize_volume(3.0) == 2.0 - def test_string_input(self): - """字符串输入.""" + def test_below_min_clamped(self): + assert normalize_volume(-1.0) == 0.0 + + def test_zero(self): + assert normalize_volume(0) == 0.0 + + def test_string_number(self): assert normalize_volume("0.5") == 0.5 def test_invalid_string(self): - """无效字符串默认 1.0.""" assert normalize_volume("abc") == 1.0 -class TestDbConversion: - """dB 转换测试.""" +# ── db_to_linear / linear_to_db ───────────────────────────────────────────── - def test_0_db_is_unity(self): - """0 dB = 1.0.""" - assert db_to_linear(0) == pytest.approx(1.0) - def test_negative_db(self): - """负 dB < 1.""" - assert db_to_linear(-6) == pytest.approx(0.5, rel=0.01) +class TestDbConversions: + def test_zero_db_is_one(self): + assert abs(db_to_linear(0) - 1.0) < 0.001 - def test_positive_db(self): - """正 dB > 1.""" - assert db_to_linear(6) == pytest.approx(2.0, rel=0.01) + def test_negative_db_less_than_one(self): + assert db_to_linear(-20) < 1.0 - def test_round_trip(self): - """往返转换.""" + def test_positive_db_greater_than_one(self): + assert db_to_linear(20) > 1.0 + + def test_roundtrip_conversion(self): original = 0.5 db = linear_to_db(original) - result = db_to_linear(db) - assert result == pytest.approx(original) + back = db_to_linear(db) + assert abs(back - original) < 0.001 - def test_zero_linear_is_negative_inf(self): - """零线性值 = -inf dB.""" + def test_20db_is_10x(self): + # 20dB = 10倍 + assert abs(db_to_linear(20) - 10.0) < 0.001 + + def test_linear_zero_is_neg_inf(self): assert math.isinf(linear_to_db(0)) assert linear_to_db(0) < 0 + def test_linear_negative_is_neg_inf(self): + assert math.isinf(linear_to_db(-1)) -# ───────────────────────────────────────────────────────────────────────────── -# 轨道排序与过滤测试 -# ───────────────────────────────────────────────────────────────────────────── + +# ── sort_tracks_by_priority ───────────────────────────────────────────────── class TestSortTracksByPriority: - """轨道优先级排序测试.""" - - def test_sorted_by_priority(self): - """按优先级排序.""" + def test_sorted_ascending(self): tracks = [ - {"priority": 10, "name": "high"}, - {"priority": 1, "name": "highest"}, - {"priority": 100, "name": "low"}, + {"name": "c", "priority": 3}, + {"name": "a", "priority": 1}, + {"name": "b", "priority": 2}, ] result = sort_tracks_by_priority(tracks) - assert result[0]["name"] == "highest" - assert result[1]["name"] == "high" - assert result[2]["name"] == "low" + assert [t["name"] for t in result] == ["a", "b", "c"] def test_default_priority_100(self): - """默认优先级 100.""" tracks = [ - {"priority": 50, "name": "mid"}, - {"name": "default"}, + {"name": "low", "priority": 50}, + {"name": "default"}, # 默认100 + {"name": "high", "priority": 150}, ] result = sort_tracks_by_priority(tracks) - assert result[0]["name"] == "mid" + assert result[0]["name"] == "low" assert result[1]["name"] == "default" + assert result[2]["name"] == "high" - def test_same_preserves_order(self): - """同优先级保持顺序.""" + def test_same_priority_stable(self): tracks = [ - {"priority": 10, "name": "first"}, - {"priority": 10, "name": "second"}, + {"name": "first", "priority": 5}, + {"name": "second", "priority": 5}, + {"name": "third", "priority": 5}, ] result = sort_tracks_by_priority(tracks) - assert result[0]["name"] == "first" - assert result[1]["name"] == "second" + assert [t["name"] for t in result] == ["first", "second", "third"] def test_empty_list(self): - """空列表.""" assert sort_tracks_by_priority([]) == [] + def test_original_not_modified(self): + tracks = [{"priority": 3}, {"priority": 1}] + original = list(tracks) + sort_tracks_by_priority(tracks) + assert tracks == original + + +# ── filter_enabled_tracks ─────────────────────────────────────────────────── + class TestFilterEnabledTracks: - """启用轨道过滤测试.""" - def test_all_enabled(self): - """全部启用.""" - tracks = [{"enabled": True}, {"enabled": True}] - assert len(filter_enabled_tracks(tracks)) == 2 + tracks = [{"name": "a", "enabled": True}, {"name": "b"}] + result = filter_enabled_tracks(tracks) + assert len(result) == 2 - def test_mixed(self): - """混合.""" + def test_some_disabled(self): tracks = [ - {"enabled": True, "name": "a"}, - {"enabled": False, "name": "b"}, + {"name": "a", "enabled": True}, + {"name": "b", "enabled": False}, + {"name": "c", "enabled": "false"}, + {"name": "d", "enabled": 0}, ] result = filter_enabled_tracks(tracks) assert len(result) == 1 assert result[0]["name"] == "a" - def test_default_enabled(self): - """默认启用.""" - tracks = [{"name": "a"}] - assert len(filter_enabled_tracks(tracks)) == 1 + def test_all_disabled(self): + tracks = [ + {"name": "a", "enabled": False}, + {"name": "b", "enabled": "false"}, + ] + assert filter_enabled_tracks(tracks) == [] def test_empty_list(self): - """空列表.""" assert filter_enabled_tracks([]) == [] + def test_string_true_enabled(self): + tracks = [{"name": "a", "enabled": "true"}] + result = filter_enabled_tracks(tracks) + assert len(result) == 1 + + +# ── count_track_types ─────────────────────────────────────────────────────── + class TestCountTrackTypes: - """轨道类型统计测试.""" - - def test_mixed_types(self): - """混合类型.""" + def test_multiple_types(self): tracks = [ {"track_type": "bgm"}, - {"track_type": "voiceover"}, + {"track_type": "voice"}, {"track_type": "bgm"}, {"track_type": "sfx"}, + {"track_type": "bgm"}, ] - counts = count_track_types(tracks) - assert counts["bgm"] == 2 - assert counts["voiceover"] == 1 - assert counts["sfx"] == 1 + result = count_track_types(tracks) + assert result == {"bgm": 3, "voice": 1, "sfx": 1} def test_default_type(self): - """默认类型.""" - tracks = [{}] - counts = count_track_types(tracks) - assert counts["unknown"] == 1 + tracks = [{"name": "a"}, {"track_type": "bgm"}] + result = count_track_types(tracks) + assert result["unknown"] == 1 + assert result["bgm"] == 1 def test_empty_list(self): - """空列表.""" assert count_track_types([]) == {} -# ───────────────────────────────────────────────────────────────────────────── -# 配置验证测试 -# ───────────────────────────────────────────────────────────────────────────── +# ── validate_audio_track ──────────────────────────────────────────────────── class TestValidateAudioTrack: - """单轨验证测试.""" + def test_valid_with_asset_id(self): + valid, errors = validate_audio_track({"asset_id": "asset_123"}) + assert valid is True + assert errors == [] - def test_valid_track(self): - """合法轨道.""" - ok, errors = validate_audio_track( - { - "audio_path": "/audio.mp3", - "volume": 0.8, - "fade_in": 1.0, - "fade_out": 2.0, - } - ) - assert ok is True - assert len(errors) == 0 + def test_valid_with_audio_path(self): + valid, errors = validate_audio_track({"audio_path": "/tmp/a.mp3"}) + assert valid is True + assert errors == [] - def test_missing_path(self): - """缺路径.""" - ok, errors = validate_audio_track({}) - assert ok is False - assert any("audio_path" in e or "asset_id" in e for e in errors) + def test_missing_source(self): + valid, errors = validate_audio_track({}) + assert valid is False + assert any("audio_path 或 asset_id" in e for e in errors) def test_negative_volume(self): - """负音量.""" - ok, errors = validate_audio_track( - { - "audio_path": "/a.mp3", - "volume": -1, - } - ) - assert ok is False + valid, errors = validate_audio_track({"asset_id": "a", "volume": -1}) + assert valid is False + assert any("volume" in e for e in errors) + + def test_volume_too_high(self): + valid, errors = validate_audio_track({"asset_id": "a", "volume": 3.0}) + assert valid is False + assert any("volume" in e for e in errors) + + def test_invalid_volume_string(self): + valid, errors = validate_audio_track({"asset_id": "a", "volume": "abc"}) + assert valid is False assert any("volume" in e for e in errors) def test_negative_fade_in(self): - """负淡入.""" - ok, errors = validate_audio_track( - { - "audio_path": "/a.mp3", - "fade_in": -1, - } - ) - assert ok is False + valid, errors = validate_audio_track({"asset_id": "a", "fade_in": -1}) + assert valid is False assert any("fade_in" in e for e in errors) def test_negative_fade_out(self): - """负淡出.""" - ok, errors = validate_audio_track( - { - "audio_path": "/a.mp3", - "fade_out": -1, - } - ) - assert ok is False + valid, errors = validate_audio_track({"asset_id": "a", "fade_out": -1}) + assert valid is False assert any("fade_out" in e for e in errors) - def test_invalid_volume_type(self): - """无效音量类型.""" - ok, errors = validate_audio_track( + def test_invalid_fade_in_string(self): + valid, errors = validate_audio_track({"asset_id": "a", "fade_in": "abc"}) + assert valid is False + assert any("fade_in" in e for e in errors) + + def test_invalid_start_time(self): + valid, errors = validate_audio_track({"asset_id": "a", "start_time": "abc"}) + assert valid is False + assert any("start_time" in e for e in errors) + + def test_multiple_errors(self): + valid, errors = validate_audio_track( { - "audio_path": "/a.mp3", - "volume": "loud", + "volume": "abc", + "fade_in": "def", + "start_time": "ghi", } ) - assert ok is False - assert any("volume" in e for e in errors) + assert valid is False + assert len(errors) >= 4 # source + volume + fade_in + start_time - def test_with_asset_id(self): - """有 asset_id 无 audio_path 也合法.""" - ok, errors = validate_audio_track({"asset_id": "123"}) - assert ok is True + +# ── validate_mix_config ───────────────────────────────────────────────────── class TestValidateMixConfig: - """混音配置验证测试.""" - def test_valid_config(self): - """合法配置.""" - ok, errors = validate_mix_config( - { - "tracks": [ - {"audio_path": "/a.mp3", "volume": 0.5}, - {"audio_path": "/b.mp3", "volume": 0.8}, - ], - "target_duration": 60, - } - ) - assert ok is True + config = { + "tracks": [{"audio_path": "/a.mp3", "volume": 1.0}], + "target_duration": 10.0, + } + valid, errors = validate_mix_config(config) + assert valid is True + assert errors == [] + + def test_no_tracks(self): + valid, errors = validate_mix_config({}) + assert valid is False + assert any("至少需要一条轨道" in e for e in errors) def test_empty_tracks(self): - """空轨道列表.""" - ok, errors = validate_mix_config({"tracks": []}) - assert ok is False - assert any("至少需要" in e for e in errors) + valid, errors = validate_mix_config({"tracks": []}) + assert valid is False + assert any("至少需要一条轨道" in e for e in errors) - def test_invalid_track(self): - """无效轨道.""" - ok, errors = validate_mix_config( - { - "tracks": [ - {"audio_path": "/a.mp3"}, - {}, # 无效 - ], - } - ) - assert ok is False - assert len(errors) >= 1 + def test_invalid_track_errors_prefixed(self): + config = {"tracks": [{"volume": "abc"}]} + valid, errors = validate_mix_config(config) + assert valid is False + assert any(e.startswith("第1轨：") for e in errors) + + def test_multiple_invalid_tracks(self): + config = { + "tracks": [ + {"volume": "bad"}, + {"audio_path": "/a.mp3", "fade_in": "bad"}, + ] + } + valid, errors = validate_mix_config(config) + assert valid is False + track1_errors = [e for e in errors if e.startswith("第1轨：")] + track2_errors = [e for e in errors if e.startswith("第2轨：")] + assert len(track1_errors) >= 1 + assert len(track2_errors) >= 1 def test_negative_target_duration(self): - """负目标时长.""" - ok, errors = validate_mix_config( - { - "tracks": [{"audio_path": "/a.mp3"}], - "target_duration": -10, - } - ) - assert ok is False + config = { + "tracks": [{"audio_path": "/a.mp3"}], + "target_duration": -5, + } + valid, errors = validate_mix_config(config) + assert valid is False + assert any("target_duration" in e for e in errors) + + def test_invalid_target_duration(self): + config = { + "tracks": [{"audio_path": "/a.mp3"}], + "target_duration": "abc", + } + valid, errors = validate_mix_config(config) + assert valid is False assert any("target_duration" in e for e in errors) -# ───────────────────────────────────────────────────────────────────────────── -# 工具函数测试 -# ───────────────────────────────────────────────────────────────────────────── +# ── calculate_total_tracks ────────────────────────────────────────────────── class TestCalculateTotalTracks: - """总轨道数计算测试.""" + def test_with_main_default(self): + config = {"tracks": [{}, {}, {}]} + assert calculate_total_tracks(config) == 4 # 3 + 1主 - def test_with_main(self): - """含主音频.""" - assert calculate_total_tracks({"tracks": [1, 2, 3]}) == 4 + def test_with_main_explicit(self): + config = {"tracks": [{}, {}], "has_main_audio": True} + assert calculate_total_tracks(config) == 3 def test_without_main(self): - """不含主音频.""" - assert ( - calculate_total_tracks( - { - "tracks": [1, 2], - "has_main_audio": False, - } - ) - == 2 - ) + config = {"tracks": [{}, {}], "has_main_audio": False} + assert calculate_total_tracks(config) == 2 - def test_empty_tracks_with_main(self): - """无轨道，只有主音频.""" - assert calculate_total_tracks({"tracks": []}) == 1 + def test_no_tracks_with_main(self): + config = {"tracks": [], "has_main_audio": True} + assert calculate_total_tracks(config) == 1 + + def test_empty_config(self): + assert calculate_total_tracks({}) == 1 + + +# ── estimate_mix_duration ─────────────────────────────────────────────────── class TestEstimateMixDuration: - """混音时长估算测试.""" + def test_single_track(self): + tracks = [{"start_time": 0, "duration": 10}] + assert estimate_mix_duration(tracks) == 10.0 - def test_multiple_tracks(self): - """多轨道取最长结束时间.""" + def test_multiple_tracks_takes_max(self): tracks = [ {"start_time": 0, "duration": 10}, - {"start_time": 5, "duration": 20}, # 结束 25 - {"start_time": 2, "duration": 5}, + {"start_time": 5, "duration": 20}, # end=25 + {"start_time": 2, "duration": 8}, # end=10 ] - assert estimate_mix_duration(tracks) == pytest.approx(25.0) + assert estimate_mix_duration(tracks) == 25.0 def test_empty_list(self): - """空列表.""" assert estimate_mix_duration([]) == 0.0 def test_zero_duration_tracks_ignored(self): - """零时长轨道忽略.""" tracks = [ {"start_time": 0, "duration": 0}, - {"start_time": 5, "duration": 10}, + {"start_time": 5, "duration": 0}, ] - assert estimate_mix_duration(tracks) == pytest.approx(15.0) + assert estimate_mix_duration(tracks) == 0.0 + + def test_invalid_values_skipped(self): + tracks = [ + {"start_time": "abc", "duration": 10}, + {"start_time": 0, "duration": "xyz"}, + {"start_time": 2, "duration": 5}, + ] + assert estimate_mix_duration(tracks) == 7.0 + + def test_negative_start_time(self): + tracks = [{"start_time": -5, "duration": 10}] # end=5 + assert estimate_mix_duration(tracks) == 5.0 + + def test_string_numbers(self): + tracks = [{"start_time": "2.5", "duration": "3.5"}] + assert estimate_mix_duration(tracks) == 6.0 diff --git a/tests/unit/test_pagination.py b/tests/unit/test_pagination.py index 1d7c55e30..06adbe903 100755 --- a/tests/unit/test_pagination.py +++ b/tests/unit/test_pagination.py @@ -1,9 +1,6 @@ -"""通用分页器单元测试.""" - -from __future__ import annotations +"""pagination 单元测试.""" import pytest -from pydantic import ValidationError from packages.application.common.pagination import ( PaginatedResponse, @@ -12,377 +9,171 @@ from packages.application.common.pagination import ( paginate, ) +# ── PaginationParams ──────────────────────────────────────────────────────── + class TestPaginationParams: - """PaginationParams 测试""" - def test_default_values(self): - """默认值正确""" params = PaginationParams() assert params.page == 1 assert params.page_size == 20 - def test_offset_first_page(self): - """第一页 offset 为 0""" + def test_custom_values(self): + params = PaginationParams(page=3, page_size=50) + assert params.page == 3 + assert params.page_size == 50 + + def test_offset_calculation(self): params = PaginationParams(page=1, page_size=20) assert params.offset == 0 - def test_offset_second_page(self): - """第二页 offset 计算正确""" - params = PaginationParams(page=2, page_size=20) - assert params.offset == 20 + params = PaginationParams(page=3, page_size=20) + assert params.offset == 40 - def test_offset_custom_page_size(self): - """自定义 page_size 的 offset""" - params = PaginationParams(page=3, page_size=10) - assert params.offset == 20 + params = PaginationParams(page=10, page_size=50) + assert params.offset == 450 def test_limit_equals_page_size(self): - """limit 等于 page_size""" - params = PaginationParams(page_size=50) - assert params.limit == 50 + params = PaginationParams(page_size=30) + assert params.limit == 30 def test_page_must_be_at_least_1(self): - """page 不能小于 1""" - with pytest.raises(ValidationError): + with pytest.raises(ValueError): PaginationParams(page=0) - def test_page_negative_raises(self): - """page 不能为负数""" - with pytest.raises(ValidationError): - PaginationParams(page=-1) - def test_page_size_must_be_at_least_1(self): - """page_size 不能小于 1""" - with pytest.raises(ValidationError): + with pytest.raises(ValueError): PaginationParams(page_size=0) def test_page_size_max_100(self): - """page_size 最大 100""" - with pytest.raises(ValidationError): + with pytest.raises(ValueError): PaginationParams(page_size=101) - def test_page_size_100_is_valid(self): - """page_size=100 是合法的""" - params = PaginationParams(page_size=100) - assert params.page_size == 100 + +# ── PaginationMeta ────────────────────────────────────────────────────────── class TestPaginationMeta: - """PaginationMeta 测试""" - def test_from_params_first_page(self): - """第一页元数据""" params = PaginationParams(page=1, page_size=10) meta = PaginationMeta.from_params(params, total=25) - assert meta.page == 1 assert meta.page_size == 10 assert meta.total == 25 - assert meta.total_pages == 3 + assert meta.total_pages == 3 # ceil(25/10) assert meta.has_next is True assert meta.has_prev is False def test_from_params_last_page(self): - """最后一页元数据""" params = PaginationParams(page=3, page_size=10) meta = PaginationMeta.from_params(params, total=25) - assert meta.page == 3 assert meta.total_pages == 3 assert meta.has_next is False assert meta.has_prev is True def test_from_params_middle_page(self): - """中间页元数据""" params = PaginationParams(page=2, page_size=10) - meta = PaginationMeta.from_params(params, total=50) - - assert meta.page == 2 - assert meta.total_pages == 5 + meta = PaginationMeta.from_params(params, total=25) assert meta.has_next is True assert meta.has_prev is True + def test_from_params_single_page(self): + params = PaginationParams(page=1, page_size=20) + meta = PaginationMeta.from_params(params, total=5) + assert meta.total_pages == 1 + assert meta.has_next is False + assert meta.has_prev is False + def test_from_params_zero_total(self): - """总数为 0 时""" params = PaginationParams(page=1, page_size=20) meta = PaginationMeta.from_params(params, total=0) - assert meta.total == 0 assert meta.total_pages == 0 assert meta.has_next is False assert meta.has_prev is False - def test_from_params_exact_multiple(self): - """总数刚好是 page_size 的整数倍""" + def test_from_params_exact_page_size(self): params = PaginationParams(page=1, page_size=10) - meta = PaginationMeta.from_params(params, total=30) - - assert meta.total_pages == 3 - - def test_from_params_single_page(self): - """单页即可放下所有数据""" - params = PaginationParams(page=1, page_size=100) - meta = PaginationMeta.from_params(params, total=50) - + meta = PaginationMeta.from_params(params, total=10) assert meta.total_pages == 1 - assert meta.has_next is False - assert meta.has_prev is False + + def test_from_params_one_extra(self): + params = PaginationParams(page=1, page_size=10) + meta = PaginationMeta.from_params(params, total=11) + assert meta.total_pages == 2 + + +# ── PaginatedResponse ─────────────────────────────────────────────────────── class TestPaginatedResponse: - """PaginatedResponse 测试""" - - def test_create_success(self): - """创建分页响应""" - params = PaginationParams(page=1, page_size=10) - data = [1, 2, 3] - - response = PaginatedResponse.create(data, params, total=25) - - assert response.data == [1, 2, 3] + def test_create_response(self): + params = PaginationParams(page=1, page_size=5) + data = [1, 2, 3, 4, 5] + response = PaginatedResponse.create(data, params, total=15) + assert response.data == data assert response.pagination.page == 1 - assert response.pagination.total == 25 + assert response.pagination.total == 15 assert response.pagination.total_pages == 3 - def test_create_empty_data(self): - """空数据分页响应""" - params = PaginationParams(page=1, page_size=20) - response = PaginatedResponse.create([], params, total=0) - assert response.data == [] - assert response.pagination.total == 0 - assert response.pagination.total_pages == 0 +# ── paginate function ─────────────────────────────────────────────────────── -class TestPaginateFunction: - """paginate 函数测试（内存分页）""" - +class TestPaginate: def test_first_page(self): - """第一页分页""" items = list(range(30)) params = PaginationParams(page=1, page_size=10) - result = paginate(items, params) - assert result.data == list(range(10)) assert result.pagination.total == 30 assert result.pagination.total_pages == 3 assert result.pagination.has_next is True assert result.pagination.has_prev is False - def test_second_page(self): - """第二页分页""" - items = list(range(30)) - params = PaginationParams(page=2, page_size=10) - - result = paginate(items, params) - - assert result.data == list(range(10, 20)) - assert result.pagination.page == 2 - def test_last_page(self): - """最后一页分页""" items = list(range(25)) params = PaginationParams(page=3, page_size=10) - result = paginate(items, params) - assert result.data == list(range(20, 25)) assert len(result.data) == 5 assert result.pagination.has_next is False + assert result.pagination.has_prev is True - def test_empty_list(self): - """空列表分页""" - params = PaginationParams(page=1, page_size=20) - result = paginate([], params) - - assert result.data == [] - assert result.pagination.total == 0 - assert result.pagination.total_pages == 0 - - def test_page_beyond_total(self): - """页码超出总数""" + def test_single_page(self): items = list(range(5)) - params = PaginationParams(page=10, page_size=10) - - result = paginate(items, params) - - assert result.data == [] - assert result.pagination.total == 5 - assert result.pagination.total_pages == 1 - - def test_custom_page_size(self): - """自定义每页数量""" - items = list(range(100)) - params = PaginationParams(page=1, page_size=50) - - result = paginate(items, params) - - assert len(result.data) == 50 - assert result.pagination.total_pages == 2 - - def test_single_item(self): - """单条数据""" - items = ["only_one"] params = PaginationParams(page=1, page_size=10) - result = paginate(items, params) - - assert result.data == ["only_one"] - assert result.pagination.total == 1 - assert result.pagination.total_pages == 1 - - def test_generic_type_preserved(self): - """泛型类型数据正确""" - items = [{"id": 1, "name": "a"}, {"id": 2, "name": "b"}] - params = PaginationParams(page=1, page_size=10) - - result = paginate(items, params) - - assert len(result.data) == 2 - assert result.data[0]["id"] == 1 - - -# ── PaginationParams 补充边界 ─────────────────────────────────────────────── - - -class TestPaginationParamsEdgeCases: - """PaginationParams 补充边界场景.""" - - def test_page_size_1_minimum(self): - """page_size=1 是允许的最小值.""" - params = PaginationParams(page_size=1) - assert params.page_size == 1 - assert params.limit == 1 - - def test_page_size_100_maximum(self): - """page_size=100 是允许的最大值.""" - params = PaginationParams(page_size=100) - assert params.page_size == 100 - - def test_offset_page_1_size_100(self): - """第1页每页100条 offset=0.""" - params = PaginationParams(page=1, page_size=100) - assert params.offset == 0 - - def test_offset_page_100_size_100(self): - """第100页每页100条 offset=9900.""" - params = PaginationParams(page=100, page_size=100) - assert params.offset == 9900 - - def test_large_page_number_accepted(self): - """极大页码（超过实际页数）允许.""" - params = PaginationParams(page=999999, page_size=20) - assert params.page == 999999 - assert params.offset == (999999 - 1) * 20 - - -# ── PaginationMeta 补充边界 ───────────────────────────────────────────────── - - -class TestPaginationMetaEdgeCases: - """PaginationMeta 补充边界场景.""" - - def test_total_0_page_1(self): - """total=0, page=1 时 total_pages=0, 无上下页.""" - params = PaginationParams(page=1, page_size=20) - meta = PaginationMeta.from_params(params, total=0) - assert meta.total_pages == 0 - assert meta.has_next is False - assert meta.has_prev is False - - def test_total_0_page_beyond(self): - """total=0, page>1 时 has_prev=True（因为page>1）.""" - params = PaginationParams(page=3, page_size=20) - meta = PaginationMeta.from_params(params, total=0) - assert meta.total_pages == 0 - assert meta.has_next is False - assert meta.has_prev is True - - def test_exact_last_page(self): - """刚好是最后一页时 has_next=False.""" - params = PaginationParams(page=5, page_size=10) - meta = PaginationMeta.from_params(params, total=50) - assert meta.total_pages == 5 - assert meta.has_next is False - assert meta.has_prev is True - - def test_one_more_than_exact(self): - """比整数页多1条时总页数+1.""" - params = PaginationParams(page=1, page_size=10) - meta = PaginationMeta.from_params(params, total=51) - assert meta.total_pages == 6 - - def test_page_exactly_total_pages(self): - """page == total_pages 时 has_next=False.""" - params = PaginationParams(page=3, page_size=10) - meta = PaginationMeta.from_params(params, total=30) - assert meta.has_next is False - - def test_total_1_page_1_size_1(self): - """1条数据1页.""" - params = PaginationParams(page=1, page_size=1) - meta = PaginationMeta.from_params(params, total=1) - assert meta.total_pages == 1 - assert meta.has_next is False - assert meta.has_prev is False - - -# ── paginate 补充边界 ────────────────────────────────────────────────────── - - -class TestPaginateEdgeCases: - """paginate 补充边界场景.""" - - def test_single_item_list(self): - """单元素列表.""" - result = paginate([42], PaginationParams(page=1, page_size=10)) - assert result.data == [42] - assert result.pagination.total == 1 - assert result.pagination.total_pages == 1 - - def test_page_exactly_last(self): - """刚好在最后一页.""" - items = list(range(25)) - result = paginate(items, PaginationParams(page=3, page_size=10)) - assert result.data == list(range(20, 25)) - assert result.pagination.has_next is False - - def test_page_past_end_returns_empty(self): - """页码超过总数返回空.""" - items = list(range(5)) - result = paginate(items, PaginationParams(page=10, page_size=10)) - assert result.data == [] - assert result.pagination.total == 5 - - def test_empty_list_page_1(self): - """空列表第1页.""" - result = paginate([], PaginationParams(page=1, page_size=10)) - assert result.data == [] - assert result.pagination.total == 0 - assert result.pagination.total_pages == 0 - - def test_page_size_1_iterates_all(self): - """page_size=1 时每页1条.""" - items = ["a", "b", "c"] - r1 = paginate(items, PaginationParams(page=1, page_size=1)) - r2 = paginate(items, PaginationParams(page=2, page_size=1)) - r3 = paginate(items, PaginationParams(page=3, page_size=1)) - assert r1.data == ["a"] - assert r2.data == ["b"] - assert r3.data == ["c"] - - def test_does_not_mutate_input(self): - """不修改输入列表.""" - items = [1, 2, 3, 4, 5] - original = items[:] - paginate(items, PaginationParams(page=1, page_size=2)) - assert items == original - - def test_page_size_greater_than_total(self): - """每页条数大于总数.""" - items = list(range(5)) - result = paginate(items, PaginationParams(page=1, page_size=100)) assert result.data == items assert result.pagination.total_pages == 1 + + def test_empty_list(self): + items = [] + params = PaginationParams(page=1, page_size=10) + result = paginate(items, params) + assert result.data == [] + assert result.pagination.total == 0 + assert result.pagination.total_pages == 0 + + def test_page_beyond_end(self): + items = list(range(5)) + params = PaginationParams(page=10, page_size=10) + result = paginate(items, params) + assert result.data == [] + assert result.pagination.total == 5 + + def test_page_size_larger_than_items(self): + items = list(range(5)) + params = PaginationParams(page=1, page_size=100) + result = paginate(items, params) + assert result.data == items + assert result.pagination.total_pages == 1 + + def test_middle_page(self): + items = list(range(100)) + params = PaginationParams(page=5, page_size=10) + result = paginate(items, params) + assert result.data == list(range(40, 50)) + assert result.pagination.has_next is True + assert result.pagination.has_prev is True diff --git a/tests/unit/test_text_splitter.py b/tests/unit/test_text_splitter.py index 3e221601e..831253df2 100755 --- a/tests/unit/test_text_splitter.py +++ b/tests/unit/test_text_splitter.py @@ -1,412 +1,100 @@ -"""文本分段工具单元测试.""" - -from __future__ import annotations - -import pytest +"""text_splitter 单元测试.""" from packages.application.tts_job.text_splitter import split_text class TestSplitText: - """split_text 函数测试""" - - def test_empty_string_returns_empty_list(self): - """空字符串返回空列表""" + def test_empty_text_returns_empty(self): assert split_text("") == [] - def test_whitespace_only_returns_empty_list(self): - """纯空白字符返回空列表""" - assert split_text(" \n \t ") == [] - - def test_short_text_returns_single_segment(self): - """短文本直接返回单段""" - text = "这是一段短文本。" - result = split_text(text, max_chars=500) - assert result == [text] - - def test_text_length_equals_max_chars(self): - """文本长度恰好等于 max_chars 时返回单段""" - text = "a" * 100 - result = split_text(text, max_chars=100) - assert len(result) == 1 - assert len(result[0]) == 100 - - def test_splits_on_sentence_boundary(self): - """在句子边界处分段""" - # 构造长文本，确保超过 max_chars - sentences = ["今天天气真好。我们一起去公园散步吧。", "公园里有很多花。还有很多小朋友在玩耍。"] * 10 - text = "".join(sentences) - - result = split_text(text, max_chars=200) - - assert len(result) >= 2 - # 每段都不超过 max_chars - for seg in result: - assert len(seg) <= 200 - - def test_all_segments_within_max_chars(self): - """所有分段都不超过 max_chars""" - text = "这是第一句话。这是第二句话。这是第三句话。这是第四句话。这是第五句话。" * 10 - - result = split_text(text, max_chars=100) - - for seg in result: - assert len(seg) <= 100 - - def test_long_single_sentence_hard_cut(self): - """超长单句会被硬切""" - text = "a" * 1000 # 没有标点 - - result = split_text(text, max_chars=200) - - assert len(result) > 1 - for seg in result: - assert len(seg) <= 200 - - def test_newline_is_sentence_end(self): - """换行符作为句子结束符""" - text = "第一行内容\n第二行内容\n第三行内容" * 10 - - result = split_text(text, max_chars=50) - - assert len(result) > 1 - for seg in result: - assert len(seg) <= 50 - - def test_chinese_punctuation(self): - """中文标点（。！？；）作为句子结束符""" - text = "你好！今天吃什么？我吃米饭；你呢？我也吃米饭。" * 10 - - result = split_text(text, max_chars=80) - - for seg in result: - assert len(seg) <= 80 - - def test_english_punctuation(self): - """英文标点（.!?;）作为句子结束符""" - text = "Hello! How are you? I'm fine; thank you. Good bye." * 10 - - result = split_text(text, max_chars=80) - - for seg in result: - assert len(seg) <= 80 - - def test_merged_short_segments(self): - """过短的段落会被合并""" - # 构造很多短句 - text = "你好。再见。谢谢。抱歉。好的。不行。可以。去吧。" * 5 # 每句3-4字 - - result = split_text(text, max_chars=100) - - # 合并后段数应该比单纯按句切的少 - assert len(result) < len(text) // 3 # 粗略估计 - for seg in result: - assert len(seg) <= 100 - - def test_preserves_content(self): - """分段后内容总和与原文基本一致（忽略strip的空白）""" - text = "这是测试文本。包含多个句子。用来验证分段正确性。" * 5 - - result = split_text(text, max_chars=50) - - # 合并所有分段，去掉空白后应该与原文去掉空白后基本一致 - combined = "".join(result).replace(" ", "") - original = text.strip().replace(" ", "") - assert combined == original - - def test_custom_max_chars(self): - """支持自定义 max_chars""" - text = "测试" * 100 # 200字 - - result_50 = split_text(text, max_chars=50) - result_100 = split_text(text, max_chars=100) - - # max_chars 越小，段数应该越多 - assert len(result_50) >= len(result_100) - - def test_single_char_text(self): - """单字符文本""" - assert split_text("好", max_chars=10) == ["好"] - - def test_text_with_only_punctuation(self): - """纯标点文本""" - text = "。。。。。。。。。。" # 10个句号 - result = split_text(text, max_chars=5) - - assert len(result) >= 1 - for seg in result: - assert len(seg) <= 5 - - def test_mixed_content(self): - """中英文混合内容""" - text = "今天的天气是 sunny and warm。我们去了 park 玩。真的很开心！" * 5 - - result = split_text(text, max_chars=80) - - for seg in result: - assert len(seg) <= 80 - - -# ── 短文本与空文本补充 ────────────────────────────────────────────────────── - - -class TestSplitTextEmptyAndShort: - """空文本与短文本补充场景.""" - - def test_whitespace_only_returns_empty(self): - """纯空白文本返回空列表.""" + def test_whitespace_only(self): assert split_text(" \n\t ") == [] - def test_single_char(self): - """单字符文本.""" - assert split_text("好", max_chars=10) == ["好"] - - def test_exactly_max_chars_no_split(self): - """刚好等于 max_chars 不分割.""" - text = "a" * 100 - result = split_text(text, max_chars=100) + def test_short_text_single_segment(self): + text = "你好世界。" + result = split_text(text, max_chars=500) assert len(result) == 1 assert result[0] == text - def test_one_over_max_chars_splits(self): - """超过 max_chars 1 个字符就会分割.""" - text = "a" * 101 + def test_exact_max_chars(self): + text = "a" * 500 + result = split_text(text, max_chars=500) + assert len(result) == 1 + assert len(result[0]) == 500 + + def test_splits_on_sentence_boundary(self): + # 两个长句子，各300字左右，超过50字阈值 + sent1 = "你" * 300 + "。" + sent2 = "我" * 300 + "。" + text = sent1 + sent2 + result = split_text(text, max_chars=500) + assert len(result) == 2 + assert result[0] == sent1 + assert result[1] == sent2 + + def test_long_sentence_hard_cut(self): + # 一个超长句子，没有句末标点，会被硬切 + text = "长" * 800 + result = split_text(text, max_chars=500) + assert len(result) >= 2 + assert all(len(seg) <= 500 for seg in result) + # 合起来应该等于原文本 + assert "".join(result) == text + + def test_short_segments_merged(self): + # 多个短句应该被合并 + sentences = [f"第{i}句。" for i in range(10)] + text = "".join(sentences) + result = split_text(text, max_chars=200) + # 每句5字左右，10句才50字，应该合并成1段 + assert len(result) < 10 + assert len(result[0]) <= 200 + + def test_preserves_content(self): + text = "今天天气真好。我们去公园玩吧！你觉得怎么样？好的，走吧。" + result = split_text(text, max_chars=20) + # 合并后内容应一致 + assert "".join(result) == text + + def test_multiple_punctuation_types(self): + # 构造足够长的文本触发分段 + text = "第一" * 30 + "。" + "第二" * 30 + "！" + "第三" * 30 + "？" + "第四" * 30 + "；" result = split_text(text, max_chars=100) assert len(result) >= 2 + assert "".join(result) == text - def test_none_raises(self): - """None 输入抛 AttributeError（strip 失败）.""" - with pytest.raises(AttributeError): - split_text(None) + def test_custom_max_chars(self): + text = "a" * 100 + "。" + "b" * 100 + "。" + result = split_text(text, max_chars=150) + assert len(result) == 2 + assert "a" in result[0] + assert "b" in result[1] - -# ── 句子边界分段补充 ────────────────────────────────────────────────────── - - -class TestSplitTextSentenceBoundaries: - """句子边界分段补充场景.""" - - def test_split_on_fullwidth_period(self): - """全角句号分段.""" - text = "第一句很长的内容。" * 20 - result = split_text(text, max_chars=60) - assert len(result) > 1 - for seg in result: - assert len(seg) <= 60 - - def test_split_on_fullwidth_question(self): - """全角问号分段.""" - text = "你知道这是为什么吗？" + "是的。" * 20 - result = split_text(text, max_chars=60) - assert len(result) > 1 - - def test_split_on_fullwidth_exclamation(self): - """全角感叹号分段.""" - text = "真是太棒了！" + "内容。" * 20 - result = split_text(text, max_chars=60) - assert len(result) > 1 - - def test_split_on_newline(self): - """换行符分段.""" - lines = ["这是第一行很长的一段文字内容" * 3 for _ in range(5)] - text = "\n".join(lines) - result = split_text(text, max_chars=80) - assert len(result) > 1 - - def test_split_on_semicolon(self): - """全角分号分段.""" - text = "第一项内容；" + "其他内容。" * 20 - result = split_text(text, max_chars=60) - assert len(result) > 1 - - def test_english_period_splits(self): - """英文句号分段.""" - text = "Hello world. " * 30 - result = split_text(text, max_chars=80) - assert len(result) > 1 - - def test_short_sentences_stay_merged(self): - """短句（都 < 50字的句子不会单独成段，会累积到一起.""" - text = "你好。我好。大家好。" - result = split_text(text, max_chars=200) - assert len(result) == 1 - - -# ── 长句强制切段补充 ────────────────────────────────────────────────────── - - -class TestSplitTextLongSentenceForce: - """超长单句强制切段补充.""" - - def test_no_punctuation_forced_split(self): - """完全没有标点的超长文本硬切.""" - text = "字" * 300 - result = split_text(text, max_chars=100) - assert len(result) == 3 - for seg in result: - assert len(seg) == 100 - - def test_force_split_preserves_content(self): - """硬切不丢字符.""" - text = "a" * 250 - result = split_text(text, max_chars=100) - assert sum(len(s) for s in result) == 250 - - def test_mixed_long_and_short(self): - """长句短句混合.""" - long_part = "非常长的句子没有标点符号" * 15 - text = long_part + "。结尾。" - result = split_text(text, max_chars=100) - assert len(result) > 1 - for seg in result: - assert len(seg) <= 100 - - -# ── 短段合并补充 ───────────────────────────────────────────────────────── - - -class TestSplitTextShortSegmentMerge: - """短段合并补充场景.""" - - def test_multiple_short_sentences_merged(self): - """多个短句合并成一段.""" - sentences = ["你好。", "我好。", "大家好。", "天气好。", "心情好。"] - text = "".join(sentences) - result = split_text(text, max_chars=200) - assert len(result) == 1 - - def test_short_tail_merged(self): - """尾部短段被合并到前一段.""" - # 前面一段接近 max_chars，尾部很短 - long_part = "一二三四五六七八九十" * 9 + "。" # ~90字 - tail = "完。" # 2字 - text = long_part + tail - result = split_text(text, max_chars=100) - # 尾部短的应该被合并 - assert len(result) <= 2 - - -# ── 边界情况补充 ───────────────────────────────────────────────────────── - - -class TestSplitTextEdgeCases: - """边界情况补充.""" - - def test_only_punctuation(self): - """纯标点符号.""" - text = "。。。。。" - result = split_text(text, max_chars=10) - assert len(result) == 1 - - def test_mixed_chinese_english(self): - """中英文混合.""" - text = "你好Hello。World!" * 20 - result = split_text(text, max_chars=100) - assert len(result) > 1 - for seg in result: - assert len(seg) <= 100 - - def test_strip_whitespace(self): - """首尾空白被去除.""" - text = " 你好世界。 " - result = split_text(text, max_chars=100) - assert result == ["你好世界。"] - - def test_total_length_preserved(self): - """分段后总长度等于原文 strip 后长度.""" - text = "这是一段用于测试的文本内容。" * 20 - result = split_text(text, max_chars=100) + def test_newline_as_sentence_end(self): + text = "第一段\n第二段\n第三段" + result = split_text(text, max_chars=50) + assert len(result) >= 1 assert "".join(result) == text.strip() - def test_custom_small_max_chars(self): - """很小的 max_chars.""" - text = "一二三四五六七八九十。" * 5 + def test_minimum_segment_length(self): + # 句子太短（<50字）不会立即分段 + text = "短句一。短句二。短句三。" + result = split_text(text, max_chars=200) + assert len(result) == 1 + + def test_trailing_content_added(self): + # 最后一段不完整的句子也要加上 + text = "完整的句子。剩余内容" + result = split_text(text, max_chars=50) + assert "".join(result) == text + + def test_no_empty_segments(self): + text = "。。。。。" # 全是标点 + result = split_text(text, max_chars=2) + assert all(len(seg) > 0 for seg in result) + + def test_chinese_and_english_mixed(self): + text = "Hello世界。这是测试Test文本。Mixed混合。" result = split_text(text, max_chars=20) - assert len(result) > 1 - for seg in result: - assert len(seg) <= 20 - - -# ── 更多边界场景补充 ───────────────────────────────────────────────────────── - - -class TestSplitTextMoreEdgeCases: - """更多边界场景补充""" - - def test_max_chars_one(self): - """max_chars=1 每个字符一段""" - text = "一二三四五" - result = split_text(text, max_chars=1) - assert len(result) == 5 - for seg in result: - assert len(seg) == 1 - - def test_consecutive_newlines(self): - """连续多个换行符""" - text = "第一段\n\n\n第二段\n\n第三段" - result = split_text(text, max_chars=100) - # 合并后应该是一段（内容不长且合并逻辑会被合并） - assert len(result) >= 1 - assert "第一段" in result[0] - for seg in result: - assert len(seg) <= 100 - - def test_only_newlines_only(self): - """只有换行符（纯空白被strip掉返回空""" - assert split_text("\n\n\n\n") == [] - - def test_leading_trailing_whitespace(self): - """首尾空白被去除""" - text = " 你好世界。 " - result = split_text(text, max_chars=100) - assert result == ["你好世界。"] - - def test_very_long_single_sentence_many_segments(self): - """超长单句被切成很多段""" - text = "字" * 1000 - result = split_text(text, max_chars=100) - assert len(result) == 10 - for seg in result: - assert len(seg) == 100 - - def test_mixed_punctuation_types(self): - """全角半角标点混合""" - text = "你好!再见。谢谢?抱歉；好的" - result = split_text(text, max_chars=200) - assert len(result) == 1 - - def test_last_segment_short_merged_to_previous(self): - """尾部极短段被合并到前一段""" - # 构造第一段接近max_chars，结尾有个短句尾巴 - long_part = "一二三四五六七八九十" * 9 + "。" # ~90字 - tail = "完" # 1字 - text = long_part + tail - result = split_text(text, max_chars=100) - # 尾巴应该被合并 - combined = "".join(result) - assert combined == text.strip() - assert len(result) <= 2 - - def test_all_short_sentences_merged_into_one(self): - """大量短句全部合并成一段""" - sentences = ["你好。", "我好。", "他好。", "大家好。", "才是真的好。"] - text = "".join(sentences) - result = split_text(text, max_chars=200) - assert len(result) == 1 - - def test_punctuation_only_long(self): - """很长的纯标点文本""" - text = "。" * 200 - result = split_text(text, max_chars=50) - assert len(result) >= 4 - for seg in result: - assert len(seg) <= 50 - - def test_tab_not_sentence_end(self): - """制表符不是句子结束符""" - text = "这是一段\t包含制表符的文本内容" + "字" * 100 - result = split_text(text, max_chars=50) - # 制表符不在句子结束符集合中，不会触发分段 - # 制表符会保留在分段内容中 - has_tab = any("\t" in seg for seg in result) - assert has_tab + assert len(result) >= 2 + assert "".join(result) == text