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。
197 lines
7.2 KiB
Python
197 lines
7.2 KiB
Python
"""nodes/ffmpeg.py 的模块级测试（数据 → 测试过程 → 验证结果）。
|
||
|
||
被测模块：`nodes/ffmpeg.py`（ffmpeg 定位 + 提音），可独立调用。
|
||
测试使用模块目录 `data/` 下的真实视频，调用真实 ffmpeg 子进程产出真实 WAV；
|
||
仅在 I/O 边界（环境变量 / PATH 查找）使用 monkeypatch。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import shutil
|
||
import subprocess
|
||
import wave
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from nodes.ffmpeg import _bundled_ffmpeg, _ffmpeg_bin, invoke
|
||
from wov_sdk.models import InvokeRequest
|
||
|
||
# 模块专用真实素材：
|
||
# - clip_with_audio.mp4：10 秒真实视频 + 真实语音音轨（提音用例的输入）；
|
||
# - subtitle_10s.mp4：仅视频无音轨（用于验证"无音频流"时的失败路径）。
|
||
DATA_DIR = Path(__file__).resolve().parent / "data"
|
||
TEST_VIDEO = DATA_DIR / "clip_with_audio.mp4"
|
||
VIDEO_WITHOUT_AUDIO = DATA_DIR / "subtitle_10s.mp4"
|
||
|
||
|
||
def _request(tmp_path: Path, video: Path | None, **params) -> InvokeRequest:
|
||
"""构造真实请求；video 为 None 时表示不传 video_uri。"""
|
||
inputs = {} if video is None else {"video_uri": str(video)}
|
||
return InvokeRequest(
|
||
run_id="run-test",
|
||
node_instance_id="ffmpeg-1",
|
||
params=params,
|
||
inputs=inputs,
|
||
output_dir=str(tmp_path / "out"),
|
||
)
|
||
|
||
|
||
def _available_ffmpeg() -> str | None:
|
||
"""返回当前环境可用的 ffmpeg 路径（用于跳过缺少 ffmpeg 的环境）。"""
|
||
found = shutil.which("ffmpeg")
|
||
if found:
|
||
return found
|
||
return _bundled_ffmpeg()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# ffmpeg 定位
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_ffmpeg_bin_prefers_explicit_env(monkeypatch, tmp_path: Path) -> None:
|
||
"""FFMPEG_BIN 环境变量最优先（部署可指定自定义二进制）。"""
|
||
# 数据：显式配置一个真实存在的假二进制路径。
|
||
fake = tmp_path / "my-ffmpeg"
|
||
fake.write_text("#!/bin/sh\n", encoding="utf-8")
|
||
monkeypatch.setenv("FFMPEG_BIN", str(fake))
|
||
|
||
# 测试过程与验证结果
|
||
assert _ffmpeg_bin() == str(fake)
|
||
|
||
|
||
def test_ffmpeg_bin_falls_back_to_path(monkeypatch) -> None:
|
||
"""未配置环境变量时使用 PATH 中的 ffmpeg。"""
|
||
# 数据：清除显式配置。
|
||
monkeypatch.delenv("FFMPEG_BIN", raising=False)
|
||
|
||
# 测试过程
|
||
resolved = _ffmpeg_bin()
|
||
|
||
# 验证结果：返回可执行的 ffmpeg（PATH 或内置二进制）。
|
||
assert resolved
|
||
assert shutil.which(resolved) is not None or Path(resolved).is_file()
|
||
|
||
|
||
def test_ffmpeg_bin_falls_back_to_bundled(monkeypatch) -> None:
|
||
"""PATH 无 ffmpeg 时回退 imageio-ffmpeg 内置二进制。"""
|
||
# 数据：清空环境变量并让 which 返回 None。
|
||
monkeypatch.delenv("FFMPEG_BIN", raising=False)
|
||
monkeypatch.setattr(shutil, "which", lambda name: None)
|
||
|
||
# 测试过程
|
||
resolved = _ffmpeg_bin()
|
||
|
||
# 验证结果：得到内置二进制路径（本环境已安装 imageio-ffmpeg）。
|
||
bundled = _bundled_ffmpeg()
|
||
assert resolved == (bundled or "ffmpeg")
|
||
|
||
|
||
def test_ffmpeg_bin_returns_plain_name_when_nothing_available(monkeypatch) -> None:
|
||
"""完全不可用时返回裸名 "ffmpeg"，由调用处统一报失败。"""
|
||
# 数据：环境变量、PATH、内置二进制都不可用。
|
||
monkeypatch.delenv("FFMPEG_BIN", raising=False)
|
||
monkeypatch.setattr(shutil, "which", lambda name: None)
|
||
monkeypatch.setattr("nodes.ffmpeg._bundled_ffmpeg", lambda: None)
|
||
|
||
# 测试过程与验证结果
|
||
assert _ffmpeg_bin() == "ffmpeg"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# invoke：真实提音
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_invoke_extracts_16k_mono_wav(tmp_path: Path) -> None:
|
||
"""真实视频 → 16kHz 单声道 WAV 产物（ASR 节点的输入契约）。"""
|
||
# 数据：模块 data/ 下带真实语音音轨的测试视频。
|
||
if _available_ffmpeg() is None:
|
||
pytest.skip("环境中没有可用 ffmpeg")
|
||
assert TEST_VIDEO.is_file(), f"缺少测试视频 {TEST_VIDEO}"
|
||
|
||
# 测试过程
|
||
response = invoke(_request(tmp_path, TEST_VIDEO))
|
||
|
||
# 验证结果：产物存在，WAV 头为 16kHz 单声道，时长与源视频一致（约 10s）。
|
||
assert response.status == "completed", response.error
|
||
audio = Path(response.outputs["audio_uri"])
|
||
assert audio.is_file() and audio.suffix == ".wav"
|
||
with wave.open(str(audio), "rb") as wav:
|
||
assert wav.getframerate() == 16000
|
||
assert wav.getnchannels() == 1
|
||
duration = wav.getnframes() / wav.getframerate()
|
||
assert 9.0 <= duration <= 11.0
|
||
|
||
|
||
def test_invoke_honors_sample_rate_and_channels_params(tmp_path: Path) -> None:
|
||
"""sample_rate / channels 参数透传到 ffmpeg（产物头体现）。"""
|
||
# 数据：显式请求 8kHz 单声道。
|
||
if _available_ffmpeg() is None:
|
||
pytest.skip("环境中没有可用 ffmpeg")
|
||
|
||
# 测试过程
|
||
response = invoke(_request(tmp_path, TEST_VIDEO, sample_rate=8000, channels=1))
|
||
|
||
# 验证结果
|
||
assert response.status == "completed", response.error
|
||
with wave.open(response.outputs["audio_uri"], "rb") as wav:
|
||
assert wav.getframerate() == 8000
|
||
|
||
|
||
def test_invoke_fails_without_video_uri(tmp_path: Path) -> None:
|
||
"""缺少 video_uri 时返回 failed 并说明原因。"""
|
||
# 数据：不传输入。
|
||
# 测试过程
|
||
response = invoke(_request(tmp_path, None))
|
||
|
||
# 验证结果
|
||
assert response.status == "failed"
|
||
assert "video_uri" in (response.error or "")
|
||
|
||
|
||
def test_invoke_fails_when_ffmpeg_missing(monkeypatch, tmp_path: Path) -> None:
|
||
"""环境无 ffmpeg 时明确失败，不抛晦涩的子进程异常。"""
|
||
# 数据：把所有 ffmpeg 来源都屏蔽。
|
||
monkeypatch.setenv("FFMPEG_BIN", "definitely-not-a-real-binary")
|
||
monkeypatch.setattr(shutil, "which", lambda name: None)
|
||
|
||
# 测试过程
|
||
response = invoke(_request(tmp_path, TEST_VIDEO))
|
||
|
||
# 验证结果
|
||
assert response.status == "failed"
|
||
assert "ffmpeg" in (response.error or "")
|
||
|
||
|
||
def test_invoke_fails_when_ffmpeg_returns_error(tmp_path: Path) -> None:
|
||
"""输入文件不是合法媒体时 ffmpeg 报错 → 节点返回 failed（不静默成功）。"""
|
||
# 数据：把文本文件伪装成视频。
|
||
broken = tmp_path / "broken.mp4"
|
||
broken.write_text("not a video", encoding="utf-8")
|
||
if _available_ffmpeg() is None:
|
||
pytest.skip("环境中没有可用 ffmpeg")
|
||
|
||
# 测试过程
|
||
response = invoke(_request(tmp_path, broken))
|
||
|
||
# 验证结果
|
||
assert response.status == "failed"
|
||
assert (response.error or "").strip()
|
||
|
||
|
||
def test_invoke_fails_when_video_has_no_audio_stream(tmp_path: Path) -> None:
|
||
"""源视频不含音频流时提音失败并返回 ffmpeg 诊断信息（不静默产出空文件）。"""
|
||
# 数据：只有视频轨的测试素材。
|
||
if _available_ffmpeg() is None:
|
||
pytest.skip("环境中没有可用 ffmpeg")
|
||
assert VIDEO_WITHOUT_AUDIO.is_file()
|
||
|
||
# 测试过程
|
||
response = invoke(_request(tmp_path, VIDEO_WITHOUT_AUDIO))
|
||
|
||
# 验证结果：失败且无残缺产物。
|
||
assert response.status == "failed"
|
||
assert not (tmp_path / "out" / "audio.wav").exists()
|