xc_agent_platform_demo/scripts/smoke_health.py

#!/usr/bin/env python3
"""Smoke: GET /health must return 200 {"status":"ok"}.

Usage:
  # Against already-running server:
  SMOKE_BASE_URL=http://127.0.0.1:8080 python3 scripts/smoke_health.py

  # Or let this script start a local server (needs STRATEGY_ID):
  STRATEGY_ID=test python3 scripts/smoke_health.py
"""

from __future__ import annotations

import json
import os
import signal
import subprocess
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]


def _get(url: str, timeout: float = 2.0) -> tuple[int, dict]:
    req = urllib.request.Request(url, method="GET")
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        body = resp.read().decode("utf-8")
        return resp.status, json.loads(body)


def _wait_health(base: str, attempts: int = 30) -> None:
    url = base.rstrip("/") + "/health"
    last_err: Exception | None = None
    for _ in range(attempts):
        try:
            code, data = _get(url)
            if code == 200 and data.get("status") == "ok":
                print(f"OK: GET {url} -> {code} {data}")
                return
            raise AssertionError(f"unexpected response: {code} {data}")
        except (urllib.error.URLError, ConnectionError, AssertionError, TimeoutError) as exc:
            last_err = exc
            time.sleep(0.2)
    raise SystemExit(f"smoke-health failed: {last_err}")


def main() -> None:
    external = os.environ.get("SMOKE_BASE_URL", "").strip()
    if external:
        _wait_health(external)
        # optional status check
        try:
            code, data = _get(external.rstrip("/") + "/status")
            print(f"OK: GET /status -> {code} {data}")
        except Exception as exc:  # noqa: BLE001
            print(f"WARN: /status check skipped: {exc}", file=sys.stderr)
        return

    if not os.environ.get("STRATEGY_ID", "").strip():
        os.environ["STRATEGY_ID"] = "smoke-test-strategy"

    port = os.environ.get("SMOKE_PORT", "18080")
    env = os.environ.copy()
    env["PORT"] = port
    env["HOST"] = "127.0.0.1"
    env["STRATEGY_ID"] = env.get("STRATEGY_ID") or "smoke-test-strategy"

    proc = subprocess.Popen(
        [sys.executable, "-m", "src.server"],
        cwd=str(ROOT),
        env=env,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.PIPE,
    )
    try:
        _wait_health(f"http://127.0.0.1:{port}")
        code, data = _get(f"http://127.0.0.1:{port}/status")
        print(f"OK: GET /status -> {code} {data}")
        assert "strategy_id_present" in data
        assert data["strategy_id_present"] is True
        # never assert on secret values — only booleans
    finally:
        proc.send_signal(signal.SIGTERM)
        try:
            proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            proc.kill()
            proc.wait(timeout=2)
        if proc.returncode not in (0, -signal.SIGTERM, signal.SIGTERM):
            # 0 or terminated by SIGTERM is fine
            err = (proc.stderr.read() or b"").decode("utf-8", errors="replace")
            if proc.returncode is not None and proc.returncode > 0:
                print(f"WARN: server exit code {proc.returncode}: {err}", file=sys.stderr)


if __name__ == "__main__":
    main()