yashprmr/MeridianFinancial

1"""2scripts/smoke_test.py3---------------------4Lightweight smoke test for the Meridian Financial serving API.5 6Tests7-----81. GET /health        — status == "ok"92. GET /metrics       — required keys present103. GET /openapi.json  — all 6 endpoints registered11 12Can be run:13  * During Docker build verification14  * As a CI post-deploy check15  * Manually against any running instance16 17Usage18-----19  python scripts/smoke_test.py [--base-url http://localhost:8000]20 21Exit codes22----------23  0  all tests passed24  1  one or more tests failed25"""26 27from __future__ import annotations28 29import argparse30import json31import sys32import urllib.error33import urllib.request34 35 36# ---------------------------------------------------------------------------37# Smoke test cases38# ---------------------------------------------------------------------------39 40def _get(url: str, timeout: int = 10) -> dict:41    """HTTP GET with a basic error wrapper."""42    req = urllib.request.Request(url, method="GET")43    with urllib.request.urlopen(req, timeout=timeout) as resp:44        return json.loads(resp.read().decode())45 46 47def test_health(base: str) -> tuple[bool, str]:48    try:49        data = _get(f"{base}/health")50        assert data.get("status") == "ok", f"Expected status=ok, got {data}"51        return True, "status == ok"52    except Exception as exc:  # noqa: BLE00153        return False, str(exc)54 55 56def test_metrics(base: str) -> tuple[bool, str]:57    try:58        data = _get(f"{base}/metrics")59        required = {"total_requests", "uptime_seconds", "prediction_distribution", "rag_retrieval_stats"}60        missing = required - set(data.keys())61        assert not missing, f"Missing keys: {missing}"62        return True, f"keys present: {sorted(required)}"63    except Exception as exc:  # noqa: BLE00164        return False, str(exc)65 66 67def test_openapi(base: str) -> tuple[bool, str]:68    try:69        data = _get(f"{base}/openapi.json")70        paths = set(data.get("paths", {}).keys())71        required_paths = {72            "/health", "/predict", "/ask-complaints",73            "/batch-score", "/customer-intel", "/metrics",74        }75        missing = required_paths - paths76        assert not missing, f"Missing endpoints: {missing}"77        return True, f"all {len(required_paths)} endpoints registered"78    except Exception as exc:  # noqa: BLE00179        return False, str(exc)80 81 82# ---------------------------------------------------------------------------83# Runner84# ---------------------------------------------------------------------------85 86TESTS = [87    ("GET /health", test_health),88    ("GET /metrics", test_metrics),89    ("GET /openapi.json", test_openapi),90]91 92 93def run_smoke_tests(base_url: str) -> int:94    """Run all smoke tests and return exit code (0=pass, 1=fail)."""95    base = base_url.rstrip("/")96    print(f"\nSmoke tests against {base}")97    print("=" * 60)98 99    passed = 0100    failed = 0101 102    for name, test_fn in TESTS:103        ok, msg = test_fn(base)104        status = "PASS" if ok else "FAIL"105        print(f"  [{status}] {name}  —  {msg}")106        if ok:107            passed += 1108        else:109            failed += 1110 111    print("=" * 60)112    print(f"Result: {passed} passed, {failed} failed\n")113    return 0 if failed == 0 else 1114 115 116def main() -> None:117    parser = argparse.ArgumentParser(description="Meridian Financial API smoke tests")118    parser.add_argument(119        "--base-url",120        default="http://localhost:8000",121        help="Base URL of the running API (default: http://localhost:8000)",122    )123    args = parser.parse_args()124    sys.exit(run_smoke_tests(args.base_url))125 126 127if __name__ == "__main__":128    main()129