248 lines
7.6 KiB
Python
Executable File
248 lines
7.6 KiB
Python
Executable File
#!/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()
|