test_stripe_integration
1 import pytest 2 import stripe 3 import json 4 import base64 5 import hmac 6 import hashlib 7 import time 8 import os 9 from unittest.mock import patch, MagicMock 10 from fastapi.testclient import TestClient 11 from datetime import datetime, timedelta 12 13 # Importer l'application principale 14 from main import app 15 from subscription_routes import webhook_secret, router as subscription_router 16 from auth_models import User, ApiKeyLevel 17 from database import update_user 18 19 # Informations de test 20 TEST_STRIPE_SECRET = 'sk_test_51XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX' 21 TEST_WEBHOOK_SECRET = 'whsec_XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX' 22 TEST_CUSTOMER_ID = 'cus_XXXXXXXXXXXX' 23 TEST_SUBSCRIPTION_ID = 'sub_XXXXXXXXXXXX' 24 TEST_PAYMENT_METHOD_ID = 'pm_XXXXXXXXXXXX' 25 TEST_PRICE_ID = 'price_1XXXXXXXXXXXXXXXXXXXbasic' 26 TEST_INVOICE_ID = 'in_XXXXXXXXXXXX' 27 28 # Client de test 29 client = TestClient(app) 30 31 # Initialisation d'un utilisateur de test 32 @pytest.fixture 33 def test_user(): 34 """Crée un utilisateur de test avec un token JWT valide.""" 35 user = User( 36 id="test123", 37 username="testuser", 38 email="test@example.com", 39 hashed_password="hashed_password123", 40 full_name="Test User", 41 roles=["user"], 42 subscription=ApiKeyLevel.FREE, 43 stripe_customer_id=TEST_CUSTOMER_ID, 44 created_at=datetime.utcnow() 45 ) 46 47 # Sauvegarder l'utilisateur dans la base de données simulée 48 from database import users_db 49 users_db[user.id] = user 50 51 # Créer un token valide pour l'utilisateur 52 from auth import create_access_token 53 access_token = create_access_token( 54 data={"sub": user.id, "name": user.full_name, "email": user.email, 55 "roles": user.roles, "api_level": user.subscription}, 56 expires_delta=timedelta(minutes=30) 57 ) 58 59 return {"user": user, "token": access_token} 60 61 @pytest.fixture 62 def test_headers(test_user): 63 """Crée les headers d'authentification pour les tests.""" 64 return { 65 "Authorization": f"Bearer {test_user['token']}" 66 } 67 68 # Mock pour Stripe 69 @pytest.fixture 70 def mock_stripe(): 71 """Mock les fonctions Stripe pour les tests.""" 72 # Mock Customer 73 customer_mock = MagicMock() 74 customer_mock.id = TEST_CUSTOMER_ID 75 customer_mock.email = "test@example.com" 76 customer_mock.invoice_settings = MagicMock() 77 customer_mock.invoice_settings.default_payment_method = TEST_PAYMENT_METHOD_ID 78 79 # Mock PaymentMethod 80 payment_method_mock = MagicMock() 81 payment_method_mock.id = TEST_PAYMENT_METHOD_ID 82 payment_method_mock.type = "card" 83 payment_method_mock.card = MagicMock() 84 payment_method_mock.card.last4 = "4242" 85 payment_method_mock.card.brand = "visa" 86 payment_method_mock.card.exp_month = 12 87 payment_method_mock.card.exp_year = 2025 88 89 payment_methods_mock = MagicMock() 90 payment_methods_mock.data = [payment_method_mock] 91 92 # Mock Price 93 price_mock = MagicMock() 94 price_mock.id = TEST_PRICE_ID 95 price_mock.unit_amount = 1900 96 price_mock.currency = "usd" 97 price_mock.recurring = MagicMock() 98 price_mock.recurring.interval = "month" 99 price_mock.product = MagicMock() 100 price_mock.product.description = "Basic Plan for API access" 101 price_mock.product.metadata = {"features": "Feature 1,Feature 2,Feature 3"} 102 103 # Mock Subscription 104 subscription_mock = MagicMock() 105 subscription_mock.id = TEST_SUBSCRIPTION_ID 106 subscription_mock.status = "active" 107 subscription_mock.current_period_end = int(time.time()) + 30 * 24 * 60 * 60 # 30 jours 108 subscription_mock.cancel_at_period_end = False 109 subscription_mock.customer = TEST_CUSTOMER_ID 110 subscription_mock.items = MagicMock() 111 subscription_mock.items.data = [MagicMock()] 112 subscription_mock.items.data[0].price = price_mock 113 subscription_mock.latest_invoice = MagicMock() 114 subscription_mock.latest_invoice.payment_intent = MagicMock() 115 subscription_mock.latest_invoice.payment_intent.client_secret = "pi_secret_XXXXXX" 116 117 subscriptions_mock = MagicMock() 118 subscriptions_mock.data = [subscription_mock] 119 120 # Mock Invoice 121 invoice_mock = MagicMock() 122 invoice_mock.id = TEST_INVOICE_ID 123 invoice_mock.number = "INV-001" 124 invoice_mock.total = 1900 125 invoice_mock.currency = "usd" 126 invoice_mock.status = "paid" 127 invoice_mock.created = int(time.time()) - 7 * 24 * 60 * 60 # 7 jours 128 invoice_mock.due_date = int(time.time()) - 7 * 24 * 60 * 60 # 7 jours 129 invoice_mock.invoice_pdf = "https://example.com/invoice.pdf" 130 131 invoices_mock = MagicMock() 132 invoices_mock.data = [invoice_mock] 133 134 # Patch les méthodes Stripe 135 with patch('stripe.Customer.create', return_value=customer_mock) as create_customer_mock, \ 136 patch('stripe.Customer.retrieve', return_value=customer_mock) as retrieve_customer_mock, \ 137 patch('stripe.Customer.modify', return_value=customer_mock) as modify_customer_mock, \ 138 patch('stripe.PaymentMethod.attach', return_value=payment_method_mock) as attach_payment_method_mock, \ 139 patch('stripe.PaymentMethod.detach', return_value=payment_method_mock) as detach_payment_method_mock, \ 140 patch('stripe.PaymentMethod.list', return_value=payment_methods_mock) as list_payment_methods_mock, \ 141 patch('stripe.Price.retrieve', return_value=price_mock) as retrieve_price_mock, \ 142 patch('stripe.Subscription.create', return_value=subscription_mock) as create_subscription_mock, \ 143 patch('stripe.Subscription.retrieve', return_value=subscription_mock) as retrieve_subscription_mock, \ 144 patch('stripe.Subscription.modify', return_value=subscription_mock) as modify_subscription_mock, \ 145 patch('stripe.Subscription.list', return_value=subscriptions_mock) as list_subscriptions_mock, \ 146 patch('stripe.Invoice.list', return_value=invoices_mock) as list_invoices_mock, \ 147 patch('stripe.Webhook.construct_event') as webhook_construct_mock: 148 149 yield { 150 "customer": customer_mock, 151 "payment_method": payment_method_mock, 152 "price": price_mock, 153 "subscription": subscription_mock, 154 "invoice": invoice_mock, 155 "create_customer": create_customer_mock, 156 "retrieve_customer": retrieve_customer_mock, 157 "modify_customer": modify_customer_mock, 158 "attach_payment_method": attach_payment_method_mock, 159 "detach_payment_method": detach_payment_method_mock, 160 "list_payment_methods": list_payment_methods_mock, 161 "retrieve_price": retrieve_price_mock, 162 "create_subscription": create_subscription_mock, 163 "retrieve_subscription": retrieve_subscription_mock, 164 "modify_subscription": modify_subscription_mock, 165 "list_subscriptions": list_subscriptions_mock, 166 "list_invoices": list_invoices_mock, 167 "webhook_construct": webhook_construct_mock 168 } 169 170 # Tests des routes d'abonnement 171 def test_get_current_subscription(test_headers, mock_stripe): 172 """Teste la récupération de l'abonnement actuel.""" 173 response = client.get("/api/subscriptions/current", headers=test_headers) 174 assert response.status_code == 200 175 data = response.json() 176 assert "subscription" in data 177 assert data["subscription"]["id"] == TEST_SUBSCRIPTION_ID 178 assert data["subscription"]["status"] == "active" 179 assert not data["subscription"]["cancel_at_period_end"] 180 181 def test_get_plans(test_headers, mock_stripe): 182 """Teste la récupération des plans disponibles.""" 183 response = client.get("/api/subscriptions/plans", headers=test_headers) 184 assert response.status_code == 200 185 data = response.json() 186 assert "plans" in data 187 assert len(data["plans"]) > 0 188 assert "price" in data["plans"][0] 189 assert "features" in data["plans"][0] 190 191 def test_get_invoices(test_headers, mock_stripe): 192 """Teste la récupération des factures.""" 193 response = client.get("/api/subscriptions/invoices", headers=test_headers) 194 assert response.status_code == 200 195 data = response.json() 196 assert "invoices" in data 197 assert len(data["invoices"]) > 0 198 assert data["invoices"][0]["id"] == TEST_INVOICE_ID 199 assert data["invoices"][0]["status"] == "paid" 200 201 def test_get_payment_methods(test_headers, mock_stripe): 202 """Teste la récupération des méthodes de paiement.""" 203 response = client.get("/api/subscriptions/payment-methods", headers=test_headers) 204 assert response.status_code == 200 205 data = response.json() 206 assert "payment_methods" in data 207 assert len(data["payment_methods"]) > 0 208 assert data["payment_methods"][0]["id"] == TEST_PAYMENT_METHOD_ID 209 assert data["payment_methods"][0]["last4"] == "4242" 210 211 def test_create_subscription(test_headers, mock_stripe): 212 """Teste la création d'un abonnement.""" 213 payload = { 214 "priceId": TEST_PRICE_ID 215 } 216 response = client.post("/api/subscriptions/create-subscription", json=payload, headers=test_headers) 217 assert response.status_code == 200 218 data = response.json() 219 assert "clientSecret" in data 220 assert "subscription" in data 221 assert data["subscription"]["id"] == TEST_SUBSCRIPTION_ID 222 assert data["subscription"]["status"] == "active" 223 224 # Vérifier que la méthode Stripe a été appelée avec les bons paramètres 225 mock_stripe["create_subscription"].assert_called_once() 226 call_args = mock_stripe["create_subscription"].call_args[1] 227 assert call_args["customer"] == TEST_CUSTOMER_ID 228 assert call_args["items"][0]["price"] == TEST_PRICE_ID 229 230 def test_cancel_subscription(test_headers, mock_stripe): 231 """Teste l'annulation d'un abonnement.""" 232 # Configurer le mock pour l'annulation 233 mock_stripe["modify_subscription"].return_value.cancel_at_period_end = True 234 235 payload = { 236 "subscriptionId": TEST_SUBSCRIPTION_ID 237 } 238 response = client.post("/api/subscriptions/cancel", json=payload, headers=test_headers) 239 assert response.status_code == 200 240 data = response.json() 241 assert "subscription" in data 242 assert data["subscription"]["id"] == TEST_SUBSCRIPTION_ID 243 assert data["subscription"]["cancel_at_period_end"] == True 244 245 # Vérifier que la méthode Stripe a été appelée avec les bons paramètres 246 mock_stripe["modify_subscription"].assert_called_once_with( 247 TEST_SUBSCRIPTION_ID, 248 cancel_at_period_end=True 249 ) 250 251 def test_reactivate_subscription(test_headers, mock_stripe): 252 """Teste la réactivation d'un abonnement.""" 253 # Configurer le mock pour la réactivation 254 mock_stripe["modify_subscription"].return_value.cancel_at_period_end = False 255 256 payload = { 257 "subscriptionId": TEST_SUBSCRIPTION_ID 258 } 259 response = client.post("/api/subscriptions/reactivate", json=payload, headers=test_headers) 260 assert response.status_code == 200 261 data = response.json() 262 assert "subscription" in data 263 assert data["subscription"]["id"] == TEST_SUBSCRIPTION_ID 264 assert data["subscription"]["cancel_at_period_end"] == False 265 266 # Vérifier que la méthode Stripe a été appelée avec les bons paramètres 267 mock_stripe["modify_subscription"].assert_called_once_with( 268 TEST_SUBSCRIPTION_ID, 269 cancel_at_period_end=False 270 ) 271 272 def test_update_payment_method(test_headers, mock_stripe): 273 """Teste la mise à jour de la méthode de paiement.""" 274 payload = { 275 "paymentMethodId": TEST_PAYMENT_METHOD_ID 276 } 277 response = client.post("/api/subscriptions/update-payment-method", json=payload, headers=test_headers) 278 assert response.status_code == 200 279 data = response.json() 280 assert data["success"] == True 281 assert "payment_method" in data 282 assert data["payment_method"]["id"] == TEST_PAYMENT_METHOD_ID 283 284 # Vérifier que les méthodes Stripe ont été appelées avec les bons paramètres 285 mock_stripe["attach_payment_method"].assert_called_once_with( 286 TEST_PAYMENT_METHOD_ID, 287 customer=TEST_CUSTOMER_ID 288 ) 289 mock_stripe["modify_customer"].assert_called_once() 290 call_args = mock_stripe["modify_customer"].call_args[1] 291 assert call_args["invoice_settings"]["default_payment_method"] == TEST_PAYMENT_METHOD_ID 292 293 def test_delete_payment_method(test_headers, mock_stripe): 294 """Teste la suppression d'une méthode de paiement.""" 295 response = client.delete(f"/api/subscriptions/payment-methods/{TEST_PAYMENT_METHOD_ID}", headers=test_headers) 296 assert response.status_code == 200 297 data = response.json() 298 assert data["success"] == True 299 300 # Vérifier que la méthode Stripe a été appelée avec les bons paramètres 301 mock_stripe["detach_payment_method"].assert_called_once_with(TEST_PAYMENT_METHOD_ID) 302 303 def test_webhook_payment_succeeded(mock_stripe): 304 """Teste le webhook de paiement réussi.""" 305 # Créer un événement Stripe simulé 306 event_data = { 307 "id": "evt_test_webhook_payment_succeeded", 308 "type": "invoice.payment_succeeded", 309 "data": { 310 "object": { 311 "id": TEST_INVOICE_ID, 312 "customer": TEST_CUSTOMER_ID, 313 "subscription": TEST_SUBSCRIPTION_ID, 314 "amount_paid": 1900, 315 "status": "paid" 316 } 317 } 318 } 319 320 # Configurer le mock pour le webhook 321 mock_stripe["webhook_construct"].return_value = event_data 322 323 # Générer une signature valide (dans un cas réel) 324 timestamp = int(time.time()) 325 payload = json.dumps(event_data) 326 signature = compute_webhook_signature(payload, timestamp, TEST_WEBHOOK_SECRET) 327 328 # Envoyer la requête webhook 329 response = client.post("/api/subscriptions/webhook", 330 headers={"Stripe-Signature": f"t={timestamp},v1={signature}"}, 331 content=payload) 332 333 assert response.status_code == 200 334 assert response.json()["status"] == "success" 335 336 # Vérifier que l'événement a été construit 337 mock_stripe["webhook_construct"].assert_called_once() 338 339 def test_webhook_subscription_updated(mock_stripe): 340 """Teste le webhook de mise à jour d'abonnement.""" 341 # Créer un événement Stripe simulé 342 event_data = { 343 "id": "evt_test_webhook_subscription_updated", 344 "type": "customer.subscription.updated", 345 "data": { 346 "object": { 347 "id": TEST_SUBSCRIPTION_ID, 348 "customer": TEST_CUSTOMER_ID, 349 "status": "active", 350 "items": { 351 "data": [ 352 { 353 "price": { 354 "id": TEST_PRICE_ID 355 } 356 } 357 ] 358 } 359 } 360 } 361 } 362 363 # Configurer le mock pour le webhook 364 mock_stripe["webhook_construct"].return_value = event_data 365 366 # Générer une signature valide (dans un cas réel) 367 timestamp = int(time.time()) 368 payload = json.dumps(event_data) 369 signature = compute_webhook_signature(payload, timestamp, TEST_WEBHOOK_SECRET) 370 371 # Envoyer la requête webhook 372 response = client.post("/api/subscriptions/webhook", 373 headers={"Stripe-Signature": f"t={timestamp},v1={signature}"}, 374 content=payload) 375 376 assert response.status_code == 200 377 assert response.json()["status"] == "success" 378 379 # Vérifier que l'événement a été construit 380 mock_stripe["webhook_construct"].assert_called_once() 381 382 # Fonctions utilitaires pour les tests 383 def compute_webhook_signature(payload, timestamp, secret): 384 """ 385 Calcule une signature de webhook Stripe valide pour les tests. 386 """ 387 signed_payload = f"{timestamp}.{payload}" 388 signature = hmac.new( 389 secret.encode('utf-8'), 390 signed_payload.encode('utf-8'), 391 hashlib.sha256 392 ).hexdigest() 393 return signature 394 395 if __name__ == "__main__": 396 # Exécution directe des tests 397 pytest.main(["-xvs", __file__])