Spaces:
Running
Running
File size: 4,260 Bytes
3493993 | 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 | #!/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())
|