From 3bdf14b5e79bb3974fa3b979469fa1a6d99913e2 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Tue, 28 Jul 2026 09:28:17 +0800 Subject: [PATCH] =?UTF-8?q?test(wave136):=20=E5=A4=9A=E8=BD=A8=E6=B7=B7?= =?UTF-8?q?=E9=9F=B3=E7=BA=AF=E9=80=BB=E8=BE=91=E6=8A=BD=E7=A6=BB=20+=2071?= =?UTF-8?q?=E5=8D=95=E6=B5=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 multi_track_mixer_pure.py:时间计算/滤镜链构建/amix混音/音量计算/轨道排序/配置验证 - 71个单测覆盖:effective_range/filter_chain/amix/volume_compensation/normalize_volume/dB转换/排序过滤/配置验证 - 0 FFmpeg依赖,纯函数可测试 --- .../multi_track_mixer_pure.py | 466 +++++++++++++ tests/unit/test_multi_track_mixer_pure.py | 638 ++++++++++++++++++ 2 files changed, 1104 insertions(+) create mode 100755 apps/worker/video_processing/multi_track_mixer_pure.py create mode 100755 tests/unit/test_multi_track_mixer_pure.py diff --git a/apps/worker/video_processing/multi_track_mixer_pure.py b/apps/worker/video_processing/multi_track_mixer_pure.py new file mode 100755 index 000000000..fc343f9f3 --- /dev/null +++ b/apps/worker/video_processing/multi_track_mixer_pure.py @@ -0,0 +1,466 @@ +"""多轨混音纯逻辑模块. + +所有函数均为纯函数,不调用 FFmpeg、不操作文件。 +便于单元测试,也方便被其他模块复用。 +""" + +from __future__ import annotations + +from typing import Any, Optional + +# ── 单轨时间计算 ────────────────────────────────────────────────────────────── + + +def calculate_effective_range( + track_start: float, + track_duration: float, + audio_duration: float, + target_duration: float, +) -> tuple[float, float, float]: + """计算轨道的有效时间范围. + + 处理: + - 轨道时长为 0 或负时用音频完整时长 + - 轨道开始在目标时长外时跳过 + - 轨道开始为负时截断开头 + + Args: + track_start: 轨道开始时间(秒),可为负 + track_duration: 轨道持续时长(秒),<=0 表示用音频全长 + audio_duration: 音频文件实际时长(秒) + target_duration: 目标总时长(秒) + + Returns: + (effective_start, need_duration, trim_start) + - effective_start: 在目标时间轴上的开始位置(>=0) + - need_duration: 需要截取的音频长度 + - trim_start: 从源音频的哪个位置开始截取 + """ + if audio_duration <= 0: + return (0.0, 0.0, 0.0) + + # 有效时长(轨道声明的时长,未被截断的) + if track_duration > 0: + effective_dur = min(track_duration, audio_duration) + else: + effective_dur = audio_duration + + effective_start = track_start + trim_start = 0.0 + + # 负的开始时间:从源音频中间开始取,轨道前段被截掉 + if effective_start < 0: + trim_start = -effective_start + # 可用时长 = 总时长 - 被截掉的前段 + effective_dur = max(0.0, effective_dur - trim_start) + effective_start = 0.0 + + # 轨道完全在目标时长之外 + if effective_start >= target_duration: + return (0.0, 0.0, 0.0) + + # 轨道完全在 0 之前 + if effective_start + effective_dur <= 0: + return (0.0, 0.0, 0.0) + + # 实际需要的源时长 + need_dur = min(effective_dur, target_duration - effective_start) + if need_dur <= 0: + return (0.0, 0.0, 0.0) + + # 调整 trim_start 不能超过音频长度 + if trim_start >= audio_duration: + return (0.0, 0.0, 0.0) + + return (effective_start, need_dur, trim_start) + + +def is_track_visible( + track_start: float, + track_duration: float, + audio_duration: float, + target_duration: float, +) -> bool: + """判断轨道是否在目标时长范围内可见(有声音). + + Args: + track_start: 轨道开始时间 + track_duration: 轨道持续时长 + audio_duration: 音频时长 + target_duration: 目标总时长 + + Returns: + 是否可见 + """ + _, need_dur, _ = calculate_effective_range(track_start, track_duration, audio_duration, target_duration) + return need_dur > 0 + + +# ── 单轨滤镜链构建 ──────────────────────────────────────────────────────────── + + +def build_track_filter_chain( + volume: float, + fade_in: float, + fade_out: float, + effective_start: float, + need_duration: float, + trim_start: float, + target_duration: float, +) -> str: + """构建单轨道预处理滤镜链. + + 处理顺序:截断 → 重置时间戳 → 音量 → 淡入 → 淡出 → 延迟 → 最终截断 → 重置时间戳 + + Args: + volume: 音量 0.0~1.0 + fade_in: 淡入时长(秒) + fade_out: 淡出时长(秒) + effective_start: 在目标轴上的开始时间 + need_duration: 需要截取的时长 + trim_start: 从源音频的哪个位置开始 + target_duration: 目标总时长 + + Returns: + 逗号分隔的滤镜字符串 + """ + filter_parts: list[str] = [] + + # 1. 截断到有效范围 + filter_parts.append(f"atrim={trim_start:.3f}:{trim_start + need_duration:.3f}") + filter_parts.append("asetpts=N/SR/TB") + + # 2. 音量调节 + safe_volume = max(0.0, min(2.0, volume)) + if abs(safe_volume - 1.0) > 0.001: + filter_parts.append(f"volume={safe_volume:.3f}") + + # 3. 淡入(必须小于总时长才有效) + if fade_in > 0 and fade_in < need_duration: + filter_parts.append(f"afade=t=in:st=0:d={fade_in:.3f}") + + # 4. 淡出 + if fade_out > 0 and fade_out < need_duration: + fade_start = need_duration - fade_out + if fade_start > 0: + filter_parts.append(f"afade=t=out:st={fade_start:.3f}:d={fade_out:.3f}") + + # 5. 时间偏移(开头静音填充) + if effective_start > 0.01: + delay_ms = int(effective_start * 1000) + filter_parts.append(f"adelay={delay_ms}|{delay_ms}") + + # 6. 最终截断到目标总时长 + filter_parts.append(f"atrim=0:{target_duration:.3f}") + filter_parts.append("asetpts=N/SR/TB") + + return ",".join(filter_parts) + + +# ── amix 混音滤镜构建 ──────────────────────────────────────────────────────── + + +def build_amix_filter(num_inputs: int, duration_mode: str = "longest") -> str: + """构建 amix 混音滤镜. + + Args: + num_inputs: 输入轨道数量 + duration_mode: 时长模式:longest / shortest / first + + Returns: + amix 滤镜字符串 + """ + if num_inputs <= 0: + return "" + + # 校验 duration_mode + if duration_mode not in ("longest", "shortest", "first"): + duration_mode = "longest" + + return f"amix=inputs={num_inputs}:duration={duration_mode}:dropout_transition=0" + + +def calculate_amix_volume_compensation(num_inputs: int) -> float: + """计算 amix 后的音量补偿系数. + + amix 会将 N 路输入每路乘以 1/N 来归一化, + 所以需要乘以 N 来补偿(简单粗暴但有效)。 + + Args: + num_inputs: 输入轨道数量 + + Returns: + 补偿系数 + """ + if num_inputs <= 1: + return 1.0 + return float(num_inputs) + + +def build_mix_filter_complex( + num_tracks: int, + has_main: bool = True, + duration_mode: str = "longest", +) -> str: + """构建完整的混音 filter_complex. + + Args: + num_tracks: 额外轨道数量 + has_main: 是否有主音频 + duration_mode: 时长模式 + + Returns: + filter_complex 字符串 + """ + total_inputs = num_tracks + (1 if has_main else 0) + if total_inputs <= 0: + return "" + + # 输入标签 + input_labels = "".join(f"[{i}:a]" for i in range(total_inputs)) + + # amix + amix = build_amix_filter(total_inputs, duration_mode) + + # 音量补偿 + compensation = calculate_amix_volume_compensation(total_inputs) + volume_filter = "" + if abs(compensation - 1.0) > 0.001: + volume_filter = f",volume={compensation}" + + return f"{input_labels}{amix}{volume_filter}[mixed]" + + +# ── 音量计算 ────────────────────────────────────────────────────────────────── + + +def normalize_volume(volume: float) -> float: + """规范化音量值. + + Args: + volume: 原始音量 + + Returns: + 规范化后的音量(0.0 ~ 2.0) + """ + if volume is None: + return 1.0 + try: + v = float(volume) + return max(0.0, min(2.0, v)) + except (ValueError, TypeError): + return 1.0 + + +def db_to_linear(db: float) -> float: + """dB 转换为线性音量. + + Args: + db: 分贝值 + + Returns: + 线性音量值 + """ + import math + + return 10 ** (db / 20.0) + + +def linear_to_db(linear: float) -> float: + """线性音量转换为 dB. + + Args: + linear: 线性音量值 + + Returns: + 分贝值 + """ + import math + + if linear <= 0: + return -float("inf") + return 20.0 * math.log10(linear) + + +# ── 轨道排序与过滤 ──────────────────────────────────────────────────────────── + + +def sort_tracks_by_priority( + tracks: list[dict], +) -> list[dict]: + """按优先级排序轨道. + + priority 数字越小优先级越高(越先播放/越底层)。 + 相同优先级保持原顺序。 + + Args: + tracks: 轨道配置列表 + + Returns: + 排序后的轨道列表 + """ + return sorted(tracks, key=lambda t: int(t.get("priority", 100))) + + +def filter_enabled_tracks(tracks: list[dict]) -> list[dict]: + """过滤出启用的轨道. + + Args: + tracks: 轨道列表 + + Returns: + 启用的轨道列表 + """ + result = [] + for t in tracks: + enabled = t.get("enabled", True) + if bool(enabled) and enabled != "false" and enabled != 0: + result.append(t) + return result + + +def count_track_types(tracks: list[dict]) -> dict[str, int]: + """统计各类型轨道数量. + + Args: + tracks: 轨道列表 + + Returns: + 类型计数字典 + """ + counts: dict[str, int] = {} + for t in tracks: + ttype = t.get("track_type", "unknown") + counts[ttype] = counts.get(ttype, 0) + 1 + return counts + + +# ── 配置验证 ────────────────────────────────────────────────────────────────── + + +def validate_audio_track(track: dict) -> tuple[bool, list[str]]: + """验证单条音轨配置. + + Args: + track: 轨道配置字典 + + Returns: + (是否合法, 错误信息列表) + """ + errors: list[str] = [] + + # 音频路径 + audio_path = track.get("audio_path", "") + if not audio_path and not track.get("asset_id"): + errors.append("轨道需要 audio_path 或 asset_id") + + # 音量范围 + volume = track.get("volume", 1.0) + try: + v = float(volume) + if v < 0: + errors.append("volume 不能为负数") + if v > 2.0: + errors.append("volume 建议不超过 2.0") + except (ValueError, TypeError): + errors.append("volume 必须是数字") + + # 淡入淡出 + fade_in = track.get("fade_in", 0) + fade_out = track.get("fade_out", 0) + try: + if float(fade_in) < 0: + errors.append("fade_in 不能为负数") + except (ValueError, TypeError): + errors.append("fade_in 必须是数字") + + try: + if float(fade_out) < 0: + errors.append("fade_out 不能为负数") + except (ValueError, TypeError): + errors.append("fade_out 必须是数字") + + # 开始时间 + start_time = track.get("start_time", 0) + try: + float(start_time) # 验证是否为数字 + except (ValueError, TypeError): + errors.append("start_time 必须是数字") + + return (len(errors) == 0, errors) + + +def validate_mix_config(config: dict) -> tuple[bool, list[str]]: + """验证混音配置. + + Args: + config: 混音配置 + + Returns: + (是否合法, 错误信息列表) + """ + errors: list[str] = [] + + tracks = config.get("tracks", []) + if not tracks: + errors.append("至少需要一条轨道") + + # 验证每条轨道 + for i, track in enumerate(tracks): + ok, track_errors = validate_audio_track(track) + if not ok: + for err in track_errors: + errors.append(f"第{i+1}轨:{err}") + + # 目标时长 + target_duration = config.get("target_duration", 0) + try: + if float(target_duration) < 0: + errors.append("target_duration 不能为负数") + except (ValueError, TypeError): + errors.append("target_duration 必须是数字") + + return (len(errors) == 0, errors) + + +# ── 工具函数 ────────────────────────────────────────────────────────────────── + + +def calculate_total_tracks(config: dict) -> int: + """计算总轨道数(含主音频). + + Args: + config: 混音配置 + + Returns: + 总轨道数 + """ + tracks = config.get("tracks", []) + has_main = config.get("has_main_audio", True) + count = len(tracks) + if has_main: + count += 1 + return count + + +def estimate_mix_duration(tracks: list[dict]) -> float: + """估算混音总时长(所有轨道的最晚结束时间). + + Args: + tracks: 轨道列表,包含 start_time 和 duration + + Returns: + 估算总时长(秒) + """ + max_end = 0.0 + for t in tracks: + try: + start = float(t.get("start_time", 0)) + dur = float(t.get("duration", 0)) + if dur > 0: + end = start + dur + if end > max_end: + max_end = end + except (ValueError, TypeError): + continue + return max_end diff --git a/tests/unit/test_multi_track_mixer_pure.py b/tests/unit/test_multi_track_mixer_pure.py new file mode 100755 index 000000000..9d6c2376b --- /dev/null +++ b/tests/unit/test_multi_track_mixer_pure.py @@ -0,0 +1,638 @@ +"""多轨混音纯逻辑单元测试.""" + +from __future__ import annotations + +import math + +import pytest +from 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, +) + +# ───────────────────────────────────────────────────────────────────────────── +# 时间计算测试 +# ───────────────────────────────────────────────────────────────────────────── + + +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 + assert trim == 0.0 + + def test_track_longer_than_audio(self): + """轨道时长超过音频长度.""" + start, dur, trim = calculate_effective_range(0, 100, 30, 60) + assert start == 0.0 + assert dur == 30.0 # 用音频全长 + + def test_zero_track_duration(self): + """轨道时长为 0(用音频全长).""" + start, dur, trim = calculate_effective_range(0, 0, 30, 60) + assert start == 0.0 + assert dur == 30.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_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_zero_audio_duration(self): + """音频时长为 0.""" + start, dur, trim = calculate_effective_range(0, 10, 0, 60) + assert dur == 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_full_target_duration(self): + """轨道覆盖整个目标时长.""" + start, dur, trim = calculate_effective_range(0, 0, 100, 60) + assert start == 0.0 + assert dur == 60.0 + + +class TestIsTrackVisible: + """轨道可见性测试.""" + + def test_visible_track(self): + """可见轨道.""" + assert is_track_visible(5, 10, 30, 60) is True + + def test_invisible_after_target(self): + """目标之后不可见.""" + assert is_track_visible(100, 10, 30, 60) is False + + def test_invisible_zero_duration(self): + """零时长不可见.""" + assert is_track_visible(0, 0, 0, 60) is False + + +# ───────────────────────────────────────────────────────────────────────────── +# 滤镜链构建测试 +# ───────────────────────────────────────────────────────────────────────────── + + +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 滤镜.""" + result = build_track_filter_chain( + volume=1.0, + fade_in=0, + fade_out=0, + effective_start=0, + need_duration=10, + trim_start=0, + target_duration=60, + ) + assert "volume=" not in result + + def test_no_fade_in(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, + fade_in=0, + fade_out=0, + effective_start=0.001, + need_duration=10, + trim_start=0, + target_duration=60, + ) + assert "adelay" not in result + + def test_with_delay(self): + """有延迟.""" + result = build_track_filter_chain( + volume=1.0, + fade_in=0, + fade_out=0, + effective_start=2.5, + need_duration=10, + trim_start=0, + target_duration=60, + ) + 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): + """从音频中间开始截取.""" + 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, + ) + assert "atrim=3.000:8.000" in result # 3.0 to 3.0+5.0 + + +# ───────────────────────────────────────────────────────────────────────────── +# amix 滤镜测试 +# ───────────────────────────────────────────────────────────────────────────── + + +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") + assert "duration=shortest" in result + + def test_invalid_duration_mode(self): + """无效模式,默认 longest.""" + result = build_amix_filter(3, "invalid") + assert "duration=longest" in result + + +class TestCalculateAmixVolumeCompensation: + """音量补偿计算测试.""" + + def test_single_track(self): + """单轨,无需补偿.""" + assert calculate_amix_volume_compensation(1) == 1.0 + + def test_two_tracks(self): + """两轨,补偿 2x.""" + assert calculate_amix_volume_compensation(2) == 2.0 + + def test_five_tracks(self): + """五轨,补偿 5x.""" + assert calculate_amix_volume_compensation(5) == 5.0 + + def test_zero_tracks(self): + """零轨,返回 1.""" + assert calculate_amix_volume_compensation(0) == 1.0 + + +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): + """无轨道无主音频.""" + assert build_mix_filter_complex(0, has_main=False) == "" + + +# ───────────────────────────────────────────────────────────────────────────── +# 音量计算测试 +# ───────────────────────────────────────────────────────────────────────────── + + +class TestNormalizeVolume: + """音量规范化测试.""" + + def test_normal_volume(self): + """正常音量.""" + assert normalize_volume(0.5) == 0.5 + + def test_none_default(self): + """None 默认 1.0.""" + assert normalize_volume(None) == 1.0 + + def test_below_zero_clamped(self): + """负值钳制到 0.""" + assert normalize_volume(-5) == 0.0 + + def test_above_max_clamped(self): + """超过上限钳制.""" + assert normalize_volume(3.0) == 2.0 + + def test_string_input(self): + """字符串输入.""" + assert normalize_volume("0.5") == 0.5 + + def test_invalid_string(self): + """无效字符串默认 1.0.""" + assert normalize_volume("abc") == 1.0 + + +class TestDbConversion: + """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) + + def test_positive_db(self): + """正 dB > 1.""" + assert db_to_linear(6) == pytest.approx(2.0, rel=0.01) + + def test_round_trip(self): + """往返转换.""" + original = 0.5 + db = linear_to_db(original) + result = db_to_linear(db) + assert result == pytest.approx(original) + + def test_zero_linear_is_negative_inf(self): + """零线性值 = -inf dB.""" + assert math.isinf(linear_to_db(0)) + assert linear_to_db(0) < 0 + + +# ───────────────────────────────────────────────────────────────────────────── +# 轨道排序与过滤测试 +# ───────────────────────────────────────────────────────────────────────────── + + +class TestSortTracksByPriority: + """轨道优先级排序测试.""" + + def test_sorted_by_priority(self): + """按优先级排序.""" + tracks = [ + {"priority": 10, "name": "high"}, + {"priority": 1, "name": "highest"}, + {"priority": 100, "name": "low"}, + ] + result = sort_tracks_by_priority(tracks) + assert result[0]["name"] == "highest" + assert result[1]["name"] == "high" + assert result[2]["name"] == "low" + + def test_default_priority_100(self): + """默认优先级 100.""" + tracks = [ + {"priority": 50, "name": "mid"}, + {"name": "default"}, + ] + result = sort_tracks_by_priority(tracks) + assert result[0]["name"] == "mid" + assert result[1]["name"] == "default" + + def test_same_preserves_order(self): + """同优先级保持顺序.""" + tracks = [ + {"priority": 10, "name": "first"}, + {"priority": 10, "name": "second"}, + ] + result = sort_tracks_by_priority(tracks) + assert result[0]["name"] == "first" + assert result[1]["name"] == "second" + + def test_empty_list(self): + """空列表.""" + assert sort_tracks_by_priority([]) == [] + + +class TestFilterEnabledTracks: + """启用轨道过滤测试.""" + + def test_all_enabled(self): + """全部启用.""" + tracks = [{"enabled": True}, {"enabled": True}] + assert len(filter_enabled_tracks(tracks)) == 2 + + def test_mixed(self): + """混合.""" + tracks = [ + {"enabled": True, "name": "a"}, + {"enabled": False, "name": "b"}, + ] + 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_empty_list(self): + """空列表.""" + assert filter_enabled_tracks([]) == [] + + +class TestCountTrackTypes: + """轨道类型统计测试.""" + + def test_mixed_types(self): + """混合类型.""" + tracks = [ + {"track_type": "bgm"}, + {"track_type": "voiceover"}, + {"track_type": "bgm"}, + {"track_type": "sfx"}, + ] + counts = count_track_types(tracks) + assert counts["bgm"] == 2 + assert counts["voiceover"] == 1 + assert counts["sfx"] == 1 + + def test_default_type(self): + """默认类型.""" + tracks = [{}] + counts = count_track_types(tracks) + assert counts["unknown"] == 1 + + def test_empty_list(self): + """空列表.""" + assert count_track_types([]) == {} + + +# ───────────────────────────────────────────────────────────────────────────── +# 配置验证测试 +# ───────────────────────────────────────────────────────────────────────────── + + +class TestValidateAudioTrack: + """单轨验证测试.""" + + 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_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_negative_volume(self): + """负音量.""" + ok, errors = validate_audio_track( + { + "audio_path": "/a.mp3", + "volume": -1, + } + ) + assert ok 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 + 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 + assert any("fade_out" in e for e in errors) + + def test_invalid_volume_type(self): + """无效音量类型.""" + ok, errors = validate_audio_track( + { + "audio_path": "/a.mp3", + "volume": "loud", + } + ) + assert ok is False + assert any("volume" in e for e in errors) + + def test_with_asset_id(self): + """有 asset_id 无 audio_path 也合法.""" + ok, errors = validate_audio_track({"asset_id": "123"}) + assert ok is True + + +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 + + def test_empty_tracks(self): + """空轨道列表.""" + ok, errors = validate_mix_config({"tracks": []}) + assert ok 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_negative_target_duration(self): + """负目标时长.""" + ok, errors = validate_mix_config( + { + "tracks": [{"audio_path": "/a.mp3"}], + "target_duration": -10, + } + ) + assert ok is False + assert any("target_duration" in e for e in errors) + + +# ───────────────────────────────────────────────────────────────────────────── +# 工具函数测试 +# ───────────────────────────────────────────────────────────────────────────── + + +class TestCalculateTotalTracks: + """总轨道数计算测试.""" + + def test_with_main(self): + """含主音频.""" + assert calculate_total_tracks({"tracks": [1, 2, 3]}) == 4 + + def test_without_main(self): + """不含主音频.""" + assert ( + calculate_total_tracks( + { + "tracks": [1, 2], + "has_main_audio": False, + } + ) + == 2 + ) + + def test_empty_tracks_with_main(self): + """无轨道,只有主音频.""" + assert calculate_total_tracks({"tracks": []}) == 1 + + +class TestEstimateMixDuration: + """混音时长估算测试.""" + + def test_multiple_tracks(self): + """多轨道取最长结束时间.""" + tracks = [ + {"start_time": 0, "duration": 10}, + {"start_time": 5, "duration": 20}, # 结束 25 + {"start_time": 2, "duration": 5}, + ] + assert estimate_mix_duration(tracks) == pytest.approx(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}, + ] + assert estimate_mix_duration(tracks) == pytest.approx(15.0) -- 2.54.0