sepehr/office_translator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(abonnements): paliers LLM Essentielle/Premium, admin modeles par forfait, statistiques revenus, emails marketing
All checks were successful
Deploy to Production / Build and Deploy (push) Successful in 2m51s
Details

Browse Source 
- Gamme officielle : Essentielle (Pro) = deepseek-v4-flash, glm-5.3-flash,
  minimax-m3 ; Premium (Business) = claude-sonnet-5, deepseek-v4-pro, glm-5.3
- Routage reel : le modele employe suit le palier IA du forfait (reglages
  admin > defaut du plan), garde anti-croisement de palier
- Nouvelle page admin « Modeles & abonnements » (matrice forfaits/paliers,
  modele par defaut, catalogue OpenRouter)
- Statistiques enrichies : revenus (30 j + total), MRR estime, credits,
  liste d'attente, paliers utilises
- Nouvelle page admin « Marketing » : audiences avec compteurs, apercu,
  envoi test obligatoire, historique, desabonnement public + en-tetes
  List-Unsubscribe ; .gitignore pour les donnees d'execution
- Interface (accueil + tarifs) alignee sur la gamme, 13 langues completes
- Tests : routage par palier (fonction + tache), marketing, revenus en base
This commit is contained in:
sepehr
2026-09-05 12:51:49 +02:00
parent 4e6850d6b1
commit 6c26712687
76 changed files with 5289 additions and 271 deletions
Download Patch File Download Diff File Expand all files Collapse all files

483
tests/test_admin_marketing.py Normal file
View File

