feat(admin): canaux IA par palier (z.ai/DeepSeek directs), emails generes par IA, reductions Stripe
All checks were successful
Deploy to Production / Build and Deploy (push) Successful in 2m40s

- Chaque modele d'un palier devient une route {canal, modele} : openrouter,
  deepseek direct, zhipu (z.ai, nouveau canal), minimax, openai, xai ;
  repli automatique sur la route suivante si la cle manque, compatibilite
  ascendante avec les anciens reglages a chaines
- Page Fournisseurs refondue en sections ; interrupteurs et chaine de
  secours retablis ; tests de connexion avec delai d'attente
- Emails de relance generes par IA (canal au choix, prompt construit depuis
  le plan marketing, HTML sane, RTL arabe/persan, jamais d'envoi automatique)
- Codes promo : validation locale avant Stripe, Coupon+PromotionCode avec
  double plafond, comptage par session (webhook+sync=1 fois), ecriture
  atomique, relier-a-Stripe, suppression, validation publique au checkout
- Page Tarifs : champ code promo (?promo=) avec verification traduite
- Tests : 1357 verts (facturation par palier testee dans le worker,
  point de contact providers/available, promos, generation)
This commit is contained in:
2026-09-05 19:46:33 +02:00
parent 6c26712687
commit 0a013c679d
47 changed files with 7210 additions and 679 deletions

570
tests/test_admin_promos.py Normal file
View File

@@ -0,0 +1,570 @@
"""
Réductions : codes promo (spec « Fournisseur par palier IA »).
- création locale (data/promo_codes.json) + Coupon/Promotion Code Stripe
quand Stripe est configuré (sinon « local_only ») ;
- validation AU CHECKOUT AVANT tout appel Stripe (actif, forfait, expiration,
plafond) avec une cause explicite ;
- incrément du compteur uniquement à la complétion du paiement.
"""
import json
import pytest
from pathlib import Path
from types import SimpleNamespace
from datetime import datetime, timedelta, timezone
import services.payment_service as payment_svc
ADMIN_LOGIN_URL = "/api/v1/admin/login"
PROMOS_URL = "/api/v1/admin/marketing/promos"
def _write_json(path: Path, data) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
def _future(days: int = 30) -> str:
return (datetime.now(timezone.utc) + timedelta(days=days)).isoformat()
@pytest.fixture
def promo_file(tmp_path: Path, monkeypatch) -> Path:
path = tmp_path / "promo_codes.json"
monkeypatch.setattr(payment_svc, "PROMO_CODES_FILE", path)
return path
@pytest.fixture
def client(promo_file, monkeypatch, tmp_path: Path):
"""TestClient avec authentification JSON (sans base de données)."""
import services.auth_service as auth_svc
from middleware.rate_limiting import RateLimitManager
monkeypatch.setattr(auth_svc, "USERS_FILE", tmp_path / "users.json")
monkeypatch.setattr(auth_svc, "USE_DATABASE", False)
monkeypatch.setattr(auth_svc, "DATABASE_AVAILABLE", False)
async def _check_request_allow(self, request):
return True, "ok", "test"
async def _check_translation_allow(self, request, file_size_mb=0):
return True, "ok"
monkeypatch.setattr(RateLimitManager, "check_request", _check_request_allow)
monkeypatch.setattr(RateLimitManager, "check_translation", _check_translation_allow)
from fastapi.testclient import TestClient
from main import app
return TestClient(app, raise_server_exceptions=True)
@pytest.fixture
def admin_headers(client, monkeypatch):
import routes.admin_routes as admin_routes_mod
monkeypatch.setattr(admin_routes_mod, "ADMIN_USERNAME", "admin")
monkeypatch.setattr(admin_routes_mod, "ADMIN_PASSWORD", "admin-secret")
monkeypatch.setattr(admin_routes_mod, "ADMIN_PASSWORD_HASH", None)
r = client.post(ADMIN_LOGIN_URL, json={"password": "admin-secret"})
assert r.status_code == 200, r.text
token = r.json()["access_token"]
return {"Authorization": f"Bearer {token}"}
# ---------------------------------------------------------------------------
# Création / listage / désactivation (endpoints admin)
# ---------------------------------------------------------------------------
def test_create_promo_without_stripe_is_local_only(client, admin_headers, promo_file):
r = client.post(
PROMOS_URL,
json={"code": "lancement-10", "type": "percent", "value": 10,
"max_redemptions": 10, "expires_at": _future()},
headers=admin_headers,
)
assert r.status_code == 201, r.text
entry = r.json()["data"]
assert entry["code"] == "LANCEMENT-10" # normalisé en majuscules
assert entry["stripe_mode"] == "local_only"
assert entry["stripe_promotion_code_id"] is None
assert entry["times_used"] == 0
r2 = client.get(PROMOS_URL, headers=admin_headers)
assert r2.status_code == 200
assert [p["code"] for p in r2.json()["data"]["promos"]] == ["LANCEMENT-10"]
def test_create_promo_with_stripe_creates_coupon_and_code(
client, admin_headers, promo_file, monkeypatch
):
monkeypatch.setattr(payment_svc, "is_stripe_configured", lambda: True)
class _FakeObj(dict):
def __getattr__(self, name):
try:
return self[name]
except KeyError as e: # pragma: no cover
raise AttributeError(name) from e
def _fake_coupon(**kwargs):
assert kwargs["percent_off"] == 15
assert kwargs["duration"] == "once"
return _FakeObj(id="coupon_1")
def _fake_promo_code(**kwargs):
assert kwargs["coupon"] == "coupon_1"
assert kwargs["max_redemptions"] == 5
assert "expires_at" in kwargs
return _FakeObj(id="promo_1")
monkeypatch.setattr(payment_svc.stripe.Coupon, "create", _fake_coupon)
monkeypatch.setattr(payment_svc.stripe.PromotionCode, "create", _fake_promo_code)
monkeypatch.setattr(payment_svc, "stripe_mode", lambda: "test")
r = client.post(
PROMOS_URL,
json={"code": "PROMO15", "type": "percent", "value": 15,
"plans": ["pro", "business"], "max_redemptions": 5,
"expires_at": _future()},
headers=admin_headers,
)
assert r.status_code == 201, r.text
entry = r.json()["data"]
assert entry["stripe_coupon_id"] == "coupon_1"
assert entry["stripe_promotion_code_id"] == "promo_1"
assert entry["stripe_mode"] == "test"
def test_create_promo_rejects_invalid_input(client, admin_headers, promo_file):
# code trop court / minuscules après normalisation impossible (pattern)
r = client.post(PROMOS_URL, json={"code": "ab", "type": "percent", "value": 10},
headers=admin_headers)
assert r.status_code == 400
assert r.json()["error"] == "INVALID_PROMO_CODE"
# code avec caractères interdits
r = client.post(PROMOS_URL, json={"code": "MAUVAIS CODE!", "type": "percent", "value": 10},
headers=admin_headers)
assert r.status_code == 400
# pourcentage hors bornes
r = client.post(PROMOS_URL, json={"code": "GRAND", "type": "percent", "value": 150},
headers=admin_headers)
assert r.status_code == 400
assert r.json()["error"] == "INVALID_PROMO_VALUE"
# montant nul
r = client.post(PROMOS_URL, json={"code": "ZERO", "type": "amount", "value": 0},
headers=admin_headers)
assert r.status_code == 400
# forfait inconnu
r = client.post(PROMOS_URL, json={"code": "PLANX", "type": "percent", "value": 10,
"plans": ["platine"]}, headers=admin_headers)
assert r.status_code == 400
assert r.json()["error"] == "INVALID_PROMO_PLAN"
# expiration dans le passé
r = client.post(PROMOS_URL, json={"code": "HIER", "type": "percent", "value": 10,
"expires_at": "2020-01-01T00:00:00+00:00"},
headers=admin_headers)
assert r.status_code == 400
assert r.json()["error"] == "INVALID_PROMO_EXPIRY"
# plafond inférieur à 1
r = client.post(PROMOS_URL, json={"code": "PLAFOND", "type": "percent", "value": 10,
"max_redemptions": 0}, headers=admin_headers)
assert r.status_code == 400
def test_create_duplicate_promo_refused(client, admin_headers, promo_file):
body = {"code": "DUP", "type": "percent", "value": 10}
assert client.post(PROMOS_URL, json=body, headers=admin_headers).status_code == 201
r = client.post(PROMOS_URL, json=body, headers=admin_headers)
assert r.status_code == 409
assert r.json()["error"] == "PROMO_ALREADY_EXISTS"
def test_deactivate_promo(client, admin_headers, promo_file, monkeypatch):
client.post(PROMOS_URL, json={"code": "BYE", "type": "percent", "value": 5},
headers=admin_headers)
r = client.post(f"{PROMOS_URL}/BYE/deactivate", headers=admin_headers)
assert r.status_code == 200
assert r.json()["data"]["active"] is False
r2 = client.post(f"{PROMOS_URL}/INCONNU/deactivate", headers=admin_headers)
assert r2.status_code == 404
def test_promos_require_admin(client):
assert client.get(PROMOS_URL).status_code == 401
# ---------------------------------------------------------------------------
# Validation au checkout — AVANT tout appel Stripe
# ---------------------------------------------------------------------------
@pytest.fixture
def checkout_mocks(monkeypatch, promo_file):
"""Stripe « configuré » mais sous surveillance : toute création de session
est capturée (et aurait levé si appelée à tort)."""
calls: dict = {"session": 0, "kwargs": None}
def _fake_session_create(**kwargs):
calls["session"] += 1
calls["kwargs"] = kwargs
return SimpleNamespace(id="cs_test_1", url="https://stripe.test/checkout")
monkeypatch.setattr(payment_svc, "is_stripe_configured", lambda: True)
monkeypatch.setattr(payment_svc.stripe.checkout.Session, "create", _fake_session_create)
monkeypatch.setattr(payment_svc, "stripe_price_ids_for_plan", lambda p: ("price_m", "price_y"))
user = SimpleNamespace(
id="u1", email="buyer@example.com", name="Buyer",
stripe_customer_id="cus_1", stripe_subscription_id=None,
subscription_status="active", plan="free", cancel_at_period_end=False,
docs_translated_this_month=0, pages_translated_this_month=0,
api_calls_this_month=0, extra_credits=0, subscription_ends_at=None,
)
monkeypatch.setattr(payment_svc, "get_user_by_id", lambda uid: user if uid == "u1" else None)
monkeypatch.setattr(payment_svc, "update_user", lambda uid, data: None)
# Pas de base de données dans les tests : la vérification d'antidoublon
# (PaymentHistory) doit échouer proprement — la fonction la tolère.
import database.connection as db_conn
def _no_db():
raise RuntimeError("pas de base de données dans les tests")
monkeypatch.setattr(db_conn, "get_sync_session", _no_db)
return calls
@pytest.mark.asyncio
async def test_checkout_with_valid_promo_applies_discount(
checkout_mocks, promo_file, monkeypatch
):
monkeypatch.setattr(payment_svc, "stripe_mode", lambda: "test")
monkeypatch.setattr(
payment_svc.stripe.Coupon, "create", lambda **kw: SimpleNamespace(id="coupon_9")
)
monkeypatch.setattr(
payment_svc.stripe.PromotionCode,
"create",
lambda **kw: SimpleNamespace(id="promo_9"),
)
payment_svc.create_promo_code(
code="SOIF10", type="percent", value=10, plans=["pro"],
max_redemptions=10, expires_at=_future(),
)
from models.subscription import PlanType
result = await payment_svc.create_checkout_session(
user_id="u1", plan=PlanType.PRO, promo_code="soif10"
)
assert "error" not in result, result
assert result["promo_code"] == "SOIF10"
kwargs = checkout_mocks["kwargs"]
assert kwargs["discounts"] == [{"promotion_code": "promo_9"}]
assert kwargs["metadata"]["promo_code"] == "SOIF10"
assert kwargs["subscription_data"]["metadata"]["promo_code"] == "SOIF10"
@pytest.mark.asyncio
async def test_checkout_rejects_expired_promo_before_stripe(
checkout_mocks, promo_file
):
payment_svc.save_promo_codes([{
"code": "VIEUX", "active": True, "type": "percent", "value": 10,
"plans": ["all"], "max_redemptions": None, "times_used": 0,
"expires_at": "2020-01-01T00:00:00+00:00",
"stripe_coupon_id": "c", "stripe_promotion_code_id": "p",
"stripe_mode": "test", "created_at": "t", "deactivated_at": None,
}])
from models.subscription import PlanType
result = await payment_svc.create_checkout_session(
user_id="u1", plan=PlanType.PRO, promo_code="VIEUX"
)
assert result.get("error_code") == "PROMO_INVALID"
assert result.get("promo_reason") == "expired"
assert checkout_mocks["session"] == 0 # aucune session Stripe créée
@pytest.mark.asyncio
async def test_checkout_rejects_exhausted_promo(checkout_mocks, promo_file):
"""Un code à 10/10 utilisations est refusé avant tout appel Stripe."""
payment_svc.save_promo_codes([{
"code": "FINI", "active": True, "type": "percent", "value": 10,
"plans": ["all"], "max_redemptions": 10, "times_used": 10,
"expires_at": None, "stripe_coupon_id": "c",
"stripe_promotion_code_id": "p", "stripe_mode": "test",
"created_at": "t", "deactivated_at": None,
}])
from models.subscription import PlanType
result = await payment_svc.create_checkout_session(
user_id="u1", plan=PlanType.FREE, promo_code="FINI"
)
assert result.get("promo_reason") == "exhausted"
assert checkout_mocks["session"] == 0
@pytest.mark.asyncio
async def test_checkout_rejects_promo_for_other_plan(checkout_mocks, promo_file):
payment_svc.save_promo_codes([{
"code": "PROONLY", "active": True, "type": "percent", "value": 10,
"plans": ["pro"], "max_redemptions": None, "times_used": 0,
"expires_at": None, "stripe_coupon_id": "c",
"stripe_promotion_code_id": "p", "stripe_mode": "test",
"created_at": "t", "deactivated_at": None,
}])
from models.subscription import PlanType
result = await payment_svc.create_checkout_session(
user_id="u1", plan=PlanType.STARTER, promo_code="PROONLY"
)
assert result.get("promo_reason") == "plan"
assert checkout_mocks["session"] == 0
@pytest.mark.asyncio
async def test_checkout_rejects_unknown_and_local_only_promos(
checkout_mocks, promo_file
):
from models.subscription import PlanType
result = await payment_svc.create_checkout_session(
user_id="u1", plan=PlanType.PRO, promo_code="NUL"
)
assert result.get("promo_reason") == "unknown"
payment_svc.save_promo_codes([{
"code": "LOCAL", "active": True, "type": "percent", "value": 10,
"plans": ["all"], "max_redemptions": None, "times_used": 0,
"expires_at": None, "stripe_coupon_id": None,
"stripe_promotion_code_id": None, "stripe_mode": "local_only",
"created_at": "t", "deactivated_at": None,
}])
result2 = await payment_svc.create_checkout_session(
user_id="u1", plan=PlanType.PRO, promo_code="LOCAL"
)
assert result2.get("promo_reason") == "local_only"
assert checkout_mocks["session"] == 0
@pytest.mark.asyncio
async def test_completion_increments_promo_counter_once_per_session(
checkout_mocks, promo_file, monkeypatch
):
"""Le compteur local n'augmente qu'à la complétion ; un webhook suivi d'une
synchronisation du MÊME paiement ne compte qu'une fois, même en mode JSON."""
monkeypatch.setattr(payment_svc, "stripe_mode", lambda: "test")
monkeypatch.setattr(
payment_svc.stripe.Coupon, "create", lambda **kw: SimpleNamespace(id="c")
)
monkeypatch.setattr(
payment_svc.stripe.PromotionCode, "create", lambda **kw: SimpleNamespace(id="p")
)
payment_svc.create_promo_code(
code="COMPTE", type="percent", value=10, max_redemptions=10
)
assert payment_svc.load_promo_codes()[0]["times_used"] == 0
session = {
"id": "cs_1",
"metadata": {"user_id": "u1", "plan": "pro", "promo_code": "COMPTE"},
"subscription": {"id": "sub_1", "current_period_end": None},
"customer": "cus_1",
"payment_intent": None,
"amount_total": 1900,
"currency": "eur",
}
# La vérification d'antidoublon touche la base : indifférent ici, la
# fonction tolère son échec.
await payment_svc.handle_checkout_completed(session)
await payment_svc.handle_checkout_completed(session) # re-livraison du webhook
# Synchronisation manuelle du MÊME paiement (même session Stripe) : 0 incrément.
await payment_svc.handle_checkout_completed(session)
promo = payment_svc.load_promo_codes()[0]
assert promo["times_used"] == 1
assert promo["counted_sessions"] == ["cs_1"]
assert promo["last_used_at"]
# Un AUTRE paiement (autre session) compte bien une seconde fois.
session2 = dict(session, id="cs_2")
await payment_svc.handle_checkout_completed(session2)
assert payment_svc.load_promo_codes()[0]["times_used"] == 2
@pytest.mark.asyncio
async def test_checkout_without_promo_has_no_discount(checkout_mocks, promo_file):
from models.subscription import PlanType
result = await payment_svc.create_checkout_session(
user_id="u1", plan=PlanType.PRO, promo_code=""
)
assert "error" not in result
assert "discounts" not in checkout_mocks["kwargs"]
assert "promo_code" not in checkout_mocks["kwargs"]["metadata"]
# ---------------------------------------------------------------------------
# Robustesse (relecture) : doublon sous verrou, Stripe en erreur, JSON corrompu
# ---------------------------------------------------------------------------
def test_duplicate_checked_under_lock_two_creates_one_409(promo_file):
"""Deux créations du même code : la seconde est rejetée par la vérification
SOUS le verrou du service (une seule entrée persistée)."""
payment_svc.create_promo_code(code="RACE1", type="percent", value=10)
with pytest.raises(payment_svc.DuplicatePromoError):
payment_svc.create_promo_code(code="RACE1", type="percent", value=10)
assert [p["code"] for p in payment_svc.load_promo_codes()] == ["RACE1"]
def test_stripe_failure_cleans_orphan_coupon_and_persists_nothing(
promo_file, monkeypatch
):
"""Coupon créé puis PromotionCode refusé : le coupon orphelin est supprimé,
l'erreur est typée et RIEN n'est écrit dans promo_codes.json."""
monkeypatch.setattr(payment_svc, "is_stripe_configured", lambda: True)
monkeypatch.setattr(
payment_svc.stripe.Coupon, "create", lambda **kw: SimpleNamespace(id="coupon_X")
)
deleted: list = []
monkeypatch.setattr(
payment_svc.stripe.Coupon, "delete", lambda cid: deleted.append(cid)
)
def _boom(**kw):
raise payment_svc.stripe.error.StripeError("card_error test")
monkeypatch.setattr(payment_svc.stripe.PromotionCode, "create", _boom)
with pytest.raises(payment_svc.StripePromoCreationError):
payment_svc.create_promo_code(code="RATEE", type="percent", value=10)
assert deleted == ["coupon_X"]
assert payment_svc.load_promo_codes() == []
def test_unreadable_limit_or_counter_means_exhausted(promo_file):
"""max_redemptions / times_used illisibles (JSON corrompu) : le code est
traité comme épuisé avec un message, jamais comme illimité."""
base = {
"code": "CORROMPU", "active": True, "type": "percent", "value": 10,
"plans": ["all"], "expires_at": None, "stripe_coupon_id": "c",
"stripe_promotion_code_id": "p", "stripe_mode": "test",
"created_at": "t", "deactivated_at": None,
}
payment_svc.save_promo_codes([dict(base, max_redemptions="n/a", times_used=0)])
with pytest.raises(payment_svc.PromoValidationError) as err:
payment_svc.validate_promo_code("CORROMPU", "pro")
assert err.value.reason == "exhausted"
payment_svc.save_promo_codes([dict(base, max_redemptions=10, times_used="beaucoup")])
with pytest.raises(payment_svc.PromoValidationError) as err2:
payment_svc.validate_promo_code("CORROMPU", "pro")
assert err2.value.reason == "exhausted"
def test_save_is_atomic_no_temp_file_left(promo_file):
payment_svc.save_promo_codes([{
"code": "ATOM", "active": True, "type": "percent", "value": 5,
}])
assert payment_svc.PROMO_CODES_FILE.exists()
assert not payment_svc.PROMO_CODES_FILE.with_name(
payment_svc.PROMO_CODES_FILE.name + ".tmp"
).exists()
# le JSON reste lisible après écriture atomique
assert payment_svc.load_promo_codes()[0]["code"] == "ATOM"
def test_create_promo_for_free_plan_refused(client, admin_headers, promo_file):
r = client.post(
PROMOS_URL,
json={"code": "FREEBIE", "type": "percent", "value": 10, "plans": ["free"]},
headers=admin_headers,
)
assert r.status_code == 400
assert r.json()["error"] == "INVALID_PROMO_PLAN"
assert "Gratuit" in r.json()["message"]
def test_delete_promo_endpoint(client, admin_headers, promo_file, monkeypatch):
client.post(PROMOS_URL, json={"code": "ADIEU", "type": "percent", "value": 5},
headers=admin_headers)
r = client.delete(f"{PROMOS_URL}/ADIEU", headers=admin_headers)
assert r.status_code == 200
assert payment_svc.load_promo_codes() == []
r2 = client.delete(f"{PROMOS_URL}/ADIEU", headers=admin_headers)
assert r2.status_code == 404
def test_link_local_only_promo_to_stripe(client, admin_headers, promo_file, monkeypatch):
monkeypatch.setattr(payment_svc, "is_stripe_configured", lambda: True)
monkeypatch.setattr(payment_svc, "stripe_mode", lambda: "test")
monkeypatch.setattr(
payment_svc.stripe.Coupon, "create", lambda **kw: SimpleNamespace(id="coupon_L")
)
monkeypatch.setattr(
payment_svc.stripe.PromotionCode,
"create",
lambda **kw: SimpleNamespace(id="promo_L"),
)
client.post(PROMOS_URL, json={"code": "LOCAL1", "type": "amount", "value": 5},
headers=admin_headers)
r = client.post(f"{PROMOS_URL}/LOCAL1/link-stripe", headers=admin_headers)
assert r.status_code == 200, r.text
entry = r.json()["data"]
assert entry["stripe_coupon_id"] == "coupon_L"
assert entry["stripe_promotion_code_id"] == "promo_L"
assert entry["stripe_mode"] == "test"
# ---------------------------------------------------------------------------
# Point de contact public : GET /api/v1/promos/validate
# ---------------------------------------------------------------------------
def test_public_validate_endpoint(promo_file, client):
payment_svc.save_promo_codes([{
"code": "VALID10", "active": True, "type": "percent", "value": 10,
"plans": ["all"], "max_redemptions": None, "times_used": 0,
"expires_at": None, "stripe_coupon_id": "c",
"stripe_promotion_code_id": "p", "stripe_mode": "test",
"created_at": "t", "deactivated_at": None,
}])
r = client.get("/api/v1/promos/validate", params={"code": "valid10", "plan": "pro"})
assert r.status_code == 200
data = r.json()["data"]
assert data["valid"] is True
assert data["percent_off"] == 10
# montant en euros
payment_svc.save_promo_codes(payment_svc.load_promo_codes() + [{
"code": "EUROS5", "active": True, "type": "amount", "value": 5,
"plans": ["pro"], "max_redemptions": None, "times_used": 0,
"expires_at": None, "stripe_coupon_id": "c",
"stripe_promotion_code_id": "p", "stripe_mode": "test",
"created_at": "t", "deactivated_at": None,
}])
r2 = client.get("/api/v1/promos/validate", params={"code": "EUROS5", "plan": "pro"})
assert r2.json()["data"]["amount_off"] == 5
# causes explicites
r3 = client.get("/api/v1/promos/validate", params={"code": "INCONNU", "plan": "pro"})
assert r3.json()["data"]["valid"] is False
assert r3.json()["data"]["reason_code"] == "unknown"
payment_svc.save_promo_codes([dict(
code="EXPIRE1", active=True, type="percent", value=10,
plans=["all"], max_redemptions=None, times_used=0,
expires_at="2020-01-01T00:00:00+00:00", stripe_coupon_id="c",
stripe_promotion_code_id="p", stripe_mode="test",
created_at="t", deactivated_at=None,
)])
r4 = client.get("/api/v1/promos/validate", params={"code": "EXPIRE1", "plan": "pro"})
assert r4.json()["data"]["reason_code"] == "expired"

View File

@@ -0,0 +1,490 @@
"""
Routage multi-canaux des paliers IA (spec « Fournisseur par palier IA »).
- Chaque route d'un palier est un couple {provider, model} : une route peut
partir par z.ai direct (zhipu), DeepSeek direct, etc. ;
- une route dont la clé API manque est sautée au profit de la suivante ;
- l'ancien réglage à chaînes reste routé via openrouter (compatibilité) ;
- la facturation suit le PALIER (Premium = 5, Essentielle = 1), pas le canal.
"""
import asyncio
import pytest
from pathlib import Path
import routes.translate_routes as tr
# ---------------------------------------------------------------------------
# Résolution unitaire : (provider, model)
# ---------------------------------------------------------------------------
def _settings(ai_tiers: dict | None = None, **providers):
from routes.admin_routes import SettingsConfig
payload: dict = {"ai_tiers": ai_tiers} if ai_tiers is not None else {}
for provider, cfg in providers.items():
payload[provider] = cfg
return SettingsConfig.model_validate(payload)
def test_route_with_configured_key_is_used(monkeypatch):
monkeypatch.delenv("ZHIPU_API_KEY", raising=False)
cfg = _settings(
{
"essential": {
"models": [],
"default_model": "",
"routes": [{"provider": "zhipu", "model": "glm-5.3-flash"}],
},
"premium": {"models": [], "default_model": ""},
},
zhipu={"api_key": "zk-test"},
)
assert tr._resolve_tier_model(cfg, "pro", premium=False) == (
"zhipu",
"glm-5.3-flash",
)
def test_route_without_key_falls_back_to_next(monkeypatch):
monkeypatch.delenv("ZHIPU_API_KEY", raising=False)
monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False)
cfg = _settings(
{
"essential": {
"models": [],
"default_model": "",
"routes": [
{"provider": "zhipu", "model": "glm-5.3-flash"},
{"provider": "openrouter", "model": "deepseek/deepseek-v4-flash"},
],
},
"premium": {"models": [], "default_model": ""},
},
openrouter={"api_key": "sk-or-test"},
)
assert tr._resolve_tier_model(cfg, "pro", premium=False) == (
"openrouter",
"deepseek/deepseek-v4-flash",
)
def test_legacy_string_models_still_route_via_openrouter(monkeypatch):
monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
cfg = _settings(
{
"essential": {"models": ["deepseek/deepseek-v4-flash"], "default_model": ""},
"premium": {"models": [], "default_model": ""},
},
openrouter={"api_key": "sk-or-test"},
)
provider, model = tr._resolve_tier_model(cfg, "pro", premium=False)
assert (provider, model) == ("openrouter", "deepseek/deepseek-v4-flash")
def test_premium_route_resolves_premium_tier(monkeypatch):
monkeypatch.delenv("ZHIPU_API_KEY", raising=False)
cfg = _settings(
{
"essential": {"models": [], "default_model": ""},
"premium": {
"models": [],
"default_model": "",
"routes": [{"provider": "deepseek", "model": "deepseek-reasoner"}],
},
},
deepseek={"api_key": "dk-test"},
)
assert tr._resolve_tier_model(cfg, "business", premium=True) == (
"deepseek",
"deepseek-reasoner",
)
# ---------------------------------------------------------------------------
# Facturation par palier (pas par canal)
# ---------------------------------------------------------------------------
def test_cost_factor_follows_tier_not_channel():
assert tr._compute_cost_factor(None, "zhipu", premium=True) == 5
assert tr._compute_cost_factor(None, "deepseek", premium=True) == 5
assert tr._compute_cost_factor(None, "zhipu", premium=False) == 1
assert tr._compute_cost_factor(None, "openrouter_premium", premium=False) == 1
def test_cost_factor_legacy_heuristic_kept_without_tier():
class _P:
_model = "anthropic/claude-sonnet-5"
assert tr._compute_cost_factor(_P(), "zai") == 5
assert tr._compute_cost_factor(None, "openrouter_premium") == 5
assert tr._compute_cost_factor(None, "openrouter") == 1
# ---------------------------------------------------------------------------
# Validation admin des routes (normalize_ai_tiers)
# ---------------------------------------------------------------------------
def test_normalize_rejects_unknown_route_provider():
from routes.admin_routes import normalize_ai_tiers
cfg = _settings(
{
"essential": {
"models": [],
"default_model": "",
"routes": [{"provider": "anthropic", "model": "claude"}],
},
"premium": {"models": [], "default_model": ""},
}
)
with pytest.raises(ValueError, match="inconnu"):
normalize_ai_tiers(cfg.ai_tiers)
def test_normalize_rejects_route_without_model():
from routes.admin_routes import normalize_ai_tiers
cfg = _settings(
{
"essential": {
"models": [],
"default_model": "",
"routes": [{"provider": "zhipu", "model": " "}],
},
"premium": {"models": [], "default_model": ""},
}
)
with pytest.raises(ValueError, match="modèle"):
normalize_ai_tiers(cfg.ai_tiers)
def test_normalize_deduplicates_routes_and_keeps_order():
from routes.admin_routes import normalize_ai_tiers
cfg = _settings(
{
"essential": {
"models": [],
"default_model": "",
"routes": [
{"provider": "zhipu", "model": "glm-5.3-flash"},
{"provider": "zhipu", "model": "glm-5.3-flash", "label": "doublon"},
{"provider": "openrouter", "model": "deepseek/deepseek-v4-flash"},
],
},
"premium": {"models": [], "default_model": ""},
}
)
normalized = normalize_ai_tiers(cfg.ai_tiers)
assert [(r.provider, r.model) for r in normalized.essential.routes] == [
("zhipu", "glm-5.3-flash"),
("openrouter", "deepseek/deepseek-v4-flash"),
]
def test_normalize_route_cross_tier_guard():
"""Une route openrouter pointant un modèle Premium dans le palier
Essentielle est refusée (anti-croisement inchangé)."""
from routes.admin_routes import normalize_ai_tiers
cfg = _settings(
{
"essential": {
"models": [],
"default_model": "",
"routes": [{"provider": "openrouter", "model": "anthropic/claude-sonnet-5"}],
},
"premium": {"models": [], "default_model": ""},
}
)
with pytest.raises(ValueError, match="Premium"):
normalize_ai_tiers(cfg.ai_tiers)
def test_normalize_empty_routes_and_models_fills_official_range():
from routes.admin_routes import normalize_ai_tiers
from models.subscription import DEFAULT_AI_MODELS_ESSENTIAL
cfg = _settings(
{
"essential": {"models": [], "default_model": ""},
"premium": {"models": [], "default_model": ""},
}
)
normalized = normalize_ai_tiers(cfg.ai_tiers)
assert normalized.essential.models == list(DEFAULT_AI_MODELS_ESSENTIAL)
assert normalized.essential.routes == []
# ---------------------------------------------------------------------------
# Intégration worker : le canal choisi instancie le bon fournisseur
# ---------------------------------------------------------------------------
@pytest.fixture
def minimal_xlsx(tmp_path):
try:
import openpyxl
wb = openpyxl.Workbook()
wb.active["A1"] = "Hello"
p = tmp_path / "minimal.xlsx"
wb.save(p)
return p
except ImportError:
pytest.skip("openpyxl required")
def _register_job(job_id: str) -> None:
tr._translation_jobs[job_id] = {
"id": job_id,
"status": "processing",
"progress_percent": 0,
"current_step": "",
"provider": "openrouter",
"created_at": "2026-09-05T00:00:00+00:00",
}
def _run_job(monkeypatch, tmp_path, minimal_xlsx, admin_settings, provider: str,
user_plan: str, user_id=None, cost_capture: dict | None = None) -> list:
"""Lance _run_translation_job avec un faux fournisseur et renvoie les
instanciations capturées (model, base_url, api_key).
``user_id`` + ``cost_capture`` : enregistre le facteur de facturation
transmis à ``record_usage`` (substitué — aucun accès à la base)."""
import routes.admin_routes as admin_routes_mod
import services.providers.openai_provider as openai_provider_mod
import services.providers.deepseek_provider as deepseek_provider_mod
captured: list = []
class FakeProvider:
def __init__(self, api_key=None, model=None, base_url=None, timeout=60, **kwargs):
self.model = model
self._model = model
self.base_url = base_url
captured.append({"model": model, "base_url": base_url, "api_key": api_key})
class FakeExcelTranslator:
def __init__(self, provider=None):
self.provider = provider
def translate_file(self, input_path, output_path, target_lang,
source_lang="auto", progress_callback=None,
translate_images=False, **kwargs):
out = Path(output_path)
out.parent.mkdir(parents=True, exist_ok=True)
out.write_bytes(minimal_xlsx.read_bytes())
def get_translation_stats(self):
return {"attempted": 1, "changed": 1}
def _fake_record_usage(uid, pages, cost_factor, reserved_docs=1):
if cost_capture is not None:
cost_capture["cost_factor"] = cost_factor
monkeypatch.setattr(openai_provider_mod, "OpenAITranslationProvider", FakeProvider)
monkeypatch.setattr(deepseek_provider_mod, "DeepSeekTranslationProvider", FakeProvider)
monkeypatch.setattr(tr, "ExcelTranslator", FakeExcelTranslator)
monkeypatch.setattr(tr, "record_usage", _fake_record_usage)
monkeypatch.setattr(
admin_routes_mod, "load_settings", lambda: admin_settings, raising=True
)
_register_job("job-routes")
try:
asyncio.run(
tr._run_translation_job(
job_id="job-routes",
input_path=Path(minimal_xlsx),
file_extension=".xlsx",
target_lang="fr",
source_lang="en",
provider=provider,
user_id=user_id,
custom_prompt=None,
glossary_id=None,
prompt_id=None,
webhook_url=None,
user_plan=user_plan,
)
)
finally:
tr._translation_jobs.pop("job-routes", None)
return captured
def test_worker_routes_zhipu_channel(monkeypatch, tmp_path, minimal_xlsx):
monkeypatch.delenv("ZHIPU_API_KEY", raising=False)
settings = _settings(
{
"essential": {
"models": [],
"default_model": "",
"routes": [{"provider": "zhipu", "model": "glm-5.3-flash"}],
},
"premium": {"models": [], "default_model": ""},
},
zhipu={"api_key": "zk-test"},
)
captured = _run_job(monkeypatch, tmp_path, minimal_xlsx, settings,
provider="openrouter", user_plan="PlanType.PRO")
assert captured, "le fournisseur n'a pas été instancié"
assert captured[0]["model"] == "glm-5.3-flash"
assert captured[0]["base_url"] == "https://api.z.ai/api/paas/v4"
def test_worker_skips_route_without_key(monkeypatch, tmp_path, minimal_xlsx):
"""Clé zhipu absente → route suivante (openrouter) comme le veut la spec."""
monkeypatch.delenv("ZHIPU_API_KEY", raising=False)
settings = _settings(
{
"essential": {
"models": [],
"default_model": "",
"routes": [
{"provider": "zhipu", "model": "glm-5.3-flash"},
{"provider": "openrouter", "model": "deepseek/deepseek-v4-flash"},
],
},
"premium": {"models": [], "default_model": ""},
},
openrouter={"api_key": "sk-or-test"},
)
captured = _run_job(monkeypatch, tmp_path, minimal_xlsx, settings,
provider="openrouter", user_plan="PlanType.PRO")
assert captured[0]["base_url"] == "https://openrouter.ai/api/v1"
assert captured[0]["model"] == "deepseek/deepseek-v4-flash"
def test_worker_legacy_settings_keep_openrouter(monkeypatch, tmp_path, minimal_xlsx):
"""Ancien réglage (chaînes) : routé via openrouter comme aujourd'hui."""
monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
settings = _settings(
{
"essential": {"models": ["deepseek/deepseek-v4-flash"], "default_model": ""},
"premium": {"models": [], "default_model": ""},
},
openrouter={"api_key": "sk-or-test"},
)
captured = _run_job(monkeypatch, tmp_path, minimal_xlsx, settings,
provider="openrouter", user_plan="PlanType.PRO")
assert captured[0]["base_url"] == "https://openrouter.ai/api/v1"
assert captured[0]["model"] == "deepseek/deepseek-v4-flash"
def test_worker_deepseek_route_uses_deepseek_provider(monkeypatch, tmp_path, minimal_xlsx):
monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False)
settings = _settings(
{
"essential": {
"models": [],
"default_model": "",
"routes": [{"provider": "deepseek", "model": "deepseek-chat"}],
},
"premium": {"models": [], "default_model": ""},
},
deepseek={"api_key": "dk-test"},
)
captured = _run_job(monkeypatch, tmp_path, minimal_xlsx, settings,
provider="openrouter", user_plan="PlanType.PRO")
assert captured[0]["base_url"] == "https://api.deepseek.com/v1"
assert captured[0]["api_key"] == "dk-test"
# ---------------------------------------------------------------------------
# Facturation exécutée dans le worker : palier Essentielle = 1, Premium = 5
# ---------------------------------------------------------------------------
def test_worker_bills_essential_tier_at_cost_1(monkeypatch, tmp_path, minimal_xlsx):
monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
settings = _settings(
{
"essential": {"models": ["deepseek/deepseek-v4-flash"], "default_model": ""},
"premium": {"models": [], "default_model": ""},
},
openrouter={"api_key": "sk-or-test"},
)
cost: dict = {}
_run_job(monkeypatch, tmp_path, minimal_xlsx, settings,
provider="openrouter", user_plan="PlanType.PRO",
user_id="u1", cost_capture=cost)
assert cost["cost_factor"] == 1
def test_worker_bills_premium_tier_at_cost_5(monkeypatch, tmp_path, minimal_xlsx):
monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
settings = _settings(
{
"essential": {"models": [], "default_model": ""},
"premium": {"models": ["anthropic/claude-sonnet-5"], "default_model": ""},
},
openrouter={"api_key": "sk-or-test"},
)
cost: dict = {}
_run_job(monkeypatch, tmp_path, minimal_xlsx, settings,
provider="openrouter_premium", user_plan="PlanType.BUSINESS",
user_id="u1", cost_capture=cost)
assert cost["cost_factor"] == 5
# ---------------------------------------------------------------------------
# Première sauvegarde depuis l'éditeur : migration fidèle hérité → routes
# ---------------------------------------------------------------------------
def test_first_editor_save_migrates_legacy_settings_without_loss(monkeypatch):
"""Des réglages à chaînes (models/default_model) convertis en routes par
l'éditeur, puis normalisés, doivent résoudre EXACTEMENT les mêmes
canaux/modèles qu'avant la migration."""
from routes.admin_routes import SettingsConfig, normalize_ai_tiers
legacy = _settings(
{
"essential": {
"models": ["z-ai/glm-5.3-flash", "deepseek/deepseek-v4-flash"],
"default_model": "z-ai/glm-5.3-flash",
},
"premium": {
"models": ["anthropic/claude-sonnet-5", "deepseek/deepseek-v4-pro"],
"default_model": "anthropic/claude-sonnet-5",
},
},
openrouter={"api_key": "sk-or-test"},
)
before_essential = tr._resolve_tier_model(legacy, "pro", premium=False)
before_premium = tr._resolve_tier_model(legacy, "business", premium=True)
def _editor_payload(tier):
"""Miroir de la conversion faite par la page Modèles : le défaut en
tête, puis le reste de la liste — le tout en routes openrouter, avec
le miroir hérité conservé (models/default_model)."""
ordered = [tier.default_model] + [
m for m in tier.models if m != tier.default_model
]
routes = [{"provider": "openrouter", "model": m} for m in ordered]
return {"models": ordered, "default_model": tier.default_model, "routes": routes}
payload = SettingsConfig.model_validate(
{
"openrouter": {"api_key": "sk-or-test"},
"ai_tiers": {
"essential": _editor_payload(legacy.ai_tiers.essential),
"premium": _editor_payload(legacy.ai_tiers.premium),
},
}
)
normalized = normalize_ai_tiers(payload.ai_tiers)
migrated = SettingsConfig.model_validate(
{"openrouter": {"api_key": "sk-or-test"}, "ai_tiers": normalized.model_dump()}
)
after_essential = tr._resolve_tier_model(migrated, "pro", premium=False)
after_premium = tr._resolve_tier_model(migrated, "business", premium=True)
assert after_essential == before_essential == ("openrouter", "z-ai/glm-5.3-flash")
assert after_premium == before_premium == ("openrouter", "anthropic/claude-sonnet-5")
# Aucune perte : tous les modèles d'origine restent résolubles dans l'ordre.
assert [r.model for r in normalized.essential.routes] == [
"z-ai/glm-5.3-flash",
"deepseek/deepseek-v4-flash",
]
assert [r.model for r in normalized.premium.routes] == [
"anthropic/claude-sonnet-5",
"deepseek/deepseek-v4-pro",
]

