Add deployed API smoke harness

scripts/core_hub_deployed_smoke.py

@@ -0,0 +1,142 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+
+import httpx
+
+from core_hub.smoke import (
+    DEFAULT_OPERATOR_TOKEN_ENV,
+    DEFAULT_RUNTIME_TOKEN_ENV,
+    SmokeError,
+    load_secret_value,
+    normalize_base_url,
+    run_smoke,
+)
+
+
+def optional_path(value: str | None) -> Path | None:
+    return Path(value) if value else None
+
+
+def emit_report(report: dict, output_path: Path | None) -> None:
+    body = json.dumps(report, indent=2, sort_keys=True)
+    if output_path is None:
+        print(body)
+        return
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    output_path.write_text(body + "\n", encoding="utf-8")
+    print(body)
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        description="Run a non-secret deployed Core Hub API smoke against CORE_HUB_BASE_URL."
+    )
+    parser.add_argument("--base-url", default=os.environ.get("CORE_HUB_BASE_URL"))
+    parser.add_argument(
+        "--operator-token-env",
+        default=os.environ.get("CORE_HUB_OPERATOR_TOKEN_ENV", DEFAULT_OPERATOR_TOKEN_ENV),
+        help="Environment variable that contains the approved operator token.",
+    )
+    parser.add_argument(
+        "--operator-token-file",
+        default=os.environ.get("CORE_HUB_OPERATOR_TOKEN_FILE"),
+        help="File containing the approved operator token. The value is never printed.",
+    )
+    parser.add_argument(
+        "--runtime-token-env",
+        default=os.environ.get("CORE_HUB_RUNTIME_TOKEN_ENV", DEFAULT_RUNTIME_TOKEN_ENV),
+        help="Optional environment variable containing a runtime token to reuse.",
+    )
+    parser.add_argument(
+        "--runtime-token-file",
+        default=os.environ.get("CORE_HUB_RUNTIME_TOKEN_FILE"),
+        help="Optional file containing a runtime token to reuse. The value is never printed.",
+    )
+    parser.add_argument(
+        "--runtime-token-output",
+        default=os.environ.get("CORE_HUB_RUNTIME_TOKEN_OUTPUT"),
+        help="Optional 0600 file path for a newly created runtime token.",
+    )
+    parser.add_argument("--output", default=os.environ.get("CORE_HUB_SMOKE_OUTPUT"))
+    parser.add_argument("--run-id", default=os.environ.get("CORE_HUB_SMOKE_RUN_ID"))
+    parser.add_argument("--timeout", type=float, default=30.0)
+    return parser
+
+
+def failure_report(base_url: str | None, reason: str) -> dict:
+    return {
+        "ok": False,
+        "baseUrl": base_url,
+        "runId": None,
+        "checks": [],
+        "bootstrap": {},
+        "failures": [reason],
+    }
+
+
+def main() -> int:
+    args = build_parser().parse_args()
+    output_path = optional_path(args.output)
+    if not args.base_url:
+        emit_report(failure_report(None, "set CORE_HUB_BASE_URL or --base-url"), output_path)
+        return 2
+
+    try:
+        base_url = normalize_base_url(args.base_url)
+    except SmokeError as exc:
+        emit_report(failure_report(None, str(exc)), output_path)
+        return 2
+
+    try:
+        operator_token = load_secret_value(
+            args.operator_token_env, optional_path(args.operator_token_file)
+        )
+    except OSError as exc:
+        emit_report(
+            failure_report(base_url, f"could not read operator token file: {exc}"),
+            output_path,
+        )
+        return 2
+    if not operator_token:
+        emit_report(
+            failure_report(
+                base_url,
+                "set CORE_HUB_OPERATOR_TOKEN_FILE or "
+                "CORE_HUB_OPERATOR_TOKEN_ENV/CORE_HUB_OPERATOR_TOKEN",
+            ),
+            output_path,
+        )
+        return 2
+
+    try:
+        runtime_token = load_secret_value(
+            args.runtime_token_env, optional_path(args.runtime_token_file)
+        )
+    except OSError as exc:
+        emit_report(
+            failure_report(base_url, f"could not read runtime token file: {exc}"),
+            output_path,
+        )
+        return 2
+    runtime_output = optional_path(args.runtime_token_output)
+
+    with httpx.Client(base_url=base_url, timeout=args.timeout) as client:
+        report = run_smoke(
+            client,
+            base_url=base_url,
+            operator_token=operator_token,
+            runtime_token=runtime_token or None,
+            runtime_token_output=runtime_output,
+            run_id=args.run_id,
+        )
+    emit_report(report, output_path)
+    return 0 if report.get("ok") else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())