286 lines
8.7 KiB
Python
Executable File
286 lines
8.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
# Created by: Claude
|
||
# Date: 2026-01-02
|
||
# Purpose: Script de test rapide pour l'API Mesh
|
||
# Refs: server/CLAUDE.md
|
||
|
||
"""
|
||
Script de test pour vérifier le fonctionnement de base de l'API Mesh.
|
||
|
||
Usage:
|
||
python test_api.py
|
||
|
||
Le serveur doit être lancé avant d'exécuter ce script.
|
||
"""
|
||
|
||
import requests
|
||
import json
|
||
from typing import Optional
|
||
|
||
# Configuration
|
||
BASE_URL = "http://localhost:8000"
|
||
|
||
|
||
class Colors:
|
||
"""Codes ANSI pour les couleurs."""
|
||
GREEN = '\033[92m'
|
||
RED = '\033[91m'
|
||
YELLOW = '\033[93m'
|
||
BLUE = '\033[94m'
|
||
END = '\033[0m'
|
||
|
||
|
||
def print_success(message: str):
|
||
"""Afficher un message de succès."""
|
||
print(f"{Colors.GREEN}✓ {message}{Colors.END}")
|
||
|
||
|
||
def print_error(message: str):
|
||
"""Afficher un message d'erreur."""
|
||
print(f"{Colors.RED}✗ {message}{Colors.END}")
|
||
|
||
|
||
def print_info(message: str):
|
||
"""Afficher un message d'information."""
|
||
print(f"{Colors.BLUE}ℹ {message}{Colors.END}")
|
||
|
||
|
||
def print_section(title: str):
|
||
"""Afficher un titre de section."""
|
||
print(f"\n{Colors.YELLOW}{'='*60}{Colors.END}")
|
||
print(f"{Colors.YELLOW}{title}{Colors.END}")
|
||
print(f"{Colors.YELLOW}{'='*60}{Colors.END}\n")
|
||
|
||
|
||
def test_health():
|
||
"""Tester le endpoint de santé."""
|
||
print_section("Test Health Check")
|
||
|
||
try:
|
||
response = requests.get(f"{BASE_URL}/health")
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
if data.get("status") == "healthy":
|
||
print_success("Health check passed")
|
||
return True
|
||
else:
|
||
print_error(f"Health check failed: {data}")
|
||
return False
|
||
except Exception as e:
|
||
print_error(f"Health check error: {str(e)}")
|
||
return False
|
||
|
||
|
||
def test_register(username: str, password: str) -> Optional[str]:
|
||
"""Tester l'enregistrement d'un utilisateur."""
|
||
print_section(f"Test Register - {username}")
|
||
|
||
try:
|
||
response = requests.post(
|
||
f"{BASE_URL}/api/auth/register",
|
||
json={
|
||
"username": username,
|
||
"password": password,
|
||
"email": f"{username}@example.com"
|
||
}
|
||
)
|
||
|
||
if response.status_code == 201:
|
||
data = response.json()
|
||
print_success(f"User registered: {data['username']}")
|
||
print_info(f"User ID: {data['user_id']}")
|
||
print_info(f"Token: {data['access_token'][:20]}...")
|
||
return data['access_token']
|
||
elif response.status_code == 400:
|
||
print_info("User already exists (expected if running multiple times)")
|
||
# Essayer de se connecter à la place
|
||
return test_login(username, password)
|
||
else:
|
||
print_error(f"Registration failed: {response.status_code} - {response.text}")
|
||
return None
|
||
|
||
except Exception as e:
|
||
print_error(f"Registration error: {str(e)}")
|
||
return None
|
||
|
||
|
||
def test_login(username: str, password: str) -> Optional[str]:
|
||
"""Tester la connexion d'un utilisateur."""
|
||
print_section(f"Test Login - {username}")
|
||
|
||
try:
|
||
response = requests.post(
|
||
f"{BASE_URL}/api/auth/login",
|
||
json={
|
||
"username": username,
|
||
"password": password
|
||
}
|
||
)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
print_success(f"User logged in: {data['username']}")
|
||
print_info(f"Token: {data['access_token'][:20]}...")
|
||
return data['access_token']
|
||
|
||
except Exception as e:
|
||
print_error(f"Login error: {str(e)}")
|
||
return None
|
||
|
||
|
||
def test_get_me(token: str):
|
||
"""Tester la récupération des infos utilisateur."""
|
||
print_section("Test Get User Info")
|
||
|
||
try:
|
||
response = requests.get(
|
||
f"{BASE_URL}/api/auth/me",
|
||
headers={"Authorization": f"Bearer {token}"}
|
||
)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
print_success("User info retrieved")
|
||
print_info(f"Username: {data['username']}")
|
||
print_info(f"User ID: {data['user_id']}")
|
||
|
||
except Exception as e:
|
||
print_error(f"Get user info error: {str(e)}")
|
||
|
||
|
||
def test_create_room(token: str, room_name: str) -> Optional[str]:
|
||
"""Tester la création d'une room."""
|
||
print_section(f"Test Create Room - {room_name}")
|
||
|
||
try:
|
||
response = requests.post(
|
||
f"{BASE_URL}/api/rooms/",
|
||
json={"name": room_name},
|
||
headers={"Authorization": f"Bearer {token}"}
|
||
)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
print_success(f"Room created: {data['name']}")
|
||
print_info(f"Room ID: {data['room_id']}")
|
||
return data['room_id']
|
||
|
||
except Exception as e:
|
||
print_error(f"Create room error: {str(e)}")
|
||
return None
|
||
|
||
|
||
def test_list_rooms(token: str):
|
||
"""Tester la liste des rooms."""
|
||
print_section("Test List Rooms")
|
||
|
||
try:
|
||
response = requests.get(
|
||
f"{BASE_URL}/api/rooms/",
|
||
headers={"Authorization": f"Bearer {token}"}
|
||
)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
print_success(f"Found {len(data)} room(s)")
|
||
for room in data:
|
||
print_info(f" - {room['name']} ({room['room_id']})")
|
||
|
||
except Exception as e:
|
||
print_error(f"List rooms error: {str(e)}")
|
||
|
||
|
||
def test_get_room(token: str, room_id: str):
|
||
"""Tester la récupération d'une room."""
|
||
print_section("Test Get Room Details")
|
||
|
||
try:
|
||
response = requests.get(
|
||
f"{BASE_URL}/api/rooms/{room_id}",
|
||
headers={"Authorization": f"Bearer {token}"}
|
||
)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
print_success("Room details retrieved")
|
||
print_info(f"Name: {data['name']}")
|
||
print_info(f"Members: {data['member_count']}")
|
||
|
||
except Exception as e:
|
||
print_error(f"Get room error: {str(e)}")
|
||
|
||
|
||
def test_request_capability(token: str, room_id: str):
|
||
"""Tester la demande de capability token."""
|
||
print_section("Test Request Capability Token")
|
||
|
||
try:
|
||
response = requests.post(
|
||
f"{BASE_URL}/api/auth/capability",
|
||
json={
|
||
"room_id": room_id,
|
||
"capabilities": ["call", "share:file"]
|
||
},
|
||
headers={"Authorization": f"Bearer {token}"}
|
||
)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
print_success("Capability token obtained")
|
||
print_info(f"Token: {data['cap_token'][:20]}...")
|
||
print_info(f"Expires in: {data['expires_in']}s")
|
||
|
||
except Exception as e:
|
||
print_error(f"Request capability error: {str(e)}")
|
||
|
||
|
||
def main():
|
||
"""Fonction principale du test."""
|
||
print(f"{Colors.BLUE}")
|
||
print("╔════════════════════════════════════════════════════════════╗")
|
||
print("║ MESH SERVER API TEST SUITE ║")
|
||
print("╚════════════════════════════════════════════════════════════╝")
|
||
print(f"{Colors.END}")
|
||
|
||
# Test de santé
|
||
if not test_health():
|
||
print_error("\n❌ Le serveur ne répond pas. Assurez-vous qu'il est démarré.")
|
||
return
|
||
|
||
# Enregistrer deux utilisateurs
|
||
user1_token = test_register("alice", "password123")
|
||
user2_token = test_register("bob", "password456")
|
||
|
||
if not user1_token or not user2_token:
|
||
print_error("\n❌ Impossible de créer les utilisateurs de test")
|
||
return
|
||
|
||
# Tester les infos utilisateur
|
||
test_get_me(user1_token)
|
||
|
||
# Créer une room
|
||
room_id = test_create_room(user1_token, "Test Room")
|
||
|
||
if room_id:
|
||
# Lister les rooms
|
||
test_list_rooms(user1_token)
|
||
|
||
# Détails de la room
|
||
test_get_room(user1_token, room_id)
|
||
|
||
# Demander un capability token
|
||
test_request_capability(user1_token, room_id)
|
||
|
||
print(f"\n{Colors.GREEN}")
|
||
print("╔════════════════════════════════════════════════════════════╗")
|
||
print("║ ✓ TESTS TERMINÉS ║")
|
||
print("╚════════════════════════════════════════════════════════════╝")
|
||
print(f"{Colors.END}")
|
||
print_info("Pour tester le WebSocket, utilisez le client web ou un outil comme wscat")
|
||
print_info("Exemple: wscat -c 'ws://localhost:8000/ws?token=YOUR_TOKEN'")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|