Initial commit: existing turf_saas codebase
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
ML Engineer
463
tests/e2e/test_scenarios.py
Normal file
463
tests/e2e/test_scenarios.py
Normal file
@@ -0,0 +1,463 @@
|
||||
"""
|
||||
Tests E2E Playwright — SaaS Turf Prédictions IA
|
||||
Sprint 8 — QA, Beta Fermee, Go/No-Go
|
||||
Ticket: HRT-34
|
||||
|
||||
Scénarios couverts :
|
||||
1. Inscription → choix plan free → voir top 3
|
||||
2. Upgrade premium → Stripe checkout → accès toutes courses
|
||||
3. Abonnement pro → export CSV → accès API
|
||||
4. Annulation abonnement → downgrade free
|
||||
|
||||
Navigateurs : Chrome, Firefox, Safari mobile (via Playwright)
|
||||
Screenshots automatiques sur échec
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
import pytest
|
||||
from playwright.async_api import async_playwright, Page, Browser, BrowserContext
|
||||
|
||||
# === Configuration ===
|
||||
BASE_URL = os.environ.get("APP_URL", "http://localhost:8792")
|
||||
SCREENSHOT_DIR = Path("tests/screenshots")
|
||||
SCREENSHOT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Stripe test card (mode test uniquement)
|
||||
STRIPE_TEST_CARD = "4242 4242 4242 4242"
|
||||
STRIPE_EXPIRY = "12/30"
|
||||
STRIPE_CVC = "123"
|
||||
STRIPE_ZIP = "75001"
|
||||
|
||||
# Credentials beta
|
||||
BETA_PROMO_CODE = "BETA2026"
|
||||
TEST_USER_EMAIL_BASE = "testqa+{}@h3r7.tech"
|
||||
TEST_USER_PASSWORD = "TestQA_2026!"
|
||||
|
||||
|
||||
def screenshot_on_fail(test_name: str, page: Page):
|
||||
"""Decorator/helper to take screenshot on test failure."""
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
path = SCREENSHOT_DIR / f"FAIL_{test_name}_{timestamp}.png"
|
||||
return str(path)
|
||||
|
||||
|
||||
# === Fixtures ===
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def event_loop():
|
||||
loop = asyncio.get_event_loop_policy().new_event_loop()
|
||||
yield loop
|
||||
loop.close()
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
async def playwright_instance():
|
||||
async with async_playwright() as p:
|
||||
yield p
|
||||
|
||||
|
||||
@pytest.fixture(params=["chromium", "firefox", "webkit"])
|
||||
async def browser(playwright_instance, request):
|
||||
browser_type = getattr(playwright_instance, request.param)
|
||||
# Webkit simule Safari mobile
|
||||
if request.param == "webkit":
|
||||
b = await browser_type.launch(headless=True)
|
||||
yield b, "safari_mobile"
|
||||
else:
|
||||
b = await browser_type.launch(headless=True)
|
||||
yield b, request.param
|
||||
await b.close()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def context_page(browser):
|
||||
b, browser_name = browser
|
||||
if browser_name == "safari_mobile":
|
||||
# Simule iPhone 13
|
||||
ctx = await b.new_context(
|
||||
viewport={"width": 390, "height": 844},
|
||||
user_agent=(
|
||||
"Mozilla/5.0 (iPhone; CPU iPhone OS 15_0 like Mac OS X) "
|
||||
"AppleWebKit/605.1.15 (KHTML, like Gecko) Version/15.0 Mobile/15E148 Safari/604.1"
|
||||
),
|
||||
)
|
||||
else:
|
||||
ctx = await b.new_context(viewport={"width": 1280, "height": 800})
|
||||
page = await ctx.new_page()
|
||||
yield page, browser_name
|
||||
await ctx.close()
|
||||
|
||||
|
||||
# === Helper functions ===
|
||||
|
||||
|
||||
async def register_user(page: Page, email: str, password: str, promo: str = None):
|
||||
"""Inscrit un nouvel utilisateur."""
|
||||
await page.goto(f"{BASE_URL}/register")
|
||||
await page.fill('[name="email"]', email)
|
||||
await page.fill('[name="password"]', password)
|
||||
await page.fill('[name="confirm_password"]', password)
|
||||
if promo:
|
||||
promo_field = page.locator('[name="promo_code"]')
|
||||
if await promo_field.count() > 0:
|
||||
await promo_field.fill(promo)
|
||||
await page.click('[type="submit"]')
|
||||
await page.wait_for_url(f"{BASE_URL}/**", timeout=10_000)
|
||||
|
||||
|
||||
async def login_user(page: Page, email: str, password: str):
|
||||
"""Connecte un utilisateur existant."""
|
||||
await page.goto(f"{BASE_URL}/login")
|
||||
await page.fill('[name="email"]', email)
|
||||
await page.fill('[name="password"]', password)
|
||||
await page.click('[type="submit"]')
|
||||
await page.wait_for_url(f"{BASE_URL}/**", timeout=10_000)
|
||||
|
||||
|
||||
async def fill_stripe_form(page: Page):
|
||||
"""Remplit le formulaire Stripe Checkout (mode test)."""
|
||||
# Stripe embeds iframe — on attend le frame
|
||||
stripe_frame = page.frame_locator('iframe[name^="__privateStripeFrame"]').first
|
||||
await stripe_frame.locator('[placeholder="Card number"]').fill(STRIPE_TEST_CARD)
|
||||
await stripe_frame.locator('[placeholder="MM / YY"]').fill(STRIPE_EXPIRY)
|
||||
await stripe_frame.locator('[placeholder="CVC"]').fill(STRIPE_CVC)
|
||||
await stripe_frame.locator('[placeholder="ZIP"]').fill(STRIPE_ZIP)
|
||||
|
||||
|
||||
# === Test Scénario 1 : Inscription → plan free → voir top 3 ===
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_inscription_plan_free_top3(context_page):
|
||||
"""
|
||||
Scénario 1 : Inscription → choix plan free → voir top 3 prédictions
|
||||
"""
|
||||
page, browser_name = context_page
|
||||
test_name = f"test_inscription_plan_free_top3_{browser_name}"
|
||||
unique_email = TEST_USER_EMAIL_BASE.format(
|
||||
f"free_{datetime.now().strftime('%H%M%S')}"
|
||||
)
|
||||
|
||||
try:
|
||||
# 1. Inscription
|
||||
await register_user(page, unique_email, TEST_USER_PASSWORD)
|
||||
|
||||
# 2. Choix du plan free (si une page de choix de plan existe)
|
||||
if "plan" in page.url or "pricing" in page.url:
|
||||
free_plan_btn = page.locator(
|
||||
'[data-plan="free"], [data-testid="plan-free"], text=Free, text=Gratuit'
|
||||
)
|
||||
await free_plan_btn.first.click()
|
||||
await page.wait_for_timeout(1000)
|
||||
|
||||
# 3. Accès au dashboard
|
||||
await page.goto(f"{BASE_URL}/dashboard")
|
||||
await page.wait_for_load_state("networkidle", timeout=15_000)
|
||||
|
||||
# 4. Vérification : le top 3 est visible
|
||||
top3_section = page.locator(
|
||||
'[data-testid="top3"], .top3, #top3, .predictions-top3'
|
||||
)
|
||||
if await top3_section.count() > 0:
|
||||
await top3_section.first.wait_for(state="visible", timeout=5_000)
|
||||
assert await top3_section.first.is_visible(), "Top 3 section non visible"
|
||||
else:
|
||||
# Fallback : chercher des cards de chevaux
|
||||
prediction_cards = page.locator(
|
||||
".prediction-card, .horse-card, [data-horse]"
|
||||
)
|
||||
count = await prediction_cards.count()
|
||||
assert count >= 1, (
|
||||
f"Aucune prédiction visible pour plan free (browser: {browser_name})"
|
||||
)
|
||||
|
||||
# 5. Vérification : pas d'accès aux routes premium
|
||||
await page.goto(f"{BASE_URL}/api/races?all=true")
|
||||
content = await page.content()
|
||||
# Un plan free ne devrait pas voir toutes les courses sans restriction
|
||||
# Le check exact dépend de l'implémentation — ici on vérifie juste la réponse 200 ou 403
|
||||
assert page.url is not None
|
||||
|
||||
print(f"✅ [{browser_name}] Scénario 1 PASS — {unique_email}")
|
||||
|
||||
except Exception as e:
|
||||
await page.screenshot(path=screenshot_on_fail(test_name, page))
|
||||
raise AssertionError(f"[{browser_name}] Scénario 1 FAIL: {e}") from e
|
||||
|
||||
|
||||
# === Test Scénario 2 : Upgrade premium → Stripe → accès toutes courses ===
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upgrade_premium_stripe(context_page):
|
||||
"""
|
||||
Scénario 2 : Upgrade premium → Stripe checkout → accès toutes courses
|
||||
"""
|
||||
page, browser_name = context_page
|
||||
test_name = f"test_upgrade_premium_{browser_name}"
|
||||
unique_email = TEST_USER_EMAIL_BASE.format(
|
||||
f"premium_{datetime.now().strftime('%H%M%S')}"
|
||||
)
|
||||
|
||||
try:
|
||||
# 1. Inscription free
|
||||
await register_user(page, unique_email, TEST_USER_PASSWORD)
|
||||
await page.goto(f"{BASE_URL}/dashboard")
|
||||
await page.wait_for_load_state("networkidle", timeout=15_000)
|
||||
|
||||
# 2. Aller à la page upgrade/pricing
|
||||
upgrade_btn = page.locator(
|
||||
'[data-testid="upgrade"], [href*="pricing"], [href*="upgrade"], text=Upgrade, text=Premium'
|
||||
)
|
||||
await upgrade_btn.first.click()
|
||||
await page.wait_for_timeout(1000)
|
||||
|
||||
# 3. Sélectionner plan premium
|
||||
premium_btn = page.locator(
|
||||
'[data-plan="premium"], [data-testid="plan-premium"], text=Premium, text=Pro'
|
||||
)
|
||||
await premium_btn.first.click()
|
||||
await page.wait_for_timeout(1000)
|
||||
|
||||
# 4. Stripe Checkout
|
||||
# Attendre la redirection vers Stripe ou l'ouverture du formulaire
|
||||
try:
|
||||
await page.wait_for_url("**/stripe.com/**", timeout=8_000)
|
||||
# Mode redirection Stripe
|
||||
await page.fill(
|
||||
'[placeholder="Card number"]', STRIPE_TEST_CARD.replace(" ", "")
|
||||
)
|
||||
await page.fill('[placeholder="MM / YY"]', STRIPE_EXPIRY)
|
||||
await page.fill('[placeholder="CVC"]', STRIPE_CVC)
|
||||
except Exception:
|
||||
# Mode embedded Stripe
|
||||
await fill_stripe_form(page)
|
||||
|
||||
await page.click(
|
||||
'[type="submit"], [data-testid="submit"], text=Pay, text=Payer'
|
||||
)
|
||||
# Attendre le retour sur l'app
|
||||
await page.wait_for_url(f"{BASE_URL}/**", timeout=30_000)
|
||||
|
||||
# 5. Vérification : accès à toutes les courses
|
||||
await page.goto(f"{BASE_URL}/dashboard")
|
||||
await page.wait_for_load_state("networkidle", timeout=15_000)
|
||||
|
||||
# Le badge premium doit être visible
|
||||
premium_badge = page.locator(
|
||||
'[data-testid="premium-badge"], .premium-badge, .badge-premium, text=Premium'
|
||||
)
|
||||
assert (
|
||||
await premium_badge.count() > 0
|
||||
or "premium" in (await page.content()).lower()
|
||||
), f"Badge premium non détecté après upgrade (browser: {browser_name})"
|
||||
|
||||
print(f"✅ [{browser_name}] Scénario 2 PASS — {unique_email}")
|
||||
|
||||
except Exception as e:
|
||||
await page.screenshot(path=screenshot_on_fail(test_name, page))
|
||||
raise AssertionError(f"[{browser_name}] Scénario 2 FAIL: {e}") from e
|
||||
|
||||
|
||||
# === Test Scénario 3 : Abonnement pro → export CSV → accès API ===
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_pro_export_csv_api_access(context_page):
|
||||
"""
|
||||
Scénario 3 : Abonnement pro → export CSV → accès API
|
||||
"""
|
||||
page, browser_name = context_page
|
||||
test_name = f"test_pro_export_csv_{browser_name}"
|
||||
unique_email = TEST_USER_EMAIL_BASE.format(
|
||||
f"pro_{datetime.now().strftime('%H%M%S')}"
|
||||
)
|
||||
|
||||
try:
|
||||
# 1. Inscription + upgrade pro (via API admin pour les tests)
|
||||
await register_user(page, unique_email, TEST_USER_PASSWORD)
|
||||
|
||||
# Promotion directe via API de test (si disponible)
|
||||
api_resp = await page.request.post(
|
||||
f"{BASE_URL}/api/test/set-plan",
|
||||
data={"email": unique_email, "plan": "pro"},
|
||||
)
|
||||
if api_resp.status != 200:
|
||||
pytest.skip("API de promotion de plan non disponible — nécessite HRT-31")
|
||||
|
||||
# 2. Se connecter
|
||||
await login_user(page, unique_email, TEST_USER_PASSWORD)
|
||||
await page.goto(f"{BASE_URL}/dashboard")
|
||||
await page.wait_for_load_state("networkidle", timeout=15_000)
|
||||
|
||||
# 3. Export CSV
|
||||
async with page.expect_download() as download_info:
|
||||
export_btn = page.locator(
|
||||
'[data-testid="export-csv"], [href*="csv"], text=Export CSV, text=Exporter CSV'
|
||||
)
|
||||
await export_btn.first.click()
|
||||
download = await download_info.value
|
||||
assert download.suggested_filename.endswith(".csv"), (
|
||||
f"Fichier téléchargé n'est pas un CSV: {download.suggested_filename}"
|
||||
)
|
||||
# Sauvegarder pour vérification
|
||||
await download.save_as(f"/tmp/qa_export_{browser_name}.csv")
|
||||
|
||||
# 4. Accès API avec clé
|
||||
api_key_section = page.locator('[data-testid="api-key"], .api-key, #api-key')
|
||||
if await api_key_section.count() > 0:
|
||||
api_key = await api_key_section.first.inner_text()
|
||||
api_key = api_key.strip()
|
||||
# Tester l'API directement
|
||||
api_resp = await page.request.get(
|
||||
f"{BASE_URL}/api/races",
|
||||
headers={"X-API-Key": api_key},
|
||||
)
|
||||
assert api_resp.status == 200, (
|
||||
f"API access refusé avec clé valide (status: {api_resp.status})"
|
||||
)
|
||||
|
||||
print(f"✅ [{browser_name}] Scénario 3 PASS — {unique_email}")
|
||||
|
||||
except Exception as e:
|
||||
await page.screenshot(path=screenshot_on_fail(test_name, page))
|
||||
raise AssertionError(f"[{browser_name}] Scénario 3 FAIL: {e}") from e
|
||||
|
||||
|
||||
# === Test Scénario 4 : Annulation abonnement → downgrade free ===
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_annulation_downgrade_free(context_page):
|
||||
"""
|
||||
Scénario 4 : Annulation abonnement → downgrade free
|
||||
"""
|
||||
page, browser_name = context_page
|
||||
test_name = f"test_annulation_downgrade_{browser_name}"
|
||||
unique_email = TEST_USER_EMAIL_BASE.format(
|
||||
f"cancel_{datetime.now().strftime('%H%M%S')}"
|
||||
)
|
||||
|
||||
try:
|
||||
# 1. Inscription + set plan premium via API test
|
||||
await register_user(page, unique_email, TEST_USER_PASSWORD)
|
||||
api_resp = await page.request.post(
|
||||
f"{BASE_URL}/api/test/set-plan",
|
||||
data={"email": unique_email, "plan": "premium"},
|
||||
)
|
||||
if api_resp.status != 200:
|
||||
pytest.skip("API de promotion de plan non disponible — nécessite HRT-31")
|
||||
|
||||
await login_user(page, unique_email, TEST_USER_PASSWORD)
|
||||
|
||||
# 2. Aller à la page de gestion d'abonnement
|
||||
await page.goto(f"{BASE_URL}/account/subscription")
|
||||
await page.wait_for_load_state("networkidle", timeout=10_000)
|
||||
|
||||
# 3. Annuler l'abonnement
|
||||
cancel_btn = page.locator(
|
||||
'[data-testid="cancel-subscription"], text=Annuler, text=Cancel subscription, '
|
||||
"text=Résilier"
|
||||
)
|
||||
await cancel_btn.first.click()
|
||||
|
||||
# Confirmation modal
|
||||
confirm_btn = page.locator(
|
||||
'[data-testid="confirm-cancel"], text=Confirmer, text=Confirm, text=Oui'
|
||||
)
|
||||
if await confirm_btn.count() > 0:
|
||||
await confirm_btn.first.click()
|
||||
|
||||
await page.wait_for_timeout(2000)
|
||||
|
||||
# 4. Vérification : retour au plan free
|
||||
await page.goto(f"{BASE_URL}/dashboard")
|
||||
await page.wait_for_load_state("networkidle", timeout=15_000)
|
||||
content = await page.content()
|
||||
|
||||
# Le badge premium ne doit plus être actif
|
||||
premium_active = page.locator(
|
||||
'[data-testid="premium-badge"].active, .premium-active'
|
||||
)
|
||||
assert await premium_active.count() == 0, (
|
||||
f"Badge premium encore actif après annulation (browser: {browser_name})"
|
||||
)
|
||||
|
||||
# Le plan free doit être indiqué
|
||||
free_indicator = page.locator(
|
||||
'[data-plan="free"], .badge-free, text=Plan Free, text=Gratuit'
|
||||
)
|
||||
# Ou simplement vérifier l'absence de mention "premium actif"
|
||||
assert (
|
||||
"annulé" in content.lower()
|
||||
or "cancelled" in content.lower()
|
||||
or "free" in content.lower()
|
||||
or await free_indicator.count() > 0
|
||||
), f"Downgrade vers free non confirmé (browser: {browser_name})"
|
||||
|
||||
print(f"✅ [{browser_name}] Scénario 4 PASS — {unique_email}")
|
||||
|
||||
except Exception as e:
|
||||
await page.screenshot(path=screenshot_on_fail(test_name, page))
|
||||
raise AssertionError(f"[{browser_name}] Scénario 4 FAIL: {e}") from e
|
||||
|
||||
|
||||
# === Test Sécurité : Plan free ne peut pas accéder routes premium ===
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_autorisation_plan_free_acces_premium_refuse(context_page):
|
||||
"""
|
||||
Test d'autorisation : un utilisateur free ne peut pas accéder aux routes premium.
|
||||
"""
|
||||
page, browser_name = context_page
|
||||
test_name = f"test_auth_plan_free_acces_bloque_{browser_name}"
|
||||
unique_email = TEST_USER_EMAIL_BASE.format(
|
||||
f"authtest_{datetime.now().strftime('%H%M%S')}"
|
||||
)
|
||||
|
||||
try:
|
||||
await register_user(page, unique_email, TEST_USER_PASSWORD)
|
||||
await login_user(page, unique_email, TEST_USER_PASSWORD)
|
||||
|
||||
# Tenter d'accéder à une route premium
|
||||
premium_routes = [
|
||||
"/api/races?all=true",
|
||||
"/api/export/csv",
|
||||
"/api/predictions/all",
|
||||
]
|
||||
|
||||
for route in premium_routes:
|
||||
resp = await page.request.get(f"{BASE_URL}{route}")
|
||||
assert resp.status in (403, 401, 402), (
|
||||
f"Route premium accessible par plan free: {route} → HTTP {resp.status} (browser: {browser_name})"
|
||||
)
|
||||
|
||||
print(f"✅ [{browser_name}] Test autorisation PASS — {unique_email}")
|
||||
|
||||
except Exception as e:
|
||||
await page.screenshot(path=screenshot_on_fail(test_name, page))
|
||||
raise AssertionError(f"[{browser_name}] Test autorisation FAIL: {e}") from e
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Exécution directe pour debug
|
||||
import subprocess
|
||||
|
||||
subprocess.run(
|
||||
[
|
||||
"python",
|
||||
"-m",
|
||||
"pytest",
|
||||
__file__,
|
||||
"-v",
|
||||
"--tb=short",
|
||||
"--html=tests/reports/e2e_report.html",
|
||||
"--self-contained-html",
|
||||
]
|
||||
)
|
||||
305
tests/load/locustfile.py
Normal file
305
tests/load/locustfile.py
Normal file
@@ -0,0 +1,305 @@
|
||||
"""
|
||||
Tests de charge Locust — SaaS Turf Prédictions IA
|
||||
Sprint 8 — QA, Beta Fermee, Go/No-Go
|
||||
Ticket: HRT-34
|
||||
|
||||
Scénarios :
|
||||
- Test normal : 100 users simultanés, 10 min
|
||||
- Test spike : 500 users en 2 min
|
||||
|
||||
Cibles :
|
||||
- p95 latence < 500ms
|
||||
- error rate < 0.1%
|
||||
|
||||
Usage :
|
||||
# Test normal (100 users, 10 min)
|
||||
locust -f tests/load/locustfile.py --host http://localhost:8792 \
|
||||
--users 100 --spawn-rate 10 --run-time 10m \
|
||||
--headless --csv tests/reports/load_normal
|
||||
|
||||
# Test spike (500 users, 2 min)
|
||||
locust -f tests/load/locustfile.py --host http://localhost:8792 \
|
||||
--users 500 --spawn-rate 250 --run-time 2m \
|
||||
--headless --csv tests/reports/load_spike
|
||||
|
||||
# Interface web
|
||||
locust -f tests/load/locustfile.py --host http://localhost:8792
|
||||
"""
|
||||
|
||||
import random
|
||||
import json
|
||||
from locust import HttpUser, TaskSet, task, between, events
|
||||
from locust.env import Environment
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# === Credentials de test ===
|
||||
TEST_USERS = [
|
||||
{"email": f"loadtest+{i}@h3r7.tech", "password": "TestLoad_2026!"}
|
||||
for i in range(50)
|
||||
]
|
||||
|
||||
|
||||
# === Helper de login ===
|
||||
class AuthMixin:
|
||||
"""Mixin pour authentification JWT."""
|
||||
|
||||
token: str = None
|
||||
|
||||
def login(self):
|
||||
"""Login et stocke le token JWT."""
|
||||
user = random.choice(TEST_USERS)
|
||||
with self.client.post(
|
||||
"/api/auth/login",
|
||||
json={"email": user["email"], "password": user["password"]},
|
||||
catch_response=True,
|
||||
name="/api/auth/login",
|
||||
) as resp:
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
self.token = data.get("access_token") or data.get("token")
|
||||
resp.success()
|
||||
else:
|
||||
self.token = None
|
||||
resp.failure(f"Login failed: {resp.status_code}")
|
||||
|
||||
def auth_headers(self):
|
||||
if self.token:
|
||||
return {"Authorization": f"Bearer {self.token}"}
|
||||
return {}
|
||||
|
||||
|
||||
# === Comportement utilisateur Free ===
|
||||
class FreePlanTaskSet(TaskSet):
|
||||
"""Simule un utilisateur Free qui consulte les prédictions top 3."""
|
||||
|
||||
def on_start(self):
|
||||
self.user.login()
|
||||
|
||||
@task(5)
|
||||
def voir_dashboard(self):
|
||||
with self.client.get(
|
||||
"/dashboard",
|
||||
catch_response=True,
|
||||
name="/dashboard",
|
||||
) as resp:
|
||||
if resp.status_code in (200, 304):
|
||||
resp.success()
|
||||
else:
|
||||
resp.failure(f"Dashboard: {resp.status_code}")
|
||||
|
||||
@task(8)
|
||||
def voir_predictions_aujourd_hui(self):
|
||||
with self.client.get(
|
||||
"/api",
|
||||
headers=self.user.auth_headers(),
|
||||
catch_response=True,
|
||||
name="/api (today predictions)",
|
||||
) as resp:
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
if isinstance(data, (list, dict)):
|
||||
resp.success()
|
||||
else:
|
||||
resp.failure("Réponse inattendue")
|
||||
elif resp.status_code in (401, 403):
|
||||
resp.success() # Normal si token expiré
|
||||
else:
|
||||
resp.failure(f"Predictions: {resp.status_code}")
|
||||
|
||||
@task(3)
|
||||
def voir_courses(self):
|
||||
with self.client.get(
|
||||
"/api/races",
|
||||
headers=self.user.auth_headers(),
|
||||
catch_response=True,
|
||||
name="/api/races",
|
||||
) as resp:
|
||||
if resp.status_code in (200, 401, 403):
|
||||
resp.success()
|
||||
else:
|
||||
resp.failure(f"Races: {resp.status_code}")
|
||||
|
||||
@task(2)
|
||||
def voir_scoring(self):
|
||||
with self.client.get(
|
||||
"/api/scoring",
|
||||
headers=self.user.auth_headers(),
|
||||
catch_response=True,
|
||||
name="/api/scoring",
|
||||
) as resp:
|
||||
if resp.status_code in (200, 401, 403):
|
||||
resp.success()
|
||||
else:
|
||||
resp.failure(f"Scoring: {resp.status_code}")
|
||||
|
||||
@task(1)
|
||||
def voir_portail(self):
|
||||
with self.client.get(
|
||||
"/",
|
||||
catch_response=True,
|
||||
name="/ (portail)",
|
||||
) as resp:
|
||||
if resp.status_code in (200, 304):
|
||||
resp.success()
|
||||
else:
|
||||
resp.failure(f"Portail: {resp.status_code}")
|
||||
|
||||
|
||||
# === Comportement utilisateur Premium ===
|
||||
class PremiumPlanTaskSet(TaskSet):
|
||||
"""Simule un utilisateur Premium avec accès complet."""
|
||||
|
||||
def on_start(self):
|
||||
self.user.login()
|
||||
|
||||
@task(4)
|
||||
def voir_dashboard_complet(self):
|
||||
with self.client.get(
|
||||
"/dashboard",
|
||||
headers=self.user.auth_headers(),
|
||||
catch_response=True,
|
||||
name="/dashboard (premium)",
|
||||
) as resp:
|
||||
if resp.status_code in (200, 304):
|
||||
resp.success()
|
||||
else:
|
||||
resp.failure(f"Dashboard premium: {resp.status_code}")
|
||||
|
||||
@task(6)
|
||||
def voir_toutes_courses(self):
|
||||
with self.client.get(
|
||||
"/api/races?all=true",
|
||||
headers=self.user.auth_headers(),
|
||||
catch_response=True,
|
||||
name="/api/races?all=true",
|
||||
) as resp:
|
||||
if resp.status_code in (200, 403):
|
||||
resp.success()
|
||||
else:
|
||||
resp.failure(f"All races: {resp.status_code}")
|
||||
|
||||
@task(3)
|
||||
def export_csv(self):
|
||||
with self.client.get(
|
||||
"/api/export/csv",
|
||||
headers=self.user.auth_headers(),
|
||||
catch_response=True,
|
||||
name="/api/export/csv",
|
||||
) as resp:
|
||||
if resp.status_code in (200, 403):
|
||||
resp.success()
|
||||
else:
|
||||
resp.failure(f"CSV export: {resp.status_code}")
|
||||
|
||||
@task(2)
|
||||
def voir_historique(self):
|
||||
with self.client.get(
|
||||
"/api/odds_history",
|
||||
headers=self.user.auth_headers(),
|
||||
catch_response=True,
|
||||
name="/api/odds_history",
|
||||
) as resp:
|
||||
if resp.status_code in (200, 403):
|
||||
resp.success()
|
||||
else:
|
||||
resp.failure(f"Odds history: {resp.status_code}")
|
||||
|
||||
@task(1)
|
||||
def appel_api_externe(self):
|
||||
"""Simule un accès API via clé (plan pro)."""
|
||||
with self.client.get(
|
||||
"/api/races",
|
||||
headers={"X-API-Key": "test-api-key-qa"},
|
||||
catch_response=True,
|
||||
name="/api/races (api-key)",
|
||||
) as resp:
|
||||
if resp.status_code in (200, 401, 403):
|
||||
resp.success()
|
||||
else:
|
||||
resp.failure(f"API key access: {resp.status_code}")
|
||||
|
||||
|
||||
# === Utilisateurs Locust ===
|
||||
|
||||
|
||||
class FreeUser(AuthMixin, HttpUser):
|
||||
"""Utilisateur plan free (70% du trafic)."""
|
||||
|
||||
tasks = [FreePlanTaskSet]
|
||||
wait_time = between(1, 4)
|
||||
weight = 70
|
||||
|
||||
|
||||
class PremiumUser(AuthMixin, HttpUser):
|
||||
"""Utilisateur plan premium (30% du trafic)."""
|
||||
|
||||
tasks = [PremiumPlanTaskSet]
|
||||
wait_time = between(0.5, 2)
|
||||
weight = 30
|
||||
|
||||
|
||||
# === Events et rapports ===
|
||||
|
||||
|
||||
@events.test_stop.add_listener
|
||||
def on_test_stop(environment: Environment, **kwargs):
|
||||
"""Analyse les résultats et écrit un rapport."""
|
||||
stats = environment.stats
|
||||
total = stats.total
|
||||
|
||||
report_lines = [
|
||||
"=" * 60,
|
||||
"RAPPORT TEST DE CHARGE — SaaS Turf IA",
|
||||
"=" * 60,
|
||||
f"Total requests : {total.num_requests}",
|
||||
f"Failures : {total.num_failures}",
|
||||
f"Error rate : {total.fail_ratio * 100:.2f}%",
|
||||
f"Avg response time : {total.avg_response_time:.1f}ms",
|
||||
f"95th percentile : {total.get_response_time_percentile(0.95):.1f}ms",
|
||||
f"99th percentile : {total.get_response_time_percentile(0.99):.1f}ms",
|
||||
f"Max response time : {total.max_response_time:.1f}ms",
|
||||
f"RPS (avg) : {total.total_rps:.2f}",
|
||||
"",
|
||||
"CRITERES DE SUCCES :",
|
||||
]
|
||||
|
||||
p95 = total.get_response_time_percentile(0.95)
|
||||
error_rate = total.fail_ratio * 100
|
||||
|
||||
p95_ok = p95 < 500
|
||||
error_ok = error_rate < 0.1
|
||||
|
||||
report_lines.append(
|
||||
f" p95 < 500ms : {'✅ PASS' if p95_ok else '❌ FAIL'} ({p95:.1f}ms)"
|
||||
)
|
||||
report_lines.append(
|
||||
f" error rate < 0.1% : {'✅ PASS' if error_ok else '❌ FAIL'} ({error_rate:.2f}%)"
|
||||
)
|
||||
report_lines.append("")
|
||||
report_lines.append(
|
||||
f"VERDICT GLOBAL : {'✅ GO' if (p95_ok and error_ok) else '❌ NO-GO'}"
|
||||
)
|
||||
report_lines.append("=" * 60)
|
||||
|
||||
report = "\n".join(report_lines)
|
||||
print(report)
|
||||
|
||||
# Écrire dans un fichier
|
||||
import os
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
Path("tests/reports").mkdir(parents=True, exist_ok=True)
|
||||
report_file = (
|
||||
f"tests/reports/load_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
|
||||
)
|
||||
with open(report_file, "w") as f:
|
||||
f.write(report)
|
||||
print(f"\nRapport sauvegardé : {report_file}")
|
||||
|
||||
# Exit code non-0 si les critères ne sont pas respectés
|
||||
if not (p95_ok and error_ok):
|
||||
logger.error("CRITÈRES DE PERFORMANCE NON RESPECTÉS — NO-GO")
|
||||
environment.process_exit_code = 1
|
||||
188
tests/run_qa.sh
Normal file
188
tests/run_qa.sh
Normal file
@@ -0,0 +1,188 @@
|
||||
#!/bin/bash
|
||||
# ============================================================
|
||||
# Script principal QA — SaaS Turf Prédictions IA
|
||||
# Sprint 8 — QA, Beta Fermee, Go/No-Go
|
||||
# Ticket: HRT-34
|
||||
#
|
||||
# Usage : bash tests/run_qa.sh [APP_URL]
|
||||
# ============================================================
|
||||
|
||||
set -e
|
||||
|
||||
APP_URL="${1:-http://localhost:8792}"
|
||||
REPORT_DIR="tests/reports"
|
||||
VENV="venv"
|
||||
TIMESTAMP=$(date +%Y%m%d_%H%M%S)
|
||||
|
||||
mkdir -p "$REPORT_DIR"
|
||||
|
||||
echo "============================================================"
|
||||
echo "QA SPRINT 8 — SaaS Turf Prédictions IA"
|
||||
echo "Target : $APP_URL"
|
||||
echo "Date : $(date)"
|
||||
echo "============================================================"
|
||||
|
||||
# Vérifier l'environnement
|
||||
if [ ! -d "$VENV" ]; then
|
||||
echo "❌ Venv non trouvé. Créer l'environnement virtuel d'abord."
|
||||
exit 1
|
||||
fi
|
||||
|
||||
source "$VENV/bin/activate"
|
||||
|
||||
# Installer les dépendances de test si nécessaire
|
||||
pip install pytest pytest-asyncio pytest-html playwright locust bandit safety 2>/dev/null | tail -5
|
||||
|
||||
# Installer les navigateurs Playwright
|
||||
python -m playwright install chromium firefox webkit 2>/dev/null || true
|
||||
|
||||
PASS=0
|
||||
FAIL=0
|
||||
RESULTS=()
|
||||
|
||||
# ============================================================
|
||||
# 1. Tests E2E Playwright
|
||||
# ============================================================
|
||||
echo ""
|
||||
echo "--- Tests E2E Playwright ---"
|
||||
if APP_URL="$APP_URL" python -m pytest tests/e2e/ \
|
||||
-v --tb=short \
|
||||
--html="$REPORT_DIR/e2e_report_${TIMESTAMP}.html" \
|
||||
--self-contained-html \
|
||||
-q 2>&1 | tee "$REPORT_DIR/e2e_output_${TIMESTAMP}.log"; then
|
||||
echo "✅ E2E : PASS"
|
||||
RESULTS+=("✅ Tests E2E Playwright : PASS")
|
||||
((PASS++))
|
||||
else
|
||||
echo "❌ E2E : FAIL"
|
||||
RESULTS+=("❌ Tests E2E Playwright : FAIL — voir $REPORT_DIR/e2e_report_${TIMESTAMP}.html")
|
||||
((FAIL++))
|
||||
fi
|
||||
|
||||
# ============================================================
|
||||
# 2. Tests de sécurité (injection SQL, JWT, autorisation)
|
||||
# ============================================================
|
||||
echo ""
|
||||
echo "--- Tests Sécurité ---"
|
||||
if APP_URL="$APP_URL" python -m pytest tests/security/test_security.py \
|
||||
-v --tb=short \
|
||||
--html="$REPORT_DIR/security_report_${TIMESTAMP}.html" \
|
||||
--self-contained-html \
|
||||
-q 2>&1 | tee "$REPORT_DIR/security_output_${TIMESTAMP}.log"; then
|
||||
echo "✅ Sécurité : PASS"
|
||||
RESULTS+=("✅ Tests Sécurité (JWT, SQL injection, autorisation) : PASS")
|
||||
((PASS++))
|
||||
else
|
||||
echo "❌ Sécurité : FAIL"
|
||||
RESULTS+=("❌ Tests Sécurité : FAIL — voir $REPORT_DIR/security_report_${TIMESTAMP}.html")
|
||||
((FAIL++))
|
||||
fi
|
||||
|
||||
# ============================================================
|
||||
# 3. Scan Bandit (vulnérabilités code Python)
|
||||
# ============================================================
|
||||
echo ""
|
||||
echo "--- Scan Bandit ---"
|
||||
if bandit -r . \
|
||||
--exclude ./venv,./tests \
|
||||
-ll \
|
||||
-f html \
|
||||
-o "$REPORT_DIR/bandit_report_${TIMESTAMP}.html" \
|
||||
2>&1 | tee "$REPORT_DIR/bandit_output_${TIMESTAMP}.log"; then
|
||||
echo "✅ Bandit : PASS"
|
||||
RESULTS+=("✅ Bandit (sécurité code) : PASS")
|
||||
((PASS++))
|
||||
else
|
||||
echo "⚠️ Bandit : vulnérabilités détectées"
|
||||
RESULTS+=("⚠️ Bandit : voir $REPORT_DIR/bandit_report_${TIMESTAMP}.html")
|
||||
fi
|
||||
|
||||
# ============================================================
|
||||
# 4. Tests de charge Locust (100 users, 3 min)
|
||||
# ============================================================
|
||||
echo ""
|
||||
echo "--- Tests de charge Locust (100 users, 3 min) ---"
|
||||
if locust \
|
||||
-f tests/load/locustfile.py \
|
||||
--host "$APP_URL" \
|
||||
--users 100 \
|
||||
--spawn-rate 10 \
|
||||
--run-time 3m \
|
||||
--headless \
|
||||
--csv "$REPORT_DIR/load_normal_${TIMESTAMP}" \
|
||||
2>&1 | tee "$REPORT_DIR/load_output_${TIMESTAMP}.log"; then
|
||||
echo "✅ Locust 100 users : PASS"
|
||||
RESULTS+=("✅ Tests de charge 100 users : PASS")
|
||||
((PASS++))
|
||||
else
|
||||
echo "❌ Locust 100 users : FAIL"
|
||||
RESULTS+=("❌ Tests de charge 100 users : FAIL — voir $REPORT_DIR/load_output_${TIMESTAMP}.log")
|
||||
((FAIL++))
|
||||
fi
|
||||
|
||||
# ============================================================
|
||||
# 5. Test spike (500 users, 2 min) — optionnel
|
||||
# ============================================================
|
||||
echo ""
|
||||
echo "--- Test spike Locust (500 users, 2 min) ---"
|
||||
if locust \
|
||||
-f tests/load/locustfile.py \
|
||||
--host "$APP_URL" \
|
||||
--users 500 \
|
||||
--spawn-rate 250 \
|
||||
--run-time 2m \
|
||||
--headless \
|
||||
--csv "$REPORT_DIR/load_spike_${TIMESTAMP}" \
|
||||
2>&1 | tee "$REPORT_DIR/spike_output_${TIMESTAMP}.log"; then
|
||||
echo "✅ Spike 500 users : PASS"
|
||||
RESULTS+=("✅ Test spike 500 users : PASS")
|
||||
((PASS++))
|
||||
else
|
||||
echo "❌ Spike 500 users : FAIL"
|
||||
RESULTS+=("❌ Test spike 500 users : FAIL")
|
||||
((FAIL++))
|
||||
fi
|
||||
|
||||
# ============================================================
|
||||
# 6. OWASP ZAP (si Docker disponible)
|
||||
# ============================================================
|
||||
if command -v docker &>/dev/null; then
|
||||
echo ""
|
||||
echo "--- OWASP ZAP ---"
|
||||
if bash tests/security/run_owasp_zap.sh "$APP_URL" 2>&1 | tee "$REPORT_DIR/zap_main_${TIMESTAMP}.log"; then
|
||||
echo "✅ OWASP ZAP : PASS"
|
||||
RESULTS+=("✅ OWASP ZAP (zero vulnérabilité critique) : PASS")
|
||||
((PASS++))
|
||||
else
|
||||
echo "❌ OWASP ZAP : FAIL"
|
||||
RESULTS+=("❌ OWASP ZAP : vulnérabilités critiques détectées")
|
||||
((FAIL++))
|
||||
fi
|
||||
else
|
||||
RESULTS+=("⚠️ OWASP ZAP : skippé (Docker non disponible)")
|
||||
fi
|
||||
|
||||
# ============================================================
|
||||
# Rapport final
|
||||
# ============================================================
|
||||
echo ""
|
||||
echo "============================================================"
|
||||
echo "RAPPORT QA FINAL — Sprint 8"
|
||||
echo "============================================================"
|
||||
for r in "${RESULTS[@]}"; do
|
||||
echo " $r"
|
||||
done
|
||||
echo ""
|
||||
echo "Passed : $PASS"
|
||||
echo "Failed : $FAIL"
|
||||
echo ""
|
||||
|
||||
if [ "$FAIL" -eq 0 ]; then
|
||||
echo "✅ VERDICT : GO — Tous les tests passent"
|
||||
echo " Lancement public autorisé"
|
||||
exit 0
|
||||
else
|
||||
echo "❌ VERDICT : NO-GO — $FAIL test(s) en échec"
|
||||
echo " Corriger avant lancement public"
|
||||
exit 1
|
||||
fi
|
||||
100
tests/security/run_owasp_zap.sh
Normal file
100
tests/security/run_owasp_zap.sh
Normal file
@@ -0,0 +1,100 @@
|
||||
#!/bin/bash
|
||||
# ============================================================
|
||||
# Script OWASP ZAP — SaaS Turf Prédictions IA
|
||||
# Sprint 8 — QA, Beta Fermee, Go/No-Go
|
||||
# Ticket: HRT-34
|
||||
#
|
||||
# Prérequis : Docker, APP en cours d'exécution
|
||||
# Usage : bash tests/security/run_owasp_zap.sh [APP_URL]
|
||||
# ============================================================
|
||||
|
||||
set -e
|
||||
|
||||
APP_URL="${1:-http://localhost:8792}"
|
||||
REPORT_DIR="tests/reports"
|
||||
ZAP_IMAGE="ghcr.io/zaproxy/zaproxy:stable"
|
||||
TIMESTAMP=$(date +%Y%m%d_%H%M%S)
|
||||
|
||||
mkdir -p "$REPORT_DIR"
|
||||
|
||||
echo "============================================================"
|
||||
echo "OWASP ZAP — Scan SaaS Turf IA"
|
||||
echo "Target : $APP_URL"
|
||||
echo "============================================================"
|
||||
|
||||
# Vérifier que Docker est disponible
|
||||
if ! command -v docker &>/dev/null; then
|
||||
echo "❌ Docker non disponible. Installer Docker pour exécuter OWASP ZAP."
|
||||
exit 1
|
||||
fi
|
||||
|
||||
# Vérifier que l'app est en cours d'exécution
|
||||
if ! curl -sf "$APP_URL" >/dev/null 2>&1; then
|
||||
echo "❌ App non accessible sur $APP_URL"
|
||||
echo " Démarrer l'application avant de lancer le scan."
|
||||
exit 1
|
||||
fi
|
||||
|
||||
echo "✅ App accessible sur $APP_URL"
|
||||
echo ""
|
||||
echo "Démarrage du scan OWASP ZAP (mode baseline)..."
|
||||
|
||||
# Scan ZAP en mode baseline (rapide, non-destructif)
|
||||
docker run --rm \
|
||||
--network host \
|
||||
-v "$(pwd)/tests/reports:/zap/wrk:rw" \
|
||||
"$ZAP_IMAGE" \
|
||||
zap-baseline.py \
|
||||
-t "$APP_URL" \
|
||||
-r "zap_report_${TIMESTAMP}.html" \
|
||||
-J "zap_report_${TIMESTAMP}.json" \
|
||||
-l WARN \
|
||||
2>&1 | tee "$REPORT_DIR/zap_output_${TIMESTAMP}.log"
|
||||
|
||||
ZAP_EXIT=$?
|
||||
|
||||
# Analyser le rapport JSON
|
||||
REPORT_JSON="$REPORT_DIR/zap_report_${TIMESTAMP}.json"
|
||||
if [ -f "$REPORT_JSON" ]; then
|
||||
CRITICAL=$(python3 -c "
|
||||
import json, sys
|
||||
try:
|
||||
data = json.load(open('$REPORT_JSON'))
|
||||
alerts = data.get('site', [{}])[0].get('alerts', [])
|
||||
critical = [a for a in alerts if a.get('riskcode', '0') == '3']
|
||||
high = [a for a in alerts if a.get('riskcode', '0') == '2']
|
||||
print(f'CRITICAL:{len(critical)},HIGH:{len(high)}')
|
||||
for a in critical:
|
||||
print(f' [CRITICAL] {a.get(\"alert\", \"?\")}')
|
||||
for a in high:
|
||||
print(f' [HIGH] {a.get(\"alert\", \"?\")}')
|
||||
except Exception as e:
|
||||
print(f'CRITICAL:?,HIGH:? (parse error: {e})')
|
||||
" 2>/dev/null || echo "CRITICAL:?,HIGH:?")
|
||||
|
||||
echo ""
|
||||
echo "============================================================"
|
||||
echo "RÉSULTATS OWASP ZAP"
|
||||
echo "============================================================"
|
||||
echo "$CRITICAL"
|
||||
|
||||
CRIT_COUNT=$(echo "$CRITICAL" | head -1 | cut -d: -f2 | cut -d, -f1)
|
||||
if [ "$CRIT_COUNT" = "0" ]; then
|
||||
echo ""
|
||||
echo "✅ PASS — Zéro vulnérabilité critique"
|
||||
else
|
||||
echo ""
|
||||
echo "❌ FAIL — $CRIT_COUNT vulnérabilité(s) critique(s) détectée(s)"
|
||||
echo " Action requise avant Go/No-Go"
|
||||
fi
|
||||
else
|
||||
echo "⚠️ Rapport JSON non généré"
|
||||
fi
|
||||
|
||||
echo ""
|
||||
echo "Rapport HTML : $REPORT_DIR/zap_report_${TIMESTAMP}.html"
|
||||
echo "Rapport JSON : $REPORT_DIR/zap_report_${TIMESTAMP}.json"
|
||||
echo "Log complet : $REPORT_DIR/zap_output_${TIMESTAMP}.log"
|
||||
echo "============================================================"
|
||||
|
||||
exit $ZAP_EXIT
|
||||
320
tests/security/test_security.py
Normal file
320
tests/security/test_security.py
Normal file
@@ -0,0 +1,320 @@
|
||||
"""
|
||||
Tests de sécurité — SaaS Turf Prédictions IA
|
||||
Sprint 8 — QA, Beta Fermee, Go/No-Go
|
||||
Ticket: HRT-34
|
||||
|
||||
Couverture :
|
||||
- Test injection SQL sur tous les inputs
|
||||
- Test authentification : JWT expiration, refresh, logout
|
||||
- Test autorisation : plan free ne peut pas accéder routes premium
|
||||
- (OWASP ZAP est exécuté séparément via script shell)
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
import time
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
|
||||
BASE_URL = os.environ.get("APP_URL", "http://localhost:8792")
|
||||
|
||||
# === Payloads injection SQL ===
|
||||
SQL_INJECTION_PAYLOADS = [
|
||||
"' OR '1'='1",
|
||||
"' OR 1=1--",
|
||||
"'; DROP TABLE users;--",
|
||||
"' UNION SELECT null,null,null--",
|
||||
"1'; SELECT * FROM users--",
|
||||
"admin'--",
|
||||
"' OR 'x'='x",
|
||||
"1 OR 1=1",
|
||||
"%27 OR %271%27=%271",
|
||||
]
|
||||
|
||||
# === Payloads XSS ===
|
||||
XSS_PAYLOADS = [
|
||||
"<script>alert(1)</script>",
|
||||
"<img src=x onerror=alert(1)>",
|
||||
"javascript:alert(1)",
|
||||
"<svg onload=alert(1)>",
|
||||
'"><script>alert(document.cookie)</script>',
|
||||
]
|
||||
|
||||
|
||||
# === Helpers ===
|
||||
|
||||
|
||||
def get_token(email: str, password: str) -> str | None:
|
||||
"""Obtenir un token JWT."""
|
||||
try:
|
||||
resp = requests.post(
|
||||
f"{BASE_URL}/api/auth/login",
|
||||
json={"email": email, "password": password},
|
||||
timeout=5,
|
||||
)
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
return data.get("access_token") or data.get("token")
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
# === Tests injection SQL ===
|
||||
|
||||
|
||||
class TestSQLInjection:
|
||||
"""Tests d'injection SQL sur les endpoints publics et authentifiés."""
|
||||
|
||||
@pytest.mark.parametrize("payload", SQL_INJECTION_PAYLOADS)
|
||||
def test_injection_login_email(self, payload):
|
||||
"""L'injection SQL dans le champ email ne doit pas fonctionner."""
|
||||
resp = requests.post(
|
||||
f"{BASE_URL}/api/auth/login",
|
||||
json={"email": payload, "password": "anything"},
|
||||
timeout=5,
|
||||
)
|
||||
assert resp.status_code not in (200,), (
|
||||
f"Injection SQL acceptée dans email: payload={payload!r}, status={resp.status_code}"
|
||||
)
|
||||
# Vérifier qu'aucune donnée sensible n'est exposée
|
||||
body = resp.text.lower()
|
||||
for keyword in [
|
||||
"sqlite_master",
|
||||
"table_name",
|
||||
"column_name",
|
||||
"password",
|
||||
"hash",
|
||||
]:
|
||||
assert keyword not in body, (
|
||||
f"Données sensibles exposées dans la réponse: keyword={keyword!r}"
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize("payload", SQL_INJECTION_PAYLOADS)
|
||||
def test_injection_search_query(self, payload):
|
||||
"""L'injection SQL dans les paramètres de recherche ne doit pas fonctionner."""
|
||||
resp = requests.get(
|
||||
f"{BASE_URL}/api/races",
|
||||
params={"q": payload, "date": payload},
|
||||
timeout=5,
|
||||
)
|
||||
# On accepte 200 (résultat vide) ou 400/422 (validation), mais pas 500
|
||||
assert resp.status_code != 500, (
|
||||
f"Erreur serveur sur injection SQL dans recherche: payload={payload!r}"
|
||||
)
|
||||
body = resp.text.lower()
|
||||
for keyword in ["sqlite_master", "syntax error", "table_name"]:
|
||||
assert keyword not in body, f"Fuite SQL dans réponse: keyword={keyword!r}"
|
||||
|
||||
@pytest.mark.parametrize("payload", SQL_INJECTION_PAYLOADS)
|
||||
def test_injection_register_fields(self, payload):
|
||||
"""L'injection SQL dans les champs d'inscription ne doit pas passer."""
|
||||
resp = requests.post(
|
||||
f"{BASE_URL}/api/auth/register",
|
||||
json={
|
||||
"email": f"test@test.com",
|
||||
"password": payload,
|
||||
"name": payload,
|
||||
},
|
||||
timeout=5,
|
||||
)
|
||||
assert resp.status_code != 500, (
|
||||
f"Erreur serveur sur injection dans register: payload={payload!r}"
|
||||
)
|
||||
|
||||
|
||||
# === Tests authentification JWT ===
|
||||
|
||||
|
||||
class TestJWTAuthentication:
|
||||
"""Tests JWT : expiration, refresh, logout."""
|
||||
|
||||
def test_jwt_expiration_token_invalide(self):
|
||||
"""Un token expiré doit être rejeté."""
|
||||
# Token JWT expiré fabriqué manuellement (exp dans le passé)
|
||||
# Header: {"alg": "HS256", "typ": "JWT"}
|
||||
# Payload: {"sub": "test", "exp": 1000000000} (expiry: 2001)
|
||||
expired_token = (
|
||||
"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9."
|
||||
"eyJzdWIiOiJ0ZXN0IiwiZXhwIjoxMDAwMDAwMDAwfQ."
|
||||
"invalid_signature_here"
|
||||
)
|
||||
resp = requests.get(
|
||||
f"{BASE_URL}/api/races",
|
||||
headers={"Authorization": f"Bearer {expired_token}"},
|
||||
timeout=5,
|
||||
)
|
||||
assert resp.status_code in (401, 403, 422), (
|
||||
f"Token expiré accepté: status={resp.status_code}"
|
||||
)
|
||||
|
||||
def test_jwt_token_malformé(self):
|
||||
"""Un token JWT malformé doit être rejeté."""
|
||||
for bad_token in ["not.a.jwt", "Bearer", "null", "undefined", ""]:
|
||||
resp = requests.get(
|
||||
f"{BASE_URL}/api/races",
|
||||
headers={"Authorization": f"Bearer {bad_token}"},
|
||||
timeout=5,
|
||||
)
|
||||
assert resp.status_code in (401, 403, 422, 400), (
|
||||
f"Token malformé accepté: token={bad_token!r}, status={resp.status_code}"
|
||||
)
|
||||
|
||||
def test_jwt_sans_token(self):
|
||||
"""Sans token, les routes protégées doivent retourner 401."""
|
||||
resp = requests.get(f"{BASE_URL}/api/export/csv", timeout=5)
|
||||
assert resp.status_code in (401, 403), (
|
||||
f"Route protégée accessible sans token: status={resp.status_code}"
|
||||
)
|
||||
|
||||
def test_jwt_refresh(self):
|
||||
"""Le mécanisme de refresh doit fonctionner."""
|
||||
# Tenter d'obtenir un token valide d'abord
|
||||
token = get_token("loadtest+0@h3r7.tech", "TestLoad_2026!")
|
||||
if token is None:
|
||||
pytest.skip("Utilisateur de test non créé — nécessite HRT-31")
|
||||
|
||||
resp = requests.post(
|
||||
f"{BASE_URL}/api/auth/refresh",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
timeout=5,
|
||||
)
|
||||
if resp.status_code == 404:
|
||||
pytest.skip("Route /api/auth/refresh non implémentée")
|
||||
assert resp.status_code == 200, (
|
||||
f"Refresh token échoué: status={resp.status_code}"
|
||||
)
|
||||
data = resp.json()
|
||||
assert "access_token" in data or "token" in data, (
|
||||
"Aucun token retourné par /api/auth/refresh"
|
||||
)
|
||||
|
||||
def test_jwt_logout(self):
|
||||
"""Après logout, le token doit être invalidé."""
|
||||
token = get_token("loadtest+0@h3r7.tech", "TestLoad_2026!")
|
||||
if token is None:
|
||||
pytest.skip("Utilisateur de test non créé — nécessite HRT-31")
|
||||
|
||||
# Logout
|
||||
resp = requests.post(
|
||||
f"{BASE_URL}/api/auth/logout",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
timeout=5,
|
||||
)
|
||||
if resp.status_code == 404:
|
||||
pytest.skip("Route /api/auth/logout non implémentée")
|
||||
assert resp.status_code in (200, 204), (
|
||||
f"Logout échoué: status={resp.status_code}"
|
||||
)
|
||||
|
||||
# Vérifier que le token est invalidé
|
||||
resp2 = requests.get(
|
||||
f"{BASE_URL}/api/export/csv",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
timeout=5,
|
||||
)
|
||||
assert resp2.status_code in (401, 403), (
|
||||
f"Token encore valide après logout: status={resp2.status_code}"
|
||||
)
|
||||
|
||||
|
||||
# === Tests autorisation par plan ===
|
||||
|
||||
|
||||
class TestPlanAuthorisation:
|
||||
"""Tests d'autorisation : free ne peut pas accéder aux routes premium."""
|
||||
|
||||
PREMIUM_ROUTES = [
|
||||
"/api/races?all=true",
|
||||
"/api/export/csv",
|
||||
"/api/predictions/all",
|
||||
"/api/premium/historical",
|
||||
]
|
||||
|
||||
FREE_ROUTES = [
|
||||
"/api",
|
||||
"/api/races",
|
||||
"/api/scoring",
|
||||
"/dashboard",
|
||||
]
|
||||
|
||||
def test_routes_premium_inaccessibles_sans_auth(self):
|
||||
"""Les routes premium doivent être inaccessibles sans authentification."""
|
||||
for route in self.PREMIUM_ROUTES:
|
||||
resp = requests.get(f"{BASE_URL}{route}", timeout=5)
|
||||
assert resp.status_code in (401, 403, 404), (
|
||||
f"Route premium accessible sans auth: {route} → {resp.status_code}"
|
||||
)
|
||||
|
||||
def test_routes_libres_accessibles(self):
|
||||
"""Les routes libres (portail, dashboard) doivent être accessibles."""
|
||||
for route in self.FREE_ROUTES:
|
||||
resp = requests.get(f"{BASE_URL}{route}", timeout=5)
|
||||
# On accepte tout sauf 5xx
|
||||
assert resp.status_code < 500, (
|
||||
f"Route libre retourne erreur serveur: {route} → {resp.status_code}"
|
||||
)
|
||||
|
||||
def test_plan_free_bloque_routes_premium(self):
|
||||
"""Un token free ne doit pas accéder aux routes premium."""
|
||||
token = get_token("loadtest+0@h3r7.tech", "TestLoad_2026!")
|
||||
if token is None:
|
||||
pytest.skip("Utilisateur de test non créé — nécessite HRT-31")
|
||||
|
||||
for route in self.PREMIUM_ROUTES:
|
||||
resp = requests.get(
|
||||
f"{BASE_URL}{route}",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
timeout=5,
|
||||
)
|
||||
# Plan free ne devrait pas pouvoir accéder (403 ou 402 Payment Required)
|
||||
# Si 200 — les données doivent être limitées
|
||||
if resp.status_code == 200:
|
||||
data = (
|
||||
resp.json()
|
||||
if resp.headers.get("content-type", "").startswith(
|
||||
"application/json"
|
||||
)
|
||||
else {}
|
||||
)
|
||||
# Vérifier que ce n'est pas un accès complet
|
||||
if isinstance(data, list):
|
||||
# Max 3 éléments pour plan free (top 3)
|
||||
assert len(data) <= 10, (
|
||||
f"Plan free retourne trop de données sur {route}: {len(data)} éléments"
|
||||
)
|
||||
else:
|
||||
assert resp.status_code in (403, 402, 401), (
|
||||
f"Status inattendu sur route premium pour plan free: {route} → {resp.status_code}"
|
||||
)
|
||||
|
||||
def test_injection_dans_bearer_token(self):
|
||||
"""Injection dans le token Bearer ne doit pas provoquer d'erreur 500."""
|
||||
for payload in SQL_INJECTION_PAYLOADS[:3]:
|
||||
encoded = base64.b64encode(payload.encode()).decode()
|
||||
resp = requests.get(
|
||||
f"{BASE_URL}/api/races",
|
||||
headers={"Authorization": f"Bearer {encoded}"},
|
||||
timeout=5,
|
||||
)
|
||||
assert resp.status_code != 500, (
|
||||
f"Erreur serveur sur injection dans Authorization: payload={payload!r}"
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import subprocess
|
||||
|
||||
subprocess.run(
|
||||
[
|
||||
"python",
|
||||
"-m",
|
||||
"pytest",
|
||||
__file__,
|
||||
"-v",
|
||||
"--tb=short",
|
||||
"--html=tests/reports/security_report.html",
|
||||
"--self-contained-html",
|
||||
]
|
||||
)
|
||||
Reference in New Issue
Block a user