qwen-image / scripts /verify_errors.py
Mike0021's picture
Publish Qwen 2.1 Image source f3afe8a1fd18 (part 3)
ae3bc22 verified
Raw History Blame Contribute Delete
2.56 kB
#!/usr/bin/env python3
"""Exercise bounded input failures through the real private Space API, without GPU work."""
from datetime import datetime, timezone
import json
from pathlib import Path
import sys
import time
from gradio_client import Client, handle_file
from huggingface_hub import HfApi, get_token
from PIL import Image
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from settings import DEFAULT_QUALITY, MAX_PROMPT_CHARS, MAX_SEED
space = sys.argv[1] if len(sys.argv)>1 else "Mike0021/qwen-image"
token = get_token()
if not token:
raise SystemExit("Authenticated access required")
info = HfApi(token=token).space_info(space)
if str(info.runtime.stage) != "RUNNING":
raise SystemExit("Space must be running")
out = ROOT / "artifacts" / "validation" / datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S")
out.mkdir(parents=True)
fixture = out / "over-20-megapixels.png"
Image.new("RGB", (5000, 4001), "white").save(fixture)
client = Client(space, token=token, verbose=False, analytics_enabled=False)
defaults = ["A ceramic bowl", None, DEFAULT_QUALITY, "1K", "1:1", False, 42, False]
cases = [("empty_prompt", 0, " ", "Describe the image"),
("long_prompt", 0, "x"*(MAX_PROMPT_CHARS+1), "Keep your description"),
("negative_seed", 6, -1, "whole number"),
("seed_too_large", 6, MAX_SEED+1, "whole number"),
("unknown_quality", 2, "invalid", "available quality"),
("large_reference", 1, handle_file(str(fixture)), "20 megapixels")]
records = []
for name, index, value, expected in cases:
args = defaults.copy();args[index]=value
started=time.perf_counter()
try:
client.predict(*args, api_name="/generate")
row={"case":name,"status":"failed","reason":"Invalid input unexpectedly generated an output"}
except Exception as exc:
# Store only the expected safe validation fragment, never HTTP headers.
matched=expected.lower() in str(exc).lower()
row={"case":name,"status":"passed" if matched else "unclassified_error",
"error_type":type(exc).__name__,"expected_message_fragment":expected,"matched":matched}
row["client_seconds"]=time.perf_counter()-started
records.append(row);print(json.dumps(row),flush=True)
(out/"results.json").write_text(json.dumps({"space":space,"hub_commit":info.sha,
"records":records,"fixture":"Synthetic white validation input; not a model-generated output."},indent=2))
fixture.unlink()
print(out)
if any(r['status']!='passed' for r in records):
raise SystemExit(1)