| """Manual staging smoke check for a real ComfyUI server.
|
|
|
| This script is intentionally opt-in. It never runs from pytest/CI by default and
|
| does not include customer images, S3 calls, or real image generation.
|
| """
|
| from __future__ import annotations
|
|
|
| import os
|
| import sys
|
| import urllib.error
|
| import urllib.request
|
| from io import BytesIO
|
| from pathlib import Path
|
|
|
| sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
| from PIL import Image
|
|
|
| from app.core.config import settings
|
| from app.services.comfyui_client import (
|
| AIErrorCode,
|
| ComfyUIUnavailable,
|
| FaceSwapRequest,
|
| FakeComfyUIClient,
|
| build_face_only_payload,
|
| )
|
|
|
|
|
| def _enabled() -> bool:
|
| return os.getenv("ENABLE_STAGING_COMFYUI_SMOKE", "").lower() == "true"
|
|
|
|
|
| def _dummy_png() -> bytes:
|
| buf = BytesIO()
|
| Image.new("RGB", (512, 512), (180, 180, 180)).save(buf, format="PNG")
|
| return buf.getvalue()
|
|
|
|
|
| def _check_reachable(base_url: str, timeout: int) -> None:
|
| url = f"{base_url.rstrip('/')}/system_stats"
|
| req = urllib.request.Request(url, headers={"User-Agent": "leechard-staging-smoke"})
|
| with urllib.request.urlopen(req, timeout=timeout) as resp:
|
| if resp.status >= 400:
|
| raise ComfyUIUnavailable(f"ComfyUI returned HTTP {resp.status}")
|
|
|
|
|
| def main() -> int:
|
| if not _enabled():
|
| print("SKIP: set ENABLE_STAGING_COMFYUI_SMOKE=true to contact ComfyUI.")
|
| return 0
|
|
|
| if settings.ai_provider != "comfyui":
|
| print("SKIP: AI_PROVIDER must be comfyui for staging ComfyUI smoke.")
|
| return 0
|
|
|
| if not settings.comfyui_require_server:
|
| print("SKIP: COMFYUI_REQUIRE_SERVER must be true for real server smoke.")
|
| return 0
|
|
|
| if not settings.comfyui_base_url:
|
| print("SKIP: COMFYUI_BASE_URL is not configured.")
|
| return 0
|
|
|
| req = FaceSwapRequest(
|
| image_bytes=_dummy_png(),
|
| mime="image/png",
|
| face_boxes=[(192, 160, 128, 128)],
|
| model_identity_id="virtual_default",
|
| )
|
| payload = build_face_only_payload(req)
|
| mask = payload["mask"]
|
| print("OK: face-only payload generated.")
|
| print(f"workflow_name={payload['workflow_name']}")
|
| print(f"mode={payload['mode']}")
|
| print(f"edit_region={payload['edit_region']}")
|
| print(f"denoise_outside_mask={payload['denoise_outside_mask']}")
|
| print(
|
| "mask="
|
| f"erode:{mask['face_mask_erode_px']} "
|
| f"inset:{mask['mask_inset_px']} "
|
| f"hairline:{mask['hairline_protection_px']}"
|
| )
|
|
|
| fake = FakeComfyUIClient()
|
| for simulate, expected in (
|
| ("timeout", AIErrorCode.COMFYUI_TIMEOUT.value),
|
| ("unavailable", AIErrorCode.COMFYUI_UNAVAILABLE.value),
|
| ):
|
| try:
|
| fake.face_only_transform(
|
| FaceSwapRequest(
|
| image_bytes=req.image_bytes,
|
| mime=req.mime,
|
| face_boxes=req.face_boxes,
|
| simulate=simulate,
|
| )
|
| )
|
| except Exception as exc:
|
| code = getattr(exc, "code", None)
|
| if code != expected:
|
| print(f"FAIL: simulate={simulate} returned error_code={code}")
|
| return 1
|
| print(f"OK: simulate={simulate} returned error_code={code}")
|
|
|
| try:
|
| _check_reachable(settings.comfyui_base_url, settings.comfyui_timeout_seconds)
|
| except urllib.error.URLError as exc:
|
| print(f"FAIL: ComfyUI is not reachable: {exc.reason}")
|
| return 1
|
| except TimeoutError:
|
| print("FAIL: ComfyUI reachability check timed out.")
|
| return 1
|
| except Exception as exc:
|
| print(f"FAIL: ComfyUI reachability check failed: {exc}")
|
| return 1
|
|
|
| print("OK: ComfyUI base URL is reachable.")
|
| print("NOTE: real image generation is intentionally not executed by this smoke.")
|
| return 0
|
|
|
|
|
| if __name__ == "__main__":
|
| sys.exit(main())
|
|
|
