IRIS Legacy

""" Test Hybrid Booking Workflow Integration. This script tests the complete hybrid booking workflow: 1. Search for internal attendees 2. Find available meeting times 3. Create hybrid booking session 4. Send invitation email (from organizer's account) 5. Check slot availability (simulating external user) 6. Confirm booking (simulating external user selection) Architecture: MCP-first, interface-agnostic """ import asyncio import sys import json from pathlib import Path from datetime import datetime, timedelta # Add src to path sys.path.insert(0, str(Path(__file__).parent / 'src')) from mcp_servers.mcp_client import MCPClient, MCPError async def test_hybrid_booking_workflow(): """Test complete hybrid booking workflow.""" print("=" * 60) print("🧪 TEST HYBRID BOOKING WORKFLOW") print("=" * 60) # Initialize client client = MCPClient(base_url="http://localhost:8001") print(f"\n✅ MCP Client initialized: {client.base_url}") # Test data - APPROVED ACCOUNTS ONLY (see TEST_ACCOUNTS.md) # ORGANIZER: Determined by OAuth token (authenticated user) organizer_email = None organizer_name = None internal_attendees = [] # External email - using APPROVED test account external_email = "filippo.savarese@gmail.com" external_name = "Filippo External" session_id = None # ======================================== # TEST 1: Health Check # ======================================== print("\n" + "─" * 60) print("TEST 1: Health Check") print("─" * 60) try: health = await client.health_check() print(f"✅ Health: {health.get('status')}") print(f" Service: {health.get('service')}") except Exception as e: print(f"❌ Health check failed: {e}") return False # ======================================== # TEST 2: Get MCP Info (check bookings area) # ======================================== print("\n" + "─" * 60) print("TEST 2: MCP Info (verify bookings area)") print("─" * 60) try: info = await client.get_mcp_info() areas = info.get('areas', {}) if 'bookings_operations' in areas: bookings_info = areas['bookings_operations'] actions = bookings_info.get('actions', []) print(f"✅ Bookings area found with {len(actions)} actions") # Check for hybrid booking actions hybrid_actions = [ 'create_hybrid_session', 'send_hybrid_invitation', 'get_hybrid_session', 'check_hybrid_slot', 'confirm_hybrid_booking', 'find_new_hybrid_slots' ] found_actions = [action for action in actions if action in hybrid_actions] print(f" Hybrid booking actions: {len(found_actions)}/6") for action in found_actions: print(f" ✓ {action}") if len(found_actions) != 6: print(f" ⚠️ Missing actions: {set(hybrid_actions) - set(found_actions)}") else: print("❌ Bookings area not found in MCP server!") return False except Exception as e: print(f"❌ MCP info failed: {e}") return False # ======================================== # TEST 3: Get authenticated user (OAuth token owner) # ======================================== print("\n" + "─" * 60) print("TEST 3: Get Authenticated User (Organizer)") print("─" * 60) try: auth_user_data = await client.get_authenticated_user() organizer = auth_user_data.get('user', {}) organizer_email = organizer.get('mail') or organizer.get('userPrincipalName') organizer_name = organizer.get('displayName', 'Unknown') print(f"✅ Authenticated user (token owner):") print(f" Name: {organizer_name}") print(f" Email: {organizer_email}") print(f" ID: {organizer.get('id')}") print(f" → This user IS the organizer!") if not organizer_email: print("❌ Could not determine organizer email") return False except Exception as e: print(f"❌ Failed to get authenticated user: {e}") return False # ======================================== # TEST 4: Search for internal attendees (approved accounts) # ======================================== print("\n" + "─" * 60) print("TEST 4: Search Internal Attendees (approved accounts)") print("─" * 60) try: # Search for Daniele (approved account) results = await client.search_users(query="Daniele Castellari", max_results=3) if results: print(f"✅ Found {len(results)} potential attendees:") for idx, user in enumerate(results[:2], 1): name = user.get('displayName', 'N/A') email = user.get('mail') or user.get('userPrincipalName', 'N/A') print(f" {idx}. {name} ({email})") # Add first result as internal attendee (Daniele) if idx == 1 and 'daniele.castellari@infocert.it' in email.lower(): internal_attendees.append({ "email": email, "name": name }) # Optionally search for Giovanni Albero (approved) if len(internal_attendees) < 2: results2 = await client.search_users(query="Giovanni Albero", max_results=2) if results2: for user in results2[:1]: email = user.get('mail') or user.get('userPrincipalName', 'N/A') name = user.get('displayName', 'N/A') if 'giovanni.albero@infocert.it' in email.lower(): internal_attendees.append({ "email": email, "name": name }) print(f" {len(internal_attendees)}. {name} ({email})") if not internal_attendees: print("⚠️ No internal attendees found - continuing with organizer only") else: print(f"✅ Using {len(internal_attendees)} internal attendee(s)") except Exception as e: print(f"⚠️ Search attendees failed: {e}") # ======================================== # TEST 5: Find Available Meeting Times # ======================================== print("\n" + "─" * 60) print("TEST 5: Find Available Meeting Times") print("─" * 60) try: # Collect all attendee emails attendee_emails = [organizer_email] if internal_attendees: attendee_emails.extend([att['email'] for att in internal_attendees]) print(f" Checking availability for {len(attendee_emails)} attendees:") for email in attendee_emails: print(f" • {email}") # Find 3 available slots suggestions = await client.find_meeting_times( attendees=attendee_emails, duration_minutes=30, max_results=3 ) if suggestions and len(suggestions) >= 3: print(f"\n✅ Found {len(suggestions)} available slots:") proposed_slots = [] for idx, slot in enumerate(suggestions[:3], 1): start = slot.get('start') end = slot.get('end') score = slot.get('confidence', 0) print(f" {idx}. {start} → {end} (confidence: {score})") proposed_slots.append({ "start": start, "end": end }) else: print("❌ Not enough available slots found (need 3)") return False except Exception as e: print(f"❌ Find meeting times failed: {e}") return False # ======================================== # TEST 6: Create Hybrid Booking Session # ======================================== print("\n" + "─" * 60) print("TEST 6: Create Hybrid Booking Session") print("─" * 60) try: session = await client.create_booking_session( organizer_email=organizer_email, organizer_name=organizer_name, internal_attendees=internal_attendees, external_email=external_email, external_name=external_name, proposed_slots=proposed_slots, meeting_subject="Product Demo - Test Booking", meeting_duration=30 ) session_id = session.get('session_id') booking_url = session.get('booking_url') print(f"✅ Session created successfully:") print(f" Session ID: {session_id}") print(f" Booking URL: {booking_url}") print(f" Status: {session.get('status')}") print(f" Expires at: {session.get('expires_at')}") except MCPError as e: print(f"❌ MCP Error: {e.code} - {e}") print(f" Details: {e.details}") return False except Exception as e: print(f"❌ Create session failed: {e}") return False # ======================================== # TEST 7: Send Hybrid Invitation Email # ======================================== print("\n" + "─" * 60) print("TEST 7: Send Invitation Email (from organizer)") print("─" * 60) try: result = await client.send_booking_invitation(session_id=session_id) print(f"✅ Invitation sent successfully:") print(f" Status: {result.get('status')}") print(f" Recipient: {result.get('recipient')}") print(f" From: {organizer_email} (organizer's account)") print(f" Subject: {result.get('subject', 'Meeting Invitation')}") except MCPError as e: print(f"❌ MCP Error: {e.code} - {e}") print(f" Details: {e.details}") return False except Exception as e: print(f"❌ Send invitation failed: {e}") return False # ======================================== # TEST 8: Get Booking Session (verify creation) # ======================================== print("\n" + "─" * 60) print("TEST 8: Get Booking Session (verify)") print("─" * 60) try: session = await client.get_booking_session(session_id=session_id) print(f"✅ Session retrieved successfully:") print(f" Status: {session.get('status')}") print(f" Organizer: {session.get('organizer_name')}") print(f" External: {session.get('external_email')}") print(f" Proposed slots: {len(session.get('proposed_slots', []))}") except MCPError as e: print(f"❌ MCP Error: {e.code} - {e}") print(f" Details: {e.details}") return False except Exception as e: print(f"❌ Get session failed: {e}") return False # ======================================== # TEST 9: Check Slot Availability (real-time) # ======================================== print("\n" + "─" * 60) print("TEST 9: Check Slot Availability (simulate external user)") print("─" * 60) try: print(f" Checking availability for all 3 slots...") available_slots = [] for idx in range(3): availability = await client.check_slot_availability( session_id=session_id, slot_index=idx ) is_available = availability.get('available') slot = availability.get('slot', {}) confidence = availability.get('confidence', 0) status_icon = "✓" if is_available else "✗" print(f" Slot {idx+1}: {status_icon} {slot.get('start')} (confidence: {confidence}%)") if is_available: available_slots.append(idx) if available_slots: print(f"\n✅ {len(available_slots)} slot(s) available for booking") selected_slot_index = available_slots[0] # Select first available else: print("\n⚠️ No slots available - this is expected if calendars are busy") # Still continue test with slot 0 selected_slot_index = 0 except MCPError as e: print(f"❌ MCP Error: {e.code} - {e}") print(f" Details: {e.details}") return False except Exception as e: print(f"❌ Check availability failed: {e}") return False # ======================================== # TEST 10: Confirm Booking (create calendar event) # ======================================== print("\n" + "─" * 60) print("TEST 10: Confirm Booking (create calendar event)") print("─" * 60) try: print(f" Confirming slot {selected_slot_index + 1}...") result = await client.confirm_booking( session_id=session_id, selected_slot_index=selected_slot_index ) print(f"\n✅ Booking confirmed successfully:") print(f" Status: {result.get('status')}") print(f" Event ID: {result.get('event_id')}") print(f" Selected slot: {result.get('selected_slot', {}).get('start')}") print(f" Web link: {result.get('web_link', 'N/A')[:50]}...") if result.get('online_meeting_url'): print(f" Teams link: {result.get('online_meeting_url')[:50]}...") except MCPError as e: print(f"❌ MCP Error: {e.code} - {e}") print(f" Details: {e.details}") # If error is about slot not available, this is expected behavior if 'not available' in str(e).lower(): print(f" ⚠️ This is expected if calendar is busy - workflow logic correct") else: return False except Exception as e: print(f"❌ Confirm booking failed: {e}") return False # ======================================== # TEST 11: Get Session Again (verify confirmed status) # ======================================== print("\n" + "─" * 60) print("TEST 11: Get Session (verify confirmed status)") print("─" * 60) try: session = await client.get_booking_session(session_id=session_id) status = session.get('status') confirmed_at = session.get('confirmed_at') calendar_event_id = session.get('calendar_event_id') print(f"✅ Session status verified:") print(f" Status: {status}") if status == 'confirmed': print(f" Confirmed at: {confirmed_at}") print(f" Calendar event ID: {calendar_event_id}") elif status == 'pending': print(f" ⚠️ Still pending (booking may have failed due to availability)") except Exception as e: print(f"⚠️ Get session failed: {e}") # ======================================== # TEST 12 (OPTIONAL): Find New Slots # ======================================== print("\n" + "─" * 60) print("TEST 12: Find New Slots (if original were taken)") print("─" * 60) try: print(f" Finding alternative slots...") result = await client.find_new_booking_slots(session_id=session_id) new_slots = result.get('new_slots', []) print(f"✅ Found {len(new_slots)} new alternative slots:") for idx, slot in enumerate(new_slots[:3], 1): print(f" {idx}. {slot.get('start')} → {slot.get('end')}") except MCPError as e: print(f"⚠️ Find new slots: {e.code} - {e}") # This might fail if session is already confirmed - that's OK except Exception as e: print(f"⚠️ Find new slots failed: {e}") print("\n" + "=" * 60) print("✅ HYBRID BOOKING WORKFLOW TEST COMPLETE") print("=" * 60) print(f"\n📊 Summary:") print(f" Session ID: {session_id}") print(f" Organizer: {organizer_name} ({organizer_email})") print(f" External: {external_name} ({external_email})") print(f" Internal attendees: {len(internal_attendees)}") print(f" Workflow: ✅ END-TO-END COMPLETE") return True if __name__ == "__main__": print("\n🚀 Starting Hybrid Booking Integration Tests...\n") success = asyncio.run(test_hybrid_booking_workflow()) if success: print("\n✅ All tests completed successfully!") sys.exit(0) else: print("\n❌ Tests failed!") sys.exit(1)