test_subscription.py
1 """ 2 Tests pour les fonctionnalités d'abonnement 3 ------------------------------------------ 4 Ce module teste les fonctionnalités liées aux abonnements, notamment 5 la création de sessions de paiement, la gestion des webhooks Stripe, 6 et les restrictions liées aux différents niveaux d'abonnement. 7 """ 8 9 import pytest 10 import requests 11 import json 12 import os 13 import time 14 import uuid 15 from typing import Dict, Any 16 from unittest import mock 17 18 # URL de base de l'API (ajuster selon l'environnement) 19 BASE_URL = os.environ.get("API_BASE_URL", "http://localhost:8000") 20 21 # Variable pour stocker les informations de test 22 test_data = { 23 "token": None, 24 "api_key": None, 25 "stripe_session_id": None, 26 "user_id": None, 27 "subscription_id": None, 28 "checkout_session_url": None, 29 "customer_id": None 30 } 31 32 def setup_module(): 33 """Configuration initiale pour les tests d'abonnement.""" 34 # Récupérer un token et une clé API valides 35 try: 36 # Essayer d'utiliser une variable d'environnement 37 token = os.environ.get("TEST_TOKEN") 38 api_key = os.environ.get("TEST_API_KEY") 39 40 if not token or not api_key: 41 # Importer et exécuter les tests d'authentification si nécessaire 42 from test_auth import test_register_user, test_login, test_create_api_key 43 44 # Créer un utilisateur et une clé API si nécessaire 45 test_register_user() 46 test_login() 47 test_create_api_key() 48 49 from test_auth import test_data as auth_test_data 50 token = auth_test_data["token"] 51 api_key = auth_test_data["api_key"] 52 test_data["user_id"] = auth_test_data["user_id"] 53 except Exception as e: 54 print(f"Erreur lors de la configuration des tests d'abonnement: {e}") 55 # En cas d'erreur, une clé de test doit être fournie en variable d'environnement 56 token = os.environ.get("TEST_TOKEN") 57 api_key = os.environ.get("TEST_API_KEY") 58 if not token or not api_key: 59 raise Exception("Aucun token ou clé API disponible pour les tests. Définissez TEST_TOKEN et TEST_API_KEY ou exécutez test_auth.py") 60 61 test_data["token"] = token 62 test_data["api_key"] = api_key 63 64 def get_auth_headers(): 65 """Retourne les en-têtes d'authentification avec le token.""" 66 return {"Authorization": f"Bearer {test_data['token']}"} 67 68 def get_api_key_headers(): 69 """Retourne les en-têtes d'authentification avec la clé API.""" 70 return {"X-API-Key": test_data["api_key"]} 71 72 def test_list_subscription_plans(): 73 """Teste la récupération des plans d'abonnement disponibles.""" 74 response = requests.get(f"{BASE_URL}/api/subscriptions/plans") 75 76 # Si l'endpoint n'existe pas ou les abonnements ne sont pas configurés, ignorer le test 77 if response.status_code == 404: 78 pytest.skip("L'endpoint des plans d'abonnement n'est pas disponible") 79 80 assert response.status_code == 200, f"Code de statut inattendu: {response.status_code}, {response.text}" 81 82 # Vérifier la réponse 83 data = response.json() 84 assert "plans" in data 85 assert isinstance(data["plans"], list) 86 87 # Vérifier qu'il y a au moins un plan 88 assert len(data["plans"]) > 0, "Aucun plan d'abonnement trouvé" 89 90 # Vérifier la structure des plans 91 for plan in data["plans"]: 92 assert "id" in plan 93 assert "name" in plan 94 assert "priceId" in plan or "price_id" in plan 95 assert "currency" in plan 96 # D'autres champs facultatifs peuvent être vérifiés 97 98 def test_current_subscription(): 99 """Teste la récupération de l'abonnement actuel de l'utilisateur.""" 100 # S'assurer qu'un token a été obtenu 101 assert test_data["token"] is not None, "Aucun token obtenu pour le test" 102 103 # Configurer les headers avec le token 104 headers = get_auth_headers() 105 106 # Récupérer l'abonnement actuel 107 response = requests.get(f"{BASE_URL}/api/subscriptions/current", headers=headers) 108 109 # Si l'endpoint n'existe pas, ignorer le test 110 if response.status_code == 404: 111 pytest.skip("L'endpoint d'abonnement actuel n'est pas disponible") 112 113 assert response.status_code == 200, f"Code de statut inattendu: {response.status_code}, {response.text}" 114 115 # Vérifier la réponse 116 data = response.json() 117 assert "subscription" in data 118 119 # Remarque: Si l'utilisateur n'a pas d'abonnement, subscription peut être null 120 121 def test_create_checkout_session(): 122 """Teste la création d'une session de paiement.""" 123 # S'assurer qu'un token a été obtenu 124 assert test_data["token"] is not None, "Aucun token obtenu pour le test" 125 126 # Configurer les headers avec le token 127 headers = get_auth_headers() 128 129 # Récupérer d'abord la liste des plans pour choisir un plan à tester 130 plans_response = requests.get(f"{BASE_URL}/api/subscriptions/plans") 131 132 # Si l'endpoint n'existe pas ou les abonnements ne sont pas configurés, ignorer le test 133 if plans_response.status_code == 404: 134 pytest.skip("L'endpoint des plans d'abonnement n'est pas disponible") 135 136 plans_data = plans_response.json() 137 if not plans_data.get("plans"): 138 pytest.skip("Aucun plan d'abonnement disponible pour le test") 139 140 # Choisir le premier plan disponible 141 plan_id = plans_data["plans"][0]["id"] 142 143 # Données pour la création de session de paiement 144 checkout_data = { 145 "plan": plan_id, 146 "success_url": f"{BASE_URL}/payment/success?session_id={{CHECKOUT_SESSION_ID}}", 147 "cancel_url": f"{BASE_URL}/payment/cancel" 148 } 149 150 # Envoyer la requête pour créer une session de paiement 151 response = requests.post(f"{BASE_URL}/api/subscriptions/checkout", json=checkout_data, headers=headers) 152 153 # Si l'endpoint n'existe pas ou Stripe n'est pas configuré, ignorer le test 154 if response.status_code == 404 or response.status_code == 500 and "stripe" in response.text.lower(): 155 pytest.skip("L'endpoint de création de session n'est pas disponible ou Stripe n'est pas configuré") 156 157 assert response.status_code == 200, f"Code de statut inattendu: {response.status_code}, {response.text}" 158 159 # Vérifier la réponse 160 data = response.json() 161 assert "sessionId" in data or "session_id" in data 162 assert "url" in data 163 164 # Sauvegarder l'ID de session pour les tests suivants 165 test_data["stripe_session_id"] = data.get("sessionId") or data.get("session_id") 166 test_data["checkout_session_url"] = data["url"] 167 168 def test_mock_webhook_payment_succeeded(): 169 """Teste le traitement d'un webhook de paiement réussi (simulé).""" 170 # Ce test simule un webhook Stripe pour un paiement réussi 171 # Dans un environnement réel, ce webhook est appelé par Stripe directement 172 173 # S'assurer qu'un ID de session a été obtenu 174 if not test_data.get("stripe_session_id"): 175 pytest.skip("Aucun ID de session Stripe disponible pour le test de webhook") 176 177 # Créer un événement Stripe simulé 178 # Remarque: Ceci est une approximation, l'événement réel serait beaucoup plus complexe 179 mock_event = { 180 "id": f"evt_{uuid.uuid4().hex}", 181 "type": "invoice.payment_succeeded", 182 "data": { 183 "object": { 184 "id": f"in_{uuid.uuid4().hex}", 185 "subscription": f"sub_{uuid.uuid4().hex}", 186 "customer": test_data.get("customer_id", f"cus_{uuid.uuid4().hex}"), 187 "amount_paid": 2000, # 20.00€/$ en centimes 188 "status": "paid" 189 } 190 } 191 } 192 193 # Configurer la signature webhook (simulée) 194 timestamp = int(time.time()) 195 webhook_secret = os.environ.get("STRIPE_WEBHOOK_SECRET", "whsec_test") 196 signature = f"t={timestamp},v1=mock_signature,v0=mock_signature" 197 198 headers = { 199 "Stripe-Signature": signature 200 } 201 202 # Envoyer la requête webhook 203 response = requests.post( 204 f"{BASE_URL}/api/subscriptions/webhook", 205 json=mock_event, 206 headers=headers 207 ) 208 209 # Ce test est informatif et exploratoire 210 # Il n'est pas censé réussir dans un environnement de test sauf si le webhook est correctement simulé 211 # Nous vérifions simplement que l'endpoint répond et ne plante pas 212 if response.status_code != 404: 213 assert response.status_code in [200, 400], f"Code de statut inattendu: {response.status_code}, {response.text}" 214 print(f"Réponse du webhook: {response.status_code} - {response.text}") 215 216 def test_subscription_levels_access(): 217 """Teste l'accès aux fonctionnalités selon les niveaux d'abonnement.""" 218 # S'assurer qu'une clé API est disponible 219 assert test_data["api_key"] is not None, "Aucune clé API disponible pour le test" 220 221 # Configurer les headers avec la clé API 222 headers = get_api_key_headers() 223 224 # Tester l'accès à une fonctionnalité de base (disponible pour tous les niveaux) 225 response = requests.get(f"{BASE_URL}/api/tasks", headers=headers) 226 assert response.status_code == 200, "Accès refusé à une fonctionnalité de base" 227 228 # Tester l'accès à une fonctionnalité avancée (traitement par lots) 229 batch_data = { 230 "texts": ["Premier texte.", "Deuxième texte."], 231 "use_segmentation": True, 232 "max_new_tokens": 100 233 } 234 235 response = requests.post(f"{BASE_URL}/api/inference/batch", json=batch_data, headers=headers) 236 237 # Si le traitement par lots est limité aux abonnements premium, la réponse devrait indiquer cela 238 # Ce test permet de vérifier que les restrictions sont en place, même si nous n'avons pas accès 239 if response.status_code == 403 and "abonnement" in response.text.lower(): 240 print("Fonctionnalité de traitement par lots correctement restreinte par niveau d'abonnement") 241 elif response.status_code == 202: 242 print("Traitement par lots disponible avec le niveau d'abonnement actuel") 243 else: 244 print(f"Code de statut inattendu: {response.status_code}, {response.text}") 245 246 def test_cancel_subscription(): 247 """Teste l'annulation d'un abonnement.""" 248 # S'assurer qu'un token a été obtenu 249 assert test_data["token"] is not None, "Aucun token obtenu pour le test" 250 251 # Si aucun ID d'abonnement n'est disponible, ignorer ce test 252 if not test_data.get("subscription_id"): 253 pytest.skip("Aucun ID d'abonnement disponible pour le test d'annulation") 254 255 # Configurer les headers avec le token 256 headers = get_auth_headers() 257 258 # Données pour l'annulation d'abonnement 259 cancel_data = { 260 "subscriptionId": test_data["subscription_id"] 261 } 262 263 # Envoyer la requête pour annuler l'abonnement 264 response = requests.post(f"{BASE_URL}/api/subscriptions/cancel", json=cancel_data, headers=headers) 265 266 # Si l'endpoint n'existe pas ou Stripe n'est pas configuré, ignorer le test 267 if response.status_code == 404 or response.status_code == 500 and "stripe" in response.text.lower(): 268 pytest.skip("L'endpoint d'annulation n'est pas disponible ou Stripe n'est pas configuré") 269 270 # Si l'abonnement n'existe pas ou n'appartient pas à l'utilisateur, c'est attendu dans ce test 271 if response.status_code == 403 or response.status_code == 404: 272 print(f"Abonnement non trouvé ou non autorisé: {response.text}") 273 return 274 275 assert response.status_code == 200, f"Code de statut inattendu: {response.status_code}, {response.text}" 276 277 # Vérifier la réponse 278 data = response.json() 279 assert "subscription" in data 280 assert data["subscription"]["cancel_at_period_end"] == True 281 282 def test_reactivate_subscription(): 283 """Teste la réactivation d'un abonnement annulé.""" 284 # S'assurer qu'un token a été obtenu 285 assert test_data["token"] is not None, "Aucun token obtenu pour le test" 286 287 # Si aucun ID d'abonnement n'est disponible, ignorer ce test 288 if not test_data.get("subscription_id"): 289 pytest.skip("Aucun ID d'abonnement disponible pour le test de réactivation") 290 291 # Configurer les headers avec le token 292 headers = get_auth_headers() 293 294 # Données pour la réactivation d'abonnement 295 reactivate_data = { 296 "subscriptionId": test_data["subscription_id"] 297 } 298 299 # Envoyer la requête pour réactiver l'abonnement 300 response = requests.post(f"{BASE_URL}/api/subscriptions/reactivate", json=reactivate_data, headers=headers) 301 302 # Si l'endpoint n'existe pas ou Stripe n'est pas configuré, ignorer le test 303 if response.status_code == 404 or response.status_code == 500 and "stripe" in response.text.lower(): 304 pytest.skip("L'endpoint de réactivation n'est pas disponible ou Stripe n'est pas configuré") 305 306 # Si l'abonnement n'existe pas ou n'appartient pas à l'utilisateur, c'est attendu dans ce test 307 if response.status_code == 403 or response.status_code == 404: 308 print(f"Abonnement non trouvé ou non autorisé: {response.text}") 309 return 310 311 assert response.status_code == 200, f"Code de statut inattendu: {response.status_code}, {response.text}" 312 313 # Vérifier la réponse 314 data = response.json() 315 assert "subscription" in data 316 assert data["subscription"]["cancel_at_period_end"] == False 317 318 def test_get_invoices(): 319 """Teste la récupération des factures de l'utilisateur.""" 320 # S'assurer qu'un token a été obtenu 321 assert test_data["token"] is not None, "Aucun token obtenu pour le test" 322 323 # Configurer les headers avec le token 324 headers = get_auth_headers() 325 326 # Récupérer les factures 327 response = requests.get(f"{BASE_URL}/api/subscriptions/invoices", headers=headers) 328 329 # Si l'endpoint n'existe pas ou Stripe n'est pas configuré, ignorer le test 330 if response.status_code == 404 or response.status_code == 500 and "stripe" in response.text.lower(): 331 pytest.skip("L'endpoint des factures n'est pas disponible ou Stripe n'est pas configuré") 332 333 assert response.status_code == 200, f"Code de statut inattendu: {response.status_code}, {response.text}" 334 335 # Vérifier la réponse 336 data = response.json() 337 assert "invoices" in data 338 assert isinstance(data["invoices"], list) 339 340 def test_payment_methods(): 341 """Teste la récupération des méthodes de paiement de l'utilisateur.""" 342 # S'assurer qu'un token a été obtenu 343 assert test_data["token"] is not None, "Aucun token obtenu pour le test" 344 345 # Configurer les headers avec le token 346 headers = get_auth_headers() 347 348 # Récupérer les méthodes de paiement 349 response = requests.get(f"{BASE_URL}/api/subscriptions/payment-methods", headers=headers) 350 351 # Si l'endpoint n'existe pas ou Stripe n'est pas configuré, ignorer le test 352 if response.status_code == 404 or response.status_code == 500 and "stripe" in response.text.lower(): 353 pytest.skip("L'endpoint des méthodes de paiement n'est pas disponible ou Stripe n'est pas configuré") 354 355 assert response.status_code == 200, f"Code de statut inattendu: {response.status_code}, {response.text}" 356 357 # Vérifier la réponse 358 data = response.json() 359 assert "payment_methods" in data 360 assert isinstance(data["payment_methods"], list) 361 362 if __name__ == "__main__": 363 # Initialiser les tests 364 setup_module() 365 366 # Exécuter les tests manuellement 367 test_list_subscription_plans() 368 test_current_subscription() 369 370 try: 371 test_create_checkout_session() 372 except Exception as e: 373 print(f"Erreur lors de la création de session de paiement: {e}") 374 375 try: 376 test_mock_webhook_payment_succeeded() 377 except Exception as e: 378 print(f"Erreur lors du test du webhook: {e}") 379 380 test_subscription_levels_access() 381 382 try: 383 test_cancel_subscription() 384 test_reactivate_subscription() 385 except Exception as e: 386 print(f"Erreur lors des tests d'annulation/réactivation: {e}") 387 388 test_get_invoices() 389 test_payment_methods() 390 391 print("Tous les tests d'abonnement ont réussi!")