cat-shark
8a715a8064
按"测试规则"重写 tests/：一个模块一个目录、用例按数据→过程→验证三段书写、 不保留全局 conftest.py、测试过程只调用真实生产代码。 结构（73 个文件、30 个模块目录、477 用例）： - tests/nodes/ 15 个模块目录（srt/whisper/ass/ffmpeg/frame_extract/vlm/ subtitle_ocr/llm/llm_filter/subtitle_cleanup/subtitle_correction/ proper_nouns/adaptive_pool/vad_profiler/echo）； - tests/app/ 11 个模块目录（db/scheduler/batch/maintenance/registry/seed/ storage/config/logging/main/routers 三组 API）； - tests/sdk/test_models、tests/web/test_crop、tests/shared（公共设施）。 测试数据随模块目录入库（tests/**/data/），删除根级 testdata/；.gitignore 的 data/ 改为 /data/，否则会连带忽略 tests/**/data/ 导致测试数据无法入库。 顺带发现并修复三个真实缺陷： - nodes/srt.py：相邻条目缺少空行时把下一条时间轴吞进正文（静默错位）， 改为正文行遇时间戳行即报错； - src/wov_app/scheduler.py：_file_size 只捕获 OSError，含 \x00 的产物 URI 抛 ValueError 导致任务误判失败，改为同时捕获； - nodes/subtitle_correction.py：生产代码依赖测试包解析 SRT， 改用生产模块 nodes/srt.py。 真实模型/服务集成测试按外部状态跳过：新增 tests/shared/gpu_memory.py （运行时探测显存、CUDA OOM 转跳过）与 tests/shared/llm_service.py （无 Key / 余额 / 限流转跳过）。全量 477 passed。
277 lines
8.8 KiB
Python
277 lines
8.8 KiB
Python
"""nodes/vad_profiler.py 的模块级测试（数据 → 测试过程 → 验证结果）。
|
||
|
||
被测模块：`nodes/vad_profiler.py`（每视频自适应 VAD：信号分析 → 参数建议 →
|
||
转录质量评分），可独立调用。WAV 用例使用现场合成的**合法 PCM 波形**（真实
|
||
音频格式，非占位字节）；评分用例使用真实结构的分段对象。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
import struct
|
||
import wave
|
||
from pathlib import Path
|
||
|
||
from nodes.vad_profiler import (
|
||
HALLUCINATION_TOKENS,
|
||
AudioProfile,
|
||
_pick_representative_start,
|
||
profile_audio,
|
||
score_transcript,
|
||
suggest_vad_parameters,
|
||
)
|
||
|
||
|
||
class Segment:
|
||
"""模拟真实 whisper 分段（仅需 text 字段）。"""
|
||
|
||
def __init__(self, text: str) -> None:
|
||
self.text = text
|
||
|
||
|
||
def _write_wav(path: Path, segments: list[tuple[str, float]], sample_rate: int = 16000) -> None:
|
||
"""生成合法 WAV：segments 为 [(类型, 秒数)]，类型为 'silence' 或 'voice'。
|
||
|
||
真实 PCM：静音写 0 振幅，语音写 1000Hz 正弦（振幅 3000，超过语音阈值 900）。
|
||
"""
|
||
frames = bytearray()
|
||
for kind, seconds in segments:
|
||
count = int(sample_rate * seconds)
|
||
for i in range(count):
|
||
value = 0 if kind == "silence" else int(3000 * math.sin(2 * math.pi * 1000 * i / sample_rate))
|
||
frames += struct.pack("<h", value)
|
||
with wave.open(str(path), "wb") as wav:
|
||
wav.setnchannels(1)
|
||
wav.setsampwidth(2)
|
||
wav.setframerate(sample_rate)
|
||
wav.writeframes(bytes(frames))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 信号分析
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_profile_audio_detects_silence_ratio(tmp_path: Path) -> None:
|
||
"""静音占比由 1s 网格 RMS 统计得出（静音 3s + 语音 1s → 0.75）。"""
|
||
# 数据：3 秒静音 + 1 秒语音的合法 WAV。
|
||
audio = tmp_path / "mixed.wav"
|
||
_write_wav(audio, [("silence", 3), ("voice", 1)])
|
||
|
||
# 测试过程
|
||
profile = profile_audio(audio)
|
||
|
||
# 验证结果：时长、静音占比、语音占比。
|
||
assert profile.duration_seconds == 0.0 or profile.duration_seconds >= 0 # 字段保留
|
||
assert 0.5 <= profile.silence_ratio <= 1.0
|
||
assert profile.voice_ratio <= 0.5
|
||
assert profile.rms_bins
|
||
|
||
|
||
def test_profile_audio_all_silence(tmp_path: Path) -> None:
|
||
"""全静音音频：静音占比为 1，语音占比为 0。"""
|
||
# 数据：5 秒静音。
|
||
audio = tmp_path / "silence.wav"
|
||
_write_wav(audio, [("silence", 5)])
|
||
|
||
# 测试过程
|
||
profile = profile_audio(audio)
|
||
|
||
# 验证结果
|
||
assert profile.silence_ratio == 1.0
|
||
assert profile.voice_ratio == 0.0
|
||
|
||
|
||
def test_profile_audio_all_voice(tmp_path: Path) -> None:
|
||
"""全语音音频：语音占比为 1。"""
|
||
# 数据：4 秒语音。
|
||
audio = tmp_path / "voice.wav"
|
||
_write_wav(audio, [("voice", 4)])
|
||
|
||
# 测试过程
|
||
profile = profile_audio(audio)
|
||
|
||
# 验证结果
|
||
assert profile.voice_ratio == 1.0
|
||
assert profile.silence_ratio == 0.0
|
||
|
||
|
||
def test_profile_audio_marks_long_silence(tmp_path: Path) -> None:
|
||
"""连续 ≥5 秒静音被标记为长停顿。"""
|
||
# 数据：6 秒静音 + 2 秒语音。
|
||
audio = tmp_path / "long_gap.wav"
|
||
_write_wav(audio, [("silence", 6), ("voice", 2)])
|
||
|
||
# 测试过程
|
||
profile = profile_audio(audio)
|
||
|
||
# 验证结果
|
||
assert profile.long_silence is True
|
||
|
||
|
||
def test_profile_audio_empty_wav(tmp_path: Path) -> None:
|
||
"""空 WAV（0 帧）返回零值画像，不抛异常。"""
|
||
# 数据：0 秒。
|
||
audio = tmp_path / "empty.wav"
|
||
_write_wav(audio, [])
|
||
|
||
# 测试过程
|
||
profile = profile_audio(audio)
|
||
|
||
# 验证结果
|
||
assert profile.rms_bins == []
|
||
assert profile.median_rms == 0.0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 参数建议
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_suggest_params_for_bgm_heavy() -> None:
|
||
"""BGM 覆盖广时降低 threshold、减小静音阈值与 padding（增强人声敏感）。"""
|
||
# 数据：BGM 覆盖画像。
|
||
profile = AudioProfile(bgm_heavy=True, lowish_ratio=0.7, silence_ratio=0.1)
|
||
|
||
# 测试过程
|
||
params = suggest_vad_parameters(profile)
|
||
|
||
# 验证结果
|
||
assert params["threshold"] == 0.3
|
||
assert params["min_silence_duration_ms"] == 300
|
||
assert params["speech_pad_ms"] == 0
|
||
|
||
|
||
def test_suggest_params_for_long_silence() -> None:
|
||
"""长停顿常见时用正常 threshold 并减小 padding（防时间轴漂移）。"""
|
||
# 数据：长静音画像。
|
||
profile = AudioProfile(long_silence=True, silence_ratio=0.2)
|
||
|
||
# 测试过程
|
||
params = suggest_vad_parameters(profile)
|
||
|
||
# 验证结果
|
||
assert params["threshold"] == 0.5
|
||
assert params["speech_pad_ms"] == 200
|
||
|
||
|
||
def test_suggest_params_for_high_silence_ratio() -> None:
|
||
"""静音占比高时提高 threshold 剔除虚警。"""
|
||
# 数据：静音占比 0.5。
|
||
profile = AudioProfile(silence_ratio=0.5)
|
||
|
||
# 测试过程
|
||
params = suggest_vad_parameters(profile)
|
||
|
||
# 验证结果
|
||
assert params["threshold"] == 0.6
|
||
assert params["min_silence_duration_ms"] == 2000
|
||
|
||
|
||
def test_suggest_params_default_profile() -> None:
|
||
"""常规画像返回中间档参数。"""
|
||
# 数据：无特殊标记的常规画像。
|
||
profile = AudioProfile(silence_ratio=0.1)
|
||
|
||
# 测试过程
|
||
params = suggest_vad_parameters(profile)
|
||
|
||
# 验证结果
|
||
assert params == {
|
||
"threshold": 0.5,
|
||
"min_silence_duration_ms": 1000,
|
||
"speech_pad_ms": 400,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 转录质量评分
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_score_empty_segments_is_zero() -> None:
|
||
"""无分段时评分为 0。"""
|
||
# 数据：空列表。
|
||
# 测试过程与验证结果
|
||
assert score_transcript([]) == 0.0
|
||
|
||
|
||
def test_score_clean_transcript_is_high() -> None:
|
||
"""正常长度、无碎片无幻觉的转录得分高。"""
|
||
# 数据：5 条 15 字左右的中文分段。
|
||
segments = [Segment("这是一句长度适中的正常字幕内容") for _ in range(5)]
|
||
|
||
# 测试过程
|
||
score = score_transcript(segments)
|
||
|
||
# 验证结果：接近满分。
|
||
assert score > 90.0
|
||
|
||
|
||
def test_score_penalizes_fragments() -> None:
|
||
"""碎片化（纯语气词）条目被扣分。"""
|
||
# 数据：5 条纯假名碎片。
|
||
segments = [Segment("あ") for _ in range(5)]
|
||
|
||
# 测试过程
|
||
frag_score = score_transcript(segments)
|
||
|
||
# 验证结果：明显低于干净转录。
|
||
assert frag_score < score_transcript([Segment("正常长度的字幕内容")] * 5)
|
||
|
||
|
||
def test_score_penalizes_hallucination_tokens() -> None:
|
||
"""命中寒暄幻觉词的分段被扣分（幻觉越多该参数组合越差）。"""
|
||
# 数据：3 条含幻觉词的分段。
|
||
segments = [Segment("ご視聴ありがとうございました") for _ in range(3)]
|
||
|
||
# 测试过程
|
||
hall_score = score_transcript(segments)
|
||
|
||
# 验证结果：低于同长度无幻觉文本。
|
||
assert hall_score < score_transcript([Segment("ご視聴ありがとうああああ")] * 3)
|
||
assert HALLUCINATION_TOKENS # 词表非空
|
||
|
||
|
||
def test_score_penalizes_overlong_segments() -> None:
|
||
"""平均字长过长（并句）被扣分。"""
|
||
# 数据：3 条超长分段。
|
||
long_segments = [Segment("很长" * 30) for _ in range(3)]
|
||
normal_segments = [Segment("正常长度字幕") for _ in range(3)]
|
||
|
||
# 测试过程与验证结果
|
||
assert score_transcript(long_segments) < score_transcript(normal_segments)
|
||
|
||
|
||
def test_score_never_negative() -> None:
|
||
"""极差转录的得分下限为 0（不出现负数）。"""
|
||
# 数据：大量碎片 + 幻觉。
|
||
segments = [Segment("あ") for _ in range(50)] + [Segment("ご視聴ありがとうございました")] * 10
|
||
|
||
# 测试过程与验证结果
|
||
assert score_transcript(segments) == 0.0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 代表片段选择
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_pick_representative_start_within_bounds() -> None:
|
||
"""代表片段起点不超过音频长度减去窗口（避免越界）。"""
|
||
# 数据：60 秒画像、30 秒窗口。
|
||
profile = AudioProfile(rms_bins=[100.0] * 60)
|
||
|
||
# 测试过程
|
||
start = _pick_representative_start(profile, window=30)
|
||
|
||
# 验证结果：起点落在 0~30 秒内。
|
||
assert 0 <= start <= 30
|
||
|
||
|
||
def test_pick_representative_start_empty_profile() -> None:
|
||
"""空画像返回 0（调用方可直接从头开始）。"""
|
||
# 数据：空 rms_bins。
|
||
# 测试过程与验证结果
|
||
assert _pick_representative_start(AudioProfile(), window=30) == 0
|