svrnty-hermes-webui-plugin/scripts/boot-smoke.py

#!/usr/bin/env python3
"""boot-smoke.py — start hermes-webui + plugin, curl every plugin endpoint.

Exit 0 if every endpoint returns its expected status, 1 otherwise. Used by
upstream-sync.py and as a one-shot manual check after install.

Usage:
  python3 boot-smoke.py                    # start webui + smoke + stop
  python3 boot-smoke.py --no-start         # webui already running; just smoke
  python3 boot-smoke.py --base http://...  # smoke against custom base URL
"""
import argparse
import json
import os
import signal
import socket
import subprocess
import sys
import time
from pathlib import Path
from urllib.request import Request, urlopen
from urllib.error import URLError

PLUGIN_REPO = Path(__file__).resolve().parent.parent

# Endpoints we expect after the plugin is loaded. Status codes and content
# checks are minimal — this is "did it boot", not "is it correct".
SMOKE = [
    {"path": "/healthz", "expect": [200], "kind": "vanilla"},
    {"path": "/api/vault/status", "expect": [200, 401, 403], "kind": "plugin"},
    {"path": "/plugins/svrnty/app.css", "expect": [200], "kind": "plugin-static"},
    {"path": "/plugins/svrnty/app.js", "expect": [200], "kind": "plugin-static"},
]


def _free_port():
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    port = s.getsockname()[1]
    s.close()
    return port


def _wait_for(url, timeout=20):
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            with urlopen(url, timeout=1) as r:
                if r.status < 500:
                    return True
        except URLError:
            pass
        except Exception:
            pass
        time.sleep(0.3)
    return False


def _hit(base, path):
    url = base.rstrip("/") + path
    try:
        with urlopen(url, timeout=5) as r:
            return r.status, r.read()[:200]
    except URLError as e:
        if hasattr(e, "code"):
            return e.code, b""
        return None, str(e).encode()
    except Exception as e:
        return None, str(e).encode()


def smoke(base):
    rows = []
    failed = 0
    for s in SMOKE:
        status, _body = _hit(base, s["path"])
        ok = status in s["expect"]
        rows.append({"path": s["path"], "status": status, "kind": s["kind"], "ok": ok})
        if not ok:
            failed += 1
    return rows, failed


def main():
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    ap.add_argument("--no-start", action="store_true",
                    help="Assume hermes-webui is already running; just curl")
    ap.add_argument("--base", default=None,
                    help="Base URL (default http://127.0.0.1:<port>)")
    ap.add_argument("--webui-dir",
                    default=str(PLUGIN_REPO.parent / "hermes-webui"),
                    help="Path to hermes-webui repo")
    args = ap.parse_args()

    proc = None
    base = args.base or "http://127.0.0.1:8787"

    if not args.no_start:
        port = _free_port()
        base = f"http://127.0.0.1:{port}"
        env = os.environ.copy()
        env["HERMES_WEBUI_PYTHON_PLUGIN"] = "svrnty_hermes_webui_plugin"
        env["PORT"] = str(port)
        # Best-effort: start under the agent venv if it exists; else system python.
        py = Path(args.webui_dir) / "venv" / "bin" / "python"
        cmd = [str(py) if py.exists() else "python3", "bootstrap.py", "--foreground"]
        proc = subprocess.Popen(
            cmd, cwd=args.webui_dir, env=env,
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
            preexec_fn=os.setsid,
        )
        if not _wait_for(f"{base}/healthz", timeout=30):
            print(f"FAIL: webui did not respond at {base}/healthz within 30s",
                  file=sys.stderr)
            os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
            sys.exit(1)

    try:
        rows, failed = smoke(base)
    finally:
        if proc is not None:
            os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
            try:
                proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                os.killpg(os.getpgid(proc.pid), signal.SIGKILL)

    print(json.dumps({"base": base, "rows": rows, "failed": failed}, indent=2))
    sys.exit(0 if failed == 0 else 1)


if __name__ == "__main__":
    main()