Shaman
52 lines
1.5 KiB
Python
52 lines
1.5 KiB
Python
import struct
|
|
import zlib
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.config import Settings
|
|
from app.main import create_app
|
|
|
|
COMFY = "http://comfy.test"
|
|
FAKE_KEY = "sk-test-1234"
|
|
|
|
|
|
def settings(**overrides) -> Settings:
|
|
base = dict(
|
|
comfy_base_url=COMFY,
|
|
adapter_api_key=FAKE_KEY,
|
|
validate_templates_on_server=False,
|
|
poll_interval=0.0,
|
|
execution_timeout=5.0,
|
|
)
|
|
base.update(overrides)
|
|
return Settings(**base)
|
|
|
|
|
|
def png_bytes(w: int = 64, h: int = 64) -> bytes:
|
|
def chunk(t: bytes, d: bytes) -> bytes:
|
|
return struct.pack(">I", len(d)) + t + d + struct.pack(">I", zlib.crc32(t + d))
|
|
|
|
ihdr = struct.pack(">IIBBBBB", w, h, 8, 2, 0, 0, 0)
|
|
iend = b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", ihdr) + chunk(b"IDAT", zlib.compress(b"\x00" * (w * 3 + 1) * h)) + chunk(b"IEND", b"")
|
|
return iend
|
|
|
|
|
|
def png_size(data: bytes) -> tuple[int, int]:
|
|
return struct.unpack(">II", data[16:24])
|
|
|
|
|
|
@pytest.fixture
|
|
async def client():
|
|
"""Async client bound to the app with lifespan running
|
|
(templates loaded, comfy client targeting http://comfy.test)."""
|
|
import contextlib
|
|
|
|
app = create_app(settings())
|
|
async with contextlib.AsyncExitStack() as stack:
|
|
await stack.enter_async_context(app.router.lifespan_context(app))
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://adapter.test") as ac:
|
|
ac.headers["Authorization"] = f"Bearer {FAKE_KEY}"
|
|
yield ac
|