Spaces:
Running
Running
File size: 4,698 Bytes
6fe13ea e6cfeb7 6fe13ea | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 | """
Drive the generated REST endpoints against a running app.
This layer exists because `test_pipelines.py` calls `WorkflowExecutor` directly
and therefore skips gradio's *output component* postprocessing — which is
exactly where an image port's value shape matters. A bare `data:` URI passes
the executor happily and then dies in the endpoint with
OSError: [Errno 22] Invalid argument: '...\\data:image\\png;base64,...'
because gradio treats the URI as a filename. Only these tests catch that.
python apps/05_workflow1111/app.py # in another shell
python apps/05_workflow1111/test_api.py [url] # default 127.0.0.1:7865
Pass `--local` to exercise only the endpoints that need no Hugging Face token.
"""
import os
import sys
import time
import warnings
warnings.filterwarnings("ignore")
import logging # noqa: E402
logging.disable(logging.CRITICAL)
for _s in (sys.stdout, sys.stderr):
try:
_s.reconfigure(encoding="utf-8", errors="replace")
except (AttributeError, ValueError):
pass
from gradio_client import Client, handle_file # noqa: E402
HERE = os.path.dirname(os.path.abspath(__file__))
SAMPLES = os.path.join(HERE, "samples")
args = [a for a in sys.argv[1:] if not a.startswith("--")]
LOCAL_ONLY = "--local" in sys.argv
URL = args[0] if args else "http://127.0.0.1:7865"
def sample(name):
path = os.path.join(SAMPLES, name)
if not os.path.exists(path):
raise SystemExit(f"missing sample {name} — run make_samples.py first")
return handle_file(path)
# (api_name, args, needs_token)
CASES = [
("/annotator_map", lambda: [sample("control.jpg")], False),
# `/png_info` covers pipelines 10 AND 11: they share the `PNG to inspect`
# reference, so they are one weakly-connected component and therefore one
# endpoint (`subject_groups`), returning all four outputs and generating a
# video on every call. Hence four arguments and needs_token=True — the
# *canvas* still runs each output on its own, so this is an API-surface
# consequence only.
("/png_info", lambda: [sample("with_parameters.png"),
"the snow drifts past and the fox turns its head",
"Cinematic", "Slow push-in"], True),
("/upscaled_local", lambda: [sample("extras.jpg")], True), # also hits 2 Spaces
("/image", lambda: ["a red fox in a snowy pine forest", "", "Cinematic",
"enhance fine detail"], True),
("/edited_image", lambda: [sample("init_image.jpg"),
"make it a snowy winter scene"], True),
("/generated_prompt", lambda: ["a lighthouse in a storm"], True),
("/recovered_prompt", lambda: [sample("interrogate.jpg")], True),
("/detected_objects", lambda: [sample("detect.jpg")], True),
("/x_y_grid", lambda: ["a lone tree on a hill",
"at sunrise | in a storm | at night | in fog"], True),
]
def describe(value):
if isinstance(value, (list, tuple)):
return " | ".join(describe(v) for v in value)
if isinstance(value, dict):
value = value.get("path") or value.get("url") or str(value)
if isinstance(value, str) and os.path.isfile(value):
try:
from PIL import Image
with Image.open(value) as im:
return f"image {im.width}×{im.height} {im.format}"
except Exception:
return f"file ({os.path.getsize(value) // 1024} KB)"
text = str(value).replace("\n", " ")
return f"text: {text[:64]}"
def main():
client = Client(URL, verbose=False)
available = set(client.view_api(return_format="dict",
print_info=False)["named_endpoints"])
print(f"\n{URL} — {len(available)} endpoints\n")
passed = failed = skipped = 0
for name, build, needs_token in CASES:
if name not in available:
print(f" skip {name:20} (not exposed)")
skipped += 1
continue
if LOCAL_ONLY and needs_token:
skipped += 1
continue
t0 = time.time()
try:
result = client.predict(*build(), api_name=name)
except Exception as e:
failed += 1
print(f" FAIL {name:20} ({time.time() - t0:6.1f}s) "
f"{type(e).__name__}: {str(e)[:120]}")
else:
passed += 1
print(f" ok {name:20} ({time.time() - t0:6.1f}s) {describe(result)}")
sys.stdout.flush()
print(f"\n{'=' * 66}\n {passed} passed, {failed} failed, {skipped} skipped\n{'=' * 66}")
return 1 if failed else 0
if __name__ == "__main__":
sys.exit(main())
|