#!/usr/bin/env python3
"""
UCall Control-Plane E2E Test — Verified against live server
Uses real JWT tokens from registered users, generates ephemeral endpoint UUIDs.
"""

import asyncio
import json
import time
import sys
import uuid
import urllib.request
import urllib.error
from typing import Optional

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

SERVER_HTTP = "http://127.0.0.1:8080"
SERVER_WS   = "ws://127.0.0.1:8080/api/v1/ws"

# Real tokens from registration (refreshed)
TOKEN_A  = "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJmYmFmZTBmNS03M2Q3LTRlYTUtYjc1OS1kZjVkNjA1NmRlNDYiLCJ1bml2ZXJzYWxfaWQiOiI5MDI3OTQ2NzMwOTUiLCJhcHBfaWQiOm51bGwsImVuZHBvaW50X2lkIjpudWxsLCJyb2xlIjoidXNlciIsInRva2VuX3R5cGUiOiJhY2Nlc3MiLCJleHAiOjE3OTA5MjM0OTcsImlhdCI6MTc5MDkyMjU5N30.itNOhIvbdQXGq-QIxRqlUE75rdy2e10N3yKhk8VjKE4"
TOKEN_B  = "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJiZjU5MTlhOC1iNzBhLTQxZDctYjk2MS1hZjMzZTU0MTBmOTgiLCJ1bml2ZXJzYWxfaWQiOiIyODc3MTE5MDg2MDMiLCJhcHBfaWQiOm51bGwsImVuZHBvaW50X2lkIjpudWxsLCJyb2xlIjoidXNlciIsInRva2VuX3R5cGUiOiJhY2Nlc3MiLCJleHAiOjE3OTA5MjM0OTcsImlhdCI6MTc5MDkyMjU5N30.g_hXTS_ChAbTzX89CmOnESVbv3Bltof-vCluBdRBq40"
UID_A    = "902794673095"
UID_B    = "287711908603"

PASS_T = "✅"
FAIL_T = "❌"
INFO_T = "ℹ️ "
results: list[tuple[str, bool]] = []


def check(label: str, ok: bool, detail: str = "") -> bool:
    icon = PASS_T if ok else FAIL_T
    results.append((label, ok))
    suffix = f"  ({detail})" if detail else ""
    print(f"  {icon}  {label}{suffix}")
    return ok


# ─────────────────────────────────────────────────────────────────────────────
# HTTP checks
# ─────────────────────────────────────────────────────────────────────────────
def http_get(path: str, token: Optional[str] = None):
    url = f"{SERVER_HTTP}{path}"
    req = urllib.request.Request(url)
    if token:
        req.add_header("Authorization", f"Bearer {token}")
    with urllib.request.urlopen(req, timeout=5) as r:
        return r.status, r.read().decode()


def verify_http():
    print("\n━━  HTTP Endpoints  ━━")

    # /health
    try:
        status, body = http_get("/health")
        check("GET /health → 200", status == 200, f"status={status}, body={body.strip()!r}")
    except Exception as e:
        check("GET /health → 200", False, str(e))

    # /metrics
    try:
        status, body = http_get("/metrics")
        has_connections = "ucall_active_connections" in body
        has_requests    = "ucall_requests_total" in body
        check("GET /metrics → 200", status == 200, f"status={status}")
        check("  metrics has ucall_active_connections", has_connections)
        check("  metrics has ucall_requests_total",    has_requests)
    except Exception as e:
        check("GET /metrics → 200", False, str(e))


