"""Test suites. Each takes (results, ctx) and records rows via harness.check. ctx carries the interpreter tag, the gradio major and the bundle root. """ import json import os import platform import socket import subprocess import sys import time from harness import (Server, Skip, Warn, api_root, call_endpoint, check, free_port, http, http_json, threaded) S_ENV = "environment" S_INST = "install" S_FUNC = "functional" S_OFF = "offline" S_PERF = "performance" # ============================================================================= # 1. Environment # ============================================================================= def suite_environment(results, ctx): print("\n-- environment --") with check(results, S_ENV, "OS is RHEL-family 9") as c: rel = "" for path in ("/etc/redhat-release", "/etc/os-release"): if os.path.exists(path): rel = open(path).read() break c.detail = rel.splitlines()[0][:70] if rel else "unknown" if not rel: raise Skip("no /etc/redhat-release or /etc/os-release") if " 9" not in rel and 'VERSION_ID="9' not in rel: raise AssertionError("not a version 9 release: %s" % c.detail) with check(results, S_ENV, "architecture is x86_64") as c: c.detail = platform.machine() if platform.machine() != "x86_64": raise AssertionError("bundled wheels are x86_64 only") with check(results, S_ENV, "glibc >= 2.17 (manylinux_2_17 wheels)") as c: try: ver = platform.libc_ver()[1] or "" parts = tuple(int(x) for x in ver.split(".")[:2]) except Exception: raise Skip("could not determine glibc version") c.detail = "glibc %s" % ver if parts < (2, 17): raise AssertionError("glibc %s is too old for the wheels" % ver) with check(results, S_ENV, "interpreter matches the requested set") as c: c.detail = "python %s" % platform.python_version() want = ctx["py"] got = "%d.%d" % sys.version_info[:2] if got != want: raise AssertionError("expected %s, running %s" % (want, got)) with check(results, S_ENV, "no proxy variables capture loopback") as c: proxies = {k: v for k, v in os.environ.items() if k.lower() in ("http_proxy", "https_proxy", "all_proxy")} if not proxies: c.detail = "no proxy set" else: no_proxy = (os.environ.get("no_proxy", "") + "," + os.environ.get("NO_PROXY", "")) c.detail = "proxy set; no_proxy=%s" % (no_proxy.strip(",") or "(empty)") if "127.0.0.1" not in no_proxy and "localhost" not in no_proxy: raise AssertionError( "proxy set without loopback in no_proxy; gradio's startup " "self-check will fail with 'localhost is not accessible'") with check(results, S_ENV, "SELinux state") as c: try: out = subprocess.run(["getenforce"], capture_output=True, text=True, timeout=10).stdout.strip() except Exception: raise Skip("getenforce not available") c.detail = out # Enforcing is fine and expected; recorded for the report only. with check(results, S_ENV, "a loopback port can be bound") as c: p = free_port() c.detail = "bound 127.0.0.1:%d" % p with check(results, S_ENV, "writable temp dir for uploads") as c: import tempfile d = os.environ.get("GRADIO_TEMP_DIR") or tempfile.gettempdir() probe = os.path.join(d, ".gradio-write-probe") os.makedirs(d, exist_ok=True) with open(probe, "w") as fh: fh.write("x") os.unlink(probe) c.detail = d # ============================================================================= # 2. Install / pins # ============================================================================= def suite_install(results, ctx): print("\n-- install --") import gradio with check(results, S_INST, "import gradio") as c: c.detail = "gradio %s" % gradio.__version__ with check(results, S_INST, "import gradio_client") as c: import gradio_client c.detail = "gradio_client %s" % gradio_client.__version__ with check(results, S_INST, "expected gradio version for this interpreter") as c: want = ctx["expect_gradio"] c.detail = "%s (expected %s)" % (gradio.__version__, want) if gradio.__version__ != want: raise AssertionError("version drift") with check(results, S_INST, "core dependencies import") as c: import fastapi, httpx, pandas, numpy, PIL, pydantic, uvicorn # noqa c.detail = "fastapi %s, pydantic %s, numpy %s" % ( fastapi.__version__, pydantic.VERSION, numpy.__version__) with check(results, S_INST, "matplotlib present for gr.Plot") as c: import matplotlib c.detail = "matplotlib %s" % matplotlib.__version__ # The two pins that gradio 4.x cannot run without. Guarding them here means # a careless rebuild is caught by the test run rather than in production. with check(results, S_INST, "pin: huggingface-hub < 1.0 (gradio 4 only)") as c: import huggingface_hub ver = huggingface_hub.__version__ c.detail = "huggingface_hub %s" % ver if ctx["major"] >= 5: raise Skip("not applicable to gradio %d" % ctx["major"]) if int(ver.split(".")[0]) >= 1: raise AssertionError( "hub 1.0 removed HfFolder, which gradio 4 imports") if not hasattr(huggingface_hub, "HfFolder"): raise AssertionError("HfFolder missing; gradio 4 will not import") with check(results, S_INST, "pin: pydantic < 2.11 (gradio 4 only)") as c: import pydantic ver = pydantic.VERSION c.detail = "pydantic %s" % ver if ctx["major"] >= 5: raise Skip("not applicable to gradio %d" % ctx["major"]) major, minor = (int(x) for x in ver.split(".")[:2]) if (major, minor) >= (2, 11): raise AssertionError( "pydantic >=2.11 breaks gradio_client's schema walker " "(TypeError: argument of type 'bool' is not iterable)") with check(results, S_INST, "every dependency resolves (pip check)") as c: out = subprocess.run([sys.executable, "-m", "pip", "check"], capture_output=True, text=True, timeout=180) c.detail = out.stdout.strip().splitlines()[0][:90] if out.stdout.strip() else "no conflicts" if out.returncode != 0: # pip check reports gradio 4's own loose pins as warnings on some # sets; only fail on genuinely missing requirements. missing = [l for l in out.stdout.splitlines() if "is not installed" in l] if missing: raise AssertionError("; ".join(missing)[:200]) raise Skip(out.stdout.strip().splitlines()[0][:120]) with check(results, S_INST, "no compiler was needed (all wheels binary)") as c: root = ctx["bundle"] wheeldir = os.path.join(root, "wheels", ctx["tag"]) if not os.path.isdir(wheeldir): raise Skip("wheels/%s not present next to the venv" % ctx["tag"]) sdists = [f for f in os.listdir(wheeldir) if f.endswith((".tar.gz", ".zip"))] c.detail = "%d wheels, %d source archives" % ( len(os.listdir(wheeldir)) - len(sdists), len(sdists)) if sdists: raise AssertionError("source archives present: %s" % sdists[:3]) # ============================================================================= # 3. Functional # ============================================================================= def suite_functional(results, ctx): print("\n-- functional --") major = ctx["major"] prefix = api_root(major) try: server = Server("kitchen_sink.py").start() except Exception as exc: results.add(S_FUNC, "launch application under test", "FAIL", str(exc)[:200]) return results.add(S_FUNC, "launch application under test", "PASS", "ready in %.1fs on port %d" % (server.startup_seconds, server.port)) try: with check(results, S_FUNC, "GET / returns 200") as c: code, body = http(server.url + "/") c.detail = "%d, %d bytes" % (code, len(body)) if code != 200: raise AssertionError("status %d" % code) with check(results, S_FUNC, "GET /config is valid JSON with components") as c: code, body = http_json(server.url + prefix + "/config") if code != 200 or not body: code, body = http_json(server.url + "/config") if code != 200 or not body: raise AssertionError("status %d" % code) n = len(body.get("components", [])) c.detail = "%d components" % n if n < 20: raise AssertionError("expected the full component set, got %d" % n) # This is the canary for the pydantic/gradio_client schema bug: it is # the endpoint that breaks first and takes the client down with it. with check(results, S_FUNC, "GET /info returns the API schema") as c: code, body = http_json(server.url + prefix + "/info") if code != 200 or not body: raise AssertionError( "status %d -- a 500 here usually means the pydantic " "schema-walker incompatibility" % code) named = body.get("named_endpoints", {}) c.detail = "%d named endpoints" % len(named) if len(named) < 5: raise AssertionError("expected >=5 named endpoints") with check(results, S_FUNC, "static frontend assets are served") as c: code, body = http(server.url + "/") import re m = re.search(r'(?:src|href)="([^"]*assets/[^"]+\.(?:js|css))"', body) if not m: raise Skip("no asset reference found in the page") asset = m.group(1) # urljoin resolves the "./assets/..." form gradio emits; naive # concatenation leaves a "./" segment that the static route 404s. from urllib.parse import urljoin url = urljoin(server.url + "/", asset) code, data = http(url) c.detail = "%s -> %d" % (asset.split("/")[-1], code) if code != 200: raise AssertionError("asset %s returned %d" % (asset, code)) with check(results, S_FUNC, "page needs no external hosts") as c: import re code, body = http(server.url + "/") urls = set(re.findall(r'(?:src|href)="(https?://[^"]+)"', body)) external = {u for u in urls if "127.0.0.1" not in u and "localhost" not in u} # Documentation links in the footer are fine; a stylesheet or # script loaded from a CDN is not, because it will never load. blocking = {u for u in external if u.endswith((".js", ".css")) or "fonts." in u} c.detail = "%d external refs, %d blocking" % (len(external), len(blocking)) if blocking: names = sorted({u.split("/")[2] for u in blocking}) if major < 5: # gradio 4 hardcodes a cdnjs script in its index.html and # its default theme pulls Google Fonts stylesheets. The # browser cannot reach either offline: fonts fall back and # the resizer script errors in the console. Fix with # scripts/patch-offline-assets.sh plus a local-font theme. raise Warn("gradio 4 requests %s; see " "scripts/patch-offline-assets.sh and use a " "local-font theme" % ", ".join(names)) raise AssertionError("page loads external assets: %s" % sorted(blocking)[:3]) with check(results, S_FUNC, "frontend templates free of external refs") as c: import glob import gradio as _gr tdir = os.path.join(os.path.dirname(_gr.__file__), "templates") hits = {} for path in glob.glob(os.path.join(tdir, "**", "*.html"), recursive=True): with open(path, encoding="utf-8", errors="replace") as fh: text = fh.read() found = {h for h in ("fonts.googleapis.com", "fonts.gstatic.com", "cdnjs.cloudflare.com") if h in text} if found: hits[os.path.basename(path)] = sorted(found) if not hits: c.detail = "clean" elif major < 5: raise Warn("%s reference %s; run " "scripts/patch-offline-assets.sh " % (", ".join(sorted(hits)), ", ".join(sorted({h for v in hits.values() for h in v})))) else: raise AssertionError("unexpected external refs: %s" % hits) with check(results, S_FUNC, "endpoint /echo round trip") as c: out = call_endpoint(server, major, "echo", ["hello"]) c.detail = repr(out[0])[:50] if out[0] != "echo:hello": raise AssertionError("got %r" % (out[0],)) with check(results, S_FUNC, "endpoint /add computes correctly") as c: out = call_endpoint(server, major, "add", [2, 40]) c.detail = "2 + 40 = %s" % out[0] if float(out[0]) != 42.0: raise AssertionError("got %r" % (out[0],)) # Driven through gradio_client rather than the blocking REST route. # On gradio 4, POST /api/ does not iterate a generator: it # returns the first yielded value and stops. The client goes through # the queue, so it sees the run to completion on both majors. with check(results, S_FUNC, "generator endpoint streams to completion") as c: from gradio_client import Client client = Client(server.url, verbose=False) out = client.predict(6, api_name="/stream") c.detail = "%r via queue" % (out,) if out != "012345": raise AssertionError("expected final chunk '012345', got %r" % (out,)) with check(results, S_FUNC, "matplotlib plot renders through gr.Plot") as c: out = call_endpoint(server, major, "plot", [12]) c.detail = type(out[0]).__name__ if not out[0]: raise AssertionError("empty plot payload") # gr.State needs a session, which a gradio_client instance holds (a # bare POST /api/bump returns 500 on gradio 4 for exactly this reason). # One client must accumulate; a second client must start over. with check(results, S_FUNC, "gr.State is per-session, not global") as c: from gradio_client import Client a = Client(server.url, verbose=False) first, second = a.predict(api_name="/bump"), a.predict(api_name="/bump") b = Client(server.url, verbose=False) other = b.predict(api_name="/bump") c.detail = "session A: %s,%s | session B: %s" % (first, second, other) if (float(first), float(second)) != (1.0, 2.0): raise AssertionError( "state did not persist within a session: %s, %s" % (first, second)) if float(other) != 1.0: raise AssertionError( "state leaked across sessions: session B saw %s" % (other,)) with check(results, S_FUNC, "concurrent requests all succeed") as c: def one(i): out = call_endpoint(server, major, "echo", ["c%d" % i]) return out[0] got = threaded(one, 16, 8) errs = [g for g in got if not g or g[0] == "err"] c.detail = "%d/16 ok" % (16 - len(errs)) if errs: raise AssertionError("%d failed, first: %s" % (len(errs), errs[0])) # --- gradio_client --------------------------------------------------- with check(results, S_FUNC, "gradio_client connects and reads schema") as c: from gradio_client import Client client = Client(server.url, verbose=False) api = client.view_api(return_format="dict") or {} named = list((api.get("named_endpoints") or {}).keys()) c.detail = "%d endpoints" % len(named) if len(named) < 5: raise AssertionError("expected >=5, got %r" % named) ctx["_client"] = client with check(results, S_FUNC, "gradio_client predict round trip") as c: client = ctx.get("_client") if client is None: raise Skip("client did not connect") out = client.predict("abc", api_name="/echo") c.detail = repr(out)[:50] if out != "echo:abc": raise AssertionError("got %r" % (out,)) with check(results, S_FUNC, "file upload and checksum round trip") as c: client = ctx.get("_client") if client is None: raise Skip("client did not connect") import hashlib import tempfile from gradio_client import handle_file payload = b"gradio offline upload test\n" * 32 fd, path = tempfile.mkstemp(suffix=".txt") with os.fdopen(fd, "wb") as fh: fh.write(payload) try: out = client.predict(handle_file(path), api_name="/checksum") finally: os.unlink(path) if isinstance(out, str): out = json.loads(out) want = hashlib.sha256(payload).hexdigest() c.detail = "%d bytes, sha256 %s" % (out.get("bytes", -1), str(out.get("sha256"))[:12]) if out.get("bytes") != len(payload): raise AssertionError("size mismatch: %r" % (out,)) if out.get("sha256") != want: raise AssertionError("checksum mismatch") finally: server.stop() # ============================================================================= # 4. Offline behaviour # ============================================================================= def suite_offline(results, ctx): print("\n-- offline behaviour --") with check(results, S_OFF, "pip install would use no index") as c: root = ctx["bundle"] wheels = os.path.join(root, "wheels", ctx["tag"]) if not os.path.isdir(wheels): raise Skip("wheels/%s not next to the venv" % ctx["tag"]) out = subprocess.run( [sys.executable, "-m", "pip", "install", "--no-index", "--find-links", wheels, "--dry-run", "gradio"], capture_output=True, text=True, timeout=300) c.detail = "resolved from local wheels" if out.returncode == 0 else "failed" if out.returncode != 0: raise AssertionError( (out.stderr or out.stdout).strip().splitlines()[-1][:160]) with check(results, S_OFF, "analytics disabled in this environment") as c: val = os.environ.get("GRADIO_ANALYTICS_ENABLED", "") c.detail = "GRADIO_ANALYTICS_ENABLED=%s" % (val or "(unset)") if val.lower() not in ("false", "0"): # The harness sets this for the apps it starts, so the run itself # is unaffected. It is the operator's own shell that will make # every launch stall, hence a warning rather than a failure. raise Warn("not set in this shell: your own launches will stall " "on the analytics call; source gradio-offline.env") with check(results, S_OFF, "huggingface hub in offline mode") as c: val = os.environ.get("HF_HUB_OFFLINE", "") c.detail = "HF_HUB_OFFLINE=%s" % (val or "(unset)") if val not in ("1", "true", "True"): raise Warn("not set in this shell: hub lookups will stall on an " "isolated host; source gradio-offline.env") with check(results, S_OFF, "app starts with outbound traffic blackholed") as c: # Route every non-loopback destination nowhere, so any outbound # dependency shows up as a hang or error instead of passing quietly. env = {"http_proxy": "http://127.0.0.1:9", "https_proxy": "http://127.0.0.1:9", "no_proxy": "127.0.0.1,localhost,::1", "NO_PROXY": "127.0.0.1,localhost,::1"} with Server("kitchen_sink.py", env=env, timeout=90) as srv: code, _ = http(srv.url + "/") c.detail = "started in %.1fs, GET / -> %d" % (srv.startup_seconds, code) if code != 200: raise AssertionError("status %d" % code) with check(results, S_OFF, "startup is not delayed by network timeouts") as c: with Server("kitchen_sink.py") as srv: c.detail = "%.1fs" % srv.startup_seconds # A launch that waits on a dead analytics/hub socket typically # costs 10s or more; a healthy offline launch is a few seconds. if srv.startup_seconds > 25: raise AssertionError( "%.1fs is long enough to suggest an outbound call is " "timing out; check gradio-offline.env" % srv.startup_seconds) # ============================================================================= # 5. Performance # ============================================================================= def suite_performance(results, ctx, quick=False): print("\n-- performance --") major = ctx["major"] prefix = api_root(major) reps = 3 if quick else 5 # 1. cold start times = [] for _ in range(2 if quick else 3): with Server("kitchen_sink.py") as srv: times.append(srv.startup_seconds) results.metric("startup time (cold, mean of %d)" % len(times), sum(times) / len(times), "s", "min %.1f / max %.1f" % (min(times), max(times))) with Server("kitchen_sink.py") as server: # 2. RSS after startup try: with open("/proc/%d/status" % server.proc.pid) as fh: rss = next(int(l.split()[1]) for l in fh if l.startswith("VmRSS:")) results.metric("resident memory after startup", rss / 1024.0, "MB", "gradio %s" % ctx["expect_gradio"]) except Exception as exc: results.metric("resident memory after startup", -1, "MB", "unavailable: %s" % exc) # 3. page latency samples = [] for _ in range(reps * 4): t0 = time.time() http(server.url + "/") samples.append((time.time() - t0) * 1000) samples.sort() results.metric("GET / latency (median)", samples[len(samples) // 2], "ms", "p95 %.1f ms" % samples[int(len(samples) * 0.95) - 1]) # 4. config latency -- grows with component count samples = [] for _ in range(reps * 4): t0 = time.time() http(server.url + prefix + "/config") samples.append((time.time() - t0) * 1000) samples.sort() results.metric("GET /config latency (median)", samples[len(samples) // 2], "ms", "p95 %.1f ms" % samples[int(len(samples) * 0.95) - 1]) # 5. serial predict throughput n = 10 if quick else 25 t0 = time.time() for i in range(n): call_endpoint(server, major, "echo", ["p%d" % i]) elapsed = time.time() - t0 results.metric("predict throughput (serial)", n / elapsed, "calls/s", "%d calls in %.1fs" % (n, elapsed)) results.metric("predict latency (serial mean)", elapsed / n * 1000, "ms", "") # 6. concurrent predict throughput n = 16 if quick else 40 workers = 8 t0 = time.time() got = threaded(lambda i: call_endpoint(server, major, "echo", ["c%d" % i]), n, workers) elapsed = time.time() - t0 errs = sum(1 for g in got if not g or g[0] == "err") results.metric("predict throughput (%d threads)" % workers, n / elapsed, "calls/s", "%d calls, %d errors" % (n, errs)) # 7. queue behaviour under a slow handler n = 8 t0 = time.time() got = threaded(lambda i: call_endpoint(server, major, "slow", [0.25], timeout=120), n, n) elapsed = time.time() - t0 errs = sum(1 for g in got if not g or g[0] == "err") results.metric("queue: %d x 0.25s handlers" % n, elapsed, "s", "serial would be %.1fs, %d errors" % (n * 0.25, errs)) # 8. streaming t0 = time.time() call_endpoint(server, major, "stream", [20]) results.metric("streaming 20 chunks end-to-end", (time.time() - t0) * 1000, "ms", "") # 9. upload throughput try: import tempfile from gradio_client import Client, handle_file client = Client(server.url, verbose=False) size = 2 * 1024 * 1024 fd, path = tempfile.mkstemp(suffix=".bin") with os.fdopen(fd, "wb") as fh: fh.write(b"x" * size) t0 = time.time() client.predict(handle_file(path), api_name="/checksum") elapsed = time.time() - t0 os.unlink(path) results.metric("upload + checksum 2 MB", size / elapsed / 1e6, "MB/s", "%.2fs round trip" % elapsed) except Exception as exc: results.metric("upload + checksum 2 MB", -1, "MB/s", "unavailable: %s" % str(exc)[:60]) # 10. plot rendering t0 = time.time() call_endpoint(server, major, "plot", [50]) results.metric("matplotlib plot render", (time.time() - t0) * 1000, "ms", "50 points, Agg backend")