#!/usr/bin/env python3 # Created by: Claude # Date: 2026-01-02 # Purpose: Script de test pour les endpoints P2P # Refs: server/CLAUDE.md """ Script de test pour vérifier les endpoints P2P du serveur Mesh. """ 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 register_user(username: str, password: str) -> Optional[str]: """Enregistrer un utilisateur et retourner le token.""" 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: return response.json()['access_token'] elif response.status_code == 400: # User exists, login instead response = requests.post( f"{BASE_URL}/api/auth/login", json={"username": username, "password": password} ) if response.status_code == 200: return response.json()['access_token'] except Exception as e: print_error(f"Registration error: {str(e)}") return None def create_room(token: str, room_name: str) -> Optional[str]: """Créer une room et retourner le room_id.""" try: response = requests.post( f"{BASE_URL}/api/rooms/", json={"name": room_name}, headers={"Authorization": f"Bearer {token}"} ) response.raise_for_status() return response.json()['room_id'] except Exception as e: print_error(f"Create room error: {str(e)}") return None def test_create_p2p_session(token: str, room_id: str): """Tester la création d'une session P2P.""" print_section("Test Create P2P Session") try: response = requests.post( f"{BASE_URL}/api/p2p/session", json={ "room_id": room_id, "target_peer_id": "peer_target_123", "kind": "file", "capabilities": ["share:file"] }, headers={"Authorization": f"Bearer {token}"} ) response.raise_for_status() data = response.json() print_success("P2P session created") print_info(f"Session ID: {data['session_id']}") print_info(f"Session Token: {data['session_token'][:30]}...") print_info(f"Kind: {data['kind']}") print_info(f"Expires at: {data['expires_at']}") return data['session_id'] except Exception as e: print_error(f"Create P2P session error: {str(e)}") if hasattr(e, 'response') and e.response is not None: print_error(f"Response: {e.response.text}") return None def test_list_p2p_sessions(token: str): """Tester la liste des sessions P2P.""" print_section("Test List P2P Sessions") try: response = requests.get( f"{BASE_URL}/api/p2p/sessions", headers={"Authorization": f"Bearer {token}"} ) response.raise_for_status() data = response.json() sessions = data.get('sessions', []) print_success(f"Found {len(sessions)} active session(s)") for session in sessions: print_info(f" - {session['session_id']} ({session['kind']})") except Exception as e: print_error(f"List P2P sessions error: {str(e)}") def test_close_p2p_session(token: str, session_id: str): """Tester la fermeture d'une session P2P.""" print_section("Test Close P2P Session") try: response = requests.delete( f"{BASE_URL}/api/p2p/session/{session_id}", headers={"Authorization": f"Bearer {token}"} ) response.raise_for_status() data = response.json() print_success(data['message']) except Exception as e: print_error(f"Close P2P session error: {str(e)}") def test_invalid_kind(token: str, room_id: str): """Tester avec un kind invalide.""" print_section("Test Invalid Session Kind") try: response = requests.post( f"{BASE_URL}/api/p2p/session", json={ "room_id": room_id, "target_peer_id": "peer_target_123", "kind": "invalid_kind", "capabilities": ["share:file"] }, headers={"Authorization": f"Bearer {token}"} ) if response.status_code == 400: print_success("Invalid kind correctly rejected") print_info(f"Error: {response.json()['detail']}") else: print_error(f"Expected 400, got {response.status_code}") except Exception as e: print_error(f"Invalid kind test error: {str(e)}") def main(): """Fonction principale du test.""" print(f"{Colors.BLUE}") print("╔════════════════════════════════════════════════════════════╗") print("║ MESH P2P API TEST SUITE ║") print("╚════════════════════════════════════════════════════════════╝") print(f"{Colors.END}") # Enregistrer un utilisateur print_section("Setup: Register User & Create Room") token = register_user("alice_p2p", "password123") if not token: print_error("\n❌ Impossible de créer l'utilisateur de test") return print_success("User registered/logged in") # Créer une room room_id = create_room(token, "P2P Test Room") if not room_id: print_error("\n❌ Impossible de créer la room de test") return print_success(f"Room created: {room_id}") # Tester la création de session P2P session_id = test_create_p2p_session(token, room_id) if session_id: # Lister les sessions actives test_list_p2p_sessions(token) # Fermer la session test_close_p2p_session(token, session_id) # Vérifier que la session a été fermée test_list_p2p_sessions(token) # Tester avec un kind invalide test_invalid_kind(token, room_id) print(f"\n{Colors.GREEN}") print("╔════════════════════════════════════════════════════════════╗") print("║ ✓ TESTS P2P TERMINÉS ║") print("╚════════════════════════════════════════════════════════════╝") print(f"{Colors.END}") if __name__ == "__main__": main()