""" BankBot AI Endpoint Validation Script ====================================== Calls every AI endpoint and asserts the response shape is correct. Usage: # From the backend/ directory with the server running: python app/scripts/test_endpoints.py Exit codes: 0 — all tests passed 1 — one or more tests failed """ import sys import json import httpx BASE_URL = "http://127.0.0.1:8000" # ─── Result tracking ────────────────────────────────────────────────────────── results = [] # list of (name, passed, detail) def record(name: str, passed: bool, detail: str = ""): results.append((name, passed, detail)) # ─── Helpers ────────────────────────────────────────────────────────────────── def get(path: str, params: dict = None): return httpx.get(f"{BASE_URL}{path}", params=params, timeout=60) def post(path: str, body: dict): return httpx.post(f"{BASE_URL}{path}", json=body, timeout=60) def assert_keys(data: dict, *keys): missing = [k for k in keys if k not in data] if missing: raise AssertionError(f"Missing keys: {missing}") # ─── Tests ──────────────────────────────────────────────────────────────────── def test_health(): r = get("/health") assert r.status_code == 200 assert r.json().get("status") == "healthy" record("GET /health", True) def test_ai_status(): r = get("/api/ai/status") assert r.status_code == 200 data = r.json() assert_keys(data, "ai_backend", "ai_available", "db_type", "cache_type") assert data["db_type"] in ("sqlite", "postgresql") assert data["cache_type"] in ("redis", "memory") record("GET /api/ai/status", True, f"backend={data['ai_backend']} db={data['db_type']} cache={data['cache_type']}") def test_twin_predict(): r = get("/api/ai/twin/predict") assert r.status_code == 200, f"HTTP {r.status_code}: {r.text}" data = r.json() assert_keys(data, "current_balance", "projected_balance", "percent_change", "net_daily", "insight", "chart_data") assert isinstance(data["chart_data"], list) and len(data["chart_data"]) >= 1 assert data["projected_balance"] >= 0.0, "projected_balance must be non-negative" record("GET /api/ai/twin/predict", True, f"balance=${data['current_balance']:,.2f} → ${data['projected_balance']:,.2f}") def test_twin_future(): r = get("/api/ai/twin/future", params={"months": 12}) assert r.status_code == 200, f"HTTP {r.status_code}: {r.text}" data = r.json() assert_keys(data, "savings_growth", "investment_growth", "debt_decline") assert len(data["savings_growth"]) >= 1 assert len(data["investment_growth"]) >= 1 record("GET /api/ai/twin/future", True, f"savings_points={len(data['savings_growth'])}") def test_twin_scenarios(): r = get("/api/ai/twin/scenarios", params={"months": 6}) assert r.status_code == 200, f"HTTP {r.status_code}: {r.text}" data = r.json() assert_keys(data, "status_quo", "frugal", "lifestyle_inflation") for key in ("status_quo", "frugal", "lifestyle_inflation"): assert "balance_projection" in data[key], f"Missing balance_projection in {key}" record("GET /api/ai/twin/scenarios", True) def test_simulate_purchase(): r = post("/api/ai/simulate/purchase", { "amount": 500.0, "merchant": "Test Store", "category": "Shopping" }) assert r.status_code == 200, f"HTTP {r.status_code}: {r.text}" data = r.json() assert_keys(data, "risk_analysis", "projected_balance", "recommendation") assert data["risk_analysis"]["risk_level"] in ("low", "medium", "high", "critical") assert data["projected_balance"] >= 0.0 record("POST /api/ai/simulate/purchase", True, f"risk={data['risk_analysis']['risk_level']}") def test_simulate_investment(): r = post("/api/ai/simulate/investment", { "monthly_sip": 200.0, "asset_type": "stock", "lump_sum": 0.0 }) assert r.status_code == 200, f"HTTP {r.status_code}: {r.text}" data = r.json() assert_keys(data, "growth_projection", "is_affordable", "risk_analysis") assert len(data["growth_projection"]) == 3, \ f"Expected 3 growth milestones (1/3/5 yr), got {len(data['growth_projection'])}" record("POST /api/ai/simulate/investment", True, f"affordable={data['is_affordable']}") def test_simulate_subscription(): # First fetch a real subscription ID from the optimize endpoint r_subs = get("/api/ai/subscriptions/optimize") assert r_subs.status_code == 200 subs = r_subs.json().get("subscriptions", []) if not subs: record("POST /api/ai/simulate/subscription", True, "skipped — no subscriptions in DB") return sub_id = subs[0]["id"] r = post("/api/ai/simulate/subscription", {"subscription_ids": [sub_id]}) assert r.status_code == 200, f"HTTP {r.status_code}: {r.text}" data = r.json() assert_keys(data, "monthly_savings", "yearly_savings", "recommendation") assert data["monthly_savings"] >= 0.0 record("POST /api/ai/simulate/subscription", True, f"monthly_savings=${data['monthly_savings']:.2f}") def test_behavior_insights(): r = get("/api/ai/behavior/insights") assert r.status_code == 200, f"HTTP {r.status_code}: {r.text}" data = r.json() assert_keys(data, "insights", "metrics") assert isinstance(data["insights"], list) and len(data["insights"]) >= 1, \ "insights must be a non-empty list" record("GET /api/ai/behavior/insights", True, f"insights={len(data['insights'])}") def test_coaching_score(): r = get("/api/ai/coaching/score") assert r.status_code == 200, f"HTTP {r.status_code}: {r.text}" data = r.json() assert_keys(data, "overall_score", "categories", "explanation", "actionable_improvements") score = data["overall_score"] assert 0 <= score <= 100, f"overall_score {score} out of [0, 100]" expected_cats = ("savings_consistency", "debt_ratio", "spending_discipline", "emergency_funds", "investments", "subscription_management") for cat in expected_cats: assert cat in data["categories"], f"Missing category: {cat}" assert len(data["actionable_improvements"]) >= 1 record("GET /api/ai/coaching/score", True, f"score={score}/100") def test_coaching_briefing(): # This endpoint calls an LLM — allow up to 120s for local Ollama inference r = httpx.get(f"{BASE_URL}/api/ai/coaching/briefing", timeout=120) assert r.status_code == 200, f"HTTP {r.status_code}: {r.text}" data = r.json() assert_keys(data, "date", "user_name", "briefing", "metrics") assert isinstance(data["briefing"], str) and len(data["briefing"]) > 10 record("GET /api/ai/coaching/briefing", True, f"briefing_len={len(data['briefing'])} chars") def test_subscriptions_optimize(): r = get("/api/ai/subscriptions/optimize") assert r.status_code == 200, f"HTTP {r.status_code}: {r.text}" data = r.json() assert_keys(data, "subscriptions", "duplicates", "unused_subscriptions", "yearly_savings_potential", "risk_analysis") record("GET /api/ai/subscriptions/optimize", True, f"subs={len(data['subscriptions'])} " f"dupes={len(data['duplicates'])} " f"unused={len(data['unused_subscriptions'])}") def test_fraud_analysis(): r = get("/api/ai/fraud/analysis") assert r.status_code == 200, f"HTTP {r.status_code}: {r.text}" data = r.json() assert_keys(data, "total_alerts", "pending_reviews", "alerts") assert isinstance(data["total_alerts"], int) record("GET /api/ai/fraud/analysis", True, f"alerts={data['total_alerts']}") def test_chat(): # This endpoint calls an LLM — allow up to 120s for local Ollama inference r = httpx.post(f"{BASE_URL}/api/ai/chat", json={"message": "What is my current savings rate?"}, timeout=120) assert r.status_code == 200, f"HTTP {r.status_code}: {r.text}" data = r.json() assert "response" in data, "Missing 'response' key" assert isinstance(data["response"], str) and len(data["response"]) > 5 record("POST /api/ai/chat", True, f"response_len={len(data['response'])} chars") # ─── Runner ─────────────────────────────────────────────────────────────────── TESTS = [ test_health, test_ai_status, test_twin_predict, test_twin_future, test_twin_scenarios, test_simulate_purchase, test_simulate_investment, test_simulate_subscription, test_behavior_insights, test_coaching_score, test_coaching_briefing, test_subscriptions_optimize, test_fraud_analysis, test_chat, ] if __name__ == "__main__": print(f"\n{'─'*60}") print(f" BankBot AI Endpoint Validation — {BASE_URL}") print(f"{'─'*60}\n") for test_fn in TESTS: name = test_fn.__name__.replace("test_", "").replace("_", " ") try: test_fn() # result already recorded inside test_fn on success except AssertionError as e: record(name, False, str(e)) except Exception as e: record(name, False, f"Exception: {e}") # ── Summary table ───────────────────────────────────────────────────────── print(f"\n{'─'*60}") print(f" {'TEST':<40} {'RESULT':<8} DETAIL") print(f"{'─'*60}") passed = 0 failed = 0 for test_name, ok, detail in results: status = "✅ PASS" if ok else "❌ FAIL" print(f" {test_name:<40} {status:<8} {detail}") if ok: passed += 1 else: failed += 1 print(f"{'─'*60}") print(f" {passed} passed | {failed} failed | {len(results)} total") print(f"{'─'*60}\n") sys.exit(0 if failed == 0 else 1)