cbpark84
1c919f1106
- run_batch(): sequential|concurrent|auto - chain(): 파이프라인 체이닝 - schema(): 스키마 자동 추출 - ResultCache: LRU+TTL 캐시 - preview_callback: 실시간 프리뷰 - upload_file/audio/video() - BridgeInputSEED: random/increment/fixed - MCPServer: MCP Tool 자동 생성
63 lines
2.3 KiB
Python
63 lines
2.3 KiB
Python
import pytest
|
|
from pathlib import Path
|
|
from client.workflow import find_bridge_inputs, patch_inputs, load_workflow
|
|
|
|
FIXTURE_PATH = Path(__file__).parent / "fixtures" / "sample_workflow.json"
|
|
|
|
|
|
def test_find_bridge_inputs():
|
|
wf = load_workflow(FIXTURE_PATH)
|
|
bridge_map = find_bridge_inputs(wf)
|
|
assert "prompt" in bridge_map
|
|
assert "steps" in bridge_map
|
|
assert "cfg" in bridge_map
|
|
assert bridge_map["prompt"] == ("1", "STRING")
|
|
assert bridge_map["steps"] == ("2", "INT")
|
|
|
|
|
|
def test_patch_inputs():
|
|
wf = load_workflow(FIXTURE_PATH)
|
|
patched = patch_inputs(wf, {"prompt": "hello world", "steps": 30})
|
|
assert patched["1"]["inputs"]["value"] == "hello world"
|
|
assert patched["2"]["inputs"]["value"] == 30
|
|
# 원본 불변 확인
|
|
original = load_workflow(FIXTURE_PATH)
|
|
assert original["1"]["inputs"]["value"] == "default prompt"
|
|
|
|
|
|
def test_patch_unknown_input_raises():
|
|
wf = load_workflow(FIXTURE_PATH)
|
|
with pytest.raises(ValueError, match="not found in workflow"):
|
|
patch_inputs(wf, {"nonexistent": "value"})
|
|
|
|
|
|
# ── BridgeInputANY 테스트 ────────────────────────────────────
|
|
|
|
def test_find_bridge_inputs_any_type():
|
|
"""BridgeInputANY 노드가 type_name='ANY'로 탐색되어야 함"""
|
|
wf = {
|
|
"1": {"class_type": "BridgeInputSTRING", "inputs": {"name": "prompt", "value": ""}},
|
|
"5": {"class_type": "BridgeInputANY", "inputs": {"name": "dynamic_input"}},
|
|
}
|
|
bridge_map = find_bridge_inputs(wf)
|
|
assert "prompt" in bridge_map
|
|
assert "dynamic_input" in bridge_map
|
|
assert bridge_map["dynamic_input"] == ("5", "ANY")
|
|
|
|
|
|
def test_patch_inputs_any_type():
|
|
"""BridgeInputANY에 어떤 값이든 주입 가능"""
|
|
wf = {
|
|
"5": {"class_type": "BridgeInputANY", "inputs": {"name": "dynamic", "value": None}},
|
|
}
|
|
patched = patch_inputs(wf, {"dynamic": [42, "hello", True]})
|
|
assert patched["5"]["inputs"]["value"] == [42, "hello", True]
|
|
|
|
|
|
def test_patch_inputs_preserves_unpatched_nodes():
|
|
"""패치하지 않은 노드는 그대로 유지"""
|
|
wf = load_workflow(FIXTURE_PATH)
|
|
patched = patch_inputs(wf, {"prompt": "changed"})
|
|
# steps는 변경하지 않았으므로 원래 값 유지
|
|
assert patched["2"]["inputs"]["value"] == 20
|