81 lines
2.4 KiB
Python
81 lines
2.4 KiB
Python
import runpy
|
|
from pathlib import Path
|
|
|
|
from wov_node_echo.__main__ import invoke
|
|
from wov_sdk.models import InvokeRequest
|
|
|
|
|
|
def test_invoke_text(tmp_path) -> None:
|
|
response = invoke(
|
|
InvokeRequest(
|
|
run_id="run_1",
|
|
node_instance_id="ni_1",
|
|
inputs={"text": "hello echo"},
|
|
output_dir=str(tmp_path),
|
|
)
|
|
)
|
|
assert response.status == "completed"
|
|
assert response.outputs["text"] == "hello echo"
|
|
assert Path(response.outputs["file_uri"]).read_text(encoding="utf-8") == "hello echo"
|
|
|
|
|
|
def test_invoke_absolute_file(tmp_path) -> None:
|
|
source = tmp_path / "input.txt"
|
|
source.write_text("from file", encoding="utf-8")
|
|
response = invoke(
|
|
InvokeRequest(
|
|
run_id="run_2",
|
|
node_instance_id="ni_2",
|
|
inputs={"file_uri": str(source)},
|
|
output_dir=str(tmp_path / "out"),
|
|
)
|
|
)
|
|
assert response.status == "completed"
|
|
assert response.outputs["text"] == "from file"
|
|
|
|
|
|
def test_invoke_relative_file(tmp_path, monkeypatch) -> None:
|
|
node_root = Path(__file__).resolve().parent.parent
|
|
source = node_root / "relative_input.txt"
|
|
source.write_text("relative", encoding="utf-8")
|
|
monkeypatch.chdir(node_root)
|
|
try:
|
|
response = invoke(
|
|
InvokeRequest(
|
|
run_id="run_3",
|
|
node_instance_id="ni_3",
|
|
inputs={"file_uri": "relative_input.txt"},
|
|
output_dir=str(tmp_path / "out"),
|
|
)
|
|
)
|
|
finally:
|
|
source.unlink()
|
|
assert response.status == "completed"
|
|
assert response.outputs["text"] == "relative"
|
|
|
|
|
|
def test_invoke_default_text(tmp_path) -> None:
|
|
response = invoke(
|
|
InvokeRequest(
|
|
run_id="run_4",
|
|
node_instance_id="ni_4",
|
|
inputs={},
|
|
output_dir=str(tmp_path),
|
|
)
|
|
)
|
|
assert response.status == "completed"
|
|
assert response.outputs["text"] == "echo"
|
|
|
|
|
|
def test_main_entrypoint(monkeypatch, tmp_path) -> None:
|
|
module_path = Path(__file__).resolve().parent.parent / "wov_node_echo" / "__main__.py"
|
|
captured = {}
|
|
|
|
def fake_run_node(manifest, handler) -> None:
|
|
captured["manifest"] = manifest
|
|
assert manifest.id == "echo"
|
|
|
|
monkeypatch.setattr("wov_sdk.server.run_node", fake_run_node)
|
|
runpy.run_path(str(module_path), run_name="__main__")
|
|
assert captured["manifest"].id == "echo"
|