test_video.py
1 import pytest 2 import requests 3 import json 4 import os 5 import time 6 import io 7 from typing import Dict, Any 8 9 # URL de base de l'API (ajuster selon l'environnement) 10 BASE_URL = os.environ.get("API_BASE_URL", "http://localhost:8000") 11 12 # Variable pour stocker les informations de test 13 test_data = { 14 "api_key": None, 15 "manipulation_task_id": None, 16 "nonverbal_task_id": None, 17 "manipulation_video_path": None, # Chemin pour la vidéo d'analyse de manipulation 18 "nonverbal_video_path": None # Chemin pour la vidéo d'analyse non-verbale 19 } 20 21 # Fichiers vidéo de test 22 SAMPLE_MANIPULATION_VIDEO_PATH = os.environ.get("SAMPLE_MANIPULATION_VIDEO_PATH", os.environ.get("SAMPLE_VIDEO_PATH")) 23 SAMPLE_NONVERBAL_VIDEO_PATH = os.environ.get("SAMPLE_NONVERBAL_VIDEO_PATH", os.environ.get("SAMPLE_VIDEO_PATH")) 24 25 def setup_module(): 26 """Configuration initiale pour les tests vidéo.""" 27 # Récupérer une clé API valide depuis les tests d'authentification ou utiliser une variable d'environnement 28 try: 29 # Essayer d'utiliser une variable d'environnement pour la clé API 30 api_key = os.environ.get("TEST_API_KEY") 31 32 if not api_key: 33 # Importer et exécuter les tests d'authentification si nécessaire 34 from test_auth import test_register_user, test_login, test_create_api_key 35 36 # Créer un utilisateur et une clé API si nécessaire 37 test_register_user() 38 test_login() 39 test_create_api_key() 40 41 from test_auth import test_data as auth_test_data 42 api_key = auth_test_data["api_key"] 43 except Exception as e: 44 # En cas d'erreur, une clé de test doit être fournie en variable d'environnement 45 api_key = os.environ.get("TEST_API_KEY") 46 if not api_key: 47 raise Exception("Aucune clé API disponible pour les tests. Définissez TEST_API_KEY ou exécutez test_auth.py") 48 49 test_data["api_key"] = api_key 50 51 def get_sample_video(video_type="manipulation"): 52 """ 53 Retourne un fichier vidéo de test selon le type demandé. 54 55 Args: 56 video_type (str): Type de vidéo ('manipulation' ou 'nonverbal') 57 58 Returns: 59 Un fichier ouvert en mode binaire 60 """ 61 video_path = None 62 63 if video_type == "manipulation": 64 video_path = SAMPLE_MANIPULATION_VIDEO_PATH 65 elif video_type == "nonverbal": 66 video_path = SAMPLE_NONVERBAL_VIDEO_PATH 67 68 if video_path and os.path.exists(video_path): 69 return open(video_path, "rb"), video_path 70 else: 71 # Pour les tests sans fichier vidéo réel, on peut utiliser un fichier minimal 72 # Ceci est juste pour tester l'API, mais ne passera pas la validation complète 73 # Vous devez définir les variables d'environnement pour des tests complets 74 video_type_name = "manipulation" if video_type == "manipulation" else "non-verbale" 75 pytest.skip(f"Aucun fichier vidéo pour analyse {video_type_name} disponible. Définissez SAMPLE_{video_type.upper()}_VIDEO_PATH.") 76 77 def test_upload_manipulation_video(): 78 """Teste le téléchargement d'un fichier vidéo pour l'analyse de manipulation.""" 79 # S'assurer qu'une clé API est disponible 80 assert test_data["api_key"] is not None, "Aucune clé API disponible pour le test" 81 82 # Configurer les headers avec la clé API 83 headers = { 84 "X-API-Key": test_data["api_key"] 85 } 86 87 # Télécharger un fichier vidéo 88 try: 89 video_file, video_path = get_sample_video("manipulation") 90 91 files = { 92 "file": ("manipulation_video.mp4", video_file, "video/mp4") 93 } 94 95 response = requests.post( 96 f"{BASE_URL}/api/video/upload", 97 headers=headers, 98 files=files 99 ) 100 101 # Fermer le fichier après utilisation 102 video_file.close() 103 104 assert response.status_code == 200, f"Code de statut inattendu: {response.status_code}, {response.text}" 105 106 # Vérifier la réponse 107 data = response.json() 108 assert "video_path" in data 109 assert "duration" in data 110 111 # Sauvegarder le chemin de la vidéo pour les tests suivants 112 test_data["manipulation_video_path"] = data["video_path"] 113 114 except Exception as e: 115 pytest.skip(f"Erreur lors du téléchargement de la vidéo de manipulation: {e}") 116 117 def test_upload_nonverbal_video(): 118 """Teste le téléchargement d'un fichier vidéo pour l'analyse non-verbale.""" 119 # S'assurer qu'une clé API est disponible 120 assert test_data["api_key"] is not None, "Aucune clé API disponible pour le test" 121 122 # Configurer les headers avec la clé API 123 headers = { 124 "X-API-Key": test_data["api_key"] 125 } 126 127 # Télécharger un fichier vidéo 128 try: 129 video_file, video_path = get_sample_video("nonverbal") 130 131 files = { 132 "file": ("nonverbal_video.mp4", video_file, "video/mp4") 133 } 134 135 response = requests.post( 136 f"{BASE_URL}/api/video/upload", 137 headers=headers, 138 files=files 139 ) 140 141 # Fermer le fichier après utilisation 142 video_file.close() 143 144 assert response.status_code == 200, f"Code de statut inattendu: {response.status_code}, {response.text}" 145 146 # Vérifier la réponse 147 data = response.json() 148 assert "video_path" in data 149 assert "duration" in data 150 151 # Sauvegarder le chemin de la vidéo pour les tests suivants 152 test_data["nonverbal_video_path"] = data["video_path"] 153 154 except Exception as e: 155 pytest.skip(f"Erreur lors du téléchargement de la vidéo non-verbale: {e}") 156 157 def test_start_manipulation_analysis(): 158 """Teste le démarrage d'une analyse de manipulation.""" 159 # S'assurer qu'une clé API et un chemin de vidéo sont disponibles 160 assert test_data["api_key"] is not None, "Aucune clé API disponible pour le test" 161 if test_data["manipulation_video_path"] is None: 162 pytest.skip("Aucun chemin de vidéo de manipulation disponible pour le test") 163 164 # Configurer les headers avec la clé API 165 headers = { 166 "X-API-Key": test_data["api_key"] 167 } 168 169 # Données pour l'analyse de manipulation 170 analysis_data = { 171 "video_path": test_data["manipulation_video_path"], 172 "transcribe": True, 173 "diarize": True, 174 "language": "fr" 175 } 176 177 # Envoyer la requête d'analyse de manipulation 178 response = requests.post( 179 f"{BASE_URL}/api/video/manipulation-analysis", 180 json=analysis_data, 181 headers=headers 182 ) 183 184 # Si le service n'est pas disponible, ignorer le test 185 if response.status_code == 503: 186 pytest.skip("Service d'analyse de manipulation non disponible") 187 188 assert response.status_code == 200, f"Code de statut inattendu: {response.status_code}, {response.text}" 189 190 # Vérifier la réponse 191 data = response.json() 192 assert "task_id" in data 193 assert data["status"] == "pending" 194 195 # Sauvegarder l'ID de tâche pour les tests suivants 196 test_data["manipulation_task_id"] = data["task_id"] 197 198 def test_start_nonverbal_analysis(): 199 """Teste le démarrage d'une analyse non-verbale.""" 200 # S'assurer qu'une clé API et un chemin de vidéo sont disponibles 201 assert test_data["api_key"] is not None, "Aucune clé API disponible pour le test" 202 if test_data["nonverbal_video_path"] is None: 203 pytest.skip("Aucun chemin de vidéo non-verbale disponible pour le test") 204 205 # Configurer les headers avec la clé API 206 headers = { 207 "X-API-Key": test_data["api_key"] 208 } 209 210 # Données pour l'analyse non-verbale 211 analysis_data = { 212 "video_path": test_data["nonverbal_video_path"], 213 "extract_frames": True, 214 "frame_count": 32, 215 "analyze_facial_expressions": True 216 } 217 218 # Envoyer la requête d'analyse non-verbale 219 response = requests.post( 220 f"{BASE_URL}/api/video/nonverbal-analysis", 221 json=analysis_data, 222 headers=headers 223 ) 224 225 # Si le service n'est pas disponible, ignorer le test 226 if response.status_code == 503: 227 pytest.skip("Service d'analyse non-verbale non disponible") 228 229 assert response.status_code == 200, f"Code de statut inattendu: {response.status_code}, {response.text}" 230 231 # Vérifier la réponse 232 data = response.json() 233 assert "task_id" in data 234 assert data["status"] == "pending" 235 236 # Sauvegarder l'ID de tâche pour les tests suivants 237 test_data["nonverbal_task_id"] = data["task_id"] 238 239 def test_get_manipulation_analysis_status(): 240 """Teste la récupération de l'état d'une tâche d'analyse de manipulation.""" 241 # S'assurer qu'une clé API et un ID de tâche sont disponibles 242 assert test_data["api_key"] is not None, "Aucune clé API disponible pour le test" 243 244 if test_data.get("manipulation_task_id") is None: 245 pytest.skip("Aucun ID de tâche d'analyse de manipulation disponible pour le test") 246 247 # Configurer les headers avec la clé API 248 headers = { 249 "X-API-Key": test_data["api_key"] 250 } 251 252 # Envoyer la requête pour récupérer l'état de la tâche 253 response = requests.get( 254 f"{BASE_URL}/api/video/tasks/{test_data['manipulation_task_id']}", 255 headers=headers 256 ) 257 assert response.status_code == 200, f"Code de statut inattendu: {response.status_code}, {response.text}" 258 259 # Vérifier la réponse 260 data = response.json() 261 assert "status" in data 262 assert "progress" in data 263 264 # La tâche peut être en attente, en cours d'exécution ou terminée 265 assert data["status"] in ["pending", "running", "completed", "failed"], f"Statut inattendu: {data['status']}" 266 267 def test_get_nonverbal_analysis_status(): 268 """Teste la récupération de l'état d'une tâche d'analyse non-verbale.""" 269 # S'assurer qu'une clé API et un ID de tâche sont disponibles 270 assert test_data["api_key"] is not None, "Aucune clé API disponible pour le test" 271 272 if test_data.get("nonverbal_task_id") is None: 273 pytest.skip("Aucun ID de tâche d'analyse non-verbale disponible pour le test") 274 275 # Configurer les headers avec la clé API 276 headers = { 277 "X-API-Key": test_data["api_key"] 278 } 279 280 # Envoyer la requête pour récupérer l'état de la tâche 281 response = requests.get( 282 f"{BASE_URL}/api/video/tasks/{test_data['nonverbal_task_id']}", 283 headers=headers 284 ) 285 assert response.status_code == 200, f"Code de statut inattendu: {response.status_code}, {response.text}" 286 287 # Vérifier la réponse 288 data = response.json() 289 assert "status" in data 290 assert "progress" in data 291 292 # La tâche peut être en attente, en cours d'exécution ou terminée 293 assert data["status"] in ["pending", "running", "completed", "failed"], f"Statut inattendu: {data['status']}" 294 295 def test_wait_for_manipulation_analysis_completion(): 296 """Teste l'attente de la fin d'une tâche d'analyse de manipulation.""" 297 # S'assurer qu'une clé API et un ID de tâche sont disponibles 298 assert test_data["api_key"] is not None, "Aucune clé API disponible pour le test" 299 300 if test_data.get("manipulation_task_id") is None: 301 pytest.skip("Aucun ID de tâche d'analyse de manipulation disponible pour le test") 302 303 # Configurer les headers avec la clé API 304 headers = { 305 "X-API-Key": test_data["api_key"] 306 } 307 308 # Attendre que la tâche soit terminée (avec un délai d'expiration) 309 max_retries = 30 # 5 minutes maximum (avec 10 secondes d'intervalle) 310 for i in range(max_retries): 311 # Envoyer la requête pour récupérer l'état de la tâche 312 response = requests.get( 313 f"{BASE_URL}/api/video/tasks/{test_data['manipulation_task_id']}", 314 headers=headers 315 ) 316 assert response.status_code == 200 317 318 # Vérifier la réponse 319 data = response.json() 320 321 # Si la tâche a échoué, afficher l'erreur 322 if data["status"] == "failed": 323 print(f"La tâche d'analyse de manipulation a échoué: {data.get('error', 'Erreur inconnue')}") 324 pytest.skip(f"La tâche d'analyse de manipulation a échoué: {data.get('error', 'Erreur inconnue')}") 325 326 # Si la tâche est terminée, le test est réussi 327 if data["status"] == "completed": 328 if "results" in data: 329 assert isinstance(data["results"], dict) 330 # Vérifier que les résultats contiennent des données 331 assert len(data["results"]) > 0 332 break 333 334 # Afficher la progression 335 print(f"Attente de la fin de l'analyse de manipulation... Progression: {data.get('progress', 0):.0f}% (tentative {i+1}/{max_retries})") 336 337 # Attendre avant de réessayer 338 time.sleep(10) 339 else: 340 pytest.skip(f"La tâche d'analyse de manipulation n'est pas terminée après {max_retries} tentatives") 341 342 def test_wait_for_nonverbal_analysis_completion(): 343 """Teste l'attente de la fin d'une tâche d'analyse non-verbale.""" 344 # S'assurer qu'une clé API et un ID de tâche sont disponibles 345 assert test_data["api_key"] is not None, "Aucune clé API disponible pour le test" 346 347 if test_data.get("nonverbal_task_id") is None: 348 pytest.skip("Aucun ID de tâche d'analyse non-verbale disponible pour le test") 349 350 # Configurer les headers avec la clé API 351 headers = { 352 "X-API-Key": test_data["api_key"] 353 } 354 355 # Attendre que la tâche soit terminée (avec un délai d'expiration) 356 max_retries = 30 # 5 minutes maximum (avec 10 secondes d'intervalle) 357 for i in range(max_retries): 358 # Envoyer la requête pour récupérer l'état de la tâche 359 response = requests.get( 360 f"{BASE_URL}/api/video/tasks/{test_data['nonverbal_task_id']}", 361 headers=headers 362 ) 363 assert response.status_code == 200 364 365 # Vérifier la réponse 366 data = response.json() 367 368 # Si la tâche a échoué, afficher l'erreur 369 if data["status"] == "failed": 370 print(f"La tâche d'analyse non-verbale a échoué: {data.get('error', 'Erreur inconnue')}") 371 pytest.skip(f"La tâche d'analyse non-verbale a échoué: {data.get('error', 'Erreur inconnue')}") 372 373 # Si la tâche est terminée, le test est réussi 374 if data["status"] == "completed": 375 if "results" in data: 376 assert isinstance(data["results"], dict) 377 # Vérifier que les résultats contiennent des données 378 assert len(data["results"]) > 0 379 break 380 381 # Afficher la progression 382 print(f"Attente de la fin de l'analyse non-verbale... Progression: {data.get('progress', 0):.0f}% (tentative {i+1}/{max_retries})") 383 384 # Attendre avant de réessayer 385 time.sleep(10) 386 else: 387 pytest.skip(f"La tâche d'analyse non-verbale n'est pas terminée après {max_retries} tentatives") 388 389 def test_get_all_video_tasks(): 390 """Teste la récupération de toutes les tâches vidéo.""" 391 # S'assurer qu'une clé API est disponible 392 assert test_data["api_key"] is not None, "Aucune clé API disponible pour le test" 393 394 # Configurer les headers avec la clé API 395 headers = { 396 "X-API-Key": test_data["api_key"] 397 } 398 399 # Envoyer la requête pour récupérer toutes les tâches vidéo 400 response = requests.get(f"{BASE_URL}/api/video/tasks", headers=headers) 401 assert response.status_code == 200 402 403 # Vérifier la réponse 404 data = response.json() 405 assert "tasks" in data 406 assert isinstance(data["tasks"], list) 407 408 # Si des tâches ont été créées précédemment, elles devraient être présentes 409 task_ids = [task["id"] for task in data["tasks"]] 410 if test_data.get("manipulation_task_id"): 411 assert test_data["manipulation_task_id"] in task_ids, "L'ID de tâche d'analyse de manipulation est absent" 412 if test_data.get("nonverbal_task_id"): 413 assert test_data["nonverbal_task_id"] in task_ids, "L'ID de tâche d'analyse non-verbale est absent" 414 415 def test_delete_video_tasks(): 416 """Teste la suppression des tâches vidéo.""" 417 # S'assurer qu'une clé API est disponible 418 assert test_data["api_key"] is not None, "Aucune clé API disponible pour le test" 419 420 # Configurer les headers avec la clé API 421 headers = { 422 "X-API-Key": test_data["api_key"] 423 } 424 425 # Supprimer la tâche d'analyse de manipulation si disponible 426 if test_data.get("manipulation_task_id"): 427 response = requests.delete( 428 f"{BASE_URL}/api/video/tasks/{test_data['manipulation_task_id']}", 429 headers=headers 430 ) 431 assert response.status_code == 200 432 433 # Vérifier que la tâche a bien été supprimée 434 response = requests.get( 435 f"{BASE_URL}/api/video/tasks/{test_data['manipulation_task_id']}", 436 headers=headers 437 ) 438 assert response.status_code == 404 439 440 # Supprimer la tâche d'analyse non-verbale si disponible 441 if test_data.get("nonverbal_task_id"): 442 response = requests.delete( 443 f"{BASE_URL}/api/video/tasks/{test_data['nonverbal_task_id']}", 444 headers=headers 445 ) 446 assert response.status_code == 200 447 448 # Vérifier que la tâche a bien été supprimée 449 response = requests.get( 450 f"{BASE_URL}/api/video/tasks/{test_data['nonverbal_task_id']}", 451 headers=headers 452 ) 453 assert response.status_code == 404 454 455 if __name__ == "__main__": 456 # Initialiser les tests 457 setup_module() 458 459 # Tests pour l'analyse de manipulation 460 test_upload_manipulation_video() 461 try: 462 test_start_manipulation_analysis() 463 test_get_manipulation_analysis_status() 464 test_wait_for_manipulation_analysis_completion() 465 except Exception as e: 466 print(f"Les tests d'analyse de manipulation ont échoué: {e}") 467 468 # Tests pour l'analyse non-verbale 469 test_upload_nonverbal_video() 470 try: 471 test_start_nonverbal_analysis() 472 test_get_nonverbal_analysis_status() 473 test_wait_for_nonverbal_analysis_completion() 474 except Exception as e: 475 print(f"Les tests d'analyse non-verbale ont échoué: {e}") 476 477 # Tests communs 478 test_get_all_video_tasks() 479 test_delete_video_tasks() 480 481 print("Tous les tests vidéo ont réussi!")