View File

@@ -52,9 +52,12 @@ def client(monkeypatch, tmp_path: Path):
def _settings(ai_tiers: dict | None = None):
from routes.admin_routes import SettingsConfig
if ai_tiers is None:
return SettingsConfig()
return SettingsConfig.model_validate({"ai_tiers": ai_tiers})
# La résolution est clé-consciente : les tests de résolution partent avec
# une clé OpenRouter disponible (comme un serveur configuré).
payload = {"openrouter": {"api_key": "sk-or-test"}}
if ai_tiers is not None:
payload["ai_tiers"] = ai_tiers
return SettingsConfig.model_validate(payload)
# ---------------------------------------------------------------------------
@@ -83,17 +86,17 @@ def test_old_single_model_keys_removed():
# ---------------------------------------------------------------------------
# Résolution par palier
# Résolution par palier — renvoie la route complète (provider, model)
# ---------------------------------------------------------------------------
def test_pro_resolves_essential_plan_default():
assert _resolve_tier_model(_settings(), "pro", premium=False) == (
"deepseek/deepseek-v4-flash"
("openrouter", "deepseek/deepseek-v4-flash")
)
def test_business_resolves_premium_plan_default():
assert _resolve_tier_model(_settings(), "business", premium=True) == (
"anthropic/claude-sonnet-5"
("openrouter", "anthropic/claude-sonnet-5")
)
@@ -101,7 +104,9 @@ def test_admin_tier_default_overrides_plan_default():
cfg = _settings(
{"essential": {"models": [], "default_model": "z-ai/glm-5.3-flash"}}
)
assert _resolve_tier_model(cfg, "pro", premium=False) == "z-ai/glm-5.3-flash"
assert _resolve_tier_model(cfg, "pro", premium=False) == (
"openrouter", "z-ai/glm-5.3-flash"
)
def test_admin_tier_models_first_used_when_default_empty():
@@ -113,7 +118,9 @@ def test_admin_tier_models_first_used_when_default_empty():
}
}
)
assert _resolve_tier_model(cfg, "pro", premium=False) == "minimax/minimax-m3"
assert _resolve_tier_model(cfg, "pro", premium=False) == (
"openrouter", "minimax/minimax-m3"
)
def test_pro_never_resolves_premium_model():
@@ -124,7 +131,8 @@ def test_pro_never_resolves_premium_model():
}
)
for plan_str in ("pro", "PlanType.PRO"):
model = _resolve_tier_model(cfg, plan_str, premium=False)
provider, model = _resolve_tier_model(cfg, plan_str, premium=False)
assert provider == "openrouter"
assert model in DEFAULT_AI_MODELS_ESSENTIAL
assert model not in DEFAULT_AI_MODELS_PREMIUM
@@ -155,3 +163,68 @@ def test_get_plans_exposes_model_lists(client):
]
assert plans["business"]["ai_models_premium"][0] == "anthropic/claude-sonnet-5"
assert "ai_model_essential" not in plans["pro"]
# ---------------------------------------------------------------------------
# Point de contact : GET /providers/available affiche le modèle (chaîne simple)
# ---------------------------------------------------------------------------
def test_available_providers_model_is_a_string(monkeypatch, tmp_path):
"""``_resolve_tier_model`` renvoie désormais (provider, model) : le champ
« model » de /providers/available doit rester une chaîne, égale au modèle
attendu pour les deux paliers (appelé ici avec un utilisateur Business,
car la liste est filtrée par forfait)."""
import routes.legacy_routes as legacy_mod
import routes.admin_routes as admin_routes_mod
settings = _settings(
{
"essential": {"models": [], "default_model": "z-ai/glm-5.3-flash"},
"premium": {"models": [], "default_model": "anthropic/claude-sonnet-5"},
}
)
# Les deux passerelles sont affichées dès lors qu'elles sont actives côté
# réglages (indépendamment des variables d'environnement de la machine).
settings.openrouter.enabled = True
settings.openrouter_premium.enabled = True
monkeypatch.setattr(admin_routes_mod, "load_settings", lambda: settings, raising=True)
from fastapi.testclient import TestClient
from middleware.rate_limiting import RateLimitManager
from models.subscription import PlanType
async def _check_request_allow(self, request):
return True, "ok", "test"
async def _check_translation_allow(self, request, file_size_mb=0):
return True, "ok"
monkeypatch.setattr(RateLimitManager, "check_request", _check_request_allow)
monkeypatch.setattr(RateLimitManager, "check_translation", _check_translation_allow)
import services.auth_service as auth_svc
monkeypatch.setattr(auth_svc, "USERS_FILE", tmp_path / "users.json")
monkeypatch.setattr(auth_svc, "USE_DATABASE", False)
monkeypatch.setattr(auth_svc, "DATABASE_AVAILABLE", False)
# Utilisateur Business factice (les moteurs Premium sont filtrés par plan).
business_user = type("U", (), {"plan": PlanType.BUSINESS, "id": "u-biz"})()
async def _fake_user():
return business_user
from main import app
app.dependency_overrides[legacy_mod.get_authenticated_user] = _fake_user
try:
with TestClient(app, raise_server_exceptions=True) as tc:
r = tc.get("/api/v1/providers/available")
finally:
app.dependency_overrides.pop(legacy_mod.get_authenticated_user, None)
assert r.status_code == 200, r.text
engines = {e["id"]: e for e in r.json()["providers"]}
assert isinstance(engines["openrouter"]["model"], str)
assert engines["openrouter"]["model"] == "z-ai/glm-5.3-flash"
assert isinstance(engines["openrouter_premium"]["model"], str)
assert engines["openrouter_premium"]["model"] == "anthropic/claude-sonnet-5"

