import requests import json import uuid import time import sys BASE_URL = "http://127.0.0.1:8080" def test_mobile_architecture_flow(): print("--- 1. Testing Destination Resolution for Non-UCall Number ---") res = requests.post(f"{BASE_URL}/api/v1/communication/resolve", json={ "destination": "+919999988888", "requested_action": "audio_call", "client_platform": "android" }) assert res.status_code == 200, f"Expected 200, got {res.status_code}: {res.text}" body = res.json()["data"] print("Non-UCall destination resolution:", json.dumps(body, indent=2)) assert body["route"] == "native_cellular" assert body["action"] == "open_system_dialer" assert body["phone_number"] == "+919999988888" # Test SMS resolution res_sms = requests.post(f"{BASE_URL}/api/v1/communication/resolve", json={ "destination": "+919999988888", "requested_action": "sms", "client_platform": "ios" }) assert res_sms.status_code == 200 sms_body = res_sms.json()["data"] print("Non-UCall SMS resolution:", json.dumps(sms_body, indent=2)) assert sms_body["route"] == "native_cellular" assert sms_body["action"] == "open_native_sms" print("\n--- 2. Registering Two UCall Users for End-to-End Hybrid Testing ---") suffix = str(int(time.time()))[-6:] phone1 = f"+91981{suffix}01" phone2 = f"+91981{suffix}02" u1_reg = requests.post(f"{BASE_URL}/api/v1/auth/register", json={ "phone_number": phone1, "display_name": "Ravi Native", "device_name": "Pixel 8 Pro", "device_platform": "android" }).json()["data"] u2_reg = requests.post(f"{BASE_URL}/api/v1/auth/register", json={ "phone_number": phone2, "display_name": "Priya Mobile", "device_name": "iPhone 15 Pro", "device_platform": "ios" }).json()["data"] u1_uid = u1_reg["universal_id"] u2_uid = u2_reg["universal_id"] token1 = u1_reg["access_token"] token2 = u2_reg["access_token"] print(f"Registered User 1: Universal ID={u1_uid}, Phone={phone1}") print(f"Registered User 2: Universal ID={u2_uid}, Phone={phone2}") print("\n--- 3. Testing Destination Resolution for UCall User (via Phone & UID) ---") # Resolve using phone number res_user = requests.post(f"{BASE_URL}/api/v1/communication/resolve", json={ "destination": phone2, "requested_action": "audio_call", "client_platform": "android" }, headers={"Authorization": f"Bearer {token1}"}) assert res_user.status_code == 200, res_user.text u2_resolved = res_user.json()["data"] print("UCall User resolution by phone:", json.dumps(u2_resolved, indent=2)) assert u2_resolved["route"] == "ucall_internet" assert u2_resolved["action"] == "create_ucall_session" assert u2_resolved["recipient"]["universal_id"] == u2_uid assert u2_resolved["capabilities"]["audio"] is True assert u2_resolved["capabilities"]["video"] is True # Resolve using 12-digit Universal ID res_uid = requests.post(f"{BASE_URL}/api/v1/communication/resolve", json={ "destination": u2_uid, "requested_action": "video_call", "client_platform": "ios" }, headers={"Authorization": f"Bearer {token1}"}) assert res_uid.status_code == 200 assert res_uid.json()["data"]["route"] == "ucall_internet" assert res_uid.json()["data"]["capabilities"]["video"] is True print("\n--- 4. Testing Privacy-Preserving Contact Synchronization & Matching ---") sync_res = requests.post(f"{BASE_URL}/api/v1/contacts/sync", json={ "contacts": [ { "client_contact_id": "contact-local-001", "phone_number": phone2, "contact_name": "Priya Mobile" }, { "client_contact_id": "contact-local-002", "phone_number": "+919111122222", "contact_name": "External Plumber" } ] }, headers={"Authorization": f"Bearer {token1}"}) assert sync_res.status_code == 200, sync_res.text matches = sync_res.json()["data"]["matches"] print("Contact Sync Results:", json.dumps(matches, indent=2)) assert len(matches) == 2 m1 = next(m for m in matches if m["client_contact_id"] == "contact-local-001") assert m1["is_ucall_user"] is True assert m1["universal_id"] == u2_uid assert m1["capabilities"]["internet_audio"] is True m2 = next(m for m in matches if m["client_contact_id"] == "contact-local-002") assert m2["is_ucall_user"] is False assert m2["capabilities"]["native_cellular_call"] is True assert m2["capabilities"]["native_sms"] is True print("\n--- 5. Testing Internet Call Creation, Rejection, and Cancellation ---") # Initiate call from U1 to U2 using Bearer auth and recipient_uid call_res = requests.post(f"{BASE_URL}/api/v1/calls", json={ "recipient_uid": u2_uid, "call_type": "audio" }, headers={"Authorization": f"Bearer {token1}"}) assert call_res.status_code == 200, call_res.text call_id = call_res.json()["data"]["call_id"] print(f"Initiated call_id: {call_id}, status: {call_res.json()['data']['status']}") # Reject call reject_res = requests.post(f"{BASE_URL}/api/v1/calls/{call_id}/reject", json={ "reason": "user_declined" }) assert reject_res.status_code == 200, reject_res.text rejected_session = reject_res.json()["data"] print(f"Call {call_id} rejected, status: {rejected_session['status']}, reason: {rejected_session['end_reason']}") assert rejected_session["status"] == "rejected" # Start another call and cancel it call2_res = requests.post(f"{BASE_URL}/api/v1/calls", json={ "recipient_uid": u2_uid, "call_type": "video" }, headers={"Authorization": f"Bearer {token1}"}) assert call2_res.status_code == 200, call2_res.text call2_id = call2_res.json()["data"]["call_id"] cancel_res = requests.post(f"{BASE_URL}/api/v1/calls/{call2_id}/cancel") assert cancel_res.status_code == 200, cancel_res.text assert cancel_res.json()["data"]["status"] == "ended" print("\n--- 6. Testing Call History API ---") history_res = requests.get(f"{BASE_URL}/api/v1/calls/history", headers={ "Authorization": f"Bearer {token1}" }) assert history_res.status_code == 200, history_res.text history = history_res.json()["data"] print(f"Retrieved {len(history)} call history records for caller:") for h in history[:3]: print(f" - Call {h['call_id']}: direction={h['direction']}, type={h['call_type']}, status={h['status']}, peer_uid={h['peer_universal_id']}") assert len(history) >= 2 assert any(h["call_id"] == call_id for h in history) assert any(h["call_id"] == call2_id for h in history) print("\n==========================================") print("ALL MOBILE CALLING & ROUTING TESTS PASSED!") print("==========================================") if __name__ == "__main__": test_mobile_architecture_flow()