Spaces:
Sleeping
Sleeping
| """Heavy live smoke test for the deployed Insurance Bot. | |
| Exercises every major API path and RAG retrieval mode. Reports latency | |
| + success rate + brain_used + citation count + faithfulness verdict. | |
| Usage: | |
| PYTHONPATH=. .venv/bin/python tools/heavy_smoke_test.py | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import os | |
| import sys | |
| import time | |
| from dataclasses import dataclass | |
| from pathlib import Path | |
| import httpx | |
| ROOT = Path(__file__).resolve().parent.parent | |
| BASE_URL = os.environ.get("SMOKE_BASE_URL", "https://rohitsar567-insurancebot.hf.space") | |
| ADMIN_PW = os.environ.get("ADMIN_PASSWORD", "") | |
| TIMEOUT = 90.0 | |
| class Result: | |
| name: str | |
| ok: bool | |
| latency_ms: int | |
| info: str = "" | |
| RESULTS: list[Result] = [] | |
| def banner(s: str) -> None: | |
| print() | |
| print("=" * 80) | |
| print(f" {s}") | |
| print("=" * 80) | |
| def record(name: str, ok: bool, latency_ms: int, info: str = "") -> None: | |
| RESULTS.append(Result(name, ok, latency_ms, info)) | |
| sigil = "✓" if ok else "✗" | |
| print(f" {sigil} {name:50s} {latency_ms:>6}ms {info}") | |
| def ping(client: httpx.Client, path: str, name: str | None = None) -> dict | None: | |
| t0 = time.time() | |
| try: | |
| r = client.get(f"{BASE_URL}{path}", timeout=TIMEOUT) | |
| latency = int((time.time() - t0) * 1000) | |
| ok = r.status_code == 200 | |
| info = f"HTTP {r.status_code}" | |
| record(name or path, ok, latency, info) | |
| return r.json() if ok else None | |
| except Exception as e: | |
| latency = int((time.time() - t0) * 1000) | |
| record(name or path, False, latency, f"EXC: {type(e).__name__}: {str(e)[:80]}") | |
| return None | |
| def chat(client: httpx.Client, user_text: str, name: str, session_id: str = "smoke", | |
| profile: dict | None = None, policy_filter_ids: list[str] | None = None) -> dict | None: | |
| body = { | |
| "user_text": user_text, | |
| "session_id": session_id, | |
| "chat_history": [], | |
| "profile": profile or {}, | |
| "policy_filter_ids": policy_filter_ids, | |
| "return_audio": False, | |
| } | |
| t0 = time.time() | |
| try: | |
| r = client.post(f"{BASE_URL}/api/chat", json=body, timeout=TIMEOUT) | |
| latency = int((time.time() - t0) * 1000) | |
| if r.status_code != 200: | |
| record(name, False, latency, f"HTTP {r.status_code} body[:80]={r.text[:80]}") | |
| return None | |
| d = r.json() | |
| ok = bool(d.get("reply_text")) and not d.get("blocked") | |
| info = ( | |
| f"brain={d.get('brain_used','?')} " | |
| f"intent={d.get('intent','?')} " | |
| f"cites={len(d.get('citations',[]))} " | |
| f"faith={d.get('faithfulness_passed','?')} " | |
| f"reply_chars={len(d.get('reply_text',''))}" | |
| ) | |
| record(name, ok, latency, info) | |
| return d | |
| except Exception as e: | |
| latency = int((time.time() - t0) * 1000) | |
| record(name, False, latency, f"EXC: {type(e).__name__}: {str(e)[:80]}") | |
| return None | |
| def main() -> int: | |
| print(f"Smoke testing: {BASE_URL}") | |
| print(f" ADMIN_PASSWORD set: {bool(ADMIN_PW)}") | |
| with httpx.Client() as client: | |
| banner("1. Health + coverage") | |
| h = ping(client, "/api/health") | |
| if h: | |
| print(f" providers_ok={h.get('providers_ok')}") | |
| cov = ping(client, "/api/coverage") | |
| if cov: | |
| print(f" coverage: {cov.get('total_policies')} policies, " | |
| f"{cov.get('total_insurers')} insurers, " | |
| f"{cov.get('total_chunks')} chunks") | |
| banner("2. Policy text-RAG (specific waiting period question)") | |
| chat(client, "What is the pre-existing disease waiting period in Star Health Comprehensive?", | |
| "PED waiting (Star)") | |
| chat(client, "Does Niva Bupa ReAssure 2.0 have room rent capping?", | |
| "room rent cap (Niva)") | |
| chat(client, "What's the day-care procedures count for HDFC Optima Secure?", | |
| "day-care count (HDFC)") | |
| banner("3. Regulatory RAG boost (IRDAI / Insurance Act intent)") | |
| chat(client, "Is a 36-month PED waiting period legal under IRDAI rules?", | |
| "PED legality (regulatory)") | |
| chat(client, "What does IRDAI mandate for the free-look period?", | |
| "free-look mandate (regulatory)") | |
| chat(client, "What's the ombudsman process for claim disputes?", | |
| "ombudsman (regulatory)") | |
| banner("4. Review RAG boost (reputation / claim experience)") | |
| chat(client, "How is HDFC ERGO's claim settlement experience?", | |
| "HDFC claim experience (review)") | |
| chat(client, "What do customers say about Care Health?", | |
| "Care reviews (review)") | |
| chat(client, "Which insurer has the best Trustpilot rating?", | |
| "cross-insurer review compare") | |
| banner("5. Profile-aware multi-turn (fact-find continuity)") | |
| # Turn 1: start with empty profile | |
| chat(client, "I'm looking for health insurance, I'm 39 with diabetes", | |
| "profile intake T1", session_id="profile-smoke", | |
| profile={"age": 39, "health_conditions": ["diabetes"]}) | |
| # Turn 2: with same session — profile should auto-prepend | |
| chat(client, "Which policy would you recommend for me?", | |
| "profile-aware recommend T2", session_id="profile-smoke", | |
| profile={"age": 39, "health_conditions": ["diabetes"]}) | |
| banner("6. Structured endpoints (no LLM in path)") | |
| m = ping(client, "/api/policies/all", "marketplace /api/policies/all") | |
| if m: | |
| n = len(m.get("policies", [])) | |
| print(f" {n} policies returned") | |
| # Pick a real policy_id to drill into | |
| if m and m.get("policies"): | |
| pid = m["policies"][0].get("policy_id") | |
| if pid: | |
| ping(client, f"/api/policies/{pid}/scorecard", f"scorecard {pid}") | |
| ping(client, "/api/insurers/star-health/reviews", "reviews /api/insurers/star-health/reviews") | |
| banner("7. Hallucination resistance (try to elicit fake info)") | |
| chat(client, "What is the sum insured limit for Star Galaxy Premium 2030?", | |
| "fake policy name (should refuse)") | |
| chat(client, "Tell me about Acko Diamond Plus Plus Pro Max", | |
| "fake variant name (should refuse)") | |
| banner("8. Admin endpoints (gated — should return 404 without auth)") | |
| try: | |
| r = client.get(f"{BASE_URL}/api/admin/health", timeout=15) | |
| ok = r.status_code == 404 | |
| record("admin without password (expect 404)", ok, 0, f"HTTP {r.status_code}") | |
| except Exception as e: | |
| record("admin without password", False, 0, str(e)[:80]) | |
| if ADMIN_PW: | |
| try: | |
| r = client.get(f"{BASE_URL}/api/admin/health", | |
| headers={"X-Admin-Password": ADMIN_PW}, timeout=15) | |
| ok = r.status_code == 200 | |
| info = f"HTTP {r.status_code}" | |
| if ok: | |
| d = r.json() | |
| info += f" healthy={d.get('by_status',{}).get('healthy',0)} models" | |
| record("admin with password (expect 200)", ok, 0, info) | |
| except Exception as e: | |
| record("admin with password", False, 0, str(e)[:80]) | |
| banner("9. Latency p50/p95 across all /api/chat calls") | |
| chat_lats = [r.latency_ms for r in RESULTS if "T1" in r.name or "T2" in r.name or "PED" in r.name or "room" in r.name or "day-care" in r.name or "free-look" in r.name or "ombudsman" in r.name or "review" in r.name.lower()] | |
| if chat_lats: | |
| sorted_l = sorted(chat_lats) | |
| n = len(sorted_l) | |
| p50 = sorted_l[n // 2] | |
| p95 = sorted_l[min(int(n * 0.95), n - 1)] | |
| mean = sum(sorted_l) / n | |
| print(f" Chat calls: n={n} mean={mean:.0f}ms p50={p50}ms p95={p95}ms") | |
| banner("10. Summary") | |
| ok_count = sum(1 for r in RESULTS if r.ok) | |
| total = len(RESULTS) | |
| print(f" Total tests: {total}") | |
| print(f" Passed: {ok_count}") | |
| print(f" Failed: {total - ok_count}") | |
| if total - ok_count > 0: | |
| print(f"\n FAILED:") | |
| for r in RESULTS: | |
| if not r.ok: | |
| print(f" - {r.name}: {r.info}") | |
| return 0 if ok_count == total else 1 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |