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 자동 생성
310 lines
11 KiB
Python
310 lines
11 KiB
Python
"""
|
|
client/result.py 유닛 테스트.
|
|
async 메서드는 pytest-asyncio로 테스트.
|
|
HTTP 호출은 aioresponses로 mock.
|
|
"""
|
|
|
|
import base64
|
|
import io
|
|
import json
|
|
import pytest
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
from PIL import Image
|
|
|
|
from client.result import BridgeResult
|
|
|
|
|
|
def _make_png_bytes(width=32, height=32, color=(100, 150, 200)) -> bytes:
|
|
"""테스트용 PNG bytes 생성"""
|
|
img = Image.new("RGB", (width, height), color)
|
|
buf = io.BytesIO()
|
|
img.save(buf, format="PNG")
|
|
return buf.getvalue()
|
|
|
|
|
|
def _make_inline_image_output(width=32, height=32) -> dict:
|
|
"""inline base64 포함된 이미지 출력 딕셔너리"""
|
|
png = _make_png_bytes(width, height)
|
|
return {
|
|
"__type__": "image",
|
|
"url": "/comfybridge/file/pid/img",
|
|
"data": base64.b64encode(png).decode(),
|
|
"width": width,
|
|
"height": height,
|
|
"size_bytes": len(png),
|
|
"format": "PNG",
|
|
}
|
|
|
|
|
|
def _make_url_only_image_output(width=3840, height=2160) -> dict:
|
|
"""URL만 있는 대형 이미지 (4K, inline 없음)"""
|
|
return {
|
|
"__type__": "image",
|
|
"url": "/comfybridge/file/pid/4k",
|
|
"width": width,
|
|
"height": height,
|
|
"size_bytes": 20_000_000,
|
|
"format": "PNG",
|
|
}
|
|
|
|
|
|
def _make_audio_output() -> dict:
|
|
return {
|
|
"__type__": "audio",
|
|
"url": "/comfybridge/file/pid/audio",
|
|
"format": "WAV",
|
|
"sample_rate": 44100,
|
|
"channels": 2,
|
|
"duration": 30.5,
|
|
"size_bytes": 5_000_000,
|
|
}
|
|
|
|
|
|
# ── 기본 접근 테스트 ─────────────────────────────────────────
|
|
|
|
class TestBridgeResultBasic:
|
|
def test_get_scalar_string(self):
|
|
r = BridgeResult({"caption": "hello"})
|
|
assert r.get("caption") == "hello"
|
|
|
|
def test_get_scalar_int(self):
|
|
r = BridgeResult({"steps": 20})
|
|
assert r.get("steps") == 20
|
|
|
|
def test_get_missing_returns_default(self):
|
|
r = BridgeResult({})
|
|
assert r.get("missing") is None
|
|
assert r.get("missing", 42) == 42
|
|
|
|
def test_keys(self):
|
|
r = BridgeResult({"a": 1, "b": 2, "c": 3})
|
|
assert set(r.keys()) == {"a", "b", "c"}
|
|
|
|
def test_type_of_scalar(self):
|
|
r = BridgeResult({"caption": "text", "score": 0.9})
|
|
assert r.type_of("caption") == "str"
|
|
assert r.type_of("score") == "float"
|
|
|
|
def test_type_of_image(self):
|
|
r = BridgeResult({"img": _make_inline_image_output()})
|
|
assert r.type_of("img") == "image"
|
|
|
|
def test_type_of_audio(self):
|
|
r = BridgeResult({"snd": _make_audio_output()})
|
|
assert r.type_of("snd") == "audio"
|
|
|
|
def test_type_of_missing(self):
|
|
r = BridgeResult({})
|
|
assert r.type_of("missing") is None
|
|
|
|
def test_has_inline_true(self):
|
|
r = BridgeResult({"img": _make_inline_image_output()})
|
|
assert r.has_inline("img") is True
|
|
|
|
def test_has_inline_false_no_data(self):
|
|
r = BridgeResult({"img": _make_url_only_image_output()})
|
|
assert r.has_inline("img") is False
|
|
|
|
def test_has_inline_false_scalar(self):
|
|
r = BridgeResult({"cap": "text"})
|
|
assert r.has_inline("cap") is False
|
|
|
|
def test_has_url_true(self):
|
|
r = BridgeResult({"img": _make_inline_image_output()})
|
|
assert r.has_url("img") is True
|
|
|
|
def test_has_url_false_scalar(self):
|
|
r = BridgeResult({"cap": "text"})
|
|
assert r.has_url("cap") is False
|
|
|
|
|
|
# ── 동기 메서드 테스트 ───────────────────────────────────────
|
|
|
|
class TestBridgeResultSync:
|
|
def test_as_bytes_from_inline_base64(self):
|
|
output = _make_inline_image_output()
|
|
r = BridgeResult({"img": output})
|
|
raw = r.as_bytes("img")
|
|
assert raw[:8] == b"\x89PNG\r\n\x1a\n" # PNG magic
|
|
|
|
def test_as_pil_from_inline_base64(self):
|
|
output = _make_inline_image_output(64, 64)
|
|
r = BridgeResult({"img": output})
|
|
img = r.as_pil("img")
|
|
assert img.size == (64, 64)
|
|
|
|
def test_as_bytes_from_raw_bytes(self):
|
|
png = _make_png_bytes()
|
|
r = BridgeResult({"img": png})
|
|
assert r.as_bytes("img") == png
|
|
|
|
def test_as_bytes_missing_raises_key_error(self):
|
|
r = BridgeResult({})
|
|
with pytest.raises(KeyError, match="not found"):
|
|
r.as_bytes("missing")
|
|
|
|
def test_as_bytes_url_only_raises_helpful_error(self):
|
|
"""4K 이미지처럼 inline 없으면 명확한 에러 메시지"""
|
|
r = BridgeResult({"big": _make_url_only_image_output()}, base_url="http://localhost:8188")
|
|
with pytest.raises(RuntimeError, match="download"):
|
|
r.as_bytes("big")
|
|
|
|
def test_as_pil_url_only_raises(self):
|
|
r = BridgeResult({"big": _make_url_only_image_output()}, base_url="http://localhost:8188")
|
|
with pytest.raises(RuntimeError, match="download"):
|
|
r.as_pil("big")
|
|
|
|
|
|
# ── 비동기 메서드 테스트 ─────────────────────────────────────
|
|
|
|
class TestBridgeResultAsync:
|
|
@pytest.mark.asyncio
|
|
async def test_as_bytes_async_from_inline(self):
|
|
output = _make_inline_image_output()
|
|
r = BridgeResult({"img": output})
|
|
raw = await r.as_bytes_async("img")
|
|
assert raw[:8] == b"\x89PNG\r\n\x1a\n"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_as_pil_async_from_inline(self):
|
|
output = _make_inline_image_output(128, 128)
|
|
r = BridgeResult({"img": output})
|
|
img = await r.as_pil_async("img")
|
|
assert img.size == (128, 128)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_as_bytes_async_url_fallback(self):
|
|
"""inline 없으면 URL에서 다운로드"""
|
|
png = _make_png_bytes()
|
|
output = _make_url_only_image_output()
|
|
r = BridgeResult({"big": output}, base_url="http://localhost:8188")
|
|
|
|
# _fetch_url을 mock으로 교체
|
|
r._fetch_url = AsyncMock(return_value=png)
|
|
raw = await r.as_bytes_async("big")
|
|
assert raw == png
|
|
r._fetch_url.assert_called_once_with("http://localhost:8188/comfybridge/file/pid/4k")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_as_pil_async_url_fallback(self):
|
|
png = _make_png_bytes(64, 64)
|
|
output = _make_url_only_image_output()
|
|
r = BridgeResult({"big": output}, base_url="http://localhost:8188")
|
|
r._fetch_url = AsyncMock(return_value=png)
|
|
img = await r.as_pil_async("big")
|
|
assert img.size == (64, 64)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_inline_no_network(self):
|
|
"""inline 있으면 네트워크 호출 없이 반환"""
|
|
output = _make_inline_image_output()
|
|
r = BridgeResult({"img": output})
|
|
r._fetch_url = AsyncMock(side_effect=AssertionError("Should not call network"))
|
|
raw = await r.download("img")
|
|
assert raw[:8] == b"\x89PNG\r\n\x1a\n"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_url_calls_fetch(self):
|
|
"""URL only이면 _fetch_url 호출"""
|
|
png = _make_png_bytes()
|
|
output = _make_url_only_image_output()
|
|
r = BridgeResult({"big": output}, base_url="http://localhost:8188")
|
|
r._fetch_url = AsyncMock(return_value=png)
|
|
raw = await r.download("big")
|
|
assert raw == png
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_audio(self):
|
|
"""오디오도 동일하게 다운로드"""
|
|
wav_bytes = b"RIFF" + b"\x00" * 100
|
|
output = _make_audio_output()
|
|
r = BridgeResult({"snd": output}, base_url="http://localhost:8188")
|
|
r._fetch_url = AsyncMock(return_value=wav_bytes)
|
|
raw = await r.download("snd")
|
|
assert raw == wav_bytes
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_missing_raises(self):
|
|
r = BridgeResult({}, base_url="http://localhost:8188")
|
|
with pytest.raises(KeyError, match="not found"):
|
|
await r.download("missing")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_save_writes_file(self, tmp_path):
|
|
"""save()가 파일을 올바르게 저장"""
|
|
png = _make_png_bytes(32, 32)
|
|
output = _make_inline_image_output()
|
|
r = BridgeResult({"img": output})
|
|
out_path = tmp_path / "output.png"
|
|
result_path = await r.save("img", out_path)
|
|
assert result_path == out_path
|
|
assert out_path.read_bytes()[:8] == b"\x89PNG\r\n\x1a\n"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_save_creates_parent_dirs(self, tmp_path):
|
|
"""save()가 중간 디렉토리를 자동 생성"""
|
|
output = _make_inline_image_output()
|
|
r = BridgeResult({"img": output})
|
|
nested = tmp_path / "a" / "b" / "c" / "out.png"
|
|
await r.save("img", nested)
|
|
assert nested.exists()
|
|
|
|
|
|
# ── 메타 메서드 테스트 ───────────────────────────────────────
|
|
|
|
class TestBridgeResultMeta:
|
|
def test_info_excludes_data_field(self):
|
|
"""info()는 대용량 data(base64) 필드 제외"""
|
|
output = _make_inline_image_output()
|
|
r = BridgeResult({"img": output})
|
|
info = r.info("img")
|
|
assert "data" not in info
|
|
assert "url" in info
|
|
assert "__type__" in info
|
|
|
|
def test_info_scalar(self):
|
|
r = BridgeResult({"score": 0.95})
|
|
info = r.info("score")
|
|
assert info["value"] == 0.95
|
|
|
|
def test_summary_returns_all(self):
|
|
r = BridgeResult({
|
|
"caption": "text",
|
|
"img": _make_inline_image_output(),
|
|
})
|
|
summary = r.summary()
|
|
assert "caption" in summary
|
|
assert "img" in summary
|
|
assert "data" not in summary["img"] # base64 제외
|
|
|
|
def test_repr_contains_type(self):
|
|
r = BridgeResult({"img": _make_inline_image_output()})
|
|
rep = repr(r)
|
|
assert "image" in rep
|
|
assert "img" in rep
|
|
assert "inline" in rep
|
|
|
|
def test_repr_url_only(self):
|
|
r = BridgeResult({"big": _make_url_only_image_output()})
|
|
rep = repr(r)
|
|
assert "url" in rep
|
|
assert "inline" not in rep
|
|
|
|
|
|
# ── BridgeResult 초기화 테스트 ───────────────────────────────
|
|
|
|
class TestBridgeResultInit:
|
|
def test_default_base_url_empty(self):
|
|
r = BridgeResult({})
|
|
assert r._base_url == ""
|
|
|
|
def test_base_url_trailing_slash_stripped(self):
|
|
r = BridgeResult({}, base_url="http://localhost:8188/")
|
|
assert r._base_url == "http://localhost:8188"
|
|
|
|
def test_timeout_stored(self):
|
|
r = BridgeResult({}, timeout=120.0)
|
|
assert r._timeout == 120.0
|