"""Live model smoke tests for Tiny Narrator. Run against a running local app to verify configured model endpoints: python scripts/live_smoke.py --base-url http://127.0.0.1:7860 Checks: - /api/runtime-status: confirms reader brain, speech, vision, image generation status - /api/describe-image: confirms MiniCPM-V-4.6 image description path - /api/generate-image: confirms Modal Klein image generation path Exit codes: 0 - all configured checks passed or skipped 1 - one or more configured checks failed Secrets (API keys, tokens) are never printed. """ from __future__ import annotations import argparse import json import sys import time import urllib.error import urllib.request def _get_json(url: str, timeout: int = 30) -> dict: """Fetch a JSON GET endpoint and return the parsed payload.""" request = urllib.request.Request(url, method="GET") with urllib.request.urlopen(request, timeout=timeout) as response: return json.loads(response.read().decode("utf-8")) def _post_json(url: str, payload: dict, timeout: int = 60) -> dict: """POST JSON to an endpoint and return the parsed response.""" body = json.dumps(payload).encode("utf-8") request = urllib.request.Request( url, data=body, headers={"Content-Type": "application/json"}, method="POST", ) with urllib.request.urlopen(request, timeout=timeout) as response: return json.loads(response.read().decode("utf-8")) def _format_elapsed(ms: float) -> str: if ms < 1000: return f"{round(ms)} ms" return f"{ms / 1000:.2f} s" def _status_label(status: str) -> str: return { "online": "PASS", "fallback-ready": "SKIP", "fallback": "SKIP", "placeholder-ready": "SKIP", "offline": "FAIL", }.get(status, "SKIP") def _role_status(runtime_status: dict[str, dict], role: str) -> str: return str(runtime_status.get(role, {}).get("status", "unknown")) def _role_is_configured(runtime_status: dict[str, dict], role: str) -> bool: details = runtime_status.get(role, {}) if details.get("configured") is True: return True return bool(details.get("endpoint") or details.get("base_url")) def check_runtime_status(base_url: str) -> dict[str, dict]: """Fetch runtime status and return per-role metadata.""" start = time.perf_counter() try: payload = _get_json(f"{base_url}/api/runtime-status") elapsed = (time.perf_counter() - start) * 1000 roles = { "reader_brain": payload.get("reader_brain", {}), "vision": payload.get("vision", {}), "speech": payload.get("speech", {}), "image_generation": payload.get("image_generation", {}), } labels = ", ".join(f"{role}={_role_status(roles, role)}" for role in roles) print(f" runtime-status {_format_elapsed(elapsed):>10} {labels}") return roles except (OSError, urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc: elapsed = (time.perf_counter() - start) * 1000 print(f" runtime-status {_format_elapsed(elapsed):>10} FAIL ({exc.__class__.__name__})") return {} def check_describe_image(base_url: str, runtime_status: dict[str, dict]) -> bool: """Smoke test /api/describe-image for a known article image.""" vision_status = _role_status(runtime_status, "vision") vision_configured = _role_is_configured(runtime_status, "vision") if not vision_configured and vision_status in {"fallback-ready", "unknown"}: print(f" describe-image {'SKIP':>10} vision not configured") return True if vision_configured and vision_status != "online": print(f" describe-image {'FAIL':>10} vision configured but status={vision_status}") return False start = time.perf_counter() try: payload = _post_json( f"{base_url}/api/describe-image", {"image_id": "desk-reader", "caption": "article reader"}, timeout=90, ) elapsed = (time.perf_counter() - start) * 1000 runtime = payload.get("runtime", "unknown") alt_text = payload.get("alt_text", "") truncated = alt_text[:80] + ("..." if len(alt_text) > 80 else "") if alt_text else "(empty)" if runtime == "minicpm-v4.6": print(f" describe-image {_format_elapsed(elapsed):>10} PASS runtime={runtime} alt={truncated}") return True elif runtime == "fallback": print(f" describe-image {_format_elapsed(elapsed):>10} FAIL runtime={runtime} (expected live)") return False else: print(f" describe-image {_format_elapsed(elapsed):>10} FAIL runtime={runtime} (unexpected)") return False except (OSError, urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc: elapsed = (time.perf_counter() - start) * 1000 print(f" describe-image {_format_elapsed(elapsed):>10} FAIL ({exc.__class__.__name__})") return False def check_generate_image(base_url: str, runtime_status: dict[str, dict]) -> bool: """Smoke test /api/generate-image with a deterministic prompt.""" image_status = _role_status(runtime_status, "image_generation") image_configured = _role_is_configured(runtime_status, "image_generation") if not image_configured and image_status in {"fallback-ready", "unknown"}: print(f" generate-image {'SKIP':>10} Klein not configured") return True if image_configured and image_status != "online": print(f" generate-image {'FAIL':>10} Klein configured but status={image_status}") return False start = time.perf_counter() try: payload = _post_json( f"{base_url}/api/generate-image", {"prompt": "smoke test accessible article thumbnail", "seed": 42}, timeout=180, ) elapsed = (time.perf_counter() - start) * 1000 runtime = payload.get("runtime", "unknown") image_url = payload.get("image_url", "") if runtime == "modal-klein": print(f" generate-image {_format_elapsed(elapsed):>10} PASS runtime={runtime} url={image_url[:60]}") return True elif runtime == "fallback": print(f" generate-image {_format_elapsed(elapsed):>10} FAIL runtime={runtime} (expected live Klein)") return False else: print(f" generate-image {_format_elapsed(elapsed):>10} FAIL runtime={runtime} (unexpected)") return False except (OSError, urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc: elapsed = (time.perf_counter() - start) * 1000 print(f" generate-image {_format_elapsed(elapsed):>10} FAIL ({exc.__class__.__name__})") return False def main() -> None: parser = argparse.ArgumentParser(description="Tiny Narrator live model smoke tests") parser.add_argument("--base-url", default="http://127.0.0.1:7860", help="App base URL (default: http://127.0.0.1:7860)") args = parser.parse_args() base_url = args.base_url.rstrip("/") overall_start = time.perf_counter() print(f"Tiny Narrator live smoke tests — {base_url}") print() print("Checks:") runtime_status = check_runtime_status(base_url) if not runtime_status: print() print("Could not reach the app. Is it running?") sys.exit(1) vision_configured = _role_is_configured(runtime_status, "vision") klein_configured = _role_is_configured(runtime_status, "image_generation") describe_ok = check_describe_image(base_url, runtime_status) generate_ok = check_generate_image(base_url, runtime_status) overall_elapsed = (time.perf_counter() - overall_start) * 1000 print() failures = [] if vision_configured and not describe_ok: failures.append("describe-image") if klein_configured and not generate_ok: failures.append("generate-image") if failures: print(f"FAILED: {', '.join(failures)} ({_format_elapsed(overall_elapsed)} total)") sys.exit(1) else: configured_count = sum(1 for v in [vision_configured, klein_configured] if v) skipped_count = 2 - configured_count print(f"DONE: {configured_count} passed, {skipped_count} skipped ({_format_elapsed(overall_elapsed)} total)") sys.exit(0) if __name__ == "__main__": main()