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