Files
xiaoxia-saas/tests/unit/test_multi_track_mixer_pure.py
xiaoxia de201436ea
CI/CD Pipeline / Check if frontend-only change (push) Has been cancelled
CI/CD Pipeline / Validate - Code Quality (push) Has been cancelled
CI/CD Pipeline / Validate - Type Check (mypy) (push) Has been cancelled
CI/CD Pipeline / Validate - Migration (alembic) (push) Has been cancelled
CI/CD Pipeline / Unit Tests (push) Has been cancelled
CI/CD Pipeline / Integration Tests (push) Has been cancelled
CI/CD Pipeline / Frontend Lint (push) Has been cancelled
CI/CD Pipeline / Frontend Unit Tests (push) Has been cancelled
CI/CD Pipeline / PR Build API Image (push) Has been cancelled
CI/CD Pipeline / PR Build Web Image (push) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (push) Has been cancelled
CI/CD Pipeline / Build Staging API Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Web Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / Build Production API Image (push) Has been cancelled
CI/CD Pipeline / Build Production Web Image (push) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Production (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled
CI/CD Pipeline / Canary Release to Production (push) Has been cancelled
CI/CD Pipeline / CI Gate (push) Has been cancelled
test(wave204): text_splitter + pagination 单测补全 +37测 (#1171)
2026-07-30 00:26:00 +08:00

768 lines
25 KiB
Python
Executable File

"""multi_track_mixer_pure 单元测试."""
import math
from apps.worker.video_processing.multi_track_mixer_pure import (
build_amix_filter,
build_mix_filter_complex,
build_track_filter_chain,
calculate_amix_volume_compensation,
calculate_effective_range,
calculate_total_tracks,
count_track_types,
db_to_linear,
estimate_mix_duration,
filter_enabled_tracks,
is_track_visible,
linear_to_db,
normalize_volume,
sort_tracks_by_priority,
validate_audio_track,
validate_mix_config,
)
# ── calculate_effective_range ───────────────────────────────────────────────
class TestCalculateEffectiveRange:
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_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 need == 8.0 # 10 - 2
assert trim == 2.0
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 need == 0.0
assert trim == 0.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_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):
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_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_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(2, 5, 10, 20) is True
def test_invisible_after_target(self):
assert is_track_visible(25, 5, 10, 20) 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_chain_structure(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,
)
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_volume_clamped(self):
result = build_track_filter_chain(
volume=3.0,
fade_in=0,
fade_out=0,
effective_start=0,
need_duration=5.0,
trim_start=0,
target_duration=10.0,
)
assert "volume=2.000" in result # 钳制到2.0
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=5.0,
trim_start=0,
target_duration=10.0,
)
assert "adelay=2500|2500" in result
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.0,
trim_start=0,
target_duration=10.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
# ── build_amix_filter ───────────────────────────────────────────────────────
class TestBuildAmixFilter:
def test_zero_inputs(self):
assert build_amix_filter(0) == ""
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_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_input(self):
assert calculate_amix_volume_compensation(1) == 1.0
def test_two_inputs(self):
assert calculate_amix_volume_compensation(2) == 2.0
def test_five_inputs(self):
assert calculate_amix_volume_compensation(5) == 5.0
def test_negative_inputs(self):
assert calculate_amix_volume_compensation(-1) == 1.0
# ── build_mix_filter_complex ────────────────────────────────────────────────
class TestBuildMixFilterComplex:
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_none_returns_one(self):
assert normalize_volume(None) == 1.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_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):
assert normalize_volume("abc") == 1.0
# ── db_to_linear / linear_to_db ─────────────────────────────────────────────
class TestDbConversions:
def test_zero_db_is_one(self):
assert abs(db_to_linear(0) - 1.0) < 0.001
def test_negative_db_less_than_one(self):
assert db_to_linear(-20) < 1.0
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)
back = db_to_linear(db)
assert abs(back - original) < 0.001
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_ascending(self):
tracks = [
{"name": "c", "priority": 3},
{"name": "a", "priority": 1},
{"name": "b", "priority": 2},
]
result = sort_tracks_by_priority(tracks)
assert [t["name"] for t in result] == ["a", "b", "c"]
def test_default_priority_100(self):
tracks = [
{"name": "low", "priority": 50},
{"name": "default"}, # 默认100
{"name": "high", "priority": 150},
]
result = sort_tracks_by_priority(tracks)
assert result[0]["name"] == "low"
assert result[1]["name"] == "default"
assert result[2]["name"] == "high"
def test_same_priority_stable(self):
tracks = [
{"name": "first", "priority": 5},
{"name": "second", "priority": 5},
{"name": "third", "priority": 5},
]
result = sort_tracks_by_priority(tracks)
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 = [{"name": "a", "enabled": True}, {"name": "b"}]
result = filter_enabled_tracks(tracks)
assert len(result) == 2
def test_some_disabled(self):
tracks = [
{"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_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_multiple_types(self):
tracks = [
{"track_type": "bgm"},
{"track_type": "voice"},
{"track_type": "bgm"},
{"track_type": "sfx"},
{"track_type": "bgm"},
]
result = count_track_types(tracks)
assert result == {"bgm": 3, "voice": 1, "sfx": 1}
def test_default_type(self):
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_with_audio_path(self):
valid, errors = validate_audio_track({"audio_path": "/tmp/a.mp3"})
assert valid is True
assert 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):
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):
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):
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_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(
{
"volume": "abc",
"fade_in": "def",
"start_time": "ghi",
}
)
assert valid is False
assert len(errors) >= 4 # source + volume + fade_in + start_time
# ── validate_mix_config ─────────────────────────────────────────────────────
class TestValidateMixConfig:
def test_valid_config(self):
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):
valid, errors = validate_mix_config({"tracks": []})
assert valid is False
assert any("至少需要一条轨道" in e for e in errors)
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):
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_explicit(self):
config = {"tracks": [{}, {}], "has_main_audio": True}
assert calculate_total_tracks(config) == 3
def test_without_main(self):
config = {"tracks": [{}, {}], "has_main_audio": False}
assert calculate_total_tracks(config) == 2
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_takes_max(self):
tracks = [
{"start_time": 0, "duration": 10},
{"start_time": 5, "duration": 20}, # end=25
{"start_time": 2, "duration": 8}, # end=10
]
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": 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