test(api): étend la couverture de 16 à 82 tests
Trois domaines n'étaient pas couverts : l'historique et les courbes d'évolution, les contrôles de sécurité, et le CRUD des référentiels. - test_historique.py : bornes de dates passées en paramètres, rejet d'un format de date invalide, agrégation des séries par monitoring et par service à partir de lignes à plat. - test_securite.py : protection 401 vérifiée route par route, RBAC 403 pour Superviseur et Consultant sur chaque écriture, rejet d'un jeton signé avec une autre clé, en-têtes de sécurité, rate-limit 429 à la 6e tentative, refus d'un compte désactivé, absence du hachage dans la réponse de login, anonymisation RGPD sans suppression de ligne. - test_referentiels.py : CRUD catégories et contacts, refus 409 sur rattachement, 404 sur enregistrement inexistant, modification partielle limitée aux champs fournis, réactivation d'un monitoring, plafonnement de limit, terme de recherche transmis en paramètre et non concaténé. Corrige au passage un défaut d'isolation révélé par la suite complète : le limiteur de débit est un état global, les tests de connexion se comptabilisaient entre eux et un test échouait selon l'ordre d'exécution tout en passant fichier par fichier. La fixture limiteur_vierge le remet à zéro, et un test dédié vérifie désormais explicitement le seuil. Suite vérifiée stable sur 3 exécutions consécutives et fichier par fichier.
This commit is contained in:
@@ -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