#!/usr/bin/env python3
"""
UCall Integration Test Suite
─────────────────────────────
Runs against live ucall-server at http://127.0.0.1:8080

Usage:
    DATABASE_URL="..." cargo run -p ucall-api &
    python3 tests/integration_suite.py
"""

import asyncio
import json
import sys
import time
import uuid
import random
import urllib.request
import urllib.error
from typing import Optional
from dataclasses import dataclass, field

try:
    import websockets
except ImportError:
    import subprocess
    subprocess.check_call([sys.executable, "-m", "pip", "install", "websockets", "-q"])
    import websockets

BASE = "http://127.0.0.1:8080"
WS   = "ws://127.0.0.1:8080/api/v1/ws"

# ─── Result tracking ──────────────────────────────────────────────────────────

@dataclass
class Suite:
    results: list = field(default_factory=list)

    def ok(self, name: str, detail: str = ""):
        self.results.append((name, True, detail))
        suffix = f"  ({detail})" if detail else ""
        print(f"    ✅  {name}{suffix}")

    def fail(self, name: str, detail: str = ""):
        self.results.append((name, False, detail))
        suffix = f"  ({detail})" if detail else ""
        print(f"    ❌  {name}{suffix}")

    def check(self, name: str, cond: bool, detail: str = ""):
        if cond:
            self.ok(name, detail)
        else:
            self.fail(name, detail)
        return cond

    def summary(self) -> int:
        passed = sum(1 for _, ok, _ in self.results if ok)
        total  = len(self.results)
        failed = [(n, d) for n, ok, d in self.results if not ok]
        print()
        print("══════════════════════════════════════════════════════════════")
        print(f"  Integration Suite: {passed}/{total} passed")
        if failed:
            print("  Failed:")
            for n, d in failed:
                print(f"    • {n}  {d}")
        print("══════════════════════════════════════════════════════════════")
        return 0 if passed == total else 1


# ─── Helpers ──────────────────────────────────────────────────────────────────

def rand_phone() -> str:
    return f"+1{random.randint(2000000000, 9999999999)}"

def http_post(path: str, body: dict, token: Optional[str] = None) -> dict:
    data = json.dumps(body).encode()
    req  = urllib.request.Request(f"{BASE}{path}", data=data,
                                   headers={"Content-Type": "application/json"})
    if token:
        req.add_header("Authorization", f"Bearer {token}")
    with urllib.request.urlopen(req, timeout=10) as r:
        return json.loads(r.read())

def http_get(path: str, token: Optional[str] = None):
    req = urllib.request.Request(f"{BASE}{path}")
    if token:
        req.add_header("Authorization", f"Bearer {token}")
    try:
        with urllib.request.urlopen(req, timeout=10) as r:
            raw = r.read().decode("utf-8", errors="replace")
            try:
                return r.status, json.loads(raw)
            except Exception:
                return r.status, raw
    except urllib.error.HTTPError as e:
        return e.code, {}

def register(phone: str, name: str = "Test") -> dict:
    return http_post("/api/v1/auth/register", {"phone": phone, "display_name": name})

def ws_url(token: str) -> str:
    return f"{WS}?token={token}&endpoint_id={uuid.uuid4()}"

async def ws_recv_until(ws, method: str, timeout_s: float = 5.0) -> Optional[dict]:
    deadline = time.perf_counter() + timeout_s
    while time.perf_counter() < deadline:
        remaining = deadline - time.perf_counter()
        try:
            raw = await asyncio.wait_for(ws.recv(), timeout=max(0.05, remaining))
            msg = json.loads(raw)
            if msg.get("method") == method:
                return msg
        except asyncio.TimeoutError:
            break
        except Exception:
            break
    return None


# ─────────────────────────────────────────────────────────────────────────────
# SECTION 1 — Health & Readiness
# ─────────────────────────────────────────────────────────────────────────────

def test_health_and_metrics(s: Suite):
    print("\n━━  1. Health & Readiness  ━━")

    status, body = http_get("/health")
    s.check("GET /health → 200", status == 200, f"status={status}")

    status, body = http_get("/metrics")
    s.check("GET /metrics → 200", status == 200, f"status={status}")
    if isinstance(body, str):
        s.check("  Prometheus output present",
                "ucall_http_requests_total" in body and "ucall_active_connections" in body,
                body[:80])


