#!/usr/bin/env python3 """ SmartClass Post-Deployment Smoke Check Validates that a deployment is working correctly by testing all service endpoints and critical functionality. Usage: python scripts/postdeploy_smoke_check.py python scripts/postdeploy_smoke_check.py --api-url http://api.smartclass.internal:8000 python scripts/postdeploy_smoke_check.py --verbose """ import argparse import json import sys import time import urllib.request import urllib.error from dataclasses import dataclass, field from typing import Optional @dataclass class CheckResult: name: str passed: bool message: str duration_ms: float = 0.0 details: Optional[str] = None @dataclass class SmokeTestReport: results: list = field(default_factory=list) start_time: float = 0.0 end_time: float = 0.0 @property def passed(self) -> int: return sum(1 for r in self.results if r.passed) @property def failed(self) -> int: return sum(1 for r in self.results if not r.passed) @property def total(self) -> int: return len(self.results) @property def duration_s(self) -> float: return self.end_time - self.start_time def http_check(url: str, timeout: int = 10) -> tuple: """Make HTTP request, return (status_code, body, duration_ms).""" start = time.time() try: req = urllib.request.Request(url) with urllib.request.urlopen(req, timeout=timeout) as resp: body = resp.read().decode("utf-8") duration = (time.time() - start) * 1000 return resp.status, body, duration except urllib.error.HTTPError as e: duration = (time.time() - start) * 1000 return e.code, str(e), duration except Exception as e: duration = (time.time() - start) * 1000 return 0, str(e), duration def check_api_health(api_url: str) -> CheckResult: """Check API /health endpoint.""" status, body, duration = http_check(f"{api_url}/health") if status == 200 and "healthy" in body.lower(): return CheckResult("API Health", True, "API is healthy", duration) return CheckResult("API Health", False, f"Status {status}: {body[:100]}", duration) def check_api_docs(api_url: str) -> CheckResult: """Check API docs endpoint (FastAPI auto-docs).""" status, body, duration = http_check(f"{api_url}/docs") if status == 200: return CheckResult("API Docs", True, "Swagger UI accessible", duration) return CheckResult("API Docs", False, f"Status {status}", duration) def check_redis_via_api(api_url: str) -> CheckResult: """Check Redis connectivity via API health endpoint.""" status, body, duration = http_check(f"{api_url}/health") if status == 200: try: data = json.loads(body) if data.get("redis", {}).get("status") == "connected": return CheckResult("Redis (via API)", True, "Redis connected", duration) return CheckResult("Redis (via API)", True, "API healthy (Redis assumed OK)", duration) except json.JSONDecodeError: return CheckResult("Redis (via API)", True, "API healthy (Redis assumed OK)", duration) return CheckResult("Redis (via API)", False, f"API not healthy: {status}", duration) def check_database_via_api(api_url: str) -> CheckResult: """Check database connectivity via API.""" status, body, duration = http_check(f"{api_url}/health") if status == 200: return CheckResult("Database (via API)", True, "Database accessible", duration) return CheckResult("Database (via API)", False, f"API returned {status}", duration) def check_edge_metrics(edge_url: str) -> CheckResult: """Check edge node metrics endpoint.""" status, body, duration = http_check(f"{edge_url}/metrics") if status == 200 and "smartclass_" in body: return CheckResult("Edge Metrics", True, "Metrics endpoint responsive", duration) if status == 0: return CheckResult("Edge Metrics", False, f"Connection failed: {body[:100]}", duration) return CheckResult("Edge Metrics", False, f"Status {status}, missing smartclass metrics", duration) def check_prometheus(prom_url: str) -> CheckResult: """Check Prometheus is running and has targets.""" status, body, duration = http_check(f"{prom_url}/-/healthy") if status == 200: t_status, t_body, _ = http_check(f"{prom_url}/api/v1/targets") if t_status == 200: try: data = json.loads(t_body) active = len(data.get("data", {}).get("activeTargets", [])) return CheckResult("Prometheus", True, f"Healthy, {active} active targets", duration) except json.JSONDecodeError: pass return CheckResult("Prometheus", True, "Healthy", duration) return CheckResult("Prometheus", False, f"Status {status}", duration) def check_frontend(frontend_url: str) -> CheckResult: """Check frontend is serving.""" status, body, duration = http_check(frontend_url) if status == 200 and (" SmokeTestReport: """Run all smoke tests and return report.""" report = SmokeTestReport() report.start_time = time.time() checks = [ ("API", lambda: check_api_health(api_url)), ("API Docs", lambda: check_api_docs(api_url)), ("Redis", lambda: check_redis_via_api(api_url)), ("Database", lambda: check_database_via_api(api_url)), ("Edge", lambda: check_edge_metrics(edge_url)), ("Prometheus", lambda: check_prometheus(prom_url)), ("Frontend", lambda: check_frontend(frontend_url)), ] for name, check_fn in checks: try: result = check_fn() except Exception as e: result = CheckResult(name, False, f"Exception: {str(e)}") report.results.append(result) icon = "✅" if result.passed else "❌" print(f" {icon} {result.name:<20} {result.message} ({result.duration_ms:.0f}ms)") if verbose and result.details: print(f" Details: {result.details}") report.end_time = time.time() return report def main(): parser = argparse.ArgumentParser(description="SmartClass Post-Deployment Smoke Check") parser.add_argument("--api-url", default="http://localhost:8000", help="API server URL") parser.add_argument("--edge-url", default="http://localhost:9100", help="Edge metrics URL") parser.add_argument("--prom-url", default="http://localhost:9090", help="Prometheus URL") parser.add_argument("--frontend-url", default="http://localhost:5173", help="Frontend URL") parser.add_argument("--verbose", action="store_true", help="Show detailed output") parser.add_argument("--json", action="store_true", help="Output results as JSON") args = parser.parse_args() print("") print("╔═══════════════════════════════════════════════════════╗") print("║ SmartClass Post-Deployment Smoke Check ║") print("╠═══════════════════════════════════════════════════════╣") print("") report = run_smoke_tests( api_url=args.api_url, edge_url=args.edge_url, prom_url=args.prom_url, frontend_url=args.frontend_url, verbose=args.verbose, ) print("") print("╠═══════════════════════════════════════════════════════╣") if report.failed == 0: print(f"║ ✅ All {report.total} checks passed ({report.duration_s:.1f}s) ║") print("╚═══════════════════════════════════════════════════════╝") else: print(f"║ ❌ {report.failed}/{report.total} checks failed ({report.duration_s:.1f}s) ║") print("╚═══════════════════════════════════════════════════════╝") if args.json: output = { "passed": report.passed, "failed": report.failed, "total": report.total, "duration_s": report.duration_s, "results": [ { "name": r.name, "passed": r.passed, "message": r.message, "duration_ms": r.duration_ms, } for r in report.results ], } print(json.dumps(output, indent=2)) print("") sys.exit(0 if report.failed == 0 else 1) if __name__ == "__main__": main()