""" Test script for the enhanced appointment scheduling flow """ import asyncio import uuid import json import os from datetime import datetime from utils.appointment_state import ( update_appointment_state, get_appointment_state, start_appointment_flow, update_appointment_flow ) from utils.appointment_scheduler import schedule_appointment_from_state from utils.appointment_guidance import get_appointment_guidance, generate_appointment_summary from utils.frontend_integration import frontend_integration # Test data test_conversation_id = str(uuid.uuid4()) async def test_appointment_detection(): """Test the appointment intent detection""" print("\n=== Testing Appointment Intent Detection ===") # Test messages test_messages = [ "I want to schedule a meeting to discuss a project", "Can I book an appointment with your team?", "I'd like to talk to someone about a website project", "I need a consultation about AI development" ] for message in test_messages: print(f"\nTesting message: '{message}'") # Simulate the appointment detection # In a real scenario, this would call the appointment detection endpoint print(f"Simulating appointment detection for: '{message}'") # Start an appointment flow result = start_appointment_flow( conversation_id=test_conversation_id, confidence=0.8, suggested_topic="Website Development", extracted_info={ "topic": "Website Development" }, missing_fields=["name", "email", "phone"] ) print(f"Started appointment flow: {result}") # Get the appointment state state = get_appointment_state(test_conversation_id) print(f"Initial appointment state: {json.dumps(state, indent=2)}") break # Just test one message for brevity async def test_information_collection(): """Test the information collection process""" print("\n=== Testing Information Collection ===") # Get the current state state = get_appointment_state(test_conversation_id) missing_fields = state.get("missing_fields", []) print(f"Missing fields: {missing_fields}") # Test providing name if "name" in missing_fields: print("\nProviding name: 'John Smith'") # Update the appointment state with the name update_result = update_appointment_flow( conversation_id=test_conversation_id, extracted_info={"name": "John Smith"} ) print(f"Update result: {update_result}") # Get the updated state state = get_appointment_state(test_conversation_id) print(f"Updated state: {json.dumps(state, indent=2)}") # Get guidance for the next step guidance = get_appointment_guidance(test_conversation_id, state.get("missing_fields", [])) print(f"Guidance: {json.dumps(guidance, indent=2)}") # Test providing email if "email" in state.get("missing_fields", []): print("\nProviding email: 'john@example.com'") # Update the appointment state with the email update_result = update_appointment_flow( conversation_id=test_conversation_id, extracted_info={"email": "john@example.com"} ) print(f"Update result: {update_result}") # Get the updated state state = get_appointment_state(test_conversation_id) print(f"Updated state: {json.dumps(state, indent=2)}") # Get guidance for the next step guidance = get_appointment_guidance(test_conversation_id, state.get("missing_fields", [])) print(f"Guidance: {json.dumps(guidance, indent=2)}") # Test providing phone if "phone" in state.get("missing_fields", []): print("\nProviding phone: '123-456-7890'") # Update the appointment state with the phone update_result = update_appointment_flow( conversation_id=test_conversation_id, extracted_info={"phone": "123-456-7890"} ) print(f"Update result: {update_result}") # Get the updated state state = get_appointment_state(test_conversation_id) print(f"Updated state: {json.dumps(state, indent=2)}") # Get guidance for the next step guidance = get_appointment_guidance(test_conversation_id, state.get("missing_fields", [])) print(f"Guidance: {json.dumps(guidance, indent=2)}") async def test_enhanced_response_generation(): """Test the enhanced response generation""" print("\n=== Testing Enhanced Response Generation ===") # Test messages test_messages = [ "My name is John Smith", "My email is john@example.com", "My phone number is 123-456-7890" ] for message in test_messages: print(f"\nTesting message: '{message}'") # Process the message with the frontend integration response = await frontend_integration.process_message( message=message, conversation_id=test_conversation_id, retrieved_info={"results": []}, conversation_history="", is_new_conversation=False ) print(f"Response: {response.get('response')}") async def test_appointment_scheduling(): """Test the appointment scheduling""" print("\n=== Testing Appointment Scheduling ===") # Get the current state state = get_appointment_state(test_conversation_id) print(f"Current state: {json.dumps(state, indent=2)}") # Make sure we're ready to schedule if state.get("meeting_request_stage") != "ready_to_schedule": print("Updating state to ready_to_schedule") # Update the state to be ready to schedule update_result = update_appointment_state( test_conversation_id, { "meeting_request": True, "meeting_request_stage": "ready_to_schedule", "missing_fields": [], "extracted_info": { "name": "John Smith", "email": "john@example.com", "phone": "123-456-7890", "topic": "Website Development", "preferred_date": "2025-10-01", "preferred_time": "14:00" }, "suggested_topic": "Website Development" } ) print(f"Update result: {update_result}") # Schedule the appointment print("\nScheduling the appointment") scheduling_result = await schedule_appointment_from_state(test_conversation_id) print(f"Scheduling result: {json.dumps(scheduling_result, indent=2)}") # Generate a summary of the appointment summary = generate_appointment_summary(test_conversation_id) print(f"\nAppointment summary:\n{summary}") # Verify that appointments.json was created appointments_file = "data/appointments.json" if os.path.exists(appointments_file): with open(appointments_file, "r") as f: appointments = json.load(f) print(f"\nFound {len(appointments)} appointments in {appointments_file}") print("First appointment:") first_appointment_id = next(iter(appointments)) print(json.dumps(appointments[first_appointment_id], indent=2)) else: print(f"\nError: {appointments_file} was not created") async def main(): """Run all tests""" try: await test_appointment_detection() await test_information_collection() await test_enhanced_response_generation() await test_appointment_scheduling() except Exception as e: print(f"Error running tests: {str(e)}") if __name__ == "__main__": asyncio.run(main())