feat: 完成 ffmpeg 音频提取节点

tests/test_ffmpeg_node.py

@@ -0,0 +1,106 @@
+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:
+    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:
+    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:
+    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:
+    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:
+    response = invoke(_request(tmp_path, inputs={}))
+    assert response.status == "failed"
+    assert "video_uri" in response.error
+
+
+def test_missing_ffmpeg(tmp_path, monkeypatch) -> None:
+    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:
+    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:
+    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"