# ─────────────────────────────────────────────────────────────────────────────
# SECTION 2 — Universal ID Properties
# ─────────────────────────────────────────────────────────────────────────────

def test_universal_id_properties(s: Suite):
    print("\n━━  2. Universal ID Properties  ━━")

    seen = set()
    for i in range(10):
        r = register(rand_phone(), f"UID-{i}")
        uid = r.get("data", {}).get("universal_id", "")
        s.check(f"  UID {i+1}: exactly 12 digits", len(uid) == 12, uid)
        s.check(f"  UID {i+1}: all digits", uid.isdigit(), uid)
        s.check(f"  UID {i+1}: unique", uid not in seen, uid)
        seen.add(uid)

    # Same phone always → same UID
    phone = rand_phone()
    r1 = register(phone, "Idem A")
    r2 = register(phone, "Idem B")
    uid1 = r1["data"]["universal_id"]
    uid2 = r2["data"]["universal_id"]
    s.check("  Idempotent: same phone → same UID", uid1 == uid2, f"{uid1}=={uid2}")


# ─────────────────────────────────────────────────────────────────────────────
# SECTION 3 — Identity Resolution
# ─────────────────────────────────────────────────────────────────────────────

def test_identity_resolution(s: Suite):
    print("\n━━  3. Identity Resolution  ━━")

    r    = register(rand_phone(), "Identity")
    uid  = r["data"]["universal_id"]
    tok  = r["data"]["access_token"]

    status, body = http_get(f"/api/v1/identity/{uid}", tok)
    s.check("  Resolve valid UID → 200", status == 200)
    if isinstance(body, dict):
        ret_uid = body.get("data", {}).get("universal_id") or body.get("data", {}).get("identity", {}).get("universal_id")
        s.check("  Returned UID matches", ret_uid == uid, f"{ret_uid}=={uid}")

    # Non-existent UID
    status, _ = http_get("/api/v1/identity/999999999999", tok)
    s.check("  Non-existent UID → non-200", status != 200, f"status={status}")

    # Malformed UID (11 digits)
    status, _ = http_get("/api/v1/identity/99999999999", tok)
    s.check("  Malformed UID (11 digits) → non-200", status != 200, f"status={status}")


# ─────────────────────────────────────────────────────────────────────────────
# SECTION 4 — JWT / Authentication
# ─────────────────────────────────────────────────────────────────────────────

def test_jwt_auth(s: Suite):
    print("\n━━  4. JWT Authentication  ━━")

    r   = register(rand_phone(), "JWT Test")
    tok = r["data"]["access_token"]
    ref = r["data"]["refresh_token"]
    uid = r["data"]["universal_id"]

    # Valid token works
    status, body = http_get(f"/api/v1/identity/{uid}", tok)
    s.check("  Valid token: identity access → 200", status == 200)

    # No token rejected
    status, _ = http_get(f"/api/v1/identity/{uid}")
    s.check("  No token: rejected (401/403/422)", status in (401, 403, 422), f"status={status}")

    # Invalid token rejected
    req = urllib.request.Request(f"{BASE}/api/v1/identity/{uid}")
    req.add_header("Authorization", "Bearer this.is.not.a.jwt")
    try:
        with urllib.request.urlopen(req, timeout=5) as res:
            s.check("  Invalid token: rejected", res.status != 200, f"status={res.status}")
    except urllib.error.HTTPError as e:
        s.check("  Invalid token: rejected", e.code in (401, 403, 422), f"status={e.code}")

    # Refresh token flow
    ref_resp = http_post("/api/v1/auth/refresh", {"refresh_token": ref})
    new_tok   = ref_resp.get("data", {}).get("access_token", "")
    s.check("  Refresh token → new access token", bool(new_tok), ref_resp.get("error", ""))


# ─────────────────────────────────────────────────────────────────────────────
# SECTION 5 — WebSocket Auth
# ─────────────────────────────────────────────────────────────────────────────

