Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """Smoke-test a running MediaRouter production container using stdlib only.""" | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import sys | |
| import urllib.error | |
| import urllib.request | |
| from collections.abc import Iterable | |
| from typing import Any | |
| REQUIRED_PATHS: dict[str, frozenset[str]] = { | |
| "/v1/projects": frozenset({"get", "post"}), | |
| "/v1/projects/{project_id}": frozenset({"get", "patch", "delete"}), | |
| "/v1/projects/{project_id}/assets": frozenset({"get", "post"}), | |
| "/v1/projects/{project_id}/assets/{asset_id}": frozenset({"delete"}), | |
| "/v1/projects/{project_id}/jobs": frozenset({"get", "post"}), | |
| "/v1/projects/{project_id}/jobs/{job_id}": frozenset({"delete"}), | |
| } | |
| REQUIRED_PREFIXES = ("/v1/social", "/v1/generation") | |
| def request_json( | |
| base_url: str, path: str, *, api_key: str = "" | |
| ) -> tuple[int, dict[str, Any]]: | |
| headers = {"Accept": "application/json"} | |
| if api_key: | |
| headers["Authorization"] = f"Bearer {api_key}" | |
| request = urllib.request.Request(f"{base_url.rstrip('/')}{path}", headers=headers) | |
| try: | |
| with urllib.request.urlopen(request, timeout=15) as response: | |
| status = response.status | |
| payload = response.read() | |
| except urllib.error.HTTPError as exc: | |
| status = exc.code | |
| payload = exc.read() | |
| try: | |
| decoded = json.loads(payload) | |
| except (json.JSONDecodeError, UnicodeDecodeError) as exc: | |
| raise RuntimeError(f"{path} did not return JSON (HTTP {status})") from exc | |
| if not isinstance(decoded, dict): | |
| raise RuntimeError(f"{path} did not return a JSON object") | |
| return status, decoded | |
| def missing_operations(paths: dict[str, Any]) -> Iterable[str]: | |
| for path, required_methods in REQUIRED_PATHS.items(): | |
| operations = paths.get(path) | |
| if not isinstance(operations, dict): | |
| yield f"{path} (path missing)" | |
| continue | |
| missing = required_methods - operations.keys() | |
| for method in sorted(missing): | |
| yield f"{method.upper()} {path}" | |
| for prefix in REQUIRED_PREFIXES: | |
| if not any(path.startswith(prefix) for path in paths): | |
| yield f"{prefix}/* (route family missing)" | |
| def run(base_url: str, api_key: str) -> None: | |
| health_status, health = request_json(base_url, "/health") | |
| if health_status != 200: | |
| raise RuntimeError(f"/health returned HTTP {health_status}") | |
| if health.get("success") is not True: | |
| raise RuntimeError("/health did not return the MediaRouter success envelope") | |
| openapi_status, openapi = request_json(base_url, "/openapi.json") | |
| if openapi_status != 200: | |
| raise RuntimeError(f"/openapi.json returned HTTP {openapi_status}") | |
| paths = openapi.get("paths") | |
| if not isinstance(paths, dict): | |
| raise RuntimeError("runtime OpenAPI contains no paths object") | |
| missing = list(missing_operations(paths)) | |
| if missing: | |
| raise RuntimeError("runtime OpenAPI is incomplete: " + ", ".join(missing)) | |
| auth_status, _auth = request_json(base_url, "/v1/auth/context", api_key=api_key) | |
| expected_auth_status = 200 if api_key else 401 | |
| if auth_status != expected_auth_status: | |
| raise RuntimeError( | |
| "/v1/auth/context returned " | |
| f"HTTP {auth_status}; expected {expected_auth_status}" | |
| ) | |
| final_health_status, _final_health = request_json(base_url, "/health") | |
| if final_health_status != 200: | |
| raise RuntimeError("application stopped responding during the smoke test") | |
| print("DEPLOYMENT_SMOKE=PASS") | |
| print(f"RUNTIME_OPENAPI_PATHS={len(paths)}") | |
| print(f"AUTHENTICATION_CHECK={'AUTHENTICATED' if api_key else 'FAIL_CLOSED'}") | |
| def main() -> int: | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument("--base-url", default="http://127.0.0.1:7860") | |
| parser.add_argument( | |
| "--api-key", | |
| default="", | |
| help="Optional test key; never printed. Without it, a 401 is required.", | |
| ) | |
| args = parser.parse_args() | |
| try: | |
| run(args.base_url, args.api_key) | |
| except Exception as exc: | |
| print(f"DEPLOYMENT_SMOKE=FAIL: {exc}", file=sys.stderr) | |
| return 1 | |
| return 0 | |
| if __name__ == "__main__": | |
| raise SystemExit(main()) | |