# ─────────────────────────────────────────────────────────────────────────────
# WebSocket E2E signaling
# ─────────────────────────────────────────────────────────────────────────────
async def verify_signaling():
    print("\n━━  WebSocket Control-Plane Signaling  ━━")

    ep_id_a = str(uuid.uuid4())
    ep_id_b = str(uuid.uuid4())

    def ws_url(token: str, ep_id: str) -> str:
        return f"{SERVER_WS}?token={token}&endpoint_id={ep_id}"

    call_id       = None
    invite_ts     = None
    connected_ts  = None

    try:
        print(f"  {INFO_T} Connecting A  (uid={UID_A}, ep={ep_id_a[:8]}…)")
        async with websockets.connect(ws_url(TOKEN_A, ep_id_a), open_timeout=5) as ws_a:
            check("Client A WebSocket connected", True)

            print(f"  {INFO_T} Connecting B  (uid={UID_B}, ep={ep_id_b[:8]}…)")
            async with websockets.connect(ws_url(TOKEN_B, ep_id_b), open_timeout=5) as ws_b:
                check("Client B WebSocket connected", True)

                # ── Presence heartbeats ────────────────────────────────────
                await ws_a.send(json.dumps({"method": "presence.heartbeat", "params": {}}))
                await ws_b.send(json.dumps({"method": "presence.heartbeat", "params": {}}))
                await asyncio.sleep(0.05)
                check("A and B sent presence heartbeat", True)

                # ── A → call.invite → B ───────────────────────────────────
                invite_ts = 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"
                    }
                }))
                check("A sent call.invite", True)

                # ── B waits for call.incoming ──────────────────────────────
                try:
                    raw = await asyncio.wait_for(ws_b.recv(), timeout=5.0)
                    msg = json.loads(raw)
                    call_id = msg.get("params", {}).get("call_id") or msg.get("call_id")
                    is_incoming = msg.get("method") == "call.incoming"
                    check("B received call.incoming", is_incoming,
                          f"method={msg.get('method')}, call_id={call_id}")
                except asyncio.TimeoutError:
                    check("B received call.incoming", False, "timeout 5s")

                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"
                        }
                    }))
                    check("B sent call.accept", True)

                    # ── A waits for call.connected ─────────────────────────
                    # A may receive call.ringing first (invite ack), then call.connected
                    try:
                        is_connected = False
                        connected_ts = None
                        deadline = time.perf_counter() + 5.0
                        while time.perf_counter() < deadline:
                            remaining = deadline - time.perf_counter()
                            raw = await asyncio.wait_for(ws_a.recv(), timeout=max(0.1, remaining))
                            msg = json.loads(raw)
                            if msg.get("method") == "call.connected":
                                is_connected = True
                                connected_ts = time.perf_counter()
                                break
                        check("A received call.connected", is_connected,
                              f"last_method={msg.get('method')}" if not is_connected else "")

                        if is_connected and invite_ts and connected_ts:
                            rtt_ms = (connected_ts - invite_ts) * 1000
                            check(f"Signaling RTT < 500ms (loopback)", rtt_ms < 500,
                                  f"{rtt_ms:.1f}ms")
                    except asyncio.TimeoutError:
                        check("A received call.connected", False, "timeout 5s")

                # ── B disconnects ──────────────────────────────────────────
                await ws_b.close()
                await asyncio.sleep(0.3)
                check("B disconnected from WebSocket", True)

                # ── Check Redis flushed B's presence ──────────────────────
                import subprocess
                result = subprocess.run(
                    ["redis-cli", "keys", f"presence:*"],
                    capture_output=True, text=True, timeout=3
                )
                presence_keys = result.stdout.strip()
                b_offline = ep_id_b not in presence_keys
                check("B presence key removed from Redis after disconnect",
                      b_offline, f"remaining keys: {presence_keys!r}")

    except websockets.exceptions.InvalidStatus as e:
        check("WebSocket handshake succeeded", False, str(e))
    except OSError as e:
        check("Server reachable", False, str(e))
    except Exception as e:
        check("E2E execution error", False, type(e).__name__ + ": " + str(e))
        raise


# ─────────────────────────────────────────────────────────────────────────────
# Identity resolution
# ─────────────────────────────────────────────────────────────────────────────
def verify_identity():
    print("\n━━  Identity Resolution  ━━")
    for uid in [UID_A, UID_B]:
        try:
            status, body = http_get(f"/api/v1/identity/{uid}", TOKEN_A)
            data = json.loads(body)
            check(f"GET /identity/{uid}", status == 200 and data.get("success"),
                  f"status={status}")
        except Exception as e:
            check(f"GET /identity/{uid}", False, str(e))


# ─────────────────────────────────────────────────────────────────────────────
# Main
# ─────────────────────────────────────────────────────────────────────────────
async def main():
    print()
    print("╔══════════════════════════════════════════════════════╗")
    print("║      UCall Control-Plane — B-20 Verification        ║")
    print("╚══════════════════════════════════════════════════════╝")
    print(f"  Server : {SERVER_HTTP}")
    print(f"  WS     : {SERVER_WS}")
    print(f"  Client A uid={UID_A}")
    print(f"  Client B uid={UID_B}")

    verify_http()
    verify_identity()
    await verify_signaling()

    # ── Final summary ─────────────────────────────────────────────────────
    passed = sum(1 for _, ok in results if ok)
    total  = len(results)
    failed_labels = [label for label, ok in results if not ok]

    print()
    print("══════════════════════════════════════════════════════")
    print(f"  {passed}/{total} checks passed")
    if passed == total:
        print("  B-20 Control-Plane: ✅  ALL PASS")
    else:
        print(f"  Failed checks:")
        for lbl in failed_labels:
            print(f"    • {lbl}")
        print("  B-20 Control-Plane: ⚠️   PARTIAL")
    print("══════════════════════════════════════════════════════")

    return 0 if passed == total else 1


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