customer-agent/scripts/smoke_launch.py

#!/usr/bin/env python3
"""Launch smoke checks for Customer Agent static surfaces and optional live API."""

from __future__ import annotations

import argparse
import json
import sys
import urllib.error
import urllib.request
from pathlib import Path


def check_local_files(root: Path) -> list[str]:
    required = [
        root / "embed" / "agent.js",
        root / "embed" / "agent.css",
        root / "admin" / "index.html",
        root / "test" / "index.html",
        root / "docs" / "customer-setup-guide.md",
        root / "docs" / "SHIPPING.md",
        root / "docs" / "LAUNCH.md",
        root / "VERSION",
        root / "integrations" / "wordpress" / "synderesis-customer-agent" / "synderesis-customer-agent.php",
        root / "integrations" / "wordpress" / "synderesis-customer-agent" / "uninstall.php",
        root / "evals" / "golden_cases.json",
    ]
    missing = [str(path.relative_to(root)) for path in required if not path.is_file()]
    return missing


def http_get(url: str, timeout: float = 10.0) -> tuple[int, str]:
    req = urllib.request.Request(url, headers={"User-Agent": "SynderesisSmoke/1.5"})
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            body = resp.read(200_000).decode("utf-8", errors="replace")
            return int(resp.status), body
    except urllib.error.HTTPError as exc:
        body = exc.read(20_000).decode("utf-8", errors="replace")
        return int(exc.code), body


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--product-root",
        default=str(Path(__file__).resolve().parents[1]),
        help="Path to customer-agent/",
    )
    parser.add_argument(
        "--base-url",
        default="",
        help="Optional live base URL, e.g. http://127.0.0.1:8080 or https://www.synderesis.eu",
    )
    parser.add_argument(
        "--site-key",
        default="",
        help="Optional pk_live_ key for public config smoke",
    )
    args = parser.parse_args(argv)
    root = Path(args.product_root).resolve()
    errors: list[str] = []

    print(f"Product root: {root}")
    missing = check_local_files(root)
    if missing:
        for item in missing:
            errors.append(f"missing file: {item}")
            print(f"FAIL missing {item}")
    else:
        print("PASS local package files present")

    version = (root / "VERSION").read_text(encoding="utf-8").strip()
    print(f"VERSION {version}")

    # Offline golden evals
    eval_script = root / "scripts" / "run_evals.py"
    if eval_script.is_file():
        import subprocess

        proc = subprocess.run(
            [sys.executable, str(eval_script)],
            cwd=str(root.parents[0]),
            capture_output=True,
            text=True,
        )
        if proc.returncode == 0:
            print("PASS golden evals")
        else:
            errors.append("golden evals failed")
            print(proc.stdout)
            print(proc.stderr)
            print("FAIL golden evals")
    else:
        errors.append("run_evals.py missing")

    base = args.base_url.rstrip("/")
    static_server = None
    static_thread = None
    if not base:
        # Self-contained static smoke: serve customer-agent/ so HTTP paths can be checked
        # without a full Synderesis API process.
        import http.server
        import socketserver
        import threading

        class _Handler(http.server.SimpleHTTPRequestHandler):
            def __init__(self, *a, **k):
                super().__init__(*a, directory=str(root), **k)

            def log_message(self, format: str, *args: object) -> None:  # noqa: A003
                return

        # Bind ephemeral port on localhost
        static_server = socketserver.TCPServer(("127.0.0.1", 0), _Handler)
        port = static_server.server_address[1]
        static_thread = threading.Thread(target=static_server.serve_forever, daemon=True)
        static_thread.start()
        base = f"http://127.0.0.1:{port}"
        print(f"Started local static smoke server at {base}")
        static_paths = [
            "/embed/agent.js",
            "/embed/agent.css",
            "/admin/index.html",
            "/test/index.html",
            "/docs/customer-setup-guide.md",
            "/docs/SHIPPING.md",
            "/docs/LAUNCH.md",
            "/VERSION",
        ]
        for path in static_paths:
            url = base + path
            try:
                status, body = http_get(url)
            except Exception as exc:  # noqa: BLE001
                errors.append(f"static {path}: {exc}")
                print(f"FAIL static {path} ({exc})")
                continue
            if status != 200:
                errors.append(f"static {path}: HTTP {status}")
                print(f"FAIL static {path} HTTP {status}")
            elif path.endswith("agent.js") and "site-key" not in body and "siteKey" not in body:
                errors.append(f"static {path}: unexpected body")
                print(f"FAIL static {path} unexpected body")
            else:
                print(f"PASS static {path} HTTP {status}")
        static_server.shutdown()
        print("Stopped local static smoke server")
    else:
        paths = [
            "/health",
            "/embed/agent.js",
            "/embed/agent.css",
            "/customer-agent/admin/",
            "/customer-agent/test/",
            "/customer-agent/docs/customer-setup-guide.md",
            "/customer-agent/docs/SHIPPING.md",
            "/customer-agent/docs/LAUNCH.md",
        ]
        for path in paths:
            url = base + path
            try:
                status, body = http_get(url)
            except Exception as exc:  # noqa: BLE001
                errors.append(f"{path}: {exc}")
                print(f"FAIL {path} ({exc})")
                continue
            if status != 200:
                errors.append(f"{path}: HTTP {status}")
                print(f"FAIL {path} HTTP {status}")
            elif path.endswith("agent.js") and "SYNDERESIS" not in body.upper() and "site-key" not in body:
                errors.append(f"{path}: unexpected body")
                print(f"FAIL {path} unexpected body")
            else:
                print(f"PASS {path} HTTP {status}")

        if args.site_key:
            cfg_url = f"{base}/v1/public/agents/{args.site_key}/config"
            try:
                status, body = http_get(cfg_url)
                data = json.loads(body) if body else {}
            except Exception as exc:  # noqa: BLE001
                errors.append(f"public config: {exc}")
                print(f"FAIL public config ({exc})")
            else:
                if status == 200 and data.get("name"):
                    print(f"PASS public config agent={data.get('name')!r}")
                else:
                    errors.append(f"public config HTTP {status}")
                    print(f"FAIL public config HTTP {status}")

    if errors:
        print(f"\nSMOKE FAILED ({len(errors)} issue(s))")
        for item in errors:
            print(f"  - {item}")
        return 1
    print("\nSMOKE PASSED")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())