import urllib.request
import urllib.error
import json
import time
import uuid
import sys
import threading
import websockets
import asyncio
import pymysql

BASE_URL = "http://127.0.0.1:8080"
WS_URL = "ws://127.0.0.1:8080/api/v1/ws"
import os

def load_env_db_config():
    pw = os.environ.get("DB_PASSWORD", "")
    if not pw and os.path.exists(".env"):
        with open(".env") as f:
            for line in f:
                if line.startswith("DB_PASSWORD="):
                    pw = line.strip().split("=", 1)[1]
    return {
        "host": os.environ.get("DB_HOST", "94.136.190.60"),
        "user": os.environ.get("DB_USER", "chatpingdart_ravi"),
        "password": pw,
        "database": os.environ.get("DB_NAME", "chatpingdart_ucall"),
    }

DB_CONFIG = load_env_db_config()

def http_req(path, method="GET", payload=None, token=None, headers_extra=None):
    url = f"{BASE_URL}{path}"
    data = json.dumps(payload).encode() if payload is not None else None
    headers = {"Content-Type": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    if headers_extra:
        headers.update(headers_extra)
    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            return resp.status, json.loads(resp.read().decode())
    except urllib.error.HTTPError as e:
        body = e.read().decode()
        try:
            return e.code, json.loads(body)
        except:
            return e.code, body

print("=" * 70)
print("UCALL PRODUCTION HARDENING VERIFICATION SUITE")
print("=" * 70)

# 1. Verify Database Schema: UNIQUE KEY on token_hash
print("\n[TEST 1] Verifying UNIQUE constraint on auth_sessions.token_hash...")
conn = pymysql.connect(**DB_CONFIG)
cur = conn.cursor()
cur.execute("SHOW INDEX FROM auth_sessions WHERE Column_name = 'token_hash'")
indexes = cur.fetchall()
is_unique = any(idx[1] == 0 for idx in indexes)
assert is_unique, f"token_hash is NOT unique! {indexes}"
print("  ✓ auth_sessions.token_hash has strict UNIQUE KEY in remote MariaDB")
conn.close()

# 2. Register Two Test Users
print("\n[TEST 2] Registering Test Users for session & calling tests...")
phone_a = f"+1{int(time.time()) % 10000000000:010d}"
phone_b = f"+1{(int(time.time()) + 1) % 10000000000:010d}"

st_a, res_a = http_req("/api/v1/auth/register", "POST", {"phone": phone_a, "display_name": "User Alpha"})
assert st_a == 200, f"Register A failed: {res_a}"
user_a = res_a["data"]

st_b, res_b = http_req("/api/v1/auth/register", "POST", {"phone": phone_b, "display_name": "User Beta"})
assert st_b == 200, f"Register B failed: {res_b}"
user_b = res_b["data"]

print(f"  ✓ User Alpha registered: uid={user_a['universal_id']}, id={user_a['user_id']}")
print(f"  ✓ User Beta registered:  uid={user_b['universal_id']}, id={user_b['user_id']}")

# 3. Test Transactional Refresh-Token Rotation & Concurrent Reuse Detection
print("\n[TEST 3] Testing Concurrent Refresh Token Rotation & Family Revocation...")
ref_token = user_a["refresh_token"]

results = []
def do_refresh():
    st, res = http_req("/api/v1/auth/refresh", "POST", {"refresh_token": ref_token})
    results.append((st, res))

t1 = threading.Thread(target=do_refresh)
t2 = threading.Thread(target=do_refresh)

t1.start()
t2.start()
t1.join()
t2.join()

statuses = [r[0] for r in results]
print(f"  Concurrent refresh results: statuses={statuses}")
has_reuse_detection = any(r[0] == 401 for r in results)
print(f"  ✓ Atomic reuse detection triggered on race: {has_reuse_detection}")
assert has_reuse_detection, f"Concurrent reuse was not detected! Results: {results}"

# If reuse detected, subsequent attempt with either token must fail
st_sub, res_sub = http_req("/api/v1/auth/refresh", "POST", {"refresh_token": ref_token})
assert st_sub == 401, f"Expected 401 for revoked token family, got {st_sub}: {res_sub}"
print("  ✓ Full token family successfully invalidated upon reuse detection")

# 4. Test Single-Use WebSocket Tickets (No JWT in query string logs)
print("\n[TEST 4] Testing Single-Use WebSocket Ticket (Zero URL Token Leak)...")
# Log in user B again to get fresh tokens
st_login, res_login = http_req("/api/v1/auth/login", "POST", {"identifier": phone_b})
assert st_login == 200, f"Login failed: {res_login}"
user_b = res_login["data"]

# Generate an endpoint for User B
endpoint_b_id = str(uuid.uuid4())

# Request a single-use WS ticket
st_tick, res_tick = http_req("/api/v1/auth/ws-ticket", "POST", {
    "endpoint_id": endpoint_b_id
}, token=user_b["access_token"])
assert st_tick == 200, f"Failed to get ticket: {res_tick}"
ws_ticket = res_tick["data"]["ticket"]
assert res_tick["data"]["expires_in"] == 60
print(f"  ✓ Obtained single-use WebSocket ticket: {ws_ticket}")

# Connect via WebSocket with ticket
async def test_ws_ticket():
    uri = f"{WS_URL}?ticket={ws_ticket}&endpoint_id={endpoint_b_id}"
    async with websockets.connect(uri) as ws:
        await ws.send(json.dumps({"type": "ping", "payload": {}}))
        await asyncio.sleep(0.2)
    print("  ✓ Successfully connected to WebSocket via single-use ticket!")

    # Attempt to REUSE the ticket - MUST FAIL with HTTP 401
    try:
        async with websockets.connect(uri):
            raise AssertionError("Reused ticket connection succeeded unexpectedly!")
    except Exception as e:
        code = getattr(e, "status_code", None)
        if code is None and hasattr(e, "response"):
            code = getattr(e.response, "status_code", None)
        assert code == 401 or "401" in str(e), f"Expected 401 for reused ticket, got {e}"
        print(f"  ✓ Ticket reuse correctly rejected with status 401 (Atomic Single-Use Verified)")

asyncio.run(test_ws_ticket())

# 5. Test MySQL Idempotency for Calls & Messages
print("\n[TEST 5] Testing MySQL Idempotency for Calls and Messages...")
# User B calling User A
idem_key_call = f"idem-call-{uuid.uuid4()}"
call_payload = {
    "callee_universal_id": user_a["universal_id"],
    "call_type": "audio"
}

st_c1, res_c1 = http_req("/api/v1/calls", "POST", call_payload, token=user_b["access_token"], headers_extra={"Idempotency-Key": idem_key_call})
assert st_c1 == 200, f"Call 1 failed: {res_c1}"
call_id_1 = res_c1["data"]["call_id"]

st_c2, res_c2 = http_req("/api/v1/calls", "POST", call_payload, token=user_b["access_token"], headers_extra={"Idempotency-Key": idem_key_call})
assert st_c2 == 200, f"Call 2 failed: {res_c2}"
call_id_2 = res_c2["data"]["call_id"]

assert call_id_1 == call_id_2, f"Idempotency failed! Created duplicate call IDs: {call_id_1} vs {call_id_2}"
print(f"  ✓ Call idempotency verified: identical call_id returned ({call_id_1})")

# Test Message Idempotency
idem_key_msg = f"idem-msg-{uuid.uuid4()}"
msg_payload = {
    "recipient_uid": user_a["universal_id"],
    "message_type": "text",
    "payload": "Test idempotent message payload",
}

st_m1, res_m1 = http_req("/api/v1/messages", "POST", msg_payload, token=user_b["access_token"], headers_extra={"Idempotency-Key": idem_key_msg})
assert st_m1 == 200, f"Msg 1 failed: {res_m1}"
msg_id_1 = res_m1["data"]["id"]

st_m2, res_m2 = http_req("/api/v1/messages", "POST", msg_payload, token=user_b["access_token"], headers_extra={"Idempotency-Key": idem_key_msg})
assert st_m2 == 200, f"Msg 2 failed: {res_m2}"
msg_id_2 = res_m2["data"]["id"]

assert msg_id_1 == msg_id_2, f"Message idempotency failed! Duplicate IDs: {msg_id_1} vs {msg_id_2}"
print(f"  ✓ Message idempotency verified: identical message_id returned ({msg_id_1})")

# 6. Test Conversation Scoping, Authorization & Pagination
print("\n[TEST 6] Testing Conversation Authorization, Scoping & Pagination...")
st_conv, res_conv = http_req("/api/v1/conversations?limit=10&offset=0", "GET", token=user_b["access_token"])
assert st_conv == 200, f"Conversations failed: {res_conv}"
print(f"  ✓ Retrieved {len(res_conv['data'])} conversations for authenticated user with pagination")

st_hist, res_hist = http_req(f"/api/v1/messages/history?user2={user_a['universal_id']}&limit=5&offset=0", "GET", token=user_b["access_token"])
assert st_hist == 200, f"History failed: {res_hist}"
print(f"  ✓ Retrieved message history with limit & offset pagination")

# 7. Test WebRTC Ephemeral Coturn TURN Credentials
print("\n[TEST 7] Testing Ephemeral WebRTC TURN Credentials Generation...")
st_turn_gen, res_turn_gen = http_req("/api/v1/turn-credentials", "GET", token=user_b["access_token"])
assert st_turn_gen == 200, f"General TURN credentials failed: {res_turn_gen}"
data_turn = res_turn_gen["data"]
assert "ice_servers" in data_turn, "Missing ice_servers"
assert len(data_turn["ice_servers"]) >= 2, "Expected at least STUN and TURN server entries"
turn_entry = [s for s in data_turn["ice_servers"] if any("turn:" in u for u in s["urls"])][0]
assert turn_entry["username"] is not None
assert turn_entry["credential"] is not None
print(f"  ✓ General TURN Credentials: username={turn_entry['username']}, ttl={data_turn['ttl']}")

# Call-specific TURN credentials with ownership verification
st_turn_call, res_turn_call = http_req(f"/api/v1/calls/{call_id_1}/turn-credentials", "GET", token=user_b["access_token"])
assert st_turn_call == 200, f"Call TURN credentials failed: {res_turn_call}"
print(f"  ✓ Call-specific TURN Credentials verified for call participant {user_b['user_id']}")

# 8. Test Session Revocation & Logout
print("\n[TEST 8] Testing Logout and Session Revocation Enforcement...")
st_logout, res_logout = http_req("/api/v1/auth/logout", "POST", token=user_b["access_token"])
assert st_logout == 200, f"Logout failed: {res_logout}"
print("  ✓ User successfully logged out (all sessions revoked)")

# Verify that ticket creation is now rejected
st_tick_after, res_tick_after = http_req("/api/v1/auth/ws-ticket", "POST", token=user_b["access_token"])
assert st_tick_after == 401, f"Expected 401 after logout, got {st_tick_after}: {res_tick_after}"
print("  ✓ WebSocket ticket request correctly denied for revoked session (HTTP 401)")

print("\n" + "=" * 70)
print("ALL PRODUCTION HARDENING TESTS PASSED (100%)")
print("=" * 70)