Files
wov-node-whisper/tests/test_whisper_node.py

137 lines
4.7 KiB
Python

"""faster-whisper 转写节点测试。
通过注入假 faster_whisper 模块覆盖时间戳格式化、参数传递、SRT 生成、
异常处理和入口点启动等真实代码路径。
"""
import runpy
import sys
import types
from pathlib import Path
from wov_node_whisper.__main__ import format_timestamp, invoke
from wov_sdk.models import InvokeRequest
class FakeSegment:
"""模拟 faster-whisper 的分段对象,只提供转写测试需要的字段。"""
def __init__(self, start, end, text):
self.start = start
self.end = end
self.text = text
class FakeWhisperModel:
"""记录构造参数并返回固定分段的假 WhisperModel。"""
instances: list[tuple[tuple, dict]] = []
def __init__(self, *args, **kwargs):
# 记录每次构造参数,测试据此断言 device/compute_type 传递。
FakeWhisperModel.instances.append((args, kwargs))
self.args = args
self.kwargs = kwargs
def transcribe(self, path, **kwargs):
# 返回固定两个分段:一个普通时长,一个跨小时验证时间戳格式。
self.transcribe_args = (path, kwargs)
return (
[
FakeSegment(0, 1, "第一段"),
FakeSegment(3600.5, 3602.25, "第二段"),
],
None,
)
def _install_fake_whisper(monkeypatch, model_class=FakeWhisperModel) -> None:
"""把假 faster_whisper 模块注入 sys.modules,替代真实依赖。"""
fake_module = types.SimpleNamespace(WhisperModel=model_class)
monkeypatch.setitem(sys.modules, "faster_whisper", fake_module)
def _request(tmp_path, **overrides) -> InvokeRequest:
"""构造默认音频输入与日语参数的调用请求。"""
payload = {
"run_id": "run_1",
"node_instance_id": "ni_1",
"inputs": {"audio_uri": str(tmp_path / "audio.wav")},
"params": {"language": "ja"},
"output_dir": str(tmp_path / "out"),
}
payload.update(overrides)
return InvokeRequest(**payload)
def test_format_timestamp() -> None:
"""验证秒数到 SRT 时间戳的格式化结果。"""
assert format_timestamp(0) == "00:00:00,000"
assert format_timestamp(3600.5) == "01:00:00,500"
assert format_timestamp(61.25) == "00:01:01,250"
def test_success(tmp_path, monkeypatch) -> None:
"""验证成功转写会生成 SRT 并默认使用 auto 设备/计算类型。"""
FakeWhisperModel.instances.clear()
_install_fake_whisper(monkeypatch)
(tmp_path / "audio.wav").write_bytes(b"fake")
response = invoke(_request(tmp_path))
assert response.status == "completed"
content = Path(response.outputs["srt_uri"]).read_text(encoding="utf-8")
assert "第一段" in content
assert "01:00:00,500 --> 01:00:02,250" in content
_, kwargs = FakeWhisperModel.instances[-1]
assert kwargs["device"] == "auto"
assert kwargs["compute_type"] == "auto"
def test_compute_type_override(tmp_path, monkeypatch) -> None:
"""验证请求参数可以覆盖默认计算类型。"""
FakeWhisperModel.instances.clear()
_install_fake_whisper(monkeypatch)
(tmp_path / "audio.wav").write_bytes(b"fake")
response = invoke(_request(tmp_path, params={"language": "ja", "compute_type": "int8"}))
assert response.status == "completed"
_, kwargs = FakeWhisperModel.instances[-1]
assert kwargs["compute_type"] == "int8"
def test_model_raises(tmp_path, monkeypatch) -> None:
"""验证模型加载失败时返回 failed 与错误信息。"""
class BrokenModel:
def __init__(self, *args, **kwargs):
raise RuntimeError("model load failed")
_install_fake_whisper(monkeypatch, BrokenModel)
(tmp_path / "audio.wav").write_bytes(b"fake")
response = invoke(_request(tmp_path))
assert response.status == "failed"
assert "model load failed" in response.error
def test_missing_input(tmp_path) -> None:
"""验证缺少 audio_uri 时返回失败。"""
response = invoke(_request(tmp_path, inputs={}))
assert response.status == "failed"
def test_missing_file(tmp_path) -> None:
"""验证音频文件不存在时返回失败。"""
response = invoke(_request(tmp_path))
assert response.status == "failed"
assert "audio file not found" in response.error
def test_entrypoint(monkeypatch) -> None:
"""验证 python -m wov_node_whisper 会加载 faster-whisper manifest。"""
module_path = Path(__file__).resolve().parent.parent / "wov_node_whisper" / "__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"] == "faster-whisper"