"""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"