async def test_ws_auth(s: Suite):
    print("\n━━  5. WebSocket Authentication  ━━")

    r   = register(rand_phone(), "WS Auth")
    tok = r["data"]["access_token"]

    # Valid token
    try:
        async with websockets.connect(ws_url(tok), open_timeout=5) as ws:
            s.ok("  Valid token → WS connected")
    except Exception as e:
        s.fail("  Valid token → WS connected", str(e))

    # Invalid token
    try:
        bad_url = f"{WS}?token=bad.token.here&endpoint_id={uuid.uuid4()}"
        async with websockets.connect(bad_url, open_timeout=5):
            s.fail("  Invalid token → rejected", "connected unexpectedly")
    except websockets.exceptions.InvalidStatus as e:
        s.check("  Invalid token → 401", e.response.status_code == 401,
                f"status={e.response.status_code}")
    except Exception as e:
        s.ok("  Invalid token → rejected", str(e)[:60])

    # Missing endpoint_id
    try:
        no_ep_url = f"{WS}?token={tok}"
        async with websockets.connect(no_ep_url, open_timeout=5):
            s.fail("  Missing endpoint_id → rejected", "connected unexpectedly")
    except Exception as e:
        s.ok("  Missing endpoint_id → rejected", str(e)[:60])


# ─────────────────────────────────────────────────────────────────────────────
# SECTION 6 — Presence Lifecycle
# ─────────────────────────────────────────────────────────────────────────────

async def test_presence_lifecycle(s: Suite):
    print("\n━━  6. Presence Lifecycle  ━━")

    r   = register(rand_phone(), "Presence")
    tok = r["data"]["access_token"]

    # Connect → heartbeat → disconnect
    try:
        async with websockets.connect(ws_url(tok), open_timeout=5) as ws:
            # Heartbeat
            await ws.send(json.dumps({"method": "presence.heartbeat", "params": {}}))
            await asyncio.sleep(0.1)
            s.ok("  Heartbeat sent without error")
            # Still connected
            s.ok("  Connection alive after heartbeat")

        await asyncio.sleep(0.3)
        s.ok("  Disconnect completed cleanly")
    except Exception as e:
        s.fail("  Presence lifecycle", str(e))

    # Reconnect
    try:
        async with websockets.connect(ws_url(tok), open_timeout=5):
            s.ok("  Reconnect after disconnect: accepted")
    except Exception as e:
        s.fail("  Reconnect after disconnect", str(e))

    # Multi-device (3 endpoints, same user)
    sockets = []
    try:
        for _ in range(3):
            ws = await websockets.connect(ws_url(tok), open_timeout=5)
            sockets.append(ws)
        s.ok("  Multi-device: 3 simultaneous WS connections accepted")
    except Exception as e:
        s.fail("  Multi-device simultaneous connections", str(e))
    finally:
        for ws in sockets:
            await ws.close()


# ─────────────────────────────────────────────────────────────────────────────
# SECTION 7 — Full Call Signaling Flow
# ─────────────────────────────────────────────────────────────────────────────

async def test_call_signaling_flow(s: Suite):
    print("\n━━  7. Call Signaling Flow  ━━")

    r_a = register(rand_phone(), "Caller")
    r_b = register(rand_phone(), "Callee")
    tok_a = r_a["data"]["access_token"]
    tok_b = r_b["data"]["access_token"]
    uid_b = r_b["data"]["universal_id"]

    try:
        async with websockets.connect(ws_url(tok_a), open_timeout=5) as ws_a, \
                   websockets.connect(ws_url(tok_b), open_timeout=5) as ws_b:

            await asyncio.sleep(0.05)

            # A → call.invite
            t0 = time.perf_counter()
            await ws_a.send(json.dumps({
                "method": "call.invite",
                "params": {
                    "target_universal_id": uid_b,
                    "call_type": "audio",
                    "sdp_offer": "v=0\r\no=- 0 0 IN IP4 127.0.0.1\r\ns=-\r\n"
                }
            }))
            s.ok("  A sent call.invite")

            # B receives call.incoming
            incoming = await ws_recv_until(ws_b, "call.incoming", 5)
            call_id  = (incoming or {}).get("params", {}).get("call_id", "")
            s.check("  B received call.incoming", incoming is not None,
                    f"call_id={call_id}")

            if call_id:
                # B → call.accept
                await ws_b.send(json.dumps({
                    "method": "call.accept",
                    "params": {
                        "call_id": call_id,
                        "sdp_answer": "v=0\r\no=- 0 0 IN IP4 127.0.0.1\r\ns=-\r\n"
                    }
                }))
                s.ok("  B sent call.accept")

                # A receives call.connected
                connected = await ws_recv_until(ws_a, "call.connected", 5)
                rtt_ms = (time.perf_counter() - t0) * 1000
                s.check("  A received call.connected", connected is not None,
                        f"RTT={rtt_ms:.1f}ms")
                s.check("  Signaling RTT < 500ms (loopback)", rtt_ms < 500,
                        f"{rtt_ms:.1f}ms")

                # A → call.hangup
                await ws_a.send(json.dumps({
                    "method": "call.hangup",
                    "params": {"call_id": call_id, "reason": "completed"}
                }))
                s.ok("  A sent call.hangup")

        s.ok("  Both clients disconnected cleanly")

    except Exception as e:
        s.fail("  Call signaling flow", str(e))
        raise


