scripts/smoke_test.py

"""Test de bout en bout de l'API, sans navigateur.

Usage : DATA_DIR=/tmp/motoc-test python scripts/smoke_test.py
Repart d'une base vierge à chaque exécution.
"""

from __future__ import annotations

import os
import shutil
import sys
from pathlib import Path

TEST_DIR = Path("/tmp/motoc-smoke")
if TEST_DIR.exists():
    shutil.rmtree(TEST_DIR)
TEST_DIR.mkdir(parents=True)

SUPER = "0612345678"
USER = "0698765432"
SELFREG = "0677889900"  # bénévole qui s'inscrit sans avoir été enregistré
os.environ["DATA_DIR"] = str(TEST_DIR)
os.environ["SUPERADMIN_PHONE"] = SUPER

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from fastapi.testclient import TestClient  # noqa: E402

from app import config as app_config  # noqa: E402
from app.main import app  # noqa: E402
from app.security import limiter  # noqa: E402

passed = failed = 0


def check(label: str, condition: bool, detail: str = "") -> None:
    global passed, failed
    if condition:
        passed += 1
        print(f"  \033[32m✓\033[0m {label}")
    else:
        failed += 1
        print(f"  \033[31m✗\033[0m {label} {detail}")


def main() -> int:
    with TestClient(app) as c:
        print("\n\033[1m1. Authentification\033[0m")

        r = c.post("/api/auth/check", json={"phone": "06 12 34 56 78"})
        check("le superadmin est pré-inscrit", r.json()["state"] == "register", r.text)

        r = c.post("/api/auth/check", json={"phone": "0512345678"})
        check("un numéro fixe est refusé", r.status_code == 400)

        r = c.post("/api/auth/check", json={"phone": "0799999999"})
        body = r.json()
        check(
            "un numéro inconnu peut s'inscrire seul",
            body["state"] == "register" and body["new_account"] is True,
            r.text,
        )

        r = c.post(
            "/api/auth/register",
            json={
                "phone": SUPER,
                "first_name": "Sam",
                "last_name": "Admin",
                "nickname": "Chef",
                "pin": "482913",
            },
        )
        check("inscription du superadmin", r.status_code == 200, r.text)
        token = r.json()["token"]
        check("rôle superadmin attribué", r.json()["user"]["is_superadmin"] is True)
        A = {"Authorization": f"Bearer {token}"}

        # Inscription spontanée : un bénévole qu'aucun administrateur n'a
        # enregistré crée son compte lui-même. Il n'obtient aucun groupe.
        r = c.post(
            "/api/auth/register",
            json={
                "phone": SELFREG,
                "first_name": "Théo",
                "last_name": "Libre",
                "nickname": "",
                "pin": "482913",
            },
        )
        check("inscription spontanée acceptée", r.status_code == 200, r.text)
        selfreg_token = r.json().get("token", "")
        check("le compte spontané n'est pas superadmin", r.json()["user"]["is_superadmin"] is False)
        r = c.get("/api/groups", headers={"Authorization": f"Bearer {selfreg_token}"})
        check("un compte spontané n'a aucun groupe", r.json()["groups"] == [], r.text)

        r = c.post(
            "/api/auth/register",
            json={
                "phone": "0655555555",
                "first_name": "Faux",
                "last_name": "Admin",
                "pin": "482913",
                "role": "admin",
            },
        )
        check("profil « Admin » refusé à une inscription spontanée", r.status_code == 403, r.text)
        check(
            "et le compte n'a pas été créé",
            c.post("/api/auth/check", json={"phone": "0655555555"}).json()["new_account"] is True,
        )

        # Repli liste blanche : SELF_REGISTRATION=0 restaure l'ancien mode.
        app_config.SELF_REGISTRATION = False
        try:
            r = c.post("/api/auth/check", json={"phone": "0788888888"})
            check("liste blanche seule : numéro inconnu refusé", r.json()["state"] == "unknown", r.text)
            r = c.post(
                "/api/auth/register",
                json={"phone": "0788888888", "first_name": "A", "last_name": "B", "pin": "482913"},
            )
            check("liste blanche seule : inscription refusée", r.status_code == 403, r.text)
        finally:
            app_config.SELF_REGISTRATION = True

        for weak in ("111111", "123456", "121212", "246810"):
            rr = c.post("/api/auth/login", json={"phone": SUPER, "pin": weak})
            check(f"PIN faible « {weak} » rejeté à la connexion", rr.status_code == 401)

        print("\n\033[1m2. Groupes et sous-groupes\033[0m")

        r = c.post("/api/admin/groups", json={"name": "Festival", "description": "Général"}, headers=A)
        check("création d'un groupe principal", r.status_code == 200, r.text)
        g_root = r.json()["id"]

        r = c.post("/api/admin/groups", json={"name": "Scène 1", "parent_id": g_root}, headers=A)
        check("création d'un sous-groupe", r.status_code == 200, r.text)
        g_sub = r.json()["id"]

        r = c.post("/api/admin/groups", json={"name": "Trop profond", "parent_id": g_sub}, headers=A)
        check("hiérarchie limitée à deux niveaux", r.status_code == 400)

        print("\n\033[1m3. Comptes et affectations multiples\033[0m")

        r = c.post("/api/admin/users/bulk", json={"phones": f"{USER}\n0611111111\n0500000000"}, headers=A)
        body = r.json()
        check("import en masse", len(body["added"]) == 2 and len(body["invalid"]) == 1, r.text)

        r = c.post(
            f"/api/admin/users/{USER}/groups",
            json={"group_ids": [g_root, g_sub], "role": "member"},
            headers=A,
        )
        check("un numéro affecté à plusieurs groupes", len(r.json()["added"]) == 2, r.text)

        r = c.get(f"/api/admin/users/{USER}/groups", headers=A)
        check("les deux appartenances sont listées", len(r.json()["groups"]) == 2, r.text)

        r = c.post("/api/auth/check", json={"phone": USER})
        check("le membre peut désormais s'inscrire", r.json()["state"] == "register")

        r = c.post(
            "/api/auth/register",
            json={
                "phone": USER,
                "first_name": "Léa",
                "last_name": "Martin",
                "nickname": "Léa",
                "pin": "739154",
            },
        )
        check("inscription du membre", r.status_code == 200, r.text)
        U = {"Authorization": f"Bearer {r.json()['token']}"}

        r = c.get("/api/groups", headers=U)
        check("le membre voit ses deux groupes", len(r.json()["groups"]) == 2, r.text)

        print("\n\033[1m4. Messages\033[0m")

        r = c.post(f"/api/groups/{g_sub}/messages", json={"body": "Bonjour @tous"}, headers=U)
        check("envoi d'un message", r.status_code == 200, r.text)
        msg_id = r.json()["message"]["id"]

        r = c.post(f"/api/groups/{g_sub}/messages", json={"body": "   "}, headers=U)
        check("message vide refusé", r.status_code == 400)

        r = c.post(f"/api/groups/{g_sub}/messages", json={"body": "x" * 5000}, headers=U)
        check("message trop long refusé", r.status_code == 400)

        r = c.get(f"/api/groups/{g_sub}/messages", headers=U)
        msgs = r.json()["messages"]
        check("relecture de l'historique", any(m["id"] == msg_id for m in msgs), r.text)

        # Un groupe dont on n'est pas membre doit rester inaccessible.
        r = c.post("/api/admin/groups", json={"name": "Privé"}, headers=A)
        g_private = r.json()["id"]
        r = c.get(f"/api/groups/{g_private}/messages", headers=U)
        check("accès refusé à un groupe non rejoint (IDOR)", r.status_code == 403, r.text)

        r = c.post(f"/api/groups/{g_private}/messages", json={"body": "intrusion"}, headers=U)
        check("écriture refusée dans un groupe non rejoint", r.status_code == 403)

        r = c.delete(f"/api/messages/{msg_id}", headers=U)
        check("suppression de son propre message", r.status_code == 200, r.text)

        print("\n\033[1m5. Sécurité\033[0m")

        r = c.get("/api/admin/stats", headers=U)
        check("routes admin fermées aux membres", r.status_code == 403, r.text)

        r = c.get("/api/admin/stats")
        check("routes admin fermées sans jeton", r.status_code == 401)

        r = c.get("/api/me", headers={"Authorization": "Bearer nawak"})
        check("jeton falsifié rejeté", r.status_code == 401)

        r = c.get("/")
        check("CSP présente", "Content-Security-Policy" in r.headers)
        check("clickjacking bloqué", r.headers.get("X-Frame-Options") == "DENY")
        check("sniffing MIME bloqué", r.headers.get("X-Content-Type-Options") == "nosniff")

        # Le PIN ne doit jamais ressortir de l'API.
        r = c.get("/api/admin/users", headers=A)
        check("aucun hash de PIN exposé", "pin_hash" not in r.text and "pin_salt" not in r.text)

        # Le rate limit par numéro doit se déclencher avant l'épuisement du quota IP.
        limiter.reset()
        for _ in range(11):
            rr = c.post("/api/auth/login", json={"phone": "0611111111", "pin": "999999"})
        check("rate limit par numéro actif", rr.status_code == 429, rr.text)

        print("\n\033[1m6. Compte verrouillé et demande d'aide\033[0m")

        limiter.reset()
        for _ in range(5):
            c.post("/api/auth/login", json={"phone": USER, "pin": "000001"})
        r = c.post("/api/auth/login", json={"phone": USER, "pin": "739154"})
        check("verrouillage après 5 échecs", r.status_code == 429, r.text)

        r = c.post(
            "/api/auth/help",
            json={"phone": USER, "reason": "locked", "message": "Je n'arrive plus à entrer"},
        )
        check("demande d'aide acceptée", r.status_code == 200, r.text)

        r = c.post("/api/auth/help", json={"phone": "0777777777", "reason": "locked"})
        check("réponse identique pour un numéro inconnu", r.status_code == 200)

        r = c.get("/api/help-requests", headers=A)
        reqs = r.json()["requests"]
        check("la demande remonte à l'admin", len(reqs) == 1, r.text)

        r = c.post(f"/api/help-requests/{reqs[0]['id']}/resolve", json={"action": "unlock"}, headers=A)
        check("déverrouillage par l'admin", r.status_code == 200, r.text)

        r = c.post("/api/auth/login", json={"phone": USER, "pin": "739154"})
        check("reconnexion possible après déblocage", r.status_code == 200, r.text)

        print("\n\033[1m7. Import d'un tableur, signalement, épinglage\033[0m")

        sheet = (
            "Prénom;Nom;Téléphone portable\n"
            "Marie;LEROY;06 22 33 44 55\n"
            "Paul;GARNIER;0733445566\n"
            "Léa;Martin;0698765432\n"  # déjà inscrite : ne doit pas être écrasée
            "Sans;Numéro;\n"
        ).encode("utf-8")
        upload = {"file": ("benevoles.csv", sheet, "text/csv")}

        r = c.post("/api/admin/users/import", files=upload, data={"dry_run": "true"}, headers=A)
        body = r.json()
        check("aperçu de l'import", r.status_code == 200, r.text)
        check("colonnes détectées", body["mapping"]["phone_col"] == 2, str(body.get("mapping")))
        check("3 contacts détectés", body["detected"] == 3, str(body))
        check("2 nouveaux, 1 connu", body["new"] == 2 and body["existing"] == 1, str(body))
        check("ligne sans numéro rejetée", body["rejected_count"] == 1, str(body["rejected"]))
        check("aucun compte créé en aperçu", c.get("/api/admin/users?search=LEROY", headers=A).json()["total"] >= 0)
        before = c.get("/api/admin/users", headers=A).json()["total"]

        r = c.post(
            "/api/admin/users/import",
            files={"file": ("benevoles.csv", sheet, "text/csv")},
            data={"dry_run": "false", "group_id": str(g_root)},
            headers=A,
        )
        body = r.json()
        check("import confirmé", r.status_code == 200, r.text)
        check("2 comptes créés", body["added"] == 2, str(body))
        check("1 compte existant mis à jour", body["updated"] == 1, str(body))
        after = c.get("/api/admin/users", headers=A).json()["total"]
        check("annuaire agrandi de 2", after == before + 2, f"{before} -> {after}")

        r = c.get("/api/admin/users?search=Marie", headers=A)
        found = r.json()["users"]
        check("recherche par prénom importé", len(found) == 1, r.text)
        check("nom de famille conservé", found[0]["last_name"] == "LEROY", str(found[0]))
        check("affectation au groupe", found[0]["group_count"] == 1, str(found[0]))

        u = c.get(f"/api/admin/users?search={USER}", headers=A).json()["users"][0]
        check("profil déjà rempli non écrasé par l'import", u["first_name"] == "Léa", str(u))

        r = c.post("/api/admin/users/import", files={"file": ("x.csv", b"a;b\n1;2\n", "text/csv")}, headers=A)
        check("fichier sans téléphone refusé", r.status_code == 400, r.text)

        # Recherche puis actions du superadmin sur un bénévole.
        r = c.get("/api/admin/users?search=06 22 33", headers=A)
        check("recherche par numéro avec espaces", len(r.json()["users"]) == 1, r.text)

        r = c.post("/api/admin/users/0622334455/flag", json={"flagged": True, "reason": "à vérifier"}, headers=A)
        check("signalement d'un compte", r.status_code == 200, r.text)
        flagged = c.get("/api/admin/users?search=0622334455", headers=A).json()["users"][0]
        check("le signalement est visible", flagged["flagged"] is True, str(flagged))
        check("motif conservé", flagged["flag_reason"] == "à vérifier", str(flagged))
        check("le compte reste actif", flagged["status"] != "blocked", str(flagged))

        r = c.post("/api/admin/users/0622334455/flag", json={"flagged": False}, headers=A)
        check("levée du signalement", c.get("/api/admin/users?search=0622334455", headers=A).json()["users"][0]["flagged"] is False, r.text)

        r = c.post(f"/api/admin/users/{SELFREG}/groups", json={"group_ids": [g_sub], "role": "admin"}, headers=A)
        check("promotion admin de groupe depuis l'annuaire", r.status_code == 200, r.text)
        S = {"Authorization": f"Bearer {selfreg_token}"}
        r = c.get(f"/api/groups/{g_sub}", headers=S)
        check("le bénévole promu voit son rôle admin", r.json()["role"] == "admin", r.text)

        # Ce que peut faire un admin de groupe qui n'est pas superadmin.
        r = c.post(f"/api/groups/{g_sub}/members", json={"phone": "0744556677", "first_name": "Nouveau"}, headers=S)
        check("l'admin de groupe enregistre un numéro inconnu", r.status_code == 200, r.text)
        check("le nouveau membre est simple membre", r.json()["role"] == "member", r.text)

        r = c.post(f"/api/groups/{g_sub}/members", json={"phone": "0744556677", "role": "admin"}, headers=S)
        check("un admin de groupe ne peut pas nommer d'admin", r.json()["role"] == "member", r.text)

        r = c.delete(f"/api/groups/{g_sub}/members/0744556677", headers=S)
        check("l'admin de groupe retire un membre", r.status_code == 200, r.text)

        r = c.post(f"/api/groups/{g_sub}/messages", json={"body": "Consigne : RDV 8h au bar"}, headers=S)
        pin_id = r.json()["message"]["id"]
        r = c.post(f"/api/messages/{pin_id}/pin", json={"pinned": True}, headers=S)
        check("l'admin de groupe épingle un message", r.status_code == 200, r.text)
        check("le message porte sa date d'épinglage", r.json()["message"]["pinned_at"] is not None, r.text)

        r = c.get(f"/api/groups/{g_sub}/pins", headers=U)
        check("les épingles sont visibles par les membres", len(r.json()["messages"]) == 1, r.text)

        r = c.post(f"/api/messages/{pin_id}/pin", json={"pinned": True}, headers=U)
        check("un simple membre ne peut pas épingler", r.status_code == 403, r.text)

        r = c.delete(f"/api/messages/{pin_id}", headers=S)
        check("l'admin de groupe supprime un message", r.status_code == 200, r.text)

        print("\n\033[1m8. Accusés de lecture\033[0m")

        r = c.post(f"/api/groups/{g_sub}/messages", json={"body": "Qui a lu ?"}, headers=A)
        receipt_id = r.json()["message"]["id"]

        r = c.get(f"/api/groups/{g_sub}/receipts", headers=A)
        members = r.json()["members"]
        check("état de lecture du groupe", r.status_code == 200, r.text)
        check("un membre par ligne", len(members) >= 2, str(members))
        me = [m for m in members if m["phone"] == USER][0]
        check("le membre n'a pas encore lu", me["last_read_id"] < receipt_id, str(me))

        c.post(f"/api/groups/{g_sub}/read", json={"last_read_id": receipt_id}, headers=U)
        r = c.get(f"/api/groups/{g_sub}/receipts", headers=A)
        me = [m for m in r.json()["members"] if m["phone"] == USER][0]
        check("la lecture est enregistrée", me["last_read_id"] >= receipt_id, str(me))

        r = c.get(f"/api/groups/{g_root}/receipts", headers=U)
        check("accusés refusés hors du groupe", r.status_code == 200, r.text)

        print("\n\033[1m9. Modération\033[0m")

        r = c.post(
            f"/api/groups/{g_sub}/messages",
            json={"body": "je te fais entrer gratos ce soir"},
            headers=U,
        )
        check("le message surveillé est livré", r.status_code == 200, r.text)
        check("l'expéditeur est averti", bool(r.json()["message"].get("warning")), r.text)

        r = c.get("/api/admin/moderation", headers=A)
        alerts = r.json()["alerts"]
        check("le signalement est enregistré", len(alerts) >= 1, r.text)
        check("catégorie « accès gratuit »", alerts[0]["category"] == "acces_gratuit", str(alerts[0]))
        check("compteur à traiter", r.json()["pending"] >= 1, r.text)

        for text, ok in [
            ("c'est GR4TUIT pour nous", True),
            ("g r a t u i t", True),
            ("gratuiit", True),
            ("bonjour, RDV à 14h au camion", False),
            ("le planning est complet, merci", False),
        ]:
            rr = c.post(f"/api/groups/{g_sub}/messages", json={"body": text}, headers=U)
            got = bool(rr.json()["message"].get("warning"))
            check(f"« {text} » {'détecté' if ok else 'ignoré'}", got == ok, rr.text)

        alert_id = alerts[0]["id"]
        r = c.post(f"/api/admin/moderation/{alert_id}/review", headers=A)
        check("signalement marqué traité", r.status_code == 200, r.text)

        r = c.put(
            "/api/admin/moderation/terms",
            json={"terms": {"acces_gratuit": ["kermesse"], "revente": [], "insulte": []}},
            headers=A,
        )
        check("liste surveillée modifiable", r.json()["terms"]["acces_gratuit"] == ["kermesse"], r.text)
        rr = c.post(f"/api/groups/{g_sub}/messages", json={"body": "kermesse demain"}, headers=U)
        check("nouveau terme actif immédiatement", bool(rr.json()["message"].get("warning")), rr.text)
        rr = c.post(f"/api/groups/{g_sub}/messages", json={"body": "gratuit"}, headers=U)
        check("terme retiré de la liste ignoré", not rr.json()["message"].get("warning"), rr.text)

        r = c.get("/api/admin/moderation", headers=A)
        check("modération fermée aux non-superadmins", c.get("/api/admin/moderation", headers=U).status_code == 403)

        print("\n\033[1m10. Bac à sable\033[0m")

        # Un groupe et un compte de bac à sable, posés directement en base.
        from app import db as _db

        _db.execute(
            "INSERT INTO groups(name, parent_id, description, created_at, playground) "
            "VALUES('test_Bar', NULL, '', ?, 1)",
            (_db.now_ms(),),
        )
        pg_group = _db.query_one("SELECT id FROM groups WHERE name = 'test_Bar'")["id"]
        _db.execute(
            "INSERT INTO users(phone, first_name, last_name, nickname, pin_hash, pin_salt, "
            "status, created_at, playground) "
            "VALUES('0600000009', 'test_Zoe', 'test_BLANC', 'test_Zoe', 'x', 'y', 'active', ?, 1)",
            (_db.now_ms(),),
        )
        _db.execute(
            "INSERT INTO memberships(group_id, phone, role, joined_at) "
            "VALUES(?, '0600000009', 'member', ?)",
            (pg_group, _db.now_ms()),
        )

        r = c.post("/api/auth/check", json={"phone": "0600000009"})
        check("un compte du bac à sable est invisible", r.json()["state"] == "unknown", r.text)
        r = c.post("/api/auth/login", json={"phone": "0600000009", "pin": "482913"})
        check("connexion impossible avec un compte fictif", r.status_code == 404, r.text)
        r = c.post(
            "/api/auth/register",
            json={"phone": "0600000009", "first_name": "A", "last_name": "B", "pin": "482913"},
        )
        check("inscription impossible sur un numéro fictif", r.status_code == 403, r.text)

        r = c.get("/api/admin/users?search=test_", headers=A)
        check("absent de l'annuaire réel", r.json()["total"] == 0 or not any(
            u["phone"] == "0600000009" for u in r.json()["users"]
        ), r.text)
        r = c.get("/api/admin/groups", headers=A)
        check(
            "groupe fictif absent de la liste réelle",
            not any(g["name"] == "test_Bar" for g in r.json()["groups"]),
            r.text,
        )
        r = c.get("/api/groups", headers=A)
        check(
            "groupe fictif absent des conversations du superadmin",
            not any(g["name"] == "test_Bar" for g in r.json()["groups"]),
            r.text,
        )
        stats_before = c.get("/api/admin/stats", headers=A).json()
        check(
            "comptes fictifs exclus des statistiques",
            all(u["phone"] != "0600000009" for u in c.get("/api/admin/users", headers=A).json()["users"]),
            str(stats_before["users"]),
        )

        r = c.get("/api/admin/playground", headers=A)
        body = r.json()
        check("le bac à sable est listé", r.status_code == 200, r.text)
        check("groupe fictif présent", any(g["id"] == pg_group for g in body["groups"]), r.text)
        check("compte fictif présent", any(u["phone"] == "0600000009" for u in body["users"]), r.text)

        r = c.post(
            "/api/admin/playground/messages",
            json={"group_id": pg_group, "phone": "0600000009", "body": "Bonjour depuis le bac à sable"},
            headers=A,
        )
        check("écriture au nom d'un compte fictif", r.status_code == 200, r.text)
        check("auteur = le compte fictif", r.json()["message"]["author_label"] == "test_Zoe", r.text)

        r = c.get(f"/api/admin/playground/groups/{pg_group}/messages", headers=A)
        check("historique du bac à sable", len(r.json()["messages"]) == 1, r.text)

        r = c.post(f"/api/admin/playground/groups/{pg_group}/read", headers=A)
        check("lecture simulée", r.json()["readers"] >= 1, r.text)

        r = c.get("/api/admin/playground", headers=U)
        check("bac à sable fermé aux non-superadmins", r.status_code == 403, r.text)

        r = c.post(f"/api/admin/playground/groups/{pg_group}/clear", headers=A)
        check("conversation fictive vidable", r.status_code == 200, r.text)

        print("\n\033[1m11. Supervision machine\033[0m")

        r = c.get("/api/admin/system", headers=A)
        sysinfo = r.json()
        check("instantané machine", r.status_code == 200, r.text)
        for key in ("cpu", "memory", "disk", "network", "process", "database", "realtime"):
            check(f"section « {key} » présente", key in sysinfo, str(sysinfo.keys()))
        check("taille de base rapportée", sysinfo["database"]["size"] > 0, str(sysinfo["database"]))
        check(
            "supervision fermée aux non-superadmins",
            c.get("/api/admin/system", headers=U).status_code == 403,
        )

        print("\n\033[1m12. Tableau de bord et rétention\033[0m")

        r = c.get("/api/admin/dashboard", headers=A)
        d = r.json()
        check("le tableau de bord répond", r.status_code == 200, r.text)
        check("KPI présents", d["kpis"]["users_total"] >= 3, str(d.get("kpis")))
        check("série journalière complète", len(d["daily"]) == 14)
        check("série horaire sur 24 h", len(d["hourly"]) == 24)
        check("audit trail alimenté", len(d["audit"]) > 0)
        check("rétention à 500 messages par groupe", d["storage"]["keep_per_group"] == 500)
        check("rétention à 30 jours", d["storage"]["retention_days"] == 30)

        r = c.post("/api/admin/purge", headers=A)
        check("purge manuelle", r.status_code == 200, r.text)

        r = c.get("/api/push/key")
        check("clé VAPID publiée", len(r.json()["public_key"]) > 80, r.text)

    print(f"\n\033[1m{passed} réussis, {failed} échoués\033[0m\n")
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())