121 lines
4.7 KiB
Python
121 lines
4.7 KiB
Python
"""FFmpeg 提音节点测试。
|
|
|
|
覆盖 ffmpeg 解析优先级、成功/失败执行、缺失输入与入口点启动等真实路径。
|
|
"""
|
|
|
|
import json
|
|
import runpy
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from wov_node_ffmpeg.__main__ import _ffmpeg_bin, invoke
|
|
from wov_sdk.models import InvokeRequest
|
|
|
|
|
|
def _request(tmp_path, **overrides) -> InvokeRequest:
|
|
"""构造包含默认视频输入与提音参数的调用请求。"""
|
|
payload = {
|
|
"run_id": "run_1",
|
|
"node_instance_id": "ni_1",
|
|
"inputs": {"video_uri": str(tmp_path / "input.mp4")},
|
|
"params": {"sample_rate": 16000, "channels": 1},
|
|
"output_dir": str(tmp_path / "out"),
|
|
}
|
|
payload.update(overrides)
|
|
return InvokeRequest(**payload)
|
|
|
|
|
|
def test_success(tmp_path, monkeypatch) -> None:
|
|
"""验证成功调用会生成 audio.wav 产物。"""
|
|
def fake_run(command, **kwargs):
|
|
output = Path(command[-1])
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
output.write_bytes(b"fake wav")
|
|
return subprocess.CompletedProcess(command, 0)
|
|
|
|
monkeypatch.setattr("wov_node_ffmpeg.__main__.shutil.which", lambda _: "ffmpeg")
|
|
monkeypatch.setattr("wov_node_ffmpeg.__main__.subprocess.run", fake_run)
|
|
response = invoke(_request(tmp_path))
|
|
assert response.status == "completed"
|
|
assert Path(response.outputs["audio_uri"]).name == "audio.wav"
|
|
|
|
|
|
def test_configured_ffmpeg(tmp_path, monkeypatch) -> None:
|
|
"""验证 FFMPEG_BIN 环境变量优先于 PATH 查找。"""
|
|
fake_bin = tmp_path / "ffmpeg.exe"
|
|
fake_bin.write_bytes(b"")
|
|
monkeypatch.setenv("FFMPEG_BIN", str(fake_bin))
|
|
monkeypatch.setattr("wov_node_ffmpeg.__main__.subprocess.run", lambda *a, **k: subprocess.CompletedProcess([], 0))
|
|
response = invoke(_request(tmp_path))
|
|
assert response.status == "completed"
|
|
|
|
|
|
def test_bundled_ffmpeg_fallback(tmp_path, monkeypatch) -> None:
|
|
"""验证无系统 ffmpeg 时回退到 imageio-ffmpeg 内置二进制。"""
|
|
monkeypatch.delenv("FFMPEG_BIN", raising=False)
|
|
monkeypatch.setattr("wov_node_ffmpeg.__main__.shutil.which", lambda _: None)
|
|
bundled = _ffmpeg_bin()
|
|
assert bundled != "ffmpeg"
|
|
assert Path(bundled).is_file()
|
|
monkeypatch.setattr("wov_node_ffmpeg.__main__._ffmpeg_bin", lambda: bundled)
|
|
|
|
def fake_run(command, **kwargs):
|
|
Path(command[-1]).parent.mkdir(parents=True, exist_ok=True)
|
|
Path(command[-1]).write_bytes(b"wav")
|
|
return subprocess.CompletedProcess(command, 0)
|
|
|
|
monkeypatch.setattr("wov_node_ffmpeg.__main__.subprocess.run", fake_run)
|
|
response = invoke(_request(tmp_path))
|
|
assert response.status == "completed"
|
|
|
|
|
|
def test_bundled_ffmpeg_import_error(monkeypatch) -> None:
|
|
"""验证 imageio-ffmpeg 不可用时最终回退为 "ffmpeg" 字符串。"""
|
|
monkeypatch.delenv("FFMPEG_BIN", raising=False)
|
|
monkeypatch.setattr("wov_node_ffmpeg.__main__.shutil.which", lambda _: None)
|
|
monkeypatch.setitem(sys.modules, "imageio_ffmpeg", None)
|
|
assert _ffmpeg_bin() == "ffmpeg"
|
|
|
|
|
|
def test_missing_video_uri(tmp_path) -> None:
|
|
"""验证缺少 video_uri 时返回失败。"""
|
|
response = invoke(_request(tmp_path, inputs={}))
|
|
assert response.status == "failed"
|
|
assert "video_uri" in response.error
|
|
|
|
|
|
def test_missing_ffmpeg(tmp_path, monkeypatch) -> None:
|
|
"""验证找不到任何 ffmpeg 时返回明确失败信息。"""
|
|
monkeypatch.delenv("FFMPEG_BIN", raising=False)
|
|
monkeypatch.setattr("wov_node_ffmpeg.__main__.shutil.which", lambda _: None)
|
|
monkeypatch.setattr("wov_node_ffmpeg.__main__._bundled_ffmpeg", lambda: None)
|
|
response = invoke(_request(tmp_path))
|
|
assert response.status == "failed"
|
|
assert "ffmpeg not found" in response.error
|
|
|
|
|
|
def test_ffmpeg_failure(tmp_path, monkeypatch) -> None:
|
|
"""验证 ffmpeg 返回非零退出码时透传 stderr 错误。"""
|
|
def fake_run(command, **kwargs):
|
|
return subprocess.CompletedProcess(command, 1, stderr="boom")
|
|
|
|
monkeypatch.setattr("wov_node_ffmpeg.__main__.shutil.which", lambda _: "ffmpeg")
|
|
monkeypatch.setattr("wov_node_ffmpeg.__main__.subprocess.run", fake_run)
|
|
response = invoke(_request(tmp_path))
|
|
assert response.status == "failed"
|
|
assert "boom" in response.error
|
|
|
|
|
|
def test_entrypoint(monkeypatch) -> None:
|
|
"""验证 python -m wov_node_ffmpeg 会加载 ffmpeg-extract manifest。"""
|
|
module_path = Path(__file__).resolve().parent.parent / "wov_node_ffmpeg" / "__main__.py"
|
|
captured = {}
|
|
|
|
def fake_run_node(manifest, handler) -> None:
|
|
captured["id"] = manifest.id
|
|
|
|
monkeypatch.setattr("wov_sdk.server.run_node", fake_run_node)
|
|
runpy.run_path(str(module_path), run_name="__main__")
|
|
assert captured["id"] == "ffmpeg-extract"
|