Compare commits
6
Commits
8bf83d0a7f
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3538d77217 | ||
|
|
b7551073f5 | ||
|
|
e8b4a604e9 | ||
|
|
caa0b126b1 | ||
|
|
75f9727710 | ||
|
|
138bc3c87c |
@@ -150,6 +150,9 @@ activemq-data/
|
||||
# Environments
|
||||
.env
|
||||
.envrc
|
||||
# Configuration de déploiement : contient la chaîne de connexion et
|
||||
# la clé JWT. Seul api.env.example (sans valeurs) est versionné.
|
||||
api.env
|
||||
.venv
|
||||
env/
|
||||
venv/
|
||||
|
||||
@@ -129,9 +129,19 @@ la même information sur `/version.json`.
|
||||
```bash
|
||||
pip install pytest httpx && pytest -q
|
||||
```
|
||||
Curseur SQL simulé (aucune vraie BDD) : santé, auth (succès/échec), protection 401,
|
||||
RBAC 403, 404, validation des rôles par enum (422), refus de suppression d'un
|
||||
référentiel rattaché (409), désactivation logique d'un monitoring.
|
||||
|
||||
**82 tests**, curseur SQL simulé (aucune vraie base requise) :
|
||||
|
||||
| Fichier | Domaine testé |
|
||||
|---------|---------------|
|
||||
| `test_endpoints.py` | santé, référentiels, authentification, filtres du dashboard, rôles par enum |
|
||||
| `test_historique.py` | bornes de dates, validation du format, agrégation des courbes par monitoring et par service |
|
||||
| `test_securite.py` | protection 401 de chaque route, RBAC 403, jeton forgé rejeté, en-têtes de sécurité, rate-limit 429, compte désactivé, droits RGPD |
|
||||
| `test_referentiels.py` | CRUD complet, refus 409 sur rattachement, modification partielle, paramétrage des requêtes de recherche |
|
||||
|
||||
Le compteur anti brute-force est remis à zéro entre les tests (fixture
|
||||
`limiteur_vierge`) : c'est un état global, et sans cela les tests de connexion
|
||||
se comptabilisent entre eux et échouent selon l'ordre d'exécution.
|
||||
|
||||
## Docker & CI/CD
|
||||
|
||||
|
||||
+1
-1
@@ -77,4 +77,4 @@ docker compose up -d datasentinel-api
|
||||
## Contacts
|
||||
|
||||
- Hébergement / infra : administrateur homelab (neckfire).
|
||||
- Application / code : A. Coyaud (auteur, dépôts GitHub).
|
||||
- Application / code : A. Coyaud (auteur, dépôts Gitea — git.nfteam.ovh).
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
# ============================================================
|
||||
# Data Sentinel — modèle de configuration de l'API
|
||||
#
|
||||
# Copier ce fichier en api.env (à côté du docker-compose.yml)
|
||||
# et renseigner les valeurs. api.env n'est JAMAIS versionné :
|
||||
# il est exclu par .gitignore.
|
||||
#
|
||||
# cp Api-DataSentinel/api.env.example api.env
|
||||
# ============================================================
|
||||
|
||||
# ---- Base de données ---------------------------------------
|
||||
# Hôte SQL Server. En conteneur, le nom du service Docker
|
||||
# (ex. dev-mssql) ; en local, localhost ou HOTE\INSTANCE.
|
||||
DB_SERVER=dev-mssql
|
||||
|
||||
# Port. Laisser VIDE pour une instance nommée (HOTE\INSTANCE).
|
||||
DB_PORT=1433
|
||||
|
||||
DB_NAME=DataSentinel
|
||||
|
||||
# Compte applicatif dédié — surtout pas « sa ».
|
||||
# Création : voir README.md §3.3 (db_datareader + db_datawriter
|
||||
# sur la seule base DataSentinel).
|
||||
DB_USER=datasentinel_app
|
||||
DB_PASSWORD=
|
||||
|
||||
# Pilote ODBC embarqué dans l'image.
|
||||
DB_DRIVER=ODBC Driver 18 for SQL Server
|
||||
|
||||
# Mettre à yes pour forcer l'authentification Windows
|
||||
# (poste de développement uniquement, sans DB_USER).
|
||||
DB_TRUSTED_CONNECTION=
|
||||
|
||||
# ---- Sécurité ----------------------------------------------
|
||||
# OBLIGATOIRE en déploiement. Sans cette variable, l'API démarre
|
||||
# mais tire une clé aléatoire à chaque redémarrage : toutes les
|
||||
# sessions sont alors invalidées au moindre restart.
|
||||
# Générer : python -c "import secrets; print(secrets.token_urlsafe(64))"
|
||||
JWT_SECRET=
|
||||
|
||||
JWT_ALGORITHM=HS256
|
||||
JWT_EXPIRE_MINUTES=60
|
||||
|
||||
# ---- CORS ---------------------------------------------------
|
||||
# Origines autorisées, séparées par des virgules. Doit contenir
|
||||
# l'URL exacte du front (schéma + hôte, sans slash final).
|
||||
CORS_ORIGINS=https://datasentinel.nfteam.ovh
|
||||
|
||||
# ---- Build --------------------------------------------------
|
||||
# Injecté par la CI (SHA court du commit) et exposé par GET /health.
|
||||
APP_BUILD=local
|
||||
@@ -81,7 +81,7 @@ class Config:
|
||||
|
||||
# Paramètres API
|
||||
API_TITLE = "Data Sentinel API"
|
||||
API_VERSION = "1.0.0"
|
||||
API_VERSION = "1.1.0"
|
||||
API_DESCRIPTION = "API de monitoring de la qualité des données — XEFI"
|
||||
|
||||
# Identifiant de build injecté par la CI : permet de vérifier quelle
|
||||
|
||||
+1
-1
File diff suppressed because one or more lines are too long
-47
@@ -1,47 +0,0 @@
|
||||
import requests
|
||||
import json
|
||||
|
||||
# Configuration
|
||||
BASE_URL = "http://127.0.0.1:8000"
|
||||
|
||||
def test_get_dashboard():
|
||||
print(f"--- Test de l'endpoint: {BASE_URL}/dashboard ---")
|
||||
|
||||
try:
|
||||
# 1. Envoi de la requête
|
||||
response = requests.get(f"{BASE_URL}/dashboard")
|
||||
|
||||
# 2. Vérification du code statut
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
|
||||
if not data:
|
||||
print("L'API a répondu avec succès, mais le tableau est vide (VUE_CONSO ne contient rien).")
|
||||
return
|
||||
|
||||
# 3. Affichage formaté du tableau
|
||||
print(f"{'ID':<5} | {'NOM DU MONITORING':<40} | {'ERREURS':<8} | {'SERVICE':<15}")
|
||||
print("-" * 75)
|
||||
|
||||
for item in data:
|
||||
print(f"{item['id_monito']:<5} | {item['nom_monito'][:38]:<40} | {item['nb_erreurs']:<8} | {item['service']:<15}")
|
||||
|
||||
print(f"\nTotal de lignes récupérées : {len(data)}")
|
||||
|
||||
else:
|
||||
print(f"Erreur lors de la requête : {response.status_code}")
|
||||
print(response.text)
|
||||
|
||||
except requests.exceptions.ConnectionError:
|
||||
print("Erreur : Impossible de se connecter à l'API. Est-ce qu'uvicorn est lancé ?")
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Petit check de santé avant de tester les données
|
||||
print("Vérification de l'état de l'API...")
|
||||
health = requests.get(f"{BASE_URL}/health").json()
|
||||
print(f"Statut API : {health['api']} | Statut DB : {health['database']}\n")
|
||||
|
||||
if health['database'] == "ok":
|
||||
test_get_dashboard()
|
||||
else:
|
||||
print("Abandon du test : La base de données n'est pas accessible.")
|
||||
@@ -9,6 +9,7 @@ from fastapi.testclient import TestClient
|
||||
|
||||
import main
|
||||
from auth import create_access_token
|
||||
from rate_limit import limiter
|
||||
from routers import (
|
||||
admin,
|
||||
authentification,
|
||||
@@ -62,6 +63,21 @@ def cur(monkeypatch):
|
||||
return c
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def limiteur_vierge():
|
||||
"""
|
||||
Remet le compteur anti brute-force à zéro entre les tests.
|
||||
|
||||
Le limiteur est un état global partagé par toute l'application : sans
|
||||
remise à zéro, les tests de connexion se comptabilisent entre eux et
|
||||
finissent par recevoir un 429, avec des échecs qui dépendent de l'ordre
|
||||
d'exécution et ne se reproduisent pas fichier par fichier.
|
||||
"""
|
||||
limiter.reset()
|
||||
yield
|
||||
limiter.reset()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client():
|
||||
return TestClient(main.app)
|
||||
|
||||
@@ -0,0 +1,111 @@
|
||||
"""Historique et courbes d'évolution (curseur SQL simulé via la fixture `cur`)."""
|
||||
|
||||
|
||||
def _requetes(cur) -> str:
|
||||
"""Concatène les requêtes exécutées, pour vérifier le SQL construit."""
|
||||
return " ".join(q for q, _ in cur.executed)
|
||||
|
||||
|
||||
def _parametres(cur) -> list:
|
||||
return [p for _, args in cur.executed for p in args]
|
||||
|
||||
|
||||
def test_historique_exige_authentification(client):
|
||||
assert client.get("/historique").status_code == 401
|
||||
|
||||
|
||||
def test_historique_sans_filtre_ne_restreint_rien(client, cur, auth_headers):
|
||||
cur.description = [("id_monito",), ("nb_erreurs",)]
|
||||
cur._rows = [(1, 5)]
|
||||
r = client.get("/historique", headers=auth_headers)
|
||||
assert r.status_code == 200
|
||||
requete = _requetes(cur)
|
||||
assert "date_sauvegarde >=" not in requete
|
||||
assert "date_sauvegarde <=" not in requete
|
||||
|
||||
|
||||
def test_historique_applique_les_bornes_de_dates(client, cur, auth_headers):
|
||||
cur.description = [("id_monito",)]
|
||||
cur._rows = []
|
||||
r = client.get(
|
||||
"/historique",
|
||||
params={"date_debut": "2026-01-01", "date_fin": "2026-08-15"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert r.status_code == 200
|
||||
requete = _requetes(cur)
|
||||
assert "date_sauvegarde >= ?" in requete
|
||||
assert "date_sauvegarde <= ?" in requete
|
||||
# Les dates sont passées en paramètres, jamais concaténées dans le SQL.
|
||||
assert "2026-01-01" in _parametres(cur)
|
||||
assert "2026-08-15" in _parametres(cur)
|
||||
|
||||
|
||||
def test_historique_refuse_une_date_mal_formee(client, cur, auth_headers):
|
||||
r = client.get("/historique", params={"date_debut": "15/01/2026"}, headers=auth_headers)
|
||||
assert r.status_code == 422
|
||||
|
||||
|
||||
def test_historique_filtre_par_monitoring(client, cur, auth_headers):
|
||||
cur.description = [("id_monito",)]
|
||||
cur._rows = []
|
||||
client.get("/historique", params={"id_monito": 3}, headers=auth_headers)
|
||||
assert "id_monito = ?" in _requetes(cur)
|
||||
assert 3 in _parametres(cur)
|
||||
|
||||
|
||||
def test_evolution_monitoring_inconnu_renvoie_404(client, cur, auth_headers):
|
||||
# 999 n'est pas dans MONITO_TABLES : on refuse avant d'interroger la base.
|
||||
r = client.get("/historique/999/evolution", headers=auth_headers)
|
||||
assert r.status_code == 404
|
||||
|
||||
|
||||
def test_evolution_formate_les_points_pour_le_graphique(client, cur, auth_headers):
|
||||
cur._rows = [("2026-01-01", 12), ("2026-01-02", 9)]
|
||||
r = client.get("/historique/1/evolution", headers=auth_headers)
|
||||
assert r.status_code == 200
|
||||
corps = r.json()
|
||||
assert corps["points"] == 2
|
||||
assert corps["evolution"][0] == {"date": "2026-01-01", "nb_erreurs": 12}
|
||||
|
||||
|
||||
def test_evolution_globale_renvoie_une_serie(client, cur, auth_headers):
|
||||
cur._rows = [("2026-01-01", 56), ("2026-01-02", 54)]
|
||||
r = client.get("/evolution/global", headers=auth_headers)
|
||||
assert r.status_code == 200
|
||||
corps = r.json()
|
||||
assert corps["type"] == "GLOBAL"
|
||||
assert corps["points"] == 2
|
||||
assert corps["series"][0]["nb_erreurs"] == 56
|
||||
|
||||
|
||||
def test_evolution_par_monitoring_regroupe_par_courbe(client, cur, auth_headers):
|
||||
# Lignes à plat en base -> une série par monitoring côté API.
|
||||
cur._rows = [
|
||||
("2026-01-01", 1, "Tiers payeurs", 5),
|
||||
("2026-01-02", 1, "Tiers payeurs", 4),
|
||||
("2026-01-01", 2, "Sans RIB", 6),
|
||||
]
|
||||
r = client.get("/evolution/par-monitoring", headers=auth_headers)
|
||||
assert r.status_code == 200
|
||||
corps = r.json()
|
||||
assert corps["nb_series"] == 2
|
||||
series = {m["id_monito"]: m for m in corps["monitorings"]}
|
||||
assert len(series[1]["serie"]) == 2
|
||||
assert len(series[2]["serie"]) == 1
|
||||
assert series[1]["nom_monito"] == "Tiers payeurs"
|
||||
|
||||
|
||||
def test_evolution_par_service_regroupe_par_service(client, cur, auth_headers):
|
||||
cur._rows = [
|
||||
("2026-01-01", "Contrat", 20),
|
||||
("2026-01-02", "Contrat", 18),
|
||||
("2026-01-01", "Fournisseur", 5),
|
||||
]
|
||||
r = client.get("/evolution/par-service", headers=auth_headers)
|
||||
assert r.status_code == 200
|
||||
corps = r.json()
|
||||
assert corps["nb_series"] == 2
|
||||
par_service = {s["service"]: s["serie"] for s in corps["services"]}
|
||||
assert len(par_service["Contrat"]) == 2
|
||||
assert len(par_service["Fournisseur"]) == 1
|
||||
@@ -0,0 +1,143 @@
|
||||
"""CRUD des référentiels : catégories, contacts et nomenclature."""
|
||||
|
||||
|
||||
def _requetes(cur) -> str:
|
||||
return " ".join(q for q, _ in cur.executed)
|
||||
|
||||
|
||||
def _parametres(cur) -> list:
|
||||
return [p for _, args in cur.executed for p in args]
|
||||
|
||||
|
||||
# ------------------------------------------------------------
|
||||
# Catégories
|
||||
# ------------------------------------------------------------
|
||||
|
||||
def test_creation_categorie(client, cur, auth_headers):
|
||||
cur._one = (7,)
|
||||
r = client.post("/categories", headers=auth_headers,
|
||||
json={"intitule_categorie": "Litiges"})
|
||||
assert r.status_code == 201
|
||||
assert r.json() == {"id_categorie": 7, "intitule_categorie": "Litiges"}
|
||||
|
||||
|
||||
def test_categorie_intitule_vide_refuse(client, cur, auth_headers):
|
||||
# Contrainte min_length : on ne crée pas un référentiel sans libellé.
|
||||
r = client.post("/categories", headers=auth_headers, json={"intitule_categorie": ""})
|
||||
assert r.status_code == 422
|
||||
|
||||
|
||||
def test_suppression_categorie_refusee_si_monitorings_rattaches(client, cur, auth_headers):
|
||||
cur._one = (3,)
|
||||
r = client.delete("/categories/1", headers=auth_headers)
|
||||
assert r.status_code == 409
|
||||
assert "3 monitoring(s)" in r.json()["detail"]
|
||||
# Aucune suppression ne doit avoir été tentée.
|
||||
assert "DELETE FROM CATEGORIE" not in _requetes(cur)
|
||||
|
||||
|
||||
def test_suppression_categorie_libre_autorisee(client, cur, auth_headers):
|
||||
cur._one = (0,)
|
||||
r = client.delete("/categories/1", headers=auth_headers)
|
||||
assert r.status_code == 200
|
||||
assert "DELETE FROM CATEGORIE" in _requetes(cur)
|
||||
|
||||
|
||||
def test_modification_categorie_inexistante_renvoie_404(client, cur, auth_headers):
|
||||
cur.rowcount = 0
|
||||
r = client.put("/categories/999", headers=auth_headers,
|
||||
json={"intitule_categorie": "X"})
|
||||
assert r.status_code == 404
|
||||
|
||||
|
||||
# ------------------------------------------------------------
|
||||
# Contacts
|
||||
# ------------------------------------------------------------
|
||||
|
||||
def test_creation_contact(client, cur, auth_headers):
|
||||
cur._one = (12,)
|
||||
r = client.post("/contacts", headers=auth_headers, json={
|
||||
"id_service": 1, "intitule_contact": "Référent",
|
||||
"nom": "Dupont", "prenom": "Marie", "mail": "marie@xefi-fictif.fr",
|
||||
})
|
||||
assert r.status_code == 201
|
||||
assert r.json()["id_contact"] == 12
|
||||
assert r.json()["nom"] == "Dupont"
|
||||
|
||||
|
||||
def test_contact_champ_manquant_refuse(client, cur, auth_headers):
|
||||
r = client.post("/contacts", headers=auth_headers, json={"nom": "Dupont"})
|
||||
assert r.status_code == 422
|
||||
|
||||
|
||||
def test_suppression_contact_directe(client, cur, auth_headers):
|
||||
# Aucune table ne référence un contact : suppression sans garde-fou.
|
||||
r = client.delete("/contacts/1", headers=auth_headers)
|
||||
assert r.status_code == 200
|
||||
assert "DELETE FROM CONTACT" in _requetes(cur)
|
||||
|
||||
|
||||
def test_suppression_contact_inexistant_renvoie_404(client, cur, auth_headers):
|
||||
cur.rowcount = 0
|
||||
assert client.delete("/contacts/999", headers=auth_headers).status_code == 404
|
||||
|
||||
|
||||
# ------------------------------------------------------------
|
||||
# Nomenclature des monitorings
|
||||
# ------------------------------------------------------------
|
||||
|
||||
def test_creation_monitoring_actif_par_defaut(client, cur, auth_headers):
|
||||
cur._one = (8,)
|
||||
r = client.post("/monitorings", headers=auth_headers, json={
|
||||
"monito_intitule": "Nouveau", "id_service": 1, "id_categorie": 1,
|
||||
"table_source": "MONITO_TEST", "bdd_source": "Sage",
|
||||
})
|
||||
assert r.status_code == 201
|
||||
assert "actif" in _requetes(cur)
|
||||
|
||||
|
||||
def test_modification_partielle_ne_touche_que_les_champs_fournis(client, cur, auth_headers):
|
||||
r = client.put("/monitorings/1", headers=auth_headers,
|
||||
json={"monito_intitule": "Renommé"})
|
||||
assert r.status_code == 200
|
||||
requete = _requetes(cur)
|
||||
assert "monito_intitule = ?" in requete
|
||||
assert "bdd_source = ?" not in requete
|
||||
|
||||
|
||||
def test_modification_sans_aucun_champ_refusee(client, cur, auth_headers):
|
||||
r = client.put("/monitorings/1", headers=auth_headers, json={})
|
||||
assert r.status_code == 400
|
||||
|
||||
|
||||
def test_reactivation_possible_par_modification(client, cur, auth_headers):
|
||||
# Permet de revenir sur une désactivation.
|
||||
r = client.put("/monitorings/7", headers=auth_headers, json={"actif": True})
|
||||
assert r.status_code == 200
|
||||
assert "actif = ?" in _requetes(cur)
|
||||
assert 1 in _parametres(cur)
|
||||
|
||||
|
||||
def test_desactivation_monitoring_inexistant_renvoie_404(client, cur, auth_headers):
|
||||
cur.rowcount = 0
|
||||
assert client.delete("/monitorings/999", headers=auth_headers).status_code == 404
|
||||
|
||||
|
||||
def test_details_monitoring_recherche_sur_les_colonnes_declarees(client, cur, auth_headers):
|
||||
cur.description = [("id_ligne",)]
|
||||
cur._rows = []
|
||||
r = client.get("/monitorings/1/details", params={"search": "Lyon"}, headers=auth_headers)
|
||||
assert r.status_code == 200
|
||||
requete = _requetes(cur)
|
||||
assert "LIKE ?" in requete
|
||||
# Le terme est passé en paramètre, jamais injecté dans la requête.
|
||||
assert "%Lyon%" in _parametres(cur)
|
||||
assert "Lyon" not in requete
|
||||
|
||||
|
||||
def test_details_monitoring_limite_bornee(client, cur, auth_headers):
|
||||
# limit est plafonné à 5000 pour éviter un export non maîtrisé.
|
||||
assert client.get("/monitorings/1/details", params={"limit": 99999},
|
||||
headers=auth_headers).status_code == 422
|
||||
assert client.get("/monitorings/1/details", params={"limit": 0},
|
||||
headers=auth_headers).status_code == 422
|
||||
@@ -0,0 +1,159 @@
|
||||
"""Contrôles d'accès, en-têtes de sécurité et droits RGPD."""
|
||||
import pytest
|
||||
|
||||
from auth import create_access_token, hash_password
|
||||
from domain import UserRole
|
||||
|
||||
|
||||
def _token(role: str, username: str = "u", uid: int = 1) -> dict:
|
||||
return {"Authorization": f"Bearer {create_access_token(
|
||||
{'sub': username, 'role': role, 'uid': uid})}"}
|
||||
|
||||
|
||||
# ------------------------------------------------------------
|
||||
# Protection des routes de données
|
||||
# ------------------------------------------------------------
|
||||
|
||||
@pytest.mark.parametrize("route", [
|
||||
"/categories", "/services", "/contacts", "/monitorings",
|
||||
"/dashboard", "/dashboard/filtres", "/dashboard/summary",
|
||||
"/historique", "/evolution/global", "/admin/users", "/me/data-export",
|
||||
])
|
||||
def test_toutes_les_routes_de_donnees_exigent_un_jeton(client, route):
|
||||
assert client.get(route).status_code == 401
|
||||
|
||||
|
||||
def test_health_reste_public(client, cur):
|
||||
cur._one = (1,)
|
||||
assert client.get("/health").status_code == 200
|
||||
|
||||
|
||||
def test_jeton_invalide_rejete(client):
|
||||
r = client.get("/services", headers={"Authorization": "Bearer pas-un-vrai-jeton"})
|
||||
assert r.status_code == 401
|
||||
|
||||
|
||||
def test_jeton_signe_avec_une_autre_cle_rejete(client):
|
||||
# Un jeton forgé avec un autre secret ne doit pas ouvrir l'accès :
|
||||
# c'est ce que garantissait mal un secret de repli écrit dans le dépôt.
|
||||
from jose import jwt
|
||||
faux = jwt.encode({"sub": "pirate", "role": "Admin", "uid": 99},
|
||||
"mauvaise-cle", algorithm="HS256")
|
||||
r = client.get("/admin/users", headers={"Authorization": f"Bearer {faux}"})
|
||||
assert r.status_code == 401
|
||||
|
||||
|
||||
# ------------------------------------------------------------
|
||||
# RBAC : seuls les administrateurs écrivent
|
||||
# ------------------------------------------------------------
|
||||
|
||||
@pytest.mark.parametrize("role", [UserRole.SUPERVISEUR.value, UserRole.CONSULTANT.value])
|
||||
@pytest.mark.parametrize("methode,route,corps", [
|
||||
("post", "/services", {"nom_service": "X"}),
|
||||
("put", "/services/1", {"nom_service": "X"}),
|
||||
("delete", "/services/1", None),
|
||||
("post", "/categories", {"intitule_categorie": "X"}),
|
||||
("post", "/monitorings", {"monito_intitule": "X", "id_service": 1,
|
||||
"id_categorie": 1, "table_source": "T", "bdd_source": "Sage"}),
|
||||
("get", "/admin/users", None),
|
||||
("get", "/admin/journal", None),
|
||||
])
|
||||
def test_ecriture_interdite_hors_admin(client, cur, role, methode, route, corps):
|
||||
appel = getattr(client, methode)
|
||||
reponse = appel(route, headers=_token(role), json=corps) if corps else appel(route, headers=_token(role))
|
||||
assert reponse.status_code == 403
|
||||
|
||||
|
||||
def test_lecture_autorisee_pour_un_consultant(client, cur):
|
||||
cur.description = [("id_service",), ("nom_service",)]
|
||||
cur._rows = [(1, "Contrat")]
|
||||
r = client.get("/services", headers=_token(UserRole.CONSULTANT.value))
|
||||
assert r.status_code == 200
|
||||
|
||||
|
||||
# ------------------------------------------------------------
|
||||
# En-têtes de sécurité
|
||||
# ------------------------------------------------------------
|
||||
|
||||
@pytest.mark.parametrize("entete,valeur", [
|
||||
("X-Content-Type-Options", "nosniff"),
|
||||
("X-Frame-Options", "DENY"),
|
||||
("Referrer-Policy", "strict-origin-when-cross-origin"),
|
||||
])
|
||||
def test_entetes_de_securite_presents(client, cur, entete, valeur):
|
||||
cur._one = (1,)
|
||||
assert client.get("/health").headers[entete] == valeur
|
||||
|
||||
|
||||
def test_strict_transport_security_present(client, cur):
|
||||
cur._one = (1,)
|
||||
assert "max-age=" in client.get("/health").headers["Strict-Transport-Security"]
|
||||
|
||||
|
||||
# ------------------------------------------------------------
|
||||
# Authentification
|
||||
# ------------------------------------------------------------
|
||||
|
||||
def test_compte_desactive_ne_peut_pas_se_connecter(client, cur):
|
||||
# actif = 0 : le mot de passe est correct mais l'accès doit être refusé.
|
||||
cur._one = (1, "ancien", hash_password("Bon2026!"), "Admin", 0)
|
||||
r = client.post("/auth/login", data={"username": "ancien", "password": "Bon2026!"})
|
||||
assert r.status_code == 401
|
||||
|
||||
|
||||
def test_utilisateur_inconnu_renvoie_401(client, cur):
|
||||
cur._one = None
|
||||
r = client.post("/auth/login", data={"username": "fantome", "password": "x"})
|
||||
assert r.status_code == 401
|
||||
|
||||
|
||||
def test_le_mot_de_passe_nest_jamais_renvoye(client, cur):
|
||||
cur._one = (1, "admin", hash_password("Admin2026!"), "Admin", 1)
|
||||
r = client.post("/auth/login", data={"username": "admin", "password": "Admin2026!"})
|
||||
assert r.status_code == 200
|
||||
assert "password" not in r.text
|
||||
assert "hash" not in r.text
|
||||
|
||||
|
||||
def test_rate_limit_bloque_le_brute_force(client, cur):
|
||||
"""La 6e tentative en moins d'une minute doit être refusée (5/minute)."""
|
||||
cur._one = (1, "admin", hash_password("Admin2026!"), "Admin", 1)
|
||||
codes = [
|
||||
client.post("/auth/login", data={"username": "admin", "password": "faux"}).status_code
|
||||
for _ in range(6)
|
||||
]
|
||||
assert codes[:5] == [401] * 5
|
||||
assert codes[5] == 429
|
||||
|
||||
|
||||
def test_connexion_journalisee(client, cur):
|
||||
cur._one = (1, "admin", hash_password("Admin2026!"), "Admin", 1)
|
||||
client.post("/auth/login", data={"username": "admin", "password": "Admin2026!"})
|
||||
requetes = " ".join(q for q, _ in cur.executed)
|
||||
assert "INSERT INTO JOURNAL_AUDIT" in requetes
|
||||
# L'action passe en paramètre, pas concaténée dans la requête.
|
||||
assert "LOGIN" in [p for _, args in cur.executed for p in args]
|
||||
|
||||
|
||||
# ------------------------------------------------------------
|
||||
# RGPD
|
||||
# ------------------------------------------------------------
|
||||
|
||||
def test_export_de_mes_donnees_retourne_profil_et_journal(client, cur, auth_headers):
|
||||
cur.description = [("id_user",), ("username",)]
|
||||
cur._one = (1, "admin")
|
||||
cur._rows = [(1, "admin")]
|
||||
r = client.get("/me/data-export", headers=auth_headers)
|
||||
assert r.status_code == 200
|
||||
assert set(r.json()) == {"profile", "journal"}
|
||||
|
||||
|
||||
def test_suppression_de_compte_anonymise_au_lieu_de_supprimer(client, cur, auth_headers):
|
||||
# Droit à l'oubli : le compte est désactivé et anonymisé, la ligne reste
|
||||
# pour préserver l'intégrité du journal d'audit.
|
||||
r = client.delete("/me", headers=auth_headers)
|
||||
assert r.status_code == 200
|
||||
requetes = " ".join(q for q, _ in cur.executed)
|
||||
assert "UPDATE [USER]" in requetes
|
||||
assert "DELETE FROM [USER]" not in requetes
|
||||
assert "JOURNAL_AUDIT SET username = NULL" in requetes
|
||||
Reference in New Issue
Block a user