test_task.py
1 import pytest 2 import requests 3 import json 4 import os 5 import time 6 import uuid 7 from typing import Dict, Any, Optional, List 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 "token": None, 16 "task_ids": [], # Liste des IDs de tâches créées durant les tests 17 "text_task_id": None, 18 "batch_task_id": None, 19 "other_task_id": None, 20 "cancelled_task_id": None 21 } 22 23 def setup_module(): 24 """Configuration initiale pour les tests de tâches.""" 25 # Récupérer une clé API valide 26 try: 27 # Essayer d'utiliser une variable d'environnement pour la clé API 28 api_key = os.environ.get("TEST_API_KEY") 29 token = os.environ.get("TEST_TOKEN") 30 31 if not api_key: 32 # Importer et exécuter les tests d'authentification si nécessaire 33 from test_auth import test_register_user, test_login, test_create_api_key 34 35 # Créer un utilisateur et une clé API si nécessaire 36 test_register_user() 37 test_login() 38 test_create_api_key() 39 40 from test_auth import test_data as auth_test_data 41 api_key = auth_test_data["api_key"] 42 token = auth_test_data["token"] 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 test_data["token"] = token 51 52 # Créer quelques tâches pour les tests 53 create_test_tasks() 54 55 def get_auth_headers(): 56 """Retourne les en-têtes d'authentification avec la clé API.""" 57 return {"X-API-Key": test_data["api_key"]} 58 59 def get_token_headers(): 60 """Retourne les en-têtes d'authentification avec le token.""" 61 return {"Authorization": f"Bearer {test_data['token']}"} 62 63 def create_test_tasks(): 64 """Crée des tâches de test pour être utilisées dans les tests.""" 65 headers = get_auth_headers() 66 67 # Créer une tâche d'inférence de texte 68 try: 69 response = requests.post( 70 f"{BASE_URL}/api/inference/start", 71 json={ 72 "text": "Ceci est un test pour la gestion des tâches.", 73 "use_segmentation": True, 74 "max_new_tokens": 100 75 }, 76 headers=headers 77 ) 78 79 if response.status_code == 202: 80 data = response.json() 81 test_data["text_task_id"] = data["task_id"] 82 test_data["task_ids"].append(data["task_id"]) 83 except Exception as e: 84 print(f"Erreur lors de la création d'une tâche d'inférence: {e}") 85 86 # Tenter de créer une tâche par lots 87 try: 88 response = requests.post( 89 f"{BASE_URL}/api/inference/batch", 90 json={ 91 "texts": ["Premier texte de test.", "Deuxième texte de test."], 92 "use_segmentation": True, 93 "max_new_tokens": 100 94 }, 95 headers=headers 96 ) 97 98 if response.status_code == 202: 99 data = response.json() 100 test_data["batch_task_id"] = data["task_id"] 101 test_data["task_ids"].append(data["task_id"]) 102 except Exception as e: 103 print(f"Erreur lors de la création d'une tâche par lots: {e}") 104 105 # Créer une tâche pour l'annulation 106 try: 107 response = requests.post( 108 f"{BASE_URL}/api/inference/start", 109 json={ 110 "text": "Cette tâche sera annulée dans les tests.", 111 "use_segmentation": True, 112 "max_new_tokens": 2000 # Grand nombre pour qu'elle prenne du temps 113 }, 114 headers=headers 115 ) 116 117 if response.status_code == 202: 118 data = response.json() 119 test_data["cancelled_task_id"] = data["task_id"] 120 test_data["task_ids"].append(data["task_id"]) 121 except Exception as e: 122 print(f"Erreur lors de la création d'une tâche pour annulation: {e}") 123 124 def test_get_specific_task(): 125 """Teste la récupération d'une tâche spécifique.""" 126 # S'assurer qu'une clé API et au moins une tâche sont disponibles 127 assert test_data["api_key"] is not None, "Aucune clé API disponible pour le test" 128 if not test_data["task_ids"]: 129 pytest.skip("Aucune tâche disponible pour le test") 130 131 # Configurer les headers avec la clé API 132 headers = get_auth_headers() 133 134 # Récupérer une tâche spécifique 135 task_id = test_data["task_ids"][0] 136 response = requests.get(f"{BASE_URL}/api/tasks/{task_id}", headers=headers) 137 assert response.status_code == 200, f"Code de statut inattendu: {response.status_code}, {response.text}" 138 139 # Vérifier la réponse 140 data = response.json() 141 assert "task_id" in data 142 assert data["task_id"] == task_id 143 assert "status" in data 144 assert "type" in data 145 assert "created_at" in data 146 147 def test_get_nonexistent_task(): 148 """Teste la récupération d'une tâche qui n'existe pas.""" 149 # S'assurer qu'une clé API est disponible 150 assert test_data["api_key"] is not None, "Aucune clé API disponible pour le test" 151 152 # Configurer les headers avec la clé API 153 headers = get_auth_headers() 154 155 # Tenter de récupérer une tâche avec un ID inexistant 156 fake_task_id = str(uuid.uuid4()) 157 response = requests.get(f"{BASE_URL}/api/tasks/{fake_task_id}", headers=headers) 158 assert response.status_code == 404, f"Une tâche inexistante devrait retourner 404, mais a retourné: {response.status_code}" 159 160 def test_list_all_tasks(): 161 """Teste la récupération de la liste de toutes les tâches.""" 162 # S'assurer qu'une clé API est disponible 163 assert test_data["api_key"] is not None, "Aucune clé API disponible pour le test" 164 165 # Configurer les headers avec la clé API 166 headers = get_auth_headers() 167 168 # Récupérer la liste des tâches 169 response = requests.get(f"{BASE_URL}/api/tasks", headers=headers) 170 assert response.status_code == 200, f"Code de statut inattendu: {response.status_code}, {response.text}" 171 172 # Vérifier la réponse 173 data = response.json() 174 assert "total" in data 175 assert "tasks" in data 176 assert isinstance(data["tasks"], list) or isinstance(data["tasks"], dict) 177 178 # Vérifier que les tâches créées pendant les tests sont présentes 179 tasks_found = 0 180 181 if isinstance(data["tasks"], list): 182 task_ids = [task.get("task_id") for task in data["tasks"]] 183 for task_id in test_data["task_ids"]: 184 if task_id in task_ids: 185 tasks_found += 1 186 else: # Si c'est un dictionnaire 187 for task_id in test_data["task_ids"]: 188 if task_id in data["tasks"]: 189 tasks_found += 1 190 191 assert tasks_found > 0, "Aucune des tâches créées n'a été trouvée dans la liste" 192 193 def test_filter_tasks_by_status(): 194 """Teste le filtrage des tâches par statut.""" 195 # S'assurer qu'une clé API est disponible 196 assert test_data["api_key"] is not None, "Aucune clé API disponible pour le test" 197 198 # Configurer les headers avec la clé API 199 headers = get_auth_headers() 200 201 # Récupérer la liste des tâches en attente 202 response = requests.get(f"{BASE_URL}/api/tasks?status=pending", headers=headers) 203 assert response.status_code == 200, f"Code de statut inattendu: {response.status_code}, {response.text}" 204 205 # Vérifier la réponse 206 data = response.json() 207 208 # Vérifier que les tâches ont bien le statut demandé 209 if "tasks" in data and data["tasks"]: 210 if isinstance(data["tasks"], list): 211 for task in data["tasks"]: 212 assert task["status"] == "pending", f"Une tâche avec un statut différent a été retournée: {task['status']}" 213 else: # Si c'est un dictionnaire 214 for task_id, task in data["tasks"].items(): 215 assert task["status"] == "pending", f"Une tâche avec un statut différent a été retournée: {task['status']}" 216 217 def test_filter_tasks_by_type(): 218 """Teste le filtrage des tâches par type.""" 219 # S'assurer qu'une clé API est disponible 220 assert test_data["api_key"] is not None, "Aucune clé API disponible pour le test" 221 222 # Configurer les headers avec la clé API 223 headers = get_auth_headers() 224 225 # Récupérer la liste des tâches de type texte 226 response = requests.get(f"{BASE_URL}/api/tasks?task_type=text_inference", headers=headers) 227 assert response.status_code == 200, f"Code de statut inattendu: {response.status_code}, {response.text}" 228 229 # Vérifier la réponse 230 data = response.json() 231 232 # Vérifier que les tâches ont bien le type demandé 233 if "tasks" in data and data["tasks"]: 234 if isinstance(data["tasks"], list): 235 for task in data["tasks"]: 236 assert task["type"] == "text_inference", f"Une tâche avec un type différent a été retournée: {task['type']}" 237 else: # Si c'est un dictionnaire 238 for task_id, task in data["tasks"].items(): 239 assert task["type"] == "text_inference", f"Une tâche avec un type différent a été retournée: {task['type']}" 240 241 def test_pagination(): 242 """Teste la pagination des tâches.""" 243 # S'assurer qu'une clé API est disponible 244 assert test_data["api_key"] is not None, "Aucune clé API disponible pour le test" 245 246 # Configurer les headers avec la clé API 247 headers = get_auth_headers() 248 249 # Récupérer la première page de tâches (limité à 1 résultat) 250 response = requests.get(f"{BASE_URL}/api/tasks?limit=1&offset=0", headers=headers) 251 assert response.status_code == 200, f"Code de statut inattendu: {response.status_code}, {response.text}" 252 253 # Vérifier la réponse 254 data = response.json() 255 256 # Vérifier que la limite est respectée 257 if isinstance(data["tasks"], list): 258 assert len(data["tasks"]) <= 1, f"Plus de tâches que la limite demandée ont été retournées: {len(data['tasks'])}" 259 else: # Si c'est un dictionnaire 260 assert len(data["tasks"]) <= 1, f"Plus de tâches que la limite demandée ont été retournées: {len(data['tasks'])}" 261 262 # Récupérer la deuxième page 263 response = requests.get(f"{BASE_URL}/api/tasks?limit=1&offset=1", headers=headers) 264 assert response.status_code == 200 265 266 # Vérifier la réponse 267 data2 = response.json() 268 269 # Vérifier que la deuxième page est différente de la première 270 if "tasks" in data and "tasks" in data2 and data["tasks"] and data2["tasks"]: 271 if isinstance(data["tasks"], list) and isinstance(data2["tasks"], list): 272 first_task_id = data["tasks"][0]["task_id"] 273 second_task_id = data2["tasks"][0]["task_id"] if data2["tasks"] else None 274 275 if second_task_id: 276 assert first_task_id != second_task_id, "Les résultats paginés sont identiques" 277 278 def test_cancel_running_task(): 279 """Teste l'annulation d'une tâche en cours.""" 280 # S'assurer qu'une clé API et une tâche à annuler sont disponibles 281 assert test_data["api_key"] is not None, "Aucune clé API disponible pour le test" 282 if not test_data["cancelled_task_id"]: 283 pytest.skip("Aucune tâche disponible pour l'annulation") 284 285 # Configurer les headers avec la clé API 286 headers = get_auth_headers() 287 288 # Récupérer l'état actuel de la tâche 289 task_id = test_data["cancelled_task_id"] 290 response = requests.get(f"{BASE_URL}/api/tasks/{task_id}", headers=headers) 291 292 if response.status_code != 200: 293 pytest.skip(f"Impossible de récupérer l'état de la tâche: {response.status_code}") 294 295 data = response.json() 296 297 # Si la tâche est déjà terminée, ignorer ce test 298 if data["status"] not in ["pending", "running"]: 299 pytest.skip(f"La tâche est déjà dans l'état {data['status']} et ne peut pas être annulée") 300 301 # Annuler la tâche 302 response = requests.post(f"{BASE_URL}/api/tasks/{task_id}/cancel", headers=headers) 303 304 # Si l'endpoint d'annulation n'existe pas, essayer la suppression 305 if response.status_code == 404: 306 response = requests.delete(f"{BASE_URL}/api/tasks/{task_id}", headers=headers) 307 308 assert response.status_code in [200, 202], f"Code de statut inattendu: {response.status_code}, {response.text}" 309 310 # Vérifier que la tâche a été annulée 311 response = requests.get(f"{BASE_URL}/api/tasks/{task_id}", headers=headers) 312 assert response.status_code == 200 313 314 data = response.json() 315 assert data["status"] in ["cancelled", "deleted"], f"La tâche n'a pas été annulée correctement: {data['status']}" 316 317 def test_retry_failed_task(): 318 """Teste la réessai d'une tâche échouée.""" 319 # Cette fonctionnalité peut ne pas exister, donc on teste avec précaution 320 # S'assurer qu'une clé API est disponible 321 assert test_data["api_key"] is not None, "Aucune clé API disponible pour le test" 322 323 # Configurer les headers avec la clé API 324 headers = get_auth_headers() 325 326 # Rechercher une tâche échouée 327 response = requests.get(f"{BASE_URL}/api/tasks?status=failed", headers=headers) 328 329 if response.status_code != 200: 330 pytest.skip("Impossible de rechercher des tâches échouées") 331 332 data = response.json() 333 334 # Trouver une tâche échouée 335 failed_task_id = None 336 337 if "tasks" in data and data["tasks"]: 338 if isinstance(data["tasks"], list): 339 for task in data["tasks"]: 340 if task["status"] == "failed": 341 failed_task_id = task["task_id"] 342 break 343 else: # Si c'est un dictionnaire 344 for task_id, task in data["tasks"].items(): 345 if task["status"] == "failed": 346 failed_task_id = task_id 347 break 348 349 if not failed_task_id: 350 pytest.skip("Aucune tâche échouée trouvée pour le test") 351 352 # Tenter de réessayer la tâche 353 response = requests.post(f"{BASE_URL}/api/tasks/{failed_task_id}/retry", headers=headers) 354 355 # Ce test est exploratoire - ne pas faire échouer si l'endpoint n'existe pas 356 if response.status_code == 404: 357 pytest.skip("L'endpoint de réessai n'existe pas") 358 359 # Si l'endpoint existe, vérifier qu'il fonctionne correctement 360 assert response.status_code in [200, 202], f"Code de statut inattendu: {response.status_code}, {response.text}" 361 362 # Vérifier que la tâche a été remise à l'état "pending" ou qu'une nouvelle tâche a été créée 363 data = response.json() 364 if "task_id" in data: 365 new_task_id = data["task_id"] 366 367 # Vérifier l'état de la nouvelle tâche 368 response = requests.get(f"{BASE_URL}/api/tasks/{new_task_id}", headers=headers) 369 assert response.status_code == 200 370 371 task_data = response.json() 372 assert task_data["status"] in ["pending", "running"], f"La tâche n'a pas été correctement réessayée: {task_data['status']}" 373 374 def test_delete_task(): 375 """Teste la suppression d'une tâche.""" 376 # S'assurer qu'une clé API et au moins une tâche sont disponibles 377 assert test_data["api_key"] is not None, "Aucune clé API disponible pour le test" 378 if not test_data["task_ids"]: 379 pytest.skip("Aucune tâche disponible pour le test") 380 381 # Configurer les headers avec la clé API 382 headers = get_auth_headers() 383 384 # Sélectionner une tâche à supprimer 385 task_id_to_delete = test_data["task_ids"][-1] # Utiliser la dernière tâche créée 386 387 # Supprimer la tâche 388 response = requests.delete(f"{BASE_URL}/api/tasks/{task_id_to_delete}", headers=headers) 389 assert response.status_code == 200, f"Code de statut inattendu: {response.status_code}, {response.text}" 390 391 # Vérifier que la tâche a bien été supprimée 392 response = requests.get(f"{BASE_URL}/api/tasks/{task_id_to_delete}", headers=headers) 393 assert response.status_code == 404, f"La tâche n'a pas été correctement supprimée: {response.status_code}" 394 395 # Supprimer l'ID de la liste des tâches de test 396 test_data["task_ids"].remove(task_id_to_delete) 397 398 def test_cleanup(): 399 """Nettoie toutes les tâches créées pendant les tests.""" 400 # S'assurer qu'une clé API est disponible 401 assert test_data["api_key"] is not None, "Aucune clé API disponible pour le test" 402 403 # Configurer les headers avec la clé API 404 headers = get_auth_headers() 405 406 # Supprimer toutes les tâches de test 407 for task_id in test_data["task_ids"]: 408 try: 409 requests.delete(f"{BASE_URL}/api/tasks/{task_id}", headers=headers) 410 except Exception: 411 pass # Ignorer les erreurs lors du nettoyage 412 413 if __name__ == "__main__": 414 # Initialiser les tests 415 setup_module() 416 417 # Exécuter les tests manuellement 418 test_get_specific_task() 419 test_get_nonexistent_task() 420 test_list_all_tasks() 421 test_filter_tasks_by_status() 422 test_filter_tasks_by_type() 423 test_pagination() 424 425 try: 426 test_cancel_running_task() 427 except Exception as e: 428 print(f"Le test d'annulation a échoué: {e}") 429 430 try: 431 test_retry_failed_task() 432 except Exception as e: 433 print(f"Le test de réessai a échoué: {e}") 434 435 test_delete_task() 436 test_cleanup() 437 438 print("Tous les tests de gestion des tâches ont réussi!")