test_app.py

#!/usr/bin/env python3
"""
Test script for WhatsApp-like chat application
Demonstrates: Register → Login → Add Contact → Send Message → Status Updates
"""

import requests
import time
import json
from socketio import SimpleClient

BASE_URL = 'http://localhost:5000'

def print_step(step, message):
    print(f"\n{'='*60}")
    print(f"STEP {step}: {message}")
    print('='*60)

def test_registration():
    print_step(1, "Testing User Registration")
    
    user1_data = {
        'name': 'Test User 1',
        'email': f'testuser1_{int(time.time())}@test.com'
    }
    
    user2_data = {
        'name': 'Test User 2',
        'email': f'testuser2_{int(time.time())}@test.com'
    }
    
    print("\n📝 Registering User 1...")
    response1 = requests.post(f'{BASE_URL}/api/register', json=user1_data)
    user1 = response1.json()
    print(f"✅ User 1 registered successfully!")
    print(f"   Name: {user1['name']}")
    print(f"   ID: {user1['user_id']}")
    print(f"   Token: {user1['token'][:20]}...")
    
    print("\n📝 Registering User 2...")
    response2 = requests.post(f'{BASE_URL}/api/register', json=user2_data)
    user2 = response2.json()
    print(f"✅ User 2 registered successfully!")
    print(f"   Name: {user2['name']}")
    print(f"   ID: {user2['user_id']}")
    print(f"   Token: {user2['token'][:20]}...")
    
    return user1, user2

def test_login(user):
    print_step(2, "Testing User Login")
    
    login_data = {
        'user_id': user['user_id'],
        'token': user['token']
    }
    
    print(f"\n🔐 Logging in as {user['name']}...")
    response = requests.post(f'{BASE_URL}/api/login', json=login_data)
    result = response.json()
    
    if response.status_code == 200:
        print(f"✅ Login successful!")
        print(f"   User: {result['name']}")
        print(f"   Email: {result['email']}")
        return True
    else:
        print(f"❌ Login failed: {result.get('error')}")
        return False

def test_add_contact(user1, user2):
    print_step(3, "Testing Add Contact")
    
    contact_data = {
        'token': user1['token'],
        'contact_id': user2['user_id']
    }
    
    print(f"\n👥 {user1['name']} adding {user2['name']} as contact...")
    response = requests.post(f'{BASE_URL}/api/contacts/add', json=contact_data)
    
    if response.status_code == 200:
        result = response.json()
        print(f"✅ Contact added successfully!")
        print(f"   Contact Name: {result['name']}")
        print(f"   Contact ID: {result['contact_id']}")
        return True
    else:
        error = response.json().get('error')
        print(f"❌ Failed to add contact: {error}")
        return False

def test_get_contacts(user):
    print_step(4, "Testing Get Contacts")
    
    print(f"\n📋 Getting contacts for {user['name']}...")
    response = requests.post(f'{BASE_URL}/api/contacts', 
                            json={'token': user['token']})
    
    if response.status_code == 200:
        result = response.json()
        contacts = result['contacts']
        print(f"✅ Retrieved {len(contacts)} contact(s):")
        for contact in contacts:
            print(f"   - {contact['name']} (ID: {contact['user_id']})")
        return contacts
    else:
        print(f"❌ Failed to get contacts")
        return []

