core-hub/scripts/core_hub_deployed_smoke.py

#!/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())