1  """
  2  Configuration globale pour les tests pytest de l'API d'inférence multi-session.
  3  """
  4  
  5  import pytest
  6  import os
  7  import requests
  8  import time
  9  import sys
 10  from pathlib import Path
 11  
 12  # Ajouter le répertoire racine du projet au PYTHONPATH
 13  # Cette ligne résout le problème d'importation des modules
 14  project_root = Path(__file__).parent.parent.absolute()
 15  if str(project_root) not in sys.path:
 16      sys.path.insert(0, str(project_root))
 17      print(f"Ajout du chemin racine au sys.path: {project_root}")
 18  
 19  # URL de base de l'API (ajuster selon l'environnement)
 20  BASE_URL = os.environ.get("API_BASE_URL", "http://localhost:8000")
 21  
 22  # Variables partagées entre les tests
 23  shared_data = {
 24      "user": None,
 25      "token": None,
 26      "api_key": None
 27  }
 28  
 29  @pytest.fixture(scope="session", autouse=True)
 30  def check_api_availability():
 31      """Vérifie si l'API est disponible avant d'exécuter les tests."""
 32      try:
 33          response = requests.get(f"{BASE_URL}/api/health", timeout=5)
 34          if response.status_code != 200:
 35              pytest.skip(f"L'API n'est pas disponible ou a retourné un statut non-OK: {response.status_code}")
 36      except requests.exceptions.RequestException:
 37          pytest.skip("L'API n'est pas accessible")
 38  
 39  @pytest.fixture(scope="session")
 40  def api_url():
 41      """Retourne l'URL de base de l'API."""
 42      return BASE_URL
 43  
 44  @pytest.fixture(scope="session")
 45  def api_key():
 46      """Retourne une clé API valide pour les tests."""
 47      # Essayer d'utiliser une clé API fournie via variable d'environnement
 48      api_key = os.environ.get("TEST_API_KEY")
 49      if api_key:
 50          return api_key
 51      
 52      # Si aucune clé API n'est fournie, essayer de créer un utilisateur et une clé API
 53      if not shared_data["api_key"]:
 54          try:
 55              # Créer un utilisateur
 56              import random
 57              import string
 58              random_suffix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=8))
 59              username = f"testuser_{random_suffix}"
 60              
 61              user_data = {
 62                  "username": username,
 63                  "email": f"{username}@example.com",
 64                  "password": "Password123!",
 65                  "full_name": "Test User"
 66              }
 67              
 68              response = requests.post(f"{BASE_URL}/auth/register", json=user_data)
 69              if response.status_code != 201:
 70                  pytest.skip(f"Impossible de créer un utilisateur de test: {response.text}")
 71              
 72              user_info = response.json()
 73              shared_data["user"] = user_info
 74              
 75              # Se connecter pour obtenir un token
 76              login_data = {
 77                  "username": username,
 78                  "password": "Password123!"
 79              }
 80              
 81              response = requests.post(f"{BASE_URL}/auth/token", data=login_data)
 82              if response.status_code != 200:
 83                  pytest.skip(f"Impossible d'obtenir un token: {response.text}")
 84              
 85              token_info = response.json()
 86              shared_data["token"] = token_info["access_token"]
 87              
 88              # Créer une clé API
 89              headers = {
 90                  "Authorization": f"Bearer {shared_data['token']}"
 91              }
 92              
 93              key_data = {
 94                  "name": "Test API Key"
 95              }
 96              
 97              response = requests.post(f"{BASE_URL}/auth/api-keys", json=key_data, headers=headers)
 98              if response.status_code != 201:
 99                  pytest.skip(f"Impossible de créer une clé API: {response.text}")
100              
101              api_key_info = response.json()
102              shared_data["api_key"] = api_key_info["key"]
103              
104          except Exception as e:
105              pytest.skip(f"Erreur lors de la création d'une clé API de test: {e}")
106      
107      return shared_data["api_key"]
108  
109  @pytest.fixture(scope="session")
110  def auth_headers(api_key):
111      """Retourne les headers d'authentification avec une clé API."""
112      return {
113          "X-API-Key": api_key
114      }
115  
116  @pytest.fixture(scope="session")
117  def wait_for_completion():
118      """Fonction utilitaire pour attendre la fin d'une tâche."""
119      def _wait_for_completion(task_id, auth_headers, max_retries=30, interval=10):
120          """Attend la fin d'une tâche et retourne son résultat."""
121          for i in range(max_retries):
122              response = requests.get(f"{BASE_URL}/api/inference/{task_id}", headers=auth_headers)
123              assert response.status_code == 200
124              
125              data = response.json()
126              
127              if data["status"] == "failed":
128                  pytest.fail(f"La tâche a échoué: {data.get('error', 'Erreur inconnue')}")
129              
130              if data["status"] == "completed":
131                  return data
132                  
133              print(f"Attente de la fin de la tâche... Progression: {data.get('progress', 0):.0f}% (tentative {i+1}/{max_retries})")
134                  
135              time.sleep(interval)
136          
137          pytest.fail(f"La tâche n'est pas terminée après {max_retries} tentatives")
138      
139      return _wait_for_completion