View File

@@ -0,0 +1,460 @@
"""
Génération d'email par IA dans la page Marketing (spec « Fournisseur par
palier IA ») : prompt construit depuis MARKETING_PLAN.md, analyse robuste de
la sortie, nettoyage basique du HTML, aucun envoi automatique.
"""
import json
import pytest
from pathlib import Path
import routes.admin_routes as admin_routes_mod
ADMIN_LOGIN_URL = "/api/v1/admin/login"
GENERATE_URL = "/api/v1/admin/marketing/email/generate"
FAKE_PLAN = (
"# Plan Marketing — Office Translator (Wordly.art)\n\n"
"## 1. Positionnement & Proposition de Valeur\n"
"Office Translator (marque Wordly.art) traduit Word, Excel, PowerPoint et "
"PDF en préservant la mise en page. USP : « Traduit en place. Zéro perte "
"de mise en page. »\n\n"
"## 2. Tarification\n"
"| Plan | Mensuel |\n| Free | 0 € |\n| Starter | 9 € |\n| Pro | 19 € |\n\n"
"## 3. Actifs Marketing en Place\n"
"(cette section ne doit PAS figurer dans le prompt)\n"
)
@pytest.fixture
def promo_file(tmp_path: Path, monkeypatch) -> Path:
import services.payment_service as payment_svc
path = tmp_path / "promo_codes.json"
monkeypatch.setattr(payment_svc, "PROMO_CODES_FILE", path)
return path
@pytest.fixture
def gen_env(tmp_path: Path, monkeypatch, promo_file):
"""Plan marketing isolé + aucune clé par défaut (chaque test choisit)."""
plan_file = tmp_path / "MARKETING_PLAN.md"
plan_file.write_text(FAKE_PLAN, encoding="utf-8")
monkeypatch.setattr(admin_routes_mod, "MARKETING_PLAN_FILE", plan_file)
monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
monkeypatch.delenv("ZHIPU_API_KEY", raising=False)
monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False)
return {"plan_file": plan_file}
@pytest.fixture
def client(gen_env, monkeypatch, tmp_path: Path):
import services.auth_service as auth_svc
from middleware.rate_limiting import RateLimitManager
monkeypatch.setattr(auth_svc, "USERS_FILE", tmp_path / "users.json")
monkeypatch.setattr(auth_svc, "USE_DATABASE", False)
monkeypatch.setattr(auth_svc, "DATABASE_AVAILABLE", False)
async def _check_request_allow(self, request):
return True, "ok", "test"
async def _check_translation_allow(self, request, file_size_mb=0):
return True, "ok"
monkeypatch.setattr(RateLimitManager, "check_request", _check_request_allow)
monkeypatch.setattr(RateLimitManager, "check_translation", _check_translation_allow)
from fastapi.testclient import TestClient
from main import app
return TestClient(app, raise_server_exceptions=True)
@pytest.fixture
def admin_headers(client, monkeypatch):
monkeypatch.setattr(admin_routes_mod, "ADMIN_USERNAME", "admin")
monkeypatch.setattr(admin_routes_mod, "ADMIN_PASSWORD", "admin-secret")
monkeypatch.setattr(admin_routes_mod, "ADMIN_PASSWORD_HASH", None)
r = client.post(ADMIN_LOGIN_URL, json={"password": "admin-secret"})
assert r.status_code == 200, r.text
return {"Authorization": f"Bearer {r.json()['access_token']}"}
def _with_openrouter_key(monkeypatch):
from routes.admin_routes import SettingsConfig
monkeypatch.setattr(
admin_routes_mod, "load_settings",
lambda: SettingsConfig.model_validate({"openrouter": {"api_key": "sk-or-test"}}),
raising=True,
)
# ---------------------------------------------------------------------------
def test_generate_requires_known_channel(client, admin_headers, gen_env):
r = client.post(
GENERATE_URL,
json={"provider": "mystery", "brief": "Relance d'activation."},
headers=admin_headers,
)
assert r.status_code == 400
assert r.json()["error"] == "UNKNOWN_CHANNEL"
def test_generate_requires_configured_key(client, admin_headers, gen_env, monkeypatch):
monkeypatch.setattr(
admin_routes_mod, "load_settings",
lambda: admin_routes_mod.SettingsConfig(), raising=True,
)
r = client.post(
GENERATE_URL,
json={"provider": "openrouter", "brief": "Relance d'activation."},
headers=admin_headers,
)
assert r.status_code == 400
assert r.json()["error"] == "PROVIDER_NOT_CONFIGURED"
def test_generate_builds_prompt_from_marketing_plan(client, admin_headers, gen_env, monkeypatch):
"""Le prompt système embarque le plan marketing (positionnement + tarifs)
et le brief ; les sections suivantes du plan n'y figurent pas."""
_with_openrouter_key(monkeypatch)
captured: dict = {}
async def _fake_llm(admin_cfg, provider, model, system_prompt, user_prompt):
captured["system"] = system_prompt
captured["user"] = user_prompt
captured["provider"], captured["model"] = provider, model
return json.dumps({
"subject": "Votre document vous attend",
"html": "<html><body><p>Terminez votre traduction.</p></body></html>",
})
monkeypatch.setattr(admin_routes_mod, "_call_llm_chat", _fake_llm)
r = client.post(
GENERATE_URL,
json={"provider": "openrouter", "brief": "Relance des inscrits de la liste d'attente."},
headers=admin_headers,
)
assert r.status_code == 200, r.text
assert "Wordly.art" in captured["system"]
assert "Traduit en place" in captured["system"]
assert "19 €" in captured["system"]
assert "Actifs Marketing" not in captured["system"]
assert "liste d'attente" in captured["user"]
assert captured["model"] # modèle par défaut du canal
def test_generate_parses_fenced_json(client, admin_headers, gen_env, monkeypatch):
_with_openrouter_key(monkeypatch)
async def _fake_llm(admin_cfg, provider, model, system_prompt, user_prompt):
return (
"```json\n"
'{"subject": "Offre de lancement", "html": "<p>Bonjour !</p>"}\n'
"```"
)
monkeypatch.setattr(admin_routes_mod, "_call_llm_chat", _fake_llm)
r = client.post(
GENERATE_URL,
json={"provider": "openrouter", "brief": "b"},
headers=admin_headers,
)
assert r.status_code == 200
data = r.json()["data"]
assert data["subject"] == "Offre de lancement"
assert data["html"] == "<p>Bonjour !</p>"
def test_generate_falls_back_to_title(client, admin_headers, gen_env, monkeypatch):
"""Sortie non JSON : repli <title> = sujet, corps = HTML."""
_with_openrouter_key(monkeypatch)
async def _fake_llm(admin_cfg, provider, model, system_prompt, user_prompt):
return (
"<html><head><title>Relance</title></head>"
"<body><p>Reprenez votre traduction.</p></body></html>"
)
monkeypatch.setattr(admin_routes_mod, "_call_llm_chat", _fake_llm)
r = client.post(
GENERATE_URL,
json={"provider": "openrouter", "brief": "b"},
headers=admin_headers,
)
assert r.status_code == 200
data = r.json()["data"]
assert data["subject"] == "Relance"
assert "Reprenez votre traduction." in data["html"]
def test_generate_unparseable_output_returns_502(client, admin_headers, gen_env, monkeypatch):
_with_openrouter_key(monkeypatch)
async def _fake_llm(admin_cfg, provider, model, system_prompt, user_prompt):
return "Je ne peux pas répondre à cela."
monkeypatch.setattr(admin_routes_mod, "_call_llm_chat", _fake_llm)
r = client.post(
GENERATE_URL,
json={"provider": "openrouter", "brief": "b"},
headers=admin_headers,
)
assert r.status_code == 502
assert r.json()["error"] == "AI_GENERATION_FAILED"
def test_generate_llm_error_returns_502(client, admin_headers, gen_env, monkeypatch):
_with_openrouter_key(monkeypatch)
async def _fake_llm(admin_cfg, provider, model, system_prompt, user_prompt):
raise RuntimeError("upstream timeout")
monkeypatch.setattr(admin_routes_mod, "_call_llm_chat", _fake_llm)
r = client.post(
GENERATE_URL,
json={"provider": "openrouter", "brief": "b"},
headers=admin_headers,
)
assert r.status_code == 502
assert r.json()["error"] == "AI_GENERATION_FAILED"
def test_generate_sanitizes_html(client, admin_headers, gen_env, monkeypatch):
"""Nettoyage basique : pas de <script>, pas de gestionnaires d'événements."""
_with_openrouter_key(monkeypatch)
async def _fake_llm(admin_cfg, provider, model, system_prompt, user_prompt):
return json.dumps({
"subject": "s",
"html": "<p onclick=\"evil()\">ok</p><script>alert(1)</script>"
"<a href=\"javascript:alert(2)\">x</a>",
})
monkeypatch.setattr(admin_routes_mod, "_call_llm_chat", _fake_llm)
r = client.post(
GENERATE_URL,
json={"provider": "openrouter", "brief": "b"},
headers=admin_headers,
)
assert r.status_code == 200
html = r.json()["data"]["html"]
assert "<script" not in html
assert "onclick" not in html
assert "javascript:" not in html
def test_generate_mentions_active_promo_in_prompt(client, admin_headers, gen_env, monkeypatch):
"""Le sélecteur de promo alimente le prompt (code + valeur exacts)."""
_with_openrouter_key(monkeypatch)
import services.payment_service as payment_svc
payment_svc.save_promo_codes([{
"code": "RENTREE10", "active": True, "type": "percent", "value": 10,
"plans": ["pro"], "max_redemptions": 50, "times_used": 3,
"expires_at": None, "stripe_coupon_id": None,
"stripe_promotion_code_id": None, "stripe_mode": "local_only",
"created_at": "t", "deactivated_at": None,
}])
captured: dict = {}
async def _fake_llm(admin_cfg, provider, model, system_prompt, user_prompt):
captured["system"] = system_prompt
return json.dumps({"subject": "s", "html": "<p>x</p>"})
monkeypatch.setattr(admin_routes_mod, "_call_llm_chat", _fake_llm)
r = client.post(
GENERATE_URL,
json={"provider": "openrouter", "brief": "b", "promo_code": "rentree10"},
headers=admin_headers,
)
assert r.status_code == 200
assert "RENTREE10" in captured["system"]
assert "-10 %" in captured["system"]
def test_generate_uses_requested_channel_and_model(client, admin_headers, gen_env, monkeypatch):
from routes.admin_routes import SettingsConfig
monkeypatch.setattr(
admin_routes_mod, "load_settings",
lambda: SettingsConfig.model_validate({"zhipu": {"api_key": "zk"}}),
raising=True,
)
captured: dict = {}
async def _fake_llm(admin_cfg, provider, model, system_prompt, user_prompt):
captured["provider"], captured["model"] = provider, model
return json.dumps({"subject": "s", "html": "<p>x</p>"})
monkeypatch.setattr(admin_routes_mod, "_call_llm_chat", _fake_llm)
r = client.post(
GENERATE_URL,
json={"provider": "zhipu", "model": "glm-5.3-flash", "brief": "b"},
headers=admin_headers,
)
assert r.status_code == 200
assert captured["provider"] == "zhipu"
assert captured["model"] == "glm-5.3-flash"
def test_generate_requires_admin(client, gen_env):
r = client.post(GENERATE_URL, json={"provider": "openrouter", "brief": "b"})
assert r.status_code == 401
# ---------------------------------------------------------------------------
# Helpers d'analyse (tests unitaires directs)
# ---------------------------------------------------------------------------
def test_parse_generated_email_variants():
parse = admin_routes_mod._parse_generated_email
assert parse('{"subject":"a","html":"<p>b</p>"}') == ("a", "<p>b</p>")
assert parse('Blabla\n{"subject":"a","html":"<p>b</p>"}\nFin') == ("a", "<p>b</p>")
assert parse("pas de contenu utile") is None
assert parse("") is None
# JSON sans html → None (sujet ET html requis)
assert parse('{"subject":"a"}') is None
# ---------------------------------------------------------------------------
# Correctifs de relecture : plan manquant, promo inconnue, CTA, RTL, bornes
# ---------------------------------------------------------------------------
def test_generate_without_marketing_plan_is_refused(client, admin_headers, gen_env, monkeypatch):
"""Plan marketing absent : erreur claire, jamais de chiffres inventés."""
_with_openrouter_key(monkeypatch)
monkeypatch.setattr(
admin_routes_mod, "MARKETING_PLAN_FILE", gen_env["plan_file"].with_suffix(".absent")
)
r = client.post(
GENERATE_URL,
json={"provider": "openrouter", "brief": "b"},
headers=admin_headers,
)
assert r.status_code == 503
assert r.json()["error"] == "MARKETING_PLAN_MISSING"
def test_generate_unknown_promo_returns_promo_not_found(
client, admin_headers, gen_env, monkeypatch
):
"""Code promo inexistant ou inactif : 400 explicite, pas de génération
silencieuse sans offre."""
_with_openrouter_key(monkeypatch)
async def _fake_llm(*a, **kw): # ne doit PAS être appelé
raise AssertionError("l'appel LLM ne doit pas avoir lieu")
monkeypatch.setattr(admin_routes_mod, "_call_llm_chat", _fake_llm)
r = client.post(
GENERATE_URL,
json={"provider": "openrouter", "brief": "b", "promo_code": "NEXISTEPAS"},
headers=admin_headers,
)
assert r.status_code == 400
assert r.json()["error"] == "PROMO_NOT_FOUND"
def test_generate_prompt_contains_pricing_cta_link(
client, admin_headers, gen_env, monkeypatch
):
"""Avec une réduction : le prompt impose un bouton vers /pricing?promo=CODE."""
_with_openrouter_key(monkeypatch)
import services.payment_service as payment_svc
payment_svc.save_promo_codes([{
"code": "RENTREE10", "active": True, "type": "percent", "value": 10,
"plans": ["all"], "max_redemptions": None, "times_used": 0,
"expires_at": None, "stripe_coupon_id": None,
"stripe_promotion_code_id": None, "stripe_mode": "local_only",
"created_at": "t", "deactivated_at": None,
}])
monkeypatch.setenv("FRONTEND_URL", "https://wordly.art")
captured: dict = {}
async def _fake_llm(admin_cfg, provider, model, system_prompt, user_prompt):
captured["system"] = system_prompt
return json.dumps({"subject": "s", "html": "<p>x</p>"})
monkeypatch.setattr(admin_routes_mod, "_call_llm_chat", _fake_llm)
r = client.post(
GENERATE_URL,
json={"provider": "openrouter", "brief": "b", "promo_code": "rentree10"},
headers=admin_headers,
)
assert r.status_code == 200
assert "https://wordly.art/pricing?promo=RENTREE10" in captured["system"]
def test_generate_wraps_rtl_language_in_dir_rtl(
client, admin_headers, gen_env, monkeypatch
):
"""Arabe et persan : le HTML généré est enveloppé dans <div dir="rtl">."""
_with_openrouter_key(monkeypatch)
async def _fake_llm(admin_cfg, provider, model, system_prompt, user_prompt):
assert 'dir="rtl"' in system_prompt # consigne donnée au modèle
return json.dumps({"subject": "s", "html": "<p>مرحبا</p>"})
monkeypatch.setattr(admin_routes_mod, "_call_llm_chat", _fake_llm)
for language in ("العربية", "فارسی"):
r = client.post(
GENERATE_URL,
json={"provider": "openrouter", "brief": "b", "language": language},
headers=admin_headers,
)
assert r.status_code == 200
html = r.json()["data"]["html"]
assert html.startswith('<div dir="rtl">')
assert html.endswith("</div>")
def test_generate_field_bounds_are_enforced(client, admin_headers, gen_env):
# Les bornes Pydantic sont appliquées (422 brut ou 400 via le gestionnaire
# global RequestValidationError de l'application).
r = client.post(
GENERATE_URL,
json={"provider": "openrouter", "brief": "b", "tone": "x" * 60},
headers=admin_headers,
)
assert r.status_code in (400, 422)
r2 = client.post(
GENERATE_URL,
json={"provider": "openrouter", "brief": "b", "model": "m" * 130},
headers=admin_headers,
)
assert r2.status_code in (400, 422)
# ---------------------------------------------------------------------------
# Durcissement du nettoyage HTML (casse mixte, balises orphelines, data:)
# ---------------------------------------------------------------------------
def test_sanitize_generated_html_hardened():
sanitize = admin_routes_mod._sanitize_generated_html
# script orphelin (jamais fermé) et casse mixte
assert "<script" not in sanitize('<p>ok</p><ScRiPt>alert(1)')
# iframe / object / embed / form retirés avec leur contenu
html = sanitize('<p>a</p><iframe src="https://x.y"><p>hidden</p></iframe><p>b</p>')
assert "iframe" not in html and "hidden" not in html
html = sanitize('<object data="x"></object><embed src="y"><form action="z"></form>')
for tag in ("object", "embed", "form"):
assert tag not in html
# javascript: quel que soit le style, et data:
assert "javascript:" not in sanitize("<a href='JaVa ScRiPt:alert(1)'>x</a>")
assert "javascript:" not in sanitize("<a href='java\tscript:alert(1)'>x</a>")
assert "data:" not in sanitize("<a href='DATA:text/html;base64,xxx'>x</a>")
# le contenu sain passe intact
assert sanitize('<p style="color:#333">Bonjour</p>') == '<p style="color:#333">Bonjour</p>'
def test_marketing_plan_excerpt_truncated(tmp_path, monkeypatch):
plan = tmp_path / "MARKETING_PLAN.md"
plan.write_text("## 1. A\n" + "x" * 9000 + "\n## 3. B\n", encoding="utf-8")
monkeypatch.setattr(admin_routes_mod, "MARKETING_PLAN_FILE", plan)
excerpt = admin_routes_mod._load_marketing_plan_excerpt()
assert len(excerpt) <= admin_routes_mod.MARKETING_PLAN_PROMPT_MAX_CHARS + 10
assert excerpt.startswith("## 1.")
assert "## 3." not in excerpt