@@ -0,0 +1,483 @@
"""
Endpoints marketing (spec « Stratégie LLM par abonnement ») :
- audiences avec compteurs (liste d'attente, inactifs, par plan, tous),
- envoi test obligatoire avant tout envoi en masse,
- désabonnés exclus de tout envoi,
- historique journalisé (envoyés/échecs),
- désabonnement public.
"""
import asyncio
import json
import pytest
from pathlib import Path
from fastapi.testclient import TestClient
ADMIN_LOGIN_URL = "/api/v1/admin/login"
AUDIENCES_URL = "/api/v1/admin/marketing/audiences"
SEND_URL = "/api/v1/admin/marketing/email/send"
HISTORY_URL = "/api/v1/admin/marketing/email/history"
UNSUBSCRIBE_URL = "/api/v1/marketing/unsubscribe"
SUBJECT = "Relance test"
HTML = "<html><body><p>Bonjour !</p></body></html>"
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")
@pytest.fixture
def marketing_env(tmp_path: Path, monkeypatch):
"""Fichiers data isolés + service email simulé."""
import routes.admin_routes as admin_routes_mod
import routes.waitlist_routes as waitlist_mod
import services.email_service as email_svc
history_file = tmp_path / "marketing_emails.json"
unsub_file = tmp_path / "unsubscribes.json"
waitlist_file = tmp_path / "waitlist.json"
monkeypatch.setattr(admin_routes_mod, "MARKETING_HISTORY_FILE", history_file)
monkeypatch.setattr(admin_routes_mod, "UNSUBSCRIBES_FILE", unsub_file)
monkeypatch.setattr(waitlist_mod, "WAITLIST_FILE", waitlist_file)
monkeypatch.setattr(admin_routes_mod, "_active_campaign_hashes", set())
sent_to: list[str] = []
captured_headers: dict[str, dict] = {}
async def _fake_send(to: str, subject: str, body: str, extra_headers=None) -> bool:
sent_to.append(to)
captured_headers[to] = extra_headers or {}
return True
async def _failing_send(to: str, subject: str, body: str, extra_headers=None) -> bool:
return False
monkeypatch.setattr(email_svc, "send_email_async", _fake_send)
monkeypatch.setattr(email_svc, "is_smtp_configured", lambda: True)
def set_sender(fn):
monkeypatch.setattr(email_svc, "send_email_async", fn)
return {
"history_file": history_file,
"unsub_file": unsub_file,
"waitlist_file": waitlist_file,
"sent_to": sent_to,
"captured_headers": captured_headers,
"fail_mode": lambda: set_sender(_failing_send),
"set_sender": set_sender,
}
@pytest.fixture
def client(marketing_env, 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 main import app
return TestClient(app, raise_server_exceptions=True)
@pytest.fixture
def admin_token(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
return r.json()["access_token"]
@pytest.fixture
def auth_headers(admin_token):
return {"Authorization": f"Bearer {admin_token}"}
def _seed_users(tmp_path: Path) -> Path:
users = {
"u-pro": {
"email": "pro@example.com",
"name": "Pro User",
"plan": "pro",
"docs_translated_this_month": 5,
},
"u-idle": {
"email": "idle@example.com",
"name": "Idle User",
"plan": "starter",
"docs_translated_this_month": 0,
},
}
path = tmp_path / "users.json"
_write_json(path, users)
return path
# ---------------------------------------------------------------------------
# Audiences
# ---------------------------------------------------------------------------
def test_audiences_counts_and_unsubscribed_excluded(
client, auth_headers, marketing_env, tmp_path
):
_seed_users(tmp_path)
_write_json(
marketing_env["waitlist_file"],
[
{"email": "lead1@example.com", "interest": "pdf", "joined_at": "2026-09-01"},
{"email": "lead2@example.com", "interest": None, "joined_at": "2026-09-02"},
],
)
_write_json(marketing_env["unsub_file"], [{"email": "lead2@example.com"}])
r = client.get(AUDIENCES_URL, headers=auth_headers)
assert r.status_code == 200, r.text
data = r.json()["data"]
assert data["audiences"]["waitlist"] == 1 # lead2 désabonné → exclu
assert data["audiences"]["plan:pro"] == 1
assert data["audiences"]["inactive_30"] == 1 # idle@example.com
assert data["audiences"]["all_users"] == 2
assert data["unsubscribed_count"] == 1
def test_audiences_requires_admin(client):
assert client.get(AUDIENCES_URL).status_code == 401
# ---------------------------------------------------------------------------
# Envoi test + garde-fou « test avant envoi réel »
# ---------------------------------------------------------------------------
def _post_send(client, auth_headers, **overrides):
payload = {
"audience": "waitlist",
"subject": SUBJECT,
"html": HTML,
"test_mode": True,
}
payload.update(overrides)
return client.post(SEND_URL, json=payload, headers=auth_headers)
def test_test_send_writes_history(client, auth_headers, marketing_env):
_write_json(
marketing_env["waitlist_file"],
[{"email": "lead1@example.com"}],
)
r = _post_send(client, auth_headers, test_email="admin@example.com")
assert r.status_code == 200, r.text
assert r.json()["data"]["sent"] == 1
assert marketing_env["sent_to"] == ["admin@example.com"]
entries = json.loads(marketing_env["history_file"].read_text(encoding="utf-8"))
assert len(entries) == 1
assert entries[0]["test_mode"] is True
assert entries[0]["sent_count"] == 1
assert entries[0]["content_hash"]
def test_mass_send_requires_matching_test_first(client, auth_headers, marketing_env):
_write_json(marketing_env["waitlist_file"], [{"email": "lead1@example.com"}])
r = _post_send(client, auth_headers, test_mode=False)
assert r.status_code == 400
assert r.json()["error"] == "TEST_SEND_REQUIRED"
assert marketing_env["sent_to"] == []
def test_mass_send_after_test_excludes_unsubscribed(client, auth_headers, marketing_env):
_write_json(
marketing_env["waitlist_file"],
[
{"email": "lead1@example.com"},
{"email": "gone@example.com"},
],
)
_write_json(marketing_env["unsub_file"], [{"email": "gone@example.com"}])
# Envoi test au contenu identique (préalable obligatoire)
assert _post_send(client, auth_headers, test_email="admin@example.com").status_code == 200
marketing_env["sent_to"].clear()
r = _post_send(client, auth_headers, test_mode=False)
assert r.status_code == 202, r.text
assert r.json()["data"]["queued"] == 1 # gone@example.com exclu de la file
def test_bulk_executor_logs_history_and_excludes_unsubscribed(marketing_env):
"""L'exécuteur d'envoi de fond journalise envoyés/échecs et n'écrit
jamais à un désabonné (résolution d'audience + envoi réel)."""
import routes.admin_routes as admin_routes_mod
_write_json(
marketing_env["waitlist_file"],
[
{"email": "lead1@example.com"},
{"email": "gone@example.com"},
],
)
_write_json(marketing_env["unsub_file"], [{"email": "gone@example.com"}])
recipients = admin_routes_mod._resolve_audience("waitlist")
assert [r["email"] for r in recipients] == ["lead1@example.com"]
entry = {"id": "x", "audience": "waitlist", "test_mode": False,
"recipients_total": len(recipients), "sent_count": 0,
"failed_count": 0, "failed": [], "sent_at": "2026-09-05T00:00:00+00:00"}
asyncio.run(
admin_routes_mod._execute_bulk_send(SUBJECT, HTML, recipients, entry, "admin")
)
assert marketing_env["sent_to"] == ["lead1@example.com"]
entries = json.loads(marketing_env["history_file"].read_text(encoding="utf-8"))
assert entries[-1]["sent_count"] == 1
assert entries[-1]["failed_count"] == 0
assert "completed_at" in entries[-1]
def test_bulk_executor_continues_after_individual_smtp_failure(marketing_env):
"""Un échec SMTP individuel n'interrompt pas l'envoi (jamais d'erreur 500)."""
import routes.admin_routes as admin_routes_mod
async def _flaky_send(to: str, subject: str, body: str, extra_headers=None) -> bool:
if to == "bad@example.com":
raise RuntimeError("SMTP down")
marketing_env["sent_to"].append(to)
return True
marketing_env["set_sender"](_flaky_send)
_write_json(
marketing_env["waitlist_file"],
[{"email": "bad@example.com"}, {"email": "good@example.com"}],
)
recipients = admin_routes_mod._resolve_audience("waitlist")
entry = {"id": "y", "audience": "waitlist", "test_mode": False,
"recipients_total": len(recipients), "sent_count": 0,
"failed_count": 0, "failed": [], "sent_at": "2026-09-05T00:00:00+00:00"}
asyncio.run(
admin_routes_mod._execute_bulk_send(SUBJECT, HTML, recipients, entry, "admin")
)
assert entry["sent_count"] == 1
assert entry["failed_count"] == 1
assert entry["failed"][0]["email"] == "bad@example.com"
assert marketing_env["sent_to"] == ["good@example.com"]
def test_unsubscribed_via_public_link_is_excluded(client, auth_headers, marketing_env):
_write_json(marketing_env["waitlist_file"], [{"email": "bye@example.com"}])
r = client.get(UNSUBSCRIBE_URL, params={"email": "bye@example.com"})
assert r.status_code == 200
unsub = json.loads(marketing_env["unsub_file"].read_text(encoding="utf-8"))
assert [u["email"] for u in unsub] == ["bye@example.com"]
# Audiences vides → l'envoi est refusé proprement (pas d'erreur 500).
r2 = _post_send(client, auth_headers, test_mode=False)
assert r2.status_code == 400
assert r2.json()["error"] in ("AUDIENCE_EMPTY", "TEST_SEND_REQUIRED")
# ---------------------------------------------------------------------------
# Historique + résilience SMTP
# ---------------------------------------------------------------------------
def test_history_endpoint_returns_entries(client, auth_headers, marketing_env):
_write_json(
marketing_env["history_file"],
[
{
"id": "abc",
"sent_at": "2026-09-05T10:00:00+00:00",
"audience": "waitlist",
"subject": SUBJECT,
"test_mode": True,
"recipients_total": 1,
"sent_count": 1,
"failed_count": 0,
"failed": [],
}
],
)
r = client.get(HISTORY_URL, headers=auth_headers)
assert r.status_code == 200
entries = r.json()["data"]
assert len(entries) == 1
assert entries[0]["id"] == "abc"
def test_bulk_failure_does_not_raise_and_is_logged(client, auth_headers, marketing_env):
marketing_env["fail_mode"]()
_write_json(marketing_env["waitlist_file"], [{"email": "lead1@example.com"}])
assert _post_send(client, auth_headers, test_email="admin@example.com").status_code == 502
entry = json.loads(marketing_env["history_file"].read_text(encoding="utf-8"))[0]
assert entry["sent_count"] == 0
assert entry["failed_count"] == 1
marketing_env["sent_to"].clear()
# L'envoi test ayant échoué, l'envoi réel reste bloqué.
r = _post_send(client, auth_headers, test_mode=False)
assert r.status_code == 400
assert r.json()["error"] == "TEST_SEND_REQUIRED"
# ---------------------------------------------------------------------------
# En-têtes de délivrabilité, verrou anti-simultané, re-vérification des désabonnés
# ---------------------------------------------------------------------------
def test_bulk_send_carries_list_unsubscribe_headers(marketing_env):
"""Les emails marketing embarquent List-Unsubscribe / List-Unsubscribe-Post."""
import routes.admin_routes as admin_routes_mod
_write_json(marketing_env["waitlist_file"], [{"email": "lead1@example.com"}])
recipients = admin_routes_mod._resolve_audience("waitlist")
entry = {"id": "h1", "audience": "waitlist", "test_mode": False,
"recipients_total": 1, "sent_count": 0, "failed_count": 0,
"failed": [], "content_hash": "abc", "sent_at": "t"}
asyncio.run(
admin_routes_mod._execute_bulk_send(SUBJECT, HTML, recipients, entry, "admin")
)
headers = marketing_env["captured_headers"]["lead1@example.com"]
assert headers["List-Unsubscribe"].startswith("<http")
assert "unsubscribe?email=" in headers["List-Unsubscribe"]
from urllib.parse import unquote
assert "lead1@example.com" in unquote(headers["List-Unsubscribe"])
assert headers["List-Unsubscribe-Post"] == "List-Unsubscribe=One-Click"
# L'entrée « running » mise à jour (pas dupliquée) et clôturée.
entries = json.loads(marketing_env["history_file"].read_text(encoding="utf-8"))
assert len(entries) == 1
assert entries[0]["status"] == "completed"
def test_mass_send_writes_running_entry_then_completes(client, auth_headers, marketing_env):
_write_json(marketing_env["waitlist_file"], [{"email": "lead1@example.com"}])
assert _post_send(client, auth_headers, test_email="a@ex.com").status_code == 200
marketing_env["sent_to"].clear()
assert _post_send(client, auth_headers, test_mode=False).status_code == 202
entries = json.loads(marketing_env["history_file"].read_text(encoding="utf-8"))
bulk = [e for e in entries if not e["test_mode"]]
assert len(bulk) == 1 # une seule entrée : « running » puis mise à jour
assert bulk[0]["status"] in ("running", "completed")
def test_second_real_send_same_content_refused_while_running(client, auth_headers, marketing_env):
"""Deux campagnes simultanées de même contenu : la seconde est refusée."""
import routes.admin_routes as admin_routes_mod
_write_json(marketing_env["waitlist_file"], [{"email": "lead1@example.com"}])
assert _post_send(client, auth_headers, test_email="a@ex.com").status_code == 200
# On simule une campagne déjà en cours pour ce contenu exact.
content_hash = admin_routes_mod._marketing_content_hash(SUBJECT, HTML)
admin_routes_mod._active_campaign_hashes.add(content_hash)
try:
r = _post_send(client, auth_headers, test_mode=False)
assert r.status_code == 409
assert r.json()["error"] == "CAMPAIGN_ALREADY_RUNNING"
finally:
admin_routes_mod._active_campaign_hashes.discard(content_hash)
def test_unsubscribe_during_bulk_send_is_honored(marketing_env):
"""Un désabonnement déposé pendant la campagne stoppe l'envoi restant."""
import routes.admin_routes as admin_routes_mod
import services.email_service as email_svc
_write_json(
marketing_env["waitlist_file"],
[{"email": "a@example.com"}, {"email": "b@example.com"}],
)
async def _slow_send(to: str, subject: str, body: str, extra_headers=None) -> bool:
if to == "a@example.com":
# Se désabonne pendant que la campagne est en cours.
with admin_routes_mod._marketing_lock:
entries = admin_routes_mod._load_unsubscribes()
entries.append({"email": "b@example.com", "unsubscribed_at": "t"})
admin_routes_mod._save_unsubscribes(entries)
marketing_env["sent_to"].append(to)
return True
marketing_env["set_sender"](_slow_send)
recipients = admin_routes_mod._resolve_audience("waitlist")
entry = {"id": "h2", "audience": "waitlist", "test_mode": False,
"recipients_total": len(recipients), "sent_count": 0,
"failed_count": 0, "failed": [], "content_hash": "xyz", "sent_at": "t"}
asyncio.run(
admin_routes_mod._execute_bulk_send(SUBJECT, HTML, recipients, entry, "admin")
)
assert marketing_env["sent_to"] == ["a@example.com"] # b est passé en désabonné
assert entry["sent_count"] == 1
assert entry["skipped_unsubscribed"] == 1
assert entry["status"] == "completed"
def test_unsubscribe_page_escapes_email(client, marketing_env):
r = client.get(UNSUBSCRIBE_URL, params={"email": "<script>alert(1)</script>@x.com"})
assert r.status_code == 200
body = r.text
assert "<script>" not in body
assert "&lt;script&gt;" in body
def test_put_settings_pristine_ai_tiers_keeps_existing(client, auth_headers, monkeypatch, tmp_path):
"""Un PUT avec ai_tiers vierge ne réinitialise pas les paliers personnalisés."""
import routes.admin_routes as admin_routes_mod
settings_file = tmp_path / "provider_settings.json"
monkeypatch.setattr(admin_routes_mod, "SETTINGS_FILE", str(settings_file))
custom = {
"ai_tiers": {
"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"],
"default_model": "anthropic/claude-sonnet-5",
},
}
}
admin_routes_mod.save_settings(admin_routes_mod.SettingsConfig.model_validate(custom))
# PUT avec un bloc ai_tiers vierge (client sans gestion des paliers).
r = client.put(
"/api/v1/admin/settings",
json={"smtp": {}, "ai_tiers": {"essential": {"models": [], "default_model": ""},
"premium": {"models": [], "default_model": ""}}},
headers=auth_headers,
)
assert r.status_code == 200, r.text
saved = admin_routes_mod.load_settings()
assert saved.ai_tiers.essential.default_model == "z-ai/glm-5.3-flash"
assert saved.ai_tiers.essential.models[0] == "z-ai/glm-5.3-flash"
# Un PUT avec un défaut Essentielle appartenant à la gamme Premium → 400.
r2 = client.put(
"/api/v1/admin/settings",
json={"ai_tiers": {"essential": {"models": [], "default_model": "anthropic/claude-sonnet-5"},
"premium": {"models": [], "default_model": ""}}},
headers=auth_headers,
)
assert r2.status_code == 400
assert r2.json()["error"] == "INVALID_AI_TIER"