async def test_call_to_self_rejected(s: Suite):
    print("\n━━  7b. Self-Call Rejection  ━━")

    r   = register(rand_phone(), "Self")
    tok = r["data"]["access_token"]
    uid = r["data"]["universal_id"]

    try:
        async with websockets.connect(ws_url(tok), open_timeout=5) as ws:
            await ws.send(json.dumps({
                "method": "call.invite",
                "params": {
                    "target_universal_id": uid,
                    "call_type": "audio",
                    "sdp_offer": ""
                }
            }))
            # Should not receive call.incoming back on same connection
            result = await ws_recv_until(ws, "call.incoming", 1.0)
            s.check("  Self-call: no call.incoming delivered", result is None,
                    f"got: {result}")
    except Exception as e:
        s.fail("  Self-call rejection", str(e))


async def test_call_offline_recipient(s: Suite):
    print("\n━━  7c. Offline Recipient  ━━")

    r_a   = register(rand_phone(), "Caller2")
    r_b   = register(rand_phone(), "Offline")   # B never connects
    tok_a = r_a["data"]["access_token"]
    uid_b = r_b["data"]["universal_id"]

    try:
        async with websockets.connect(ws_url(tok_a), open_timeout=5) as ws_a:
            await ws_a.send(json.dumps({
                "method": "call.invite",
                "params": {
                    "target_universal_id": uid_b,
                    "call_type": "audio",
                    "sdp_offer": "v=0\r\n"
                }
            }))
            await asyncio.sleep(0.5)
            s.ok("  Server handled offline recipient gracefully (no crash)")
    except Exception as e:
        s.fail("  Offline recipient handling", str(e))


# ─────────────────────────────────────────────────────────────────────────────
# SECTION 8 — Disconnect / Reconnect
# ─────────────────────────────────────────────────────────────────────────────

async def test_disconnect_reconnect(s: Suite):
    print("\n━━  8. Disconnect / Reconnect  ━━")

    r   = register(rand_phone(), "Reconnect")
    tok = r["data"]["access_token"]

    for attempt in range(3):
        try:
            async with websockets.connect(ws_url(tok), open_timeout=5) as ws:
                await ws.send(json.dumps({"method": "presence.heartbeat", "params": {}}))
            await asyncio.sleep(0.2)
            s.ok(f"  Connect/disconnect cycle {attempt+1}")
        except Exception as e:
            s.fail(f"  Connect/disconnect cycle {attempt+1}", str(e))


# ─────────────────────────────────────────────────────────────────────────────
# Main
# ─────────────────────────────────────────────────────────────────────────────

async def main() -> int:
    print()
    print("╔══════════════════════════════════════════════════════════════╗")
    print("║         UCall — Integration Test Suite                      ║")
    print("║         B-20+ Verification                                  ║")
    print("╚══════════════════════════════════════════════════════════════╝")
    print(f"  Target: {BASE}")

    s = Suite()

    # Sync sections
    test_health_and_metrics(s)
    test_universal_id_properties(s)
    test_identity_resolution(s)
    test_jwt_auth(s)

    # Async sections
    await test_ws_auth(s)
    await test_presence_lifecycle(s)
    await test_call_signaling_flow(s)
    await test_call_to_self_rejected(s)
    await test_call_offline_recipient(s)
    await test_disconnect_reconnect(s)

    return s.summary()


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))