From 138bc3c87ccd345ef0975781e784792807cc9fcb Mon Sep 17 00:00:00 2001 From: Anthony Coyaud Date: Sat, 15 Aug 2026 15:01:42 +0200 Subject: [PATCH] =?UTF-8?q?test(api):=20=C3=A9tend=20la=20couverture=20de?= =?UTF-8?q?=2016=20=C3=A0=2082=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- README.md | 16 +++- tests/conftest.py | 16 ++++ tests/test_historique.py | 111 ++++++++++++++++++++++++++ tests/test_referentiels.py | 143 +++++++++++++++++++++++++++++++++ tests/test_securite.py | 159 +++++++++++++++++++++++++++++++++++++ 5 files changed, 442 insertions(+), 3 deletions(-) create mode 100644 tests/test_historique.py create mode 100644 tests/test_referentiels.py create mode 100644 tests/test_securite.py diff --git a/README.md b/README.md index 9f191b1..9f824d4 100644 --- a/README.md +++ b/README.md @@ -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 | Couverture | +|---------|------------| +| `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 diff --git a/tests/conftest.py b/tests/conftest.py index 1e34eb5..c70c678 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -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) diff --git a/tests/test_historique.py b/tests/test_historique.py new file mode 100644 index 0000000..fe97441 --- /dev/null +++ b/tests/test_historique.py @@ -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 diff --git a/tests/test_referentiels.py b/tests/test_referentiels.py new file mode 100644 index 0000000..6e6556c --- /dev/null +++ b/tests/test_referentiels.py @@ -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 diff --git a/tests/test_securite.py b/tests/test_securite.py new file mode 100644 index 0000000..0330a28 --- /dev/null +++ b/tests/test_securite.py @@ -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