#!/usr/bin/env python3 """ Post-deploy smoke test de Leonex. Verifica end-to-end via HTTP las 6 piezas criticas tras un redeploy en EasyPanel: 1. El servidor responde (HTTP 200 en raiz). 2. /api/lock-status devuelve held=False (no atascos heredados). 3. /api/alpaca-diagnose verdict OK - all green (credenciales + SDK). 4. /api/promoted-diagnose devuelve >0 promovidas (el bridge actual). 5. /api/test-custom-bridge?ticker=CEG&strategy=box_prevday_hammer_long confirma que la rama custom se cargo (custom_module_available=true). 6. dashboard/data/paper_report.json existe y se sirve con equity > 0. Output: linea por chequeo + resumen final OK/FAIL. Exit code 0 si todo OK, 1 si algo falla. USO: python tools/post_deploy_check.py python tools/post_deploy_check.py --server https://leonex.tudominio.com python tools/post_deploy_check.py --verbose """ from __future__ import annotations import argparse import json import sys import urllib.error import urllib.request DEFAULT_SERVER = "http://localhost:8765" DEFAULT_TIMEOUT = 30 def _fetch(url: str, timeout: int = DEFAULT_TIMEOUT, method: str = "GET") -> tuple[int, dict | None, str | None]: """Devuelve (http_status, json_or_none, error_text). Nunca lanza.""" try: req = urllib.request.Request(url, method=method) with urllib.request.urlopen(req, timeout=timeout) as r: body = r.read().decode("utf-8", errors="replace") try: return r.status, json.loads(body), None except json.JSONDecodeError: return r.status, None, body[:300] except urllib.error.HTTPError as e: try: body = e.read().decode("utf-8", errors="replace") return e.code, json.loads(body), None except Exception: return e.code, None, str(e) except urllib.error.URLError as e: return 0, None, f"URLError: {e}" except Exception as e: return 0, None, f"Exception: {e!r}" def check(label: str, ok: bool, detail: str = "", verbose: bool = False) -> bool: tag = "[OK] " if ok else "[FAIL]" print(f"{tag} {label}") if detail and (verbose or not ok): for line in detail.splitlines(): print(f" {line}") return ok def main() -> int: parser = argparse.ArgumentParser(description="Leonex post-deploy smoke test") parser.add_argument("--server", default=DEFAULT_SERVER, help=f"Base URL del servidor (default {DEFAULT_SERVER})") parser.add_argument("--verbose", "-v", action="store_true", help="Imprime detalles incluso si pasa el check") parser.add_argument("--custom-ticker", default="CEG") parser.add_argument("--custom-strategy", default="box_prevday_hammer_long") parser.add_argument("--custom-timeframe", default="1h") args = parser.parse_args() server = args.server.rstrip("/") print("=" * 72) print(f"Leonex post-deploy smoke test — {server}") print("=" * 72) results: list[bool] = [] # 1) Server responds status, _, err = _fetch(f"{server}/dashboard/") if status == 0: results.append(check("1/6 Server responds", False, err or "no response", args.verbose)) else: results.append(check("1/6 Server responds", status in (200, 301, 302), f"HTTP {status}", args.verbose)) # 2) Lock status status, data, err = _fetch(f"{server}/api/lock-status") if status != 200 or not data: results.append(check("2/6 /api/lock-status", False, f"HTTP {status} · {err}", args.verbose)) else: held = data.get("held") holder = data.get("holder") age = data.get("age_seconds") detail = f"held={held} holder={holder} age={age}s" ok = held is False or (age is not None and age < 300) # <5min = ok-ish results.append(check("2/6 Lock status libre", ok, detail, args.verbose)) # 3) Alpaca diagnose status, data, err = _fetch(f"{server}/api/alpaca-diagnose", method="POST", timeout=60) if status != 200 or not data: results.append(check("3/6 /api/alpaca-diagnose", False, f"HTTP {status} · {err}", args.verbose)) else: verdict = data.get("verdict", "?") ok = "OK" in verdict # Resumir solo lo mas util env = data.get("env_vars", {}) env_summary = ", ".join(f"{k}={'set' if v else 'MISSING'}" for k, v in env.items()) detail = f"verdict='{verdict}'\nenv: {env_summary}" results.append(check("3/6 Alpaca diagnose", ok, detail, args.verbose)) # 4) Promoted diagnose status, data, err = _fetch(f"{server}/api/promoted-diagnose", method="POST", timeout=60) if status != 200 or not data: results.append(check("4/6 /api/promoted-diagnose", False, f"HTTP {status} · {err}", args.verbose)) else: n_promoted = data.get("n_promoted") or len(data.get("evaluated", [])) live_now = sum(1 for e in data.get("evaluated", []) if e.get("live")) ok = (n_promoted or 0) > 0 detail = (f"promoted={n_promoted} live_today={live_now} " f"note={data.get('note', '')[:80]}") results.append(check("4/6 Promoted bridge", ok, detail, args.verbose)) # 5) Custom bridge qs = (f"ticker={args.custom_ticker}&strategy={args.custom_strategy}" f"&timeframe={args.custom_timeframe}") status, data, err = _fetch(f"{server}/api/test-custom-bridge?{qs}", timeout=60) if status != 200 or not data: results.append(check("5/6 /api/test-custom-bridge", False, f"HTTP {status} · {err}", args.verbose)) else: custom_avail = data.get("custom_module_available") is_known = data.get("is_registered_custom") ev = data.get("evaluation", {}) or {} summary = data.get("summary", "") ok = bool(custom_avail and is_known) detail = (f"custom_module_available={custom_avail}\n" f"is_registered_custom={is_known}\n" f"{summary}") results.append(check("5/6 Custom bridge cargado", ok, detail, args.verbose)) # 6) Paper report served status, data, err = _fetch( f"{server}/dashboard/data/paper_report.json") if status != 200 or not data: results.append(check("6/6 paper_report.json", False, f"HTTP {status} · {err}", args.verbose)) else: equity = data.get("equity") auto = data.get("auto_approve") sim = data.get("simulated_only") ok = bool(equity and equity > 0) detail = f"equity=${equity:,.2f} auto_approve={auto} simulated_only={sim}" results.append(check("6/6 Paper report sirviendo", ok, detail, args.verbose)) # Resumen print() print("=" * 72) n_ok = sum(results) n_total = len(results) if n_ok == n_total: print(f"RESULTADO: {n_ok}/{n_total} OK — Leonex listo para operar.") else: print(f"RESULTADO: {n_ok}/{n_total} OK — REVISAR los [FAIL] arriba.") print("=" * 72) return 0 if n_ok == n_total else 1 if __name__ == "__main__": sys.exit(main())