- ComfyUI custom nodes: BridgeInput (11 types) + BridgeOutput
- AnyType wildcard for universal output capture
- Custom server route: GET /comfybridge/results/{prompt_id}
- Python async client (ComfyBridge) + sync wrapper (ComfyBridgeSync)
- Workflow JSON parser and patcher (deepcopy, original unchanged)
- WebSocket completion detection
- BridgeResult container with PIL/bytes helpers
- Examples: basic_text_to_image.py, agent_loop.py
- Unit tests for workflow parser and client
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
32 lines
1.0 KiB
Python
32 lines
1.0 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"})
|