def test_websocket_messaging(user1, user2):
    print_step(5, "Testing WebSocket Real-time Messaging")
    
    print("\n🔌 Connecting User 1 via WebSocket...")
    client1 = SimpleClient()
    client1.connect(BASE_URL)
    client1.emit('authenticate', {'token': user1['token']})
    
    event = client1.receive(timeout=5)
    if event[0] == 'authenticated' and event[1]['success']:
        print("✅ User 1 authenticated via WebSocket")
    
    print("\n🔌 Connecting User 2 via WebSocket...")
    client2 = SimpleClient()
    client2.connect(BASE_URL)
    client2.emit('authenticate', {'token': user2['token']})
    
    event = client2.receive(timeout=5)
    if event[0] == 'authenticated' and event[1]['success']:
        print("✅ User 2 authenticated via WebSocket")
    
    print("\n💬 User 1 sending message to User 2...")
    test_message = "Hello from automated test! 👋"
    client1.emit('send_message', {
        'token': user1['token'],
        'to': user2['user_id'],
        'message': test_message
    })
    
    time.sleep(1)
    
    print("\n📨 Checking for received messages...")
    received_messages = []
    
    try:
        while True:
            event = client2.receive(timeout=2)
            print(f"   Received event: {event[0]}")
            
            if event[0] == 'new_message':
                msg = event[1]
                received_messages.append(msg)
                print(f"✅ Message received!")
                print(f"   From: {msg['from']}")
                print(f"   Message: {msg['message']}")
                print(f"   Status: {msg['status']}")
                
                print(f"\n📖 User 2 marking message as read...")
                client2.emit('mark_conversation_read', {
                    'token': user2['token'],
                    'contact_id': user1['user_id']
                })
                break
    except:
        pass
    
    time.sleep(1)
    
    print("\n🔄 Checking for status updates on User 1's side...")
    try:
        while True:
            event = client1.receive(timeout=2)
            if event[0] == 'message_status_update':
                status_update = event[1]
                print(f"✅ Status update received!")
                print(f"   Message ID: {status_update['message_id']}")
                print(f"   New Status: {status_update['status']}")
                if status_update['status'] == 'read':
                    print("\n✨ Message lifecycle complete: sent → delivered → read")
                    break
    except:
        pass
    
    print("\n🔌 Disconnecting clients...")
    client1.disconnect()
    client2.disconnect()
    
    return len(received_messages) > 0

def test_message_persistence(user1, user2):
    print_step(6, "Testing Message Persistence")
    
    print(f"\n📚 Loading message history...")
    response = requests.post(f'{BASE_URL}/api/messages', 
                            json={
                                'token': user1['token'],
                                'contact_id': user2['user_id']
                            })
    
    if response.status_code == 200:
        result = response.json()
        messages = result['messages']
        print(f"✅ Retrieved {len(messages)} message(s) from JSON storage:")
        for msg in messages[-3:]:
            print(f"   [{msg['status']}] {msg['from'][:8]}... → {msg['to'][:8]}...: {msg['message'][:30]}")
        return True
    else:
        print(f"❌ Failed to load messages")
        return False

def main():
    print("\n" + "="*60)
    print("WhatsApp-like Chat Application - Automated Test Suite")
    print("="*60)
    
    try:
        user1, user2 = test_registration()
        time.sleep(1)
        
        test_login(user1)
        time.sleep(1)
        
        test_add_contact(user1, user2)
        test_add_contact(user2, user1)
        time.sleep(1)
        
        test_get_contacts(user1)
        time.sleep(1)
        
        test_websocket_messaging(user1, user2)
        time.sleep(1)
        
        test_message_persistence(user1, user2)
        
        print("\n" + "="*60)
        print("✅ ALL TESTS COMPLETED SUCCESSFULLY!")
        print("="*60)
        print("\n📋 Test Summary:")
        print("   ✅ User Registration")
        print("   ✅ User Login")
        print("   ✅ Add Contact")
        print("   ✅ Get Contacts")
        print("   ✅ Real-time Messaging via WebSocket")
        print("   ✅ Message Status Lifecycle (sent → delivered → read)")
        print("   ✅ Message Persistence in JSON")
        print("\n💡 You can now login with these test credentials:")
        print(f"   User 1 ID: {user1['user_id']}")
        print(f"   User 1 Token: {user1['token']}")
        print(f"   User 2 ID: {user2['user_id']}")
        print(f"   User 2 Token: {user2['token']}")
        
    except Exception as e:
        print(f"\n❌ Test failed with error: {str(e)}")
        import traceback
        traceback.print_exc()

if __name__ == '__main__':
    print("\n⚠️  Make sure the server is running on http://localhost:5000")
    print("Starting tests in 3 seconds...\n")
    time.sleep(3)
    main()