243
tests/test_admin_stats_revenue.py Normal file
View File

@@ -0,0 +1,243 @@
"""
Statistiques admin enrichies (spec « Stratégie LLM par abonnement ») :
revenus encaissés, MRR estimé, paliers IA utilisés, compteur liste d'attente.
En mode JSON (sans base de données), les revenus restent à zéro sans erreur.
"""
import json
import pytest
from pathlib import Path
from fastapi.testclient import TestClient
ADMIN_LOGIN_URL = "/api/v1/admin/login"
STATS_URL = "/api/v1/admin/stats"
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")
@pytest.fixture
def stats_env(tmp_path: Path, monkeypatch):
import routes.waitlist_routes as waitlist_mod
waitlist_file = tmp_path / "waitlist.json"
monkeypatch.setattr(waitlist_mod, "WAITLIST_FILE", waitlist_file)
return {"waitlist_file": waitlist_file}
@pytest.fixture
def client(stats_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 main import app
return TestClient(app, raise_server_exceptions=True)
@pytest.fixture
def auth_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
return {"Authorization": f"Bearer {r.json()['access_token']}"}
def test_stats_include_revenue_mrr_waitlist_and_tiers(client, auth_headers, stats_env):
_write_json(stats_env["waitlist_file"], [{"email": "a@example.com"}])
r = client.get(STATS_URL, headers=auth_headers)
assert r.status_code == 200, r.text
body = r.json()
# Nouveaux blocs présents
for key in ("revenue", "mrr_estimated", "ai_tier_usage", "waitlist_count"):
assert key in body, f"clé manquante dans /admin/stats : {key}"
# Sans base de données : revenus à zéro, sans erreur
assert body["revenue"]["collected_total"] == 0.0
assert body["revenue"]["collected_30d"] == 0.0
assert body["revenue"]["credits_purchased"] == 0.0
# Compteur liste d'attente branché sur data/waitlist.json
assert body["waitlist_count"] == 1
# Paliers IA : structure complète
assert set(body["ai_tier_usage"].keys()) == {
"essential",
"premium",
"classic",
"other",
}
# Anciennes clés toujours présentes (non-régression)
assert "users" in body and "translations" in body and "cache" in body
def test_stats_mrr_estimated_from_plan_distribution(client, auth_headers, monkeypatch):
"""Le MRR estimé s'appuie sur la répartition des plans (prix mensuel effectif)."""
import services.auth_service as auth_svc
users_file = auth_svc.USERS_FILE
users_file.parent.mkdir(parents=True, exist_ok=True)
_write_json(
users_file,
{
"u1": {
"email": "p1@example.com",
"name": "P1",
"plan": "pro",
"docs_translated_this_month": 0,
},
"u2": {
"email": "p2@example.com",
"name": "P2",
"plan": "pro",
"docs_translated_this_month": 0,
},
},
)
r = client.get(STATS_URL, headers=auth_headers)
assert r.status_code == 200, r.text
body = r.json()
from services import pricing_config as pricing_cfg
from models.subscription import PlanType
monthly_pro, _ = pricing_cfg.get_effective_monthly_yearly(PlanType.PRO.value)
assert body["users"]["by_plan"].get("pro") == 2
assert body["mrr_estimated"]["by_plan"].get("pro") == pytest.approx(
round(monthly_pro * 2, 2)
)
assert body["mrr_estimated"]["total"] == pytest.approx(
round(monthly_pro * 2, 2)
)
# ---------------------------------------------------------------------------
# Revenus et paliers IA en mode base de données (SQLite de test, cf. conftest)
# ---------------------------------------------------------------------------
@pytest.fixture
def db_client(monkeypatch, tmp_path: Path):
"""TestClient avec authentification admin simple et base de test active."""
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", True)
monkeypatch.setattr(auth_svc, "DATABASE_AVAILABLE", True)
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 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)
from main import app
return TestClient(app, raise_server_exceptions=True)
def test_stats_revenue_and_tiers_from_database(db_client):
"""Paiements réussis + traductions en base → revenus et paliers non nuls."""
from database.connection import get_sync_session
from database.models import User as DBUser, PaymentHistory, Translation
with get_sync_session() as session:
user = DBUser(
email="payer@example.com",
name="Payer",
password_hash="not-a-real-hash",
)
session.add(user)
session.flush()
session.add_all(
[
PaymentHistory(
user_id=user.id,
amount_cents=1900,
currency="eur",
payment_type="subscription",
status="succeeded",
),
PaymentHistory(
user_id=user.id,
amount_cents=900,
currency="eur",
payment_type="credits",
status="succeeded",
),
PaymentHistory(
user_id=user.id,
amount_cents=5000,
currency="eur",
payment_type="subscription",
status="failed", # jamais comptabilisé
),
Translation(
user_id=user.id,
original_filename="a.xlsx",
file_type="xlsx",
target_language="fr",
provider="openrouter",
status="completed",
),
Translation(
user_id=user.id,
original_filename="b.xlsx",
file_type="xlsx",
target_language="fr",
provider="openrouter_premium",
status="completed",
),
]
)
session.commit()
r = db_client.get(STATS_URL, headers=_admin_headers(db_client))
assert r.status_code == 200, r.text
body = r.json()
assert body["revenue"]["currency"] == "EUR"
assert body["revenue"]["collected_total"] >= 28.0 # 19 € + 9 € (l'échec ignoré)
assert body["revenue"]["collected_30d"] >= 28.0
assert body["revenue"]["credits_purchased"] >= 9.0
assert body["revenue"]["payments_30d"] >= 2
assert body["ai_tier_usage"]["essential"] >= 1
assert body["ai_tier_usage"]["premium"] >= 1
def _admin_headers(client) -> dict:
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']}"}

