From a4003e71c928a2b267f29e45fbc13147637454d3 Mon Sep 17 00:00:00 2001 From: Marcello Alarcon Date: Sun, 31 May 2026 17:10:30 -0300 Subject: [PATCH] feat(skills): add `smoke` health-check command to int-stripe/linkedin/youtube Add a uniform `smoke` connectivity command to three integration skills. The command runs auth + a representative read-only step, always exits 0, and emits structured JSON so callers (CI, heartbeats, manual validation) can distinguish PASS from FAIL without parsing stderr or relying on exit codes that crash on failure. Contract: {"overall": "PASS"|"FAIL", "steps": [{"step": "...", "status": "PASS"|"FAIL", "error": "...", "duration_ms": N}], "duration_ms": N} - Always exits 0, even on failure (missing creds report overall=FAIL, never crash) - Each step wrapped in try/except; errors captured into the `error` field int-stripe, int-linkedin and int-youtube are clean reference implementations. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../int-linkedin/scripts/linkedin_client.py | 35 ++++++++++++ .../skills/int-stripe/scripts/stripe_query.py | 54 ++++++++++++++++++- .../int-youtube/scripts/youtube_client.py | 34 ++++++++++++ 3 files changed, 122 insertions(+), 1 deletion(-) diff --git a/.claude/skills/int-linkedin/scripts/linkedin_client.py b/.claude/skills/int-linkedin/scripts/linkedin_client.py index 48fcfd24..a2aa68fb 100644 --- a/.claude/skills/int-linkedin/scripts/linkedin_client.py +++ b/.claude/skills/int-linkedin/scripts/linkedin_client.py @@ -233,6 +233,41 @@ def all_accounts_summary() -> dict: result = org_followers(acc) elif cmd == "summary": result = all_accounts_summary() + elif cmd == "smoke": + import time as _time + _t0 = _time.time() + _steps = [] + _overall = "PASS" + + # step 1: auth — load accounts + _ts = _time.time() + try: + _accounts = _get_accounts() + _steps.append({"step": "auth", "status": "PASS", "duration_ms": round((_time.time() - _ts) * 1000)}) + except Exception as _e: + _steps.append({"step": "auth", "status": "FAIL", "error": str(_e)[:300], "duration_ms": round((_time.time() - _ts) * 1000)}) + _overall = "FAIL" + _accounts = [] + + # step 2: profile read + _ts = _time.time() + if not _accounts: + _steps.append({"step": "profile_read", "status": "SKIP", "duration_ms": 0}) + else: + try: + _acc = _accounts[0] + _r = profile(_acc) + if "error" in _r: + _steps.append({"step": "profile_read", "status": "FAIL", "error": str(_r["error"])[:300], "duration_ms": round((_time.time() - _ts) * 1000)}) + _overall = "FAIL" + else: + _steps.append({"step": "profile_read", "status": "PASS", "duration_ms": round((_time.time() - _ts) * 1000)}) + except Exception as _e: + _steps.append({"step": "profile_read", "status": "FAIL", "error": str(_e)[:300], "duration_ms": round((_time.time() - _ts) * 1000)}) + _overall = "FAIL" + + print(json.dumps({"overall": _overall, "steps": _steps, "duration_ms": round((_time.time() - _t0) * 1000)}, indent=2)) + sys.exit(0) else: print(f"Unknown command: {cmd}") sys.exit(1) diff --git a/.claude/skills/int-stripe/scripts/stripe_query.py b/.claude/skills/int-stripe/scripts/stripe_query.py index 49ca0168..b0f6ab43 100644 --- a/.claude/skills/int-stripe/scripts/stripe_query.py +++ b/.claude/skills/int-stripe/scripts/stripe_query.py @@ -165,10 +165,60 @@ def cmd_update(args): print(json.dumps(result, indent=2)) +def cmd_smoke(args): + """E2E gate read-only — valida auth + balance. Sempre exit 0 + JSON.""" + import time as _time + + try: + out = {"steps": [], "overall": "PASS"} + t0 = _time.monotonic() + + def _ms(since): + return round((_time.monotonic() - since) * 1000) + + # Step 1: auth (API key presente) + ts = _time.monotonic() + try: + key = os.environ.get("STRIPE_SECRET_KEY") + if not key: + out["steps"].append({"step": "auth", "status": "FAIL", "error": "STRIPE_SECRET_KEY ausente", "duration_ms": _ms(ts)}) + out["overall"] = "FAIL" + print(json.dumps({**out, "duration_ms": _ms(t0)})) + return + out["steps"].append({"step": "auth", "status": "PASS", "duration_ms": _ms(ts)}) + except Exception as exc: + out["steps"].append({"step": "auth", "status": "FAIL", "error": str(exc)[:300], "duration_ms": _ms(ts)}) + out["overall"] = "FAIL" + print(json.dumps({**out, "duration_ms": _ms(t0)})) + return + + # Step 2: retrieve balance (cheap read-only endpoint) + ts = _time.monotonic() + try: + req = urllib.request.Request( + f"{BASE_URL}/balance", + headers={"Authorization": f"Bearer {key}"}, + ) + with urllib.request.urlopen(req, timeout=15) as resp: + data = json.loads(resp.read()) + currency = data.get("available", [{}])[0].get("currency", "?") if data.get("available") else "?" + out["steps"].append({"step": "balance", "status": "PASS", "currency": currency, "duration_ms": _ms(ts)}) + except Exception as exc: + out["steps"].append({"step": "balance", "status": "FAIL", "error": str(exc)[:300], "duration_ms": _ms(ts)}) + out["overall"] = "FAIL" + + print(json.dumps({**out, "duration_ms": _ms(t0)})) + except BaseException as exc: + print(json.dumps({"overall": "FAIL", "steps": [], "error": str(exc)[:300], "duration_ms": 0})) + + def main(): parser = argparse.ArgumentParser(description="Query Stripe via REST API") sub = parser.add_subparsers(dest="command") + # smoke + sub.add_parser("smoke", help="E2E gate read-only — valida auth + balance") + # list list_p = sub.add_parser("charges"); list_p.set_defaults(command="list", resource="charges") for r in ["customers", "invoices", "subscriptions", "payment_intents", "refunds", "products", "prices", "balance_transactions"]: @@ -209,7 +259,9 @@ def main(): parser.print_help() sys.exit(1) - if args.command == "list": + if args.command == "smoke": + cmd_smoke(args) + elif args.command == "list": cmd_list(args) elif args.command == "get": cmd_get(args) diff --git a/.claude/skills/int-youtube/scripts/youtube_client.py b/.claude/skills/int-youtube/scripts/youtube_client.py index 341f8103..fbb9c95d 100644 --- a/.claude/skills/int-youtube/scripts/youtube_client.py +++ b/.claude/skills/int-youtube/scripts/youtube_client.py @@ -416,6 +416,40 @@ def all_accounts_summary() -> dict: result = comments(acc, vid_id, n) elif cmd == "summary": result = all_accounts_summary() + elif cmd == "smoke": + import time as _time + _t0 = _time.time() + _steps = [] + _overall = "PASS" + + # step 1: auth — load accounts + _ts = _time.time() + try: + _accounts = _get_accounts() + _steps.append({"step": "auth", "status": "PASS", "duration_ms": round((_time.time() - _ts) * 1000)}) + except Exception as _e: + _steps.append({"step": "auth", "status": "FAIL", "error": str(_e)[:300], "duration_ms": round((_time.time() - _ts) * 1000)}) + _overall = "FAIL" + _accounts = [] + + # step 2: channel_stats read (cheap: 1 quota unit) + _ts = _time.time() + if not _accounts: + _steps.append({"step": "channel_stats", "status": "SKIP", "duration_ms": 0}) + else: + try: + _r = channel_stats(_accounts[0]) + if "error" in _r: + _steps.append({"step": "channel_stats", "status": "FAIL", "error": str(_r["error"])[:300], "duration_ms": round((_time.time() - _ts) * 1000)}) + _overall = "FAIL" + else: + _steps.append({"step": "channel_stats", "status": "PASS", "duration_ms": round((_time.time() - _ts) * 1000)}) + except Exception as _e: + _steps.append({"step": "channel_stats", "status": "FAIL", "error": str(_e)[:300], "duration_ms": round((_time.time() - _ts) * 1000)}) + _overall = "FAIL" + + print(json.dumps({"overall": _overall, "steps": _steps, "duration_ms": round((_time.time() - _t0) * 1000)}, indent=2)) + sys.exit(0) else: print(f"Unknown command: {cmd}") sys.exit(1)