mirror of
https://github.com/rookiestar28/ComfyUI-OpenClaw.git
synced 2026-08-14 08:52:45 +00:00
feat: add pnginfo metadata api baseline
This commit is contained in:
@@ -0,0 +1,84 @@
|
||||
"""
|
||||
PNG Info API handler (R168).
|
||||
POST /openclaw/pnginfo (legacy: /moltbot/pnginfo)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Optional
|
||||
|
||||
try:
|
||||
from ..services.access_control import require_admin_token
|
||||
from ..services.aiohttp_compat import import_aiohttp_web
|
||||
from ..services.async_utils import run_in_thread
|
||||
from ..services.endpoint_manifest import (
|
||||
AuthTier,
|
||||
RiskTier,
|
||||
RoutePlane,
|
||||
endpoint_metadata,
|
||||
)
|
||||
from ..services.pnginfo import PngInfoError, parse_image_metadata
|
||||
from ..services.rate_limit import build_rate_limit_response, check_rate_limit
|
||||
except ImportError: # pragma: no cover
|
||||
from services.access_control import require_admin_token # type: ignore
|
||||
from services.aiohttp_compat import import_aiohttp_web # type: ignore
|
||||
from services.async_utils import run_in_thread # type: ignore
|
||||
from services.endpoint_manifest import ( # type: ignore
|
||||
AuthTier,
|
||||
RiskTier,
|
||||
RoutePlane,
|
||||
endpoint_metadata,
|
||||
)
|
||||
from services.pnginfo import PngInfoError, parse_image_metadata # type: ignore
|
||||
from services.rate_limit import ( # type: ignore
|
||||
build_rate_limit_response,
|
||||
check_rate_limit,
|
||||
)
|
||||
|
||||
web = import_aiohttp_web()
|
||||
|
||||
|
||||
def _json(payload: dict[str, Any], status: int = 200) -> web.Response:
|
||||
return web.json_response(payload, status=status)
|
||||
|
||||
|
||||
def _require_admin(request: web.Request) -> Optional[web.Response]:
|
||||
ok, error = require_admin_token(request)
|
||||
if ok:
|
||||
return None
|
||||
return _json({"ok": False, "error": error or "unauthorized"}, 403)
|
||||
|
||||
|
||||
@endpoint_metadata(
|
||||
auth=AuthTier.ADMIN,
|
||||
risk=RiskTier.LOW,
|
||||
summary="Parse image metadata",
|
||||
description="Extract A1111 or ComfyUI metadata from an uploaded image payload.",
|
||||
audit="pnginfo.parse",
|
||||
plane=RoutePlane.ADMIN,
|
||||
)
|
||||
async def pnginfo_handler(request: web.Request) -> web.Response:
|
||||
deny = _require_admin(request)
|
||||
if deny:
|
||||
return deny
|
||||
if not check_rate_limit(request, "admin"):
|
||||
return build_rate_limit_response(
|
||||
request,
|
||||
"admin",
|
||||
web_module=web,
|
||||
error="Rate limit exceeded",
|
||||
include_ok=False,
|
||||
)
|
||||
try:
|
||||
payload = await request.json()
|
||||
except Exception:
|
||||
return _json({"ok": False, "error": "invalid_json"}, 400)
|
||||
if not isinstance(payload, dict):
|
||||
return _json({"ok": False, "error": "invalid_payload"}, 400)
|
||||
try:
|
||||
result = await run_in_thread(parse_image_metadata, payload.get("image_b64", ""))
|
||||
except PngInfoError as exc:
|
||||
return _json({"ok": False, "error": exc.code, "detail": exc.detail}, exc.status)
|
||||
except Exception:
|
||||
return _json({"ok": False, "error": "internal_error"}, 500)
|
||||
return _json(result)
|
||||
@@ -58,6 +58,7 @@ def build_core_route_specs(
|
||||
f"{prefix}/preflight/inventory",
|
||||
handlers["inventory_handler"],
|
||||
),
|
||||
RouteSpec("POST", f"{prefix}/pnginfo", handlers["pnginfo_handler"]),
|
||||
RouteSpec("GET", f"{prefix}/checkpoints", handlers["list_checkpoints_handler"]),
|
||||
RouteSpec(
|
||||
"POST",
|
||||
|
||||
@@ -61,6 +61,7 @@ PACK_NAME = PACK_VERSION = PACK_START_TIME = LOG_FILE = get_api_key = None # ty
|
||||
metrics = tail_log = require_observability_access = check_rate_limit = trace_store = None # type: ignore
|
||||
get_executor_diagnostics = None # type: ignore
|
||||
webhook_handler = webhook_submit_handler = webhook_validate_handler = capabilities_handler = preflight_handler = None # type: ignore
|
||||
pnginfo_handler = None # type: ignore # R168
|
||||
config_get_handler = config_put_handler = llm_test_handler = llm_models_handler = llm_chat_handler = None # type: ignore
|
||||
remote_admin_page_handler = None # type: ignore # F61
|
||||
security_doctor_handler = None # type: ignore # S30
|
||||
@@ -158,6 +159,12 @@ if web is not None:
|
||||
"api.preflight_handler",
|
||||
("inventory_handler", "preflight_handler"),
|
||||
)
|
||||
(pnginfo_handler,) = import_attrs_dual(
|
||||
__package__,
|
||||
"..api.pnginfo",
|
||||
"api.pnginfo",
|
||||
("pnginfo_handler",),
|
||||
)
|
||||
(secrets_delete_handler, secrets_put_handler, secrets_status_handler) = (
|
||||
import_attrs_dual(
|
||||
__package__,
|
||||
@@ -855,6 +862,7 @@ def register_routes(server) -> None:
|
||||
"templates_list_handler": templates_list_handler,
|
||||
"preflight_handler": preflight_handler,
|
||||
"inventory_handler": inventory_handler,
|
||||
"pnginfo_handler": pnginfo_handler,
|
||||
"list_checkpoints_handler": list_checkpoints_handler,
|
||||
"create_checkpoint_handler": create_checkpoint_handler,
|
||||
"get_checkpoint_handler": get_checkpoint_handler,
|
||||
|
||||
@@ -40,8 +40,11 @@ function Get-GitDiffSnapshot {
|
||||
|
||||
function Assert-PreCommitDidNotMutateRepo {
|
||||
param(
|
||||
[Parameter(Mandatory = $true)][string]$BeforeWorktree,
|
||||
[Parameter(Mandatory = $true)][string]$BeforeIndex
|
||||
# IMPORTANT: git diff snapshots can be empty strings on a clean repo/index.
|
||||
# Do not mark these as mandatory non-empty inputs or the Windows full gate
|
||||
# fails before it can compare pre-commit mutation state.
|
||||
[AllowEmptyString()][string]$BeforeWorktree = "",
|
||||
[AllowEmptyString()][string]$BeforeIndex = ""
|
||||
)
|
||||
|
||||
$afterWorktree = Get-GitDiffSnapshot
|
||||
|
||||
@@ -10,6 +10,17 @@ if __package__ and "." in __package__:
|
||||
else: # pragma: no cover (test-only import mode)
|
||||
from config import PACK_NAME, PACK_VERSION
|
||||
|
||||
try:
|
||||
if __package__ and "." in __package__:
|
||||
from ..services.pnginfo import pnginfo_available
|
||||
else: # pragma: no cover (test-only import mode)
|
||||
from services.pnginfo import pnginfo_available # type: ignore
|
||||
except Exception: # pragma: no cover
|
||||
|
||||
def pnginfo_available() -> bool:
|
||||
return False
|
||||
|
||||
|
||||
from .runtime_profile import get_runtime_profile
|
||||
|
||||
API_VERSION = 1
|
||||
@@ -86,6 +97,7 @@ def get_capabilities() -> dict:
|
||||
"webhook_mapping": True,
|
||||
"job_events": True,
|
||||
"operator_doctor": True,
|
||||
"png_info": pnginfo_available(),
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,263 @@
|
||||
"""
|
||||
PNG Info metadata parsing service (R168).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import io
|
||||
import json
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
# CRITICAL: keep Pillow optional at import time so route bootstrap still loads
|
||||
# in environments where image parsing deps are not installed.
|
||||
try:
|
||||
from PIL import ExifTags, Image # type: ignore
|
||||
except ModuleNotFoundError: # pragma: no cover
|
||||
ExifTags = None # type: ignore
|
||||
Image = None # type: ignore
|
||||
|
||||
MAX_IMAGE_B64_LEN = 20 * 1024 * 1024 # 20MB base64 payload ceiling
|
||||
_RE_PARAM = re.compile(r'\s*(\w[\w \-/]+):\s*("(?:\\.|[^\\"])+"|[^,]*)(?:,|$)')
|
||||
_RE_IMAGE_SIZE = re.compile(r"^(\d+)x(\d+)$")
|
||||
_COMFYUI_KEYS = ("prompt", "workflow")
|
||||
_A1111_INFO_KEYS = (
|
||||
"parameters",
|
||||
"comment",
|
||||
"Comment",
|
||||
"description",
|
||||
"Description",
|
||||
"UserComment",
|
||||
"user_comment",
|
||||
)
|
||||
|
||||
|
||||
class PngInfoError(Exception):
|
||||
def __init__(self, code: str, detail: str, status: int = 400):
|
||||
super().__init__(detail)
|
||||
self.code = code
|
||||
self.detail = detail
|
||||
self.status = status
|
||||
|
||||
|
||||
def pnginfo_available() -> bool:
|
||||
return Image is not None
|
||||
|
||||
|
||||
def parse_image_metadata(image_b64: str) -> dict[str, Any]:
|
||||
if not pnginfo_available():
|
||||
raise PngInfoError(
|
||||
"pnginfo_unavailable",
|
||||
"Pillow (PIL) is required for PNG Info parsing.",
|
||||
status=503,
|
||||
)
|
||||
|
||||
payload = _decode_image_b64(image_b64)
|
||||
try:
|
||||
with Image.open(io.BytesIO(payload)) as image:
|
||||
image.load()
|
||||
text_items = _collect_text_items(image)
|
||||
except PngInfoError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
raise PngInfoError("invalid_image", "Unable to decode image metadata.") from exc
|
||||
|
||||
infotext = _extract_a1111_infotext(text_items)
|
||||
comfy_items = {
|
||||
key: value
|
||||
for key, value in text_items.items()
|
||||
if key in _COMFYUI_KEYS and value
|
||||
}
|
||||
|
||||
if infotext:
|
||||
source = "a1111"
|
||||
info = infotext
|
||||
parameters = _parse_generation_parameters(infotext)
|
||||
elif comfy_items:
|
||||
source = "comfyui"
|
||||
info = "ComfyUI metadata detected."
|
||||
parameters = {}
|
||||
else:
|
||||
source = "unknown"
|
||||
info = ""
|
||||
parameters = {}
|
||||
|
||||
items = {
|
||||
key: value
|
||||
for key, value in text_items.items()
|
||||
if value not in (None, "", {}) and key != "parameters"
|
||||
}
|
||||
|
||||
return {
|
||||
"ok": True,
|
||||
"source": source,
|
||||
"info": info,
|
||||
"parameters": parameters,
|
||||
"items": items,
|
||||
}
|
||||
|
||||
|
||||
def _decode_image_b64(image_b64: str) -> bytes:
|
||||
if not isinstance(image_b64, str) or not image_b64.strip():
|
||||
raise PngInfoError("image_b64_required", "image_b64 required")
|
||||
|
||||
if len(image_b64) > MAX_IMAGE_B64_LEN:
|
||||
raise PngInfoError(
|
||||
"image_b64_too_large",
|
||||
f"image_b64 exceeds {MAX_IMAGE_B64_LEN // 1024 // 1024}MB",
|
||||
)
|
||||
|
||||
value = image_b64.strip()
|
||||
if value.startswith("data:"):
|
||||
_, _, value = value.partition(",")
|
||||
value = re.sub(r"\s+", "", value)
|
||||
padding = len(value) % 4
|
||||
if padding:
|
||||
value += "=" * (4 - padding)
|
||||
try:
|
||||
return base64.b64decode(value, validate=True)
|
||||
except Exception as exc:
|
||||
raise PngInfoError(
|
||||
"invalid_image_b64", "image_b64 is not valid base64"
|
||||
) from exc
|
||||
|
||||
|
||||
def _collect_text_items(image) -> dict[str, Any]:
|
||||
items: dict[str, Any] = {}
|
||||
for key, value in getattr(image, "info", {}).items():
|
||||
if key == "exif":
|
||||
continue
|
||||
normalized = _normalize_metadata_value(value)
|
||||
if normalized not in (None, ""):
|
||||
items[str(key)] = normalized
|
||||
|
||||
for key, value in _collect_exif_items(image).items():
|
||||
items.setdefault(key, value)
|
||||
|
||||
return items
|
||||
|
||||
|
||||
def _collect_exif_items(image) -> dict[str, Any]:
|
||||
if ExifTags is None or not hasattr(image, "getexif"):
|
||||
return {}
|
||||
|
||||
try:
|
||||
exif = image.getexif()
|
||||
except Exception:
|
||||
return {}
|
||||
if not exif:
|
||||
return {}
|
||||
|
||||
tag_lookup = getattr(ExifTags, "TAGS", {}) or {}
|
||||
result: dict[str, Any] = {}
|
||||
for tag_id, value in exif.items():
|
||||
key = str(tag_lookup.get(tag_id, tag_id))
|
||||
normalized = _normalize_exif_value(key, value)
|
||||
if normalized not in (None, ""):
|
||||
result[key] = normalized
|
||||
return result
|
||||
|
||||
|
||||
def _normalize_exif_value(key: str, value: Any) -> Any:
|
||||
if isinstance(value, bytes):
|
||||
if key == "UserComment":
|
||||
return _decode_user_comment(value)
|
||||
return value.decode("utf-8", errors="replace").strip("\x00")
|
||||
return _normalize_metadata_value(value)
|
||||
|
||||
|
||||
def _decode_user_comment(value: bytes) -> str:
|
||||
prefixes = (b"ASCII\x00\x00\x00", b"UNICODE\x00", b"JIS\x00\x00\x00\x00\x00")
|
||||
payload = value
|
||||
for prefix in prefixes:
|
||||
if value.startswith(prefix):
|
||||
payload = value[len(prefix) :]
|
||||
break
|
||||
for encoding in ("utf-8", "utf-16", "latin-1"):
|
||||
try:
|
||||
return payload.decode(encoding).strip("\x00").strip()
|
||||
except Exception:
|
||||
continue
|
||||
return payload.decode("utf-8", errors="replace").strip("\x00").strip()
|
||||
|
||||
|
||||
def _normalize_metadata_value(value: Any) -> Any:
|
||||
if isinstance(value, bytes):
|
||||
return value.decode("utf-8", errors="replace").strip("\x00")
|
||||
if isinstance(value, (dict, list, int, float, bool)):
|
||||
return value
|
||||
if value is None:
|
||||
return None
|
||||
|
||||
text = str(value).strip()
|
||||
if not text:
|
||||
return ""
|
||||
if text[:1] in {"{", "["}:
|
||||
try:
|
||||
return json.loads(text)
|
||||
except Exception:
|
||||
return text
|
||||
return text
|
||||
|
||||
|
||||
def _extract_a1111_infotext(text_items: dict[str, Any]) -> str:
|
||||
for key in _A1111_INFO_KEYS:
|
||||
value = text_items.get(key)
|
||||
if isinstance(value, str) and value.strip():
|
||||
return value.strip()
|
||||
return ""
|
||||
|
||||
|
||||
def _parse_generation_parameters(info_text: str) -> dict[str, Any]:
|
||||
text = (info_text or "").strip()
|
||||
if not text:
|
||||
return {}
|
||||
|
||||
lines = [line.rstrip() for line in text.splitlines()]
|
||||
params_line = lines[-1] if lines else ""
|
||||
prompt_lines = lines[:-1]
|
||||
|
||||
if not _RE_PARAM.search(params_line):
|
||||
return {"positive_prompt": text}
|
||||
|
||||
positive_lines: list[str] = []
|
||||
negative_lines: list[str] = []
|
||||
in_negative = False
|
||||
for line in prompt_lines:
|
||||
if line.startswith("Negative prompt:"):
|
||||
in_negative = True
|
||||
negative_lines.append(line[len("Negative prompt:") :].lstrip())
|
||||
continue
|
||||
if in_negative:
|
||||
negative_lines.append(line)
|
||||
else:
|
||||
positive_lines.append(line)
|
||||
|
||||
parsed: dict[str, Any] = {
|
||||
"positive_prompt": "\n".join(positive_lines).strip(),
|
||||
"negative_prompt": "\n".join(negative_lines).strip(),
|
||||
}
|
||||
|
||||
for key, raw_value in _RE_PARAM.findall(params_line):
|
||||
value = _unquote(raw_value.strip())
|
||||
parsed[key] = value
|
||||
size_match = _RE_IMAGE_SIZE.match(str(value))
|
||||
if key == "Size" and size_match:
|
||||
parsed["Size-1"] = int(size_match.group(1))
|
||||
parsed["Size-2"] = int(size_match.group(2))
|
||||
|
||||
if not parsed["negative_prompt"]:
|
||||
parsed.pop("negative_prompt", None)
|
||||
if not parsed["positive_prompt"]:
|
||||
parsed.pop("positive_prompt", None)
|
||||
return parsed
|
||||
|
||||
|
||||
def _unquote(value: str) -> str:
|
||||
if len(value) >= 2 and value[0] == '"' and value[-1] == '"':
|
||||
try:
|
||||
return json.loads(value)
|
||||
except Exception:
|
||||
return value[1:-1]
|
||||
return value
|
||||
@@ -0,0 +1,86 @@
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
try:
|
||||
from aiohttp import web
|
||||
|
||||
AIOHTTP_AVAILABLE = True
|
||||
except ModuleNotFoundError:
|
||||
AIOHTTP_AVAILABLE = False
|
||||
|
||||
sys.path.append(os.getcwd())
|
||||
|
||||
|
||||
@unittest.skipIf(not AIOHTTP_AVAILABLE, "aiohttp not available")
|
||||
class TestPngInfoAPI(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_rejects_unauthorized_requests(self):
|
||||
from api.pnginfo import pnginfo_handler
|
||||
|
||||
request = AsyncMock()
|
||||
with patch("api.pnginfo.require_admin_token", return_value=(False, "Denied")):
|
||||
resp = await pnginfo_handler(request)
|
||||
self.assertEqual(resp.status, 403)
|
||||
self.assertEqual(json.loads(resp.body)["ok"], False)
|
||||
|
||||
async def test_rejects_invalid_json(self):
|
||||
from api.pnginfo import pnginfo_handler
|
||||
|
||||
request = AsyncMock()
|
||||
request.json = AsyncMock(side_effect=ValueError("bad json"))
|
||||
with (
|
||||
patch("api.pnginfo.require_admin_token", return_value=(True, None)),
|
||||
patch("api.pnginfo.check_rate_limit", return_value=True),
|
||||
):
|
||||
resp = await pnginfo_handler(request)
|
||||
self.assertEqual(resp.status, 400)
|
||||
self.assertEqual(json.loads(resp.body)["error"], "invalid_json")
|
||||
|
||||
async def test_returns_parsed_payload(self):
|
||||
from api.pnginfo import pnginfo_handler
|
||||
|
||||
request = AsyncMock()
|
||||
request.json = AsyncMock(return_value={"image_b64": "data"})
|
||||
expected = {
|
||||
"ok": True,
|
||||
"source": "a1111",
|
||||
"info": "raw infotext",
|
||||
"parameters": {"positive_prompt": "cat"},
|
||||
"items": {"Comment": "raw infotext"},
|
||||
}
|
||||
with (
|
||||
patch("api.pnginfo.require_admin_token", return_value=(True, None)),
|
||||
patch("api.pnginfo.check_rate_limit", return_value=True),
|
||||
patch("api.pnginfo.run_in_thread", return_value=expected),
|
||||
):
|
||||
resp = await pnginfo_handler(request)
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertEqual(json.loads(resp.body), expected)
|
||||
|
||||
async def test_returns_contract_error_details(self):
|
||||
from api.pnginfo import pnginfo_handler
|
||||
from services.pnginfo import PngInfoError
|
||||
|
||||
request = AsyncMock()
|
||||
request.json = AsyncMock(return_value={"image_b64": ""})
|
||||
with (
|
||||
patch("api.pnginfo.require_admin_token", return_value=(True, None)),
|
||||
patch("api.pnginfo.check_rate_limit", return_value=True),
|
||||
patch(
|
||||
"api.pnginfo.run_in_thread",
|
||||
side_effect=PngInfoError(
|
||||
"image_b64_required", "image_b64 required", status=400
|
||||
),
|
||||
),
|
||||
):
|
||||
resp = await pnginfo_handler(request)
|
||||
body = json.loads(resp.body)
|
||||
self.assertEqual(resp.status, 400)
|
||||
self.assertEqual(body["error"], "image_b64_required")
|
||||
self.assertEqual(body["detail"], "image_b64 required")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -41,6 +41,7 @@ class TestCapabilities(unittest.TestCase):
|
||||
"job_monitor",
|
||||
"callback_delivery",
|
||||
"assist_automation_compose",
|
||||
"png_info",
|
||||
]
|
||||
for feat in expected_features:
|
||||
self.assertIn(feat, features, f"Missing feature: {feat}")
|
||||
|
||||
@@ -20,6 +20,7 @@ class TestCapabilitiesContract(unittest.TestCase):
|
||||
self.assertIn("api_version", caps)
|
||||
self.assertIn("pack", caps)
|
||||
self.assertIn("features", caps)
|
||||
self.assertIn("png_info", caps["features"])
|
||||
|
||||
def test_runtime_profile_exposure(self):
|
||||
"""Verify runtime_profile reflects the actual resolved profile."""
|
||||
|
||||
@@ -20,3 +20,12 @@ class TestFullTestScriptPlaywrightBootstrap(unittest.TestCase):
|
||||
self.assertIn("npm install", content)
|
||||
self.assertIn('$env:OPENCLAW_PLAYWRIGHT_INSTALL = "1"', content)
|
||||
self.assertIn('$env:OPENCLAW_PLAYWRIGHT_BROWSERS = "chromium"', content)
|
||||
|
||||
def test_windows_precommit_snapshot_check_accepts_clean_diff_state(self):
|
||||
content = WINDOWS_SCRIPT.read_text(encoding="utf-8")
|
||||
self.assertIn('[AllowEmptyString()][string]$BeforeWorktree = ""', content)
|
||||
self.assertIn('[AllowEmptyString()][string]$BeforeIndex = ""', content)
|
||||
self.assertNotIn(
|
||||
"[Parameter(Mandatory = $true)][string]$BeforeIndex",
|
||||
content,
|
||||
)
|
||||
|
||||
@@ -0,0 +1,102 @@
|
||||
import base64
|
||||
import io
|
||||
import unittest
|
||||
|
||||
try:
|
||||
from PIL import Image, PngImagePlugin
|
||||
|
||||
PIL_AVAILABLE = True
|
||||
except ModuleNotFoundError:
|
||||
PIL_AVAILABLE = False
|
||||
|
||||
from services.pnginfo import PngInfoError, parse_image_metadata
|
||||
|
||||
|
||||
def _to_b64(image_bytes: bytes) -> str:
|
||||
return base64.b64encode(image_bytes).decode("utf-8")
|
||||
|
||||
|
||||
@unittest.skipIf(not PIL_AVAILABLE, "Pillow not available")
|
||||
class TestPngInfoService(unittest.TestCase):
|
||||
def _make_png(self, metadata: dict[str, str]) -> str:
|
||||
image = Image.new("RGB", (8, 8), color="white")
|
||||
info = PngImagePlugin.PngInfo()
|
||||
for key, value in metadata.items():
|
||||
info.add_text(key, value)
|
||||
buffer = io.BytesIO()
|
||||
image.save(buffer, format="PNG", pnginfo=info)
|
||||
return _to_b64(buffer.getvalue())
|
||||
|
||||
def _make_jpeg_with_user_comment(self, comment: str) -> str:
|
||||
image = Image.new("RGB", (8, 8), color="white")
|
||||
exif = Image.Exif()
|
||||
exif[37510] = ("ASCII\x00\x00\x00" + comment).encode("utf-8")
|
||||
buffer = io.BytesIO()
|
||||
image.save(buffer, format="JPEG", exif=exif)
|
||||
return _to_b64(buffer.getvalue())
|
||||
|
||||
def test_parse_a1111_parameters_chunk(self):
|
||||
infotext = (
|
||||
"masterpiece cat portrait\n"
|
||||
"Negative prompt: blur, lowres\n"
|
||||
"Steps: 24, Sampler: Euler a, CFG scale: 7, Seed: 42, "
|
||||
"Size: 768x512, Model: testModel, Model hash: abc123"
|
||||
)
|
||||
result = parse_image_metadata(self._make_png({"parameters": infotext}))
|
||||
self.assertEqual(result["source"], "a1111")
|
||||
self.assertEqual(result["info"], infotext)
|
||||
self.assertEqual(
|
||||
result["parameters"]["positive_prompt"], "masterpiece cat portrait"
|
||||
)
|
||||
self.assertEqual(result["parameters"]["negative_prompt"], "blur, lowres")
|
||||
self.assertEqual(result["parameters"]["Steps"], "24")
|
||||
self.assertEqual(result["parameters"]["Size-1"], 768)
|
||||
self.assertEqual(result["parameters"]["Size-2"], 512)
|
||||
|
||||
def test_parse_comment_fallback(self):
|
||||
infotext = (
|
||||
"cat\nSteps: 12, Sampler: Euler, CFG scale: 6, Seed: 11, Size: 512x512"
|
||||
)
|
||||
result = parse_image_metadata(self._make_png({"Comment": infotext}))
|
||||
self.assertEqual(result["source"], "a1111")
|
||||
self.assertEqual(result["parameters"]["positive_prompt"], "cat")
|
||||
self.assertEqual(result["parameters"]["Steps"], "12")
|
||||
|
||||
def test_parse_exif_user_comment_fallback(self):
|
||||
infotext = (
|
||||
"jpeg cat\nSteps: 15, Sampler: Euler, CFG scale: 5, Seed: 7, Size: 640x640"
|
||||
)
|
||||
result = parse_image_metadata(self._make_jpeg_with_user_comment(infotext))
|
||||
self.assertEqual(result["source"], "a1111")
|
||||
self.assertEqual(result["info"], infotext)
|
||||
self.assertEqual(result["parameters"]["Size"], "640x640")
|
||||
|
||||
def test_parse_comfyui_prompt_and_workflow_metadata(self):
|
||||
prompt = (
|
||||
'{"1":{"class_type":"KSampler","inputs":{"steps":20,"cfg":7,"seed":123}}}'
|
||||
)
|
||||
workflow = '{"nodes":[{"id":1,"type":"KSampler"}]}'
|
||||
result = parse_image_metadata(
|
||||
self._make_png({"prompt": prompt, "workflow": workflow})
|
||||
)
|
||||
self.assertEqual(result["source"], "comfyui")
|
||||
self.assertEqual(result["info"], "ComfyUI metadata detected.")
|
||||
self.assertEqual(result["parameters"], {})
|
||||
self.assertIsInstance(result["items"]["prompt"], dict)
|
||||
self.assertEqual(result["items"]["workflow"]["nodes"][0]["type"], "KSampler")
|
||||
|
||||
def test_parse_unknown_image_without_metadata(self):
|
||||
result = parse_image_metadata(self._make_png({}))
|
||||
self.assertEqual(result["source"], "unknown")
|
||||
self.assertEqual(result["info"], "")
|
||||
self.assertEqual(result["parameters"], {})
|
||||
self.assertEqual(result["items"], {})
|
||||
|
||||
def test_invalid_base64_raises_contract_error(self):
|
||||
with self.assertRaises(PngInfoError) as ctx:
|
||||
parse_image_metadata("%%%not-base64%%%")
|
||||
self.assertEqual(ctx.exception.code, "invalid_image_b64")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -45,6 +45,7 @@ class TestR151RouteRegistrars(unittest.TestCase):
|
||||
"templates_list_handler": sentinel.templates_list_handler,
|
||||
"preflight_handler": sentinel.preflight_handler,
|
||||
"inventory_handler": sentinel.inventory_handler,
|
||||
"pnginfo_handler": sentinel.pnginfo_handler,
|
||||
"list_checkpoints_handler": sentinel.list_checkpoints_handler,
|
||||
"create_checkpoint_handler": sentinel.create_checkpoint_handler,
|
||||
"get_checkpoint_handler": sentinel.get_checkpoint_handler,
|
||||
@@ -85,8 +86,9 @@ class TestR151RouteRegistrars(unittest.TestCase):
|
||||
self.assertIn(("GET", "/openclaw/health"), keys)
|
||||
self.assertIn(("POST", "/openclaw/webhook"), keys)
|
||||
self.assertIn(("GET", "/openclaw/llm/models"), keys)
|
||||
self.assertIn(("POST", "/openclaw/pnginfo"), keys)
|
||||
self.assertIn(("POST", "/openclaw/lab/experiments/{exp_id}/winner"), keys)
|
||||
self.assertEqual(49, len(specs))
|
||||
self.assertEqual(50, len(specs))
|
||||
|
||||
def test_build_assist_route_specs_preserves_expected_paths(self):
|
||||
specs = build_assist_route_specs("/moltbot", _AssistStub())
|
||||
|
||||
@@ -74,6 +74,7 @@ AUTH_CLASS_BY_ROUTE = {
|
||||
("GET", "/security/doctor"): "admin",
|
||||
("GET", "/tools"): "admin",
|
||||
("POST", "/tools/{name}/run"): "admin",
|
||||
("POST", "/pnginfo"): "admin",
|
||||
("GET", "/connector/installations"): "admin",
|
||||
("GET", "/connector/installations/resolve"): "admin",
|
||||
("GET", "/connector/installations/audit"): "admin",
|
||||
|
||||
Reference in New Issue
Block a user