157
tests/test_ai_tier_routing.py Normal file
View File

@@ -0,0 +1,157 @@
"""
Routage réel des modèles par palier IA (spec « Stratégie LLM par abonnement »).
- PLANS expose ai_models_essential / ai_models_premium (gamme officielle).
- Un plan Pro (moteur openrouter) résout toujours un modèle Essentielle,
jamais un modèle Premium.
- Le défaut admin (ai_tiers.<palier>.default_model) prime sur le défaut du
plan, sans redéploiement.
"""
import pytest
from pathlib import Path
from fastapi.testclient import TestClient
from models.subscription import (
DEFAULT_AI_MODELS_ESSENTIAL,
DEFAULT_AI_MODELS_PREMIUM,
PLANS,
PlanType,
)
from routes.translate_routes import (
_allowed_providers_for_plan,
_plan_ai_models,
_resolve_tier_model,
)
@pytest.fixture
def client(monkeypatch, tmp_path: Path):
"""TestClient with JSON auth and rate limiting disabled."""
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 main import app
return TestClient(app, raise_server_exceptions=True)
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})
# ---------------------------------------------------------------------------
# Gamme officielle dans PLANS
# ---------------------------------------------------------------------------
def test_plans_expose_official_model_ranges():
essential = PLANS[PlanType.PRO]["ai_models_essential"]
assert essential == [
"deepseek/deepseek-v4-flash",
"z-ai/glm-5.3-flash",
"minimax/minimax-m3",
]
business = PLANS[PlanType.BUSINESS]
assert business["ai_models_essential"] == essential
assert business["ai_models_premium"] == [
"anthropic/claude-sonnet-5",
"deepseek/deepseek-v4-pro",
"z-ai/glm-5.3",
]
def test_old_single_model_keys_removed():
assert "ai_model_essential" not in PLANS[PlanType.PRO]
assert "ai_model_essential" not in PLANS[PlanType.BUSINESS]
assert "ai_model_premium" not in PLANS[PlanType.BUSINESS]
# ---------------------------------------------------------------------------
# Résolution par palier
# ---------------------------------------------------------------------------
def test_pro_resolves_essential_plan_default():
assert _resolve_tier_model(_settings(), "pro", premium=False) == (
"deepseek/deepseek-v4-flash"
)
def test_business_resolves_premium_plan_default():
assert _resolve_tier_model(_settings(), "business", premium=True) == (
"anthropic/claude-sonnet-5"
)
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"
def test_admin_tier_models_first_used_when_default_empty():
cfg = _settings(
{
"essential": {
"models": ["minimax/minimax-m3", "deepseek/deepseek-v4-flash"],
"default_model": "",
}
}
)
assert _resolve_tier_model(cfg, "pro", premium=False) == "minimax/minimax-m3"
def test_pro_never_resolves_premium_model():
cfg = _settings(
{
"essential": {"models": [], "default_model": ""},
"premium": {"models": [], "default_model": ""},
}
)
for plan_str in ("pro", "PlanType.PRO"):
model = _resolve_tier_model(cfg, plan_str, premium=False)
assert model in DEFAULT_AI_MODELS_ESSENTIAL
assert model not in DEFAULT_AI_MODELS_PREMIUM
def test_pro_plan_does_not_allow_premium_provider():
allowed = _allowed_providers_for_plan(PlanType.PRO)
assert "openrouter" in allowed
assert "openrouter_premium" not in allowed
def test_plan_fallback_for_plans_without_range():
# Free/Enterprise n'ont pas de liste : retombée sur la gamme officielle.
assert _plan_ai_models(PlanType.FREE, premium=False) == DEFAULT_AI_MODELS_ESSENTIAL
assert _plan_ai_models(PlanType.ENTERPRISE, premium=True) == DEFAULT_AI_MODELS_PREMIUM
# ---------------------------------------------------------------------------
# GET /plans expose les gammes (affichage public)
# ---------------------------------------------------------------------------
def test_get_plans_exposes_model_lists(client):
r = client.get("/api/v1/auth/plans")
assert r.status_code == 200
plans = {p["id"]: p for p in r.json()["data"]["plans"]}
assert plans["pro"]["ai_models_essential"] == [
"deepseek/deepseek-v4-flash",
"z-ai/glm-5.3-flash",
"minimax/minimax-m3",
]
assert plans["business"]["ai_models_premium"][0] == "anthropic/claude-sonnet-5"
assert "ai_model_essential" not in plans["pro"]

