f6e09dbff2
- palette.py + rembg.py: implement from stubs (Pillow median-cut + rembg u2net) - vlm.py: rename Spark2→steev (Strix Halo / Ollama); bump max_tokens 1024→4096 (qwen3-vl:32b thinking mode consumes budget tokens — 4096 min for valid output) - settings.py: rename spark2_vlm_*/spark1_flux_* → vlm_*/flux_*; real defaults (steev 100.88.167.87:11434 Ollama, gx10 100.90.100.10:8188 ComfyUI) - tests/: conftest.py + test_palette.py + test_rembg.py + test_integration_e2e.py (28 unit + 10 integration; 38/38 passing — VLM raw/polished/ugc + FLUX render) - CLAUDE.md: rewrite to accurate phase status + infra + layout - requirements.txt + pyproject.toml: add Pillow, rembg, pytest-asyncio deps Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
69 lines
2.1 KiB
Python
69 lines
2.1 KiB
Python
"""Pytest port of BTE's StopgapFluxWorkflowTests + smoke for /flux/render."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from unittest.mock import patch
|
|
|
|
import httpx
|
|
from fastapi.testclient import TestClient
|
|
|
|
from svrnty_vision.routers.flux import build_stopgap_workflow
|
|
from svrnty_vision.server import app
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
def test_stopgap_workflow_includes_prompt_and_dimensions() -> None:
|
|
raw = build_stopgap_workflow("a sunlit plate of food", 1024, 1024)
|
|
assert "a sunlit plate of food" in raw
|
|
graph = json.loads(raw)
|
|
assert graph["5"]["inputs"]["width"] == 1024
|
|
assert graph["5"]["inputs"]["height"] == 1024
|
|
assert graph["10"]["inputs"]["vae_name"] == "flux2-vae.safetensors"
|
|
assert graph["11"]["inputs"]["type"] == "flux2"
|
|
assert graph["12"]["inputs"]["unet_name"] == "flux2_dev_fp8mixed.safetensors"
|
|
|
|
|
|
def test_stopgap_workflow_seeds_vary_per_call() -> None:
|
|
"""ComfyUI dedupes identical workflows (execution_cached → empty outputs)."""
|
|
a = build_stopgap_workflow("x", 512, 512)
|
|
b = build_stopgap_workflow("x", 512, 512)
|
|
assert a != b
|
|
|
|
|
|
def test_stopgap_workflow_uses_explicit_seed_when_supplied() -> None:
|
|
a = build_stopgap_workflow("x", 512, 512, seed=42)
|
|
b = build_stopgap_workflow("x", 512, 512, seed=42)
|
|
assert a == b
|
|
|
|
|
|
def test_render_requires_workflow_or_prompt() -> None:
|
|
response = client.post("/flux/render", json={"width": 512, "height": 512})
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_render_returns_502_when_gx10_unreachable() -> None:
|
|
class _StubClient:
|
|
def __init__(self, *a, **kw):
|
|
pass
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *a):
|
|
return False
|
|
|
|
async def post(self, *a, **kw):
|
|
raise httpx.ConnectError("no comfy")
|
|
|
|
async def get(self, *a, **kw):
|
|
raise httpx.ConnectError("no comfy")
|
|
|
|
with patch("svrnty_vision.routers.flux.httpx.AsyncClient", _StubClient):
|
|
response = client.post(
|
|
"/flux/render",
|
|
json={"prompt": "test", "width": 512, "height": 512},
|
|
)
|
|
assert response.status_code == 502
|