152
tests/test_worker_tier_routing.py Normal file
View File

@@ -0,0 +1,152 @@
"""
Intégration : ``_run_translation_job`` route bien le modèle par palier.
Un faux fournisseur OpenAI-compatible capture le paramètre ``model`` reçu du
worker ; ``load_settings`` (réglages admin) est substitué. On vérifie que :
- provider="openrouter" → modèle Essentielle (réglages > plan) ;
- provider="openrouter_premium" → modèle Premium ;
- un ``ai_tiers.essential`` personnalisé est honoré sans redéploiement ;
- un ``ai_tiers.essential`` contenant un modèle Premium est corrigé à l'exécution.
"""
import asyncio
import pytest
from pathlib import Path
import routes.translate_routes as tr
@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 _settings(ai_tiers: dict, api_key: str = "sk-or-test"):
from routes.admin_routes import SettingsConfig
return SettingsConfig.model_validate(
{
"openrouter": {"api_key": api_key},
"openrouter_premium": {"api_key": api_key},
"ai_tiers": ai_tiers,
}
)
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) -> list:
"""Lance _run_translation_job avec des faux traducteurs/fournisseurs et
renvoie la liste des modèles capturés par le faux fournisseur."""
import routes.admin_routes as admin_routes_mod
import services.providers.openai_provider as openai_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
captured.append(model)
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}
monkeypatch.setattr(openai_provider_mod, "OpenAITranslationProvider", FakeProvider)
monkeypatch.setattr(tr, "ExcelTranslator", FakeExcelTranslator)
monkeypatch.setattr(
admin_routes_mod, "load_settings", lambda: admin_settings, raising=True
)
_register_job("job-test")
try:
asyncio.run(
tr._run_translation_job(
job_id="job-test",
input_path=Path(minimal_xlsx),
file_extension=".xlsx",
target_lang="fr",
source_lang="en",
provider=provider,
user_id=None,
custom_prompt=None,
glossary_id=None,
prompt_id=None,
webhook_url=None,
user_plan=user_plan,
)
)
finally:
tr._translation_jobs.pop("job-test", None)
return captured
def test_openrouter_pro_uses_essential_tier_model(monkeypatch, tmp_path, minimal_xlsx):
settings = _settings({"essential": {"models": [], "default_model": ""},
"premium": {"models": [], "default_model": ""}})
captured = _run_job(monkeypatch, tmp_path, minimal_xlsx, settings,
provider="openrouter", user_plan="PlanType.PRO")
assert captured, "le faux fournisseur n'a pas été instancié"
assert captured[0] == "deepseek/deepseek-v4-flash"
assert captured[0] not in (
"anthropic/claude-sonnet-5",
"deepseek/deepseek-v4-pro",
"z-ai/glm-5.3",
)
def test_openrouter_premium_uses_premium_tier_model(monkeypatch, tmp_path, minimal_xlsx):
settings = _settings({"essential": {"models": [], "default_model": ""},
"premium": {"models": [], "default_model": ""}})
captured = _run_job(monkeypatch, tmp_path, minimal_xlsx, settings,
provider="openrouter_premium", user_plan="PlanType.BUSINESS")
assert captured, "le faux fournisseur n'a pas été instancié"
assert captured[0] == "anthropic/claude-sonnet-5"
def test_admin_tier_default_applied_without_redeployment(monkeypatch, tmp_path, minimal_xlsx):
settings = _settings({"essential": {"models": [], "default_model": "z-ai/glm-5.3-flash"},
"premium": {"models": [], "default_model": ""}})
captured = _run_job(monkeypatch, tmp_path, minimal_xlsx, settings,
provider="openrouter", user_plan="PlanType.PRO")
assert captured[0] == "z-ai/glm-5.3-flash"
def test_essential_tier_resolving_premium_model_is_corrected(monkeypatch, tmp_path, minimal_xlsx):
"""Garde d'exécution : un ai_tiers essential contaminé par un modèle
Premium ne doit pas partir en production — repli sur le défaut du plan."""
settings = _settings({"essential": {"models": [], "default_model": "anthropic/claude-sonnet-5"},
"premium": {"models": [], "default_model": ""}})
captured = _run_job(monkeypatch, tmp_path, minimal_xlsx, settings,
provider="openrouter", user_plan="PlanType.PRO")
assert captured[0] == "deepseek/deepseek-v4-flash"