Attribution du super-admin : écrire contre le schéma réel de user_roles

app/models/user_role.py

@@ -12,6 +12,25 @@
 
 
 class UserRole(Base):
+    """
+    ⚠ Ce modèle NE DÉCRIT PAS la table de production.
+
+    En production, `user_roles` vaut exactement :
+
+        CREATE TABLE public.user_roles (
+            user_id uuid NOT NULL,
+            role    public.user_role NOT NULL,   -- ENUM, pas varchar
+            PRIMARY KEY (user_id, role)
+        );
+
+    Ni `id`, ni `tenant_id`, ni `created_at`. L'écart est invisible en test,
+    puisque le schéma de test est construit depuis ce modèle : tout code écrit
+    d'après ces colonnes passe les tests et échoue en production. Écrire contre
+    `information_schema` quand on touche cette table (cf.
+    `app/scripts/grant_super_admin.py`), ou aligner le modèle sur la base par
+    une migration dédiée.
+    """
+
     __tablename__ = "user_roles"
 
     id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), primary_key=True, default=uuid4)

app/scripts/grant_super_admin.py

@@ -20,6 +20,7 @@
 from __future__ import annotations
 
 import sys
+from typing import Optional
 
 from sqlalchemy import text
 
@@ -34,16 +35,71 @@ def _bypass(conn) -> None:
     conn.execute(text(f"SET ROLE {settings.AUTH_BYPASS_ROLE}"))
 
 
+def _colonnes(conn) -> set[str]:
+    """
+    Colonnes réelles de `user_roles`, relevées en base et non déduites du modèle.
+
+    Le modèle `app/models/user_role.py` déclare `id`, `tenant_id` et
+    `created_at` ; la base de production n'a que `(user_id, role)`. Comme le
+    schéma de test est construit depuis les modèles, l'écart est invisible en
+    test et n'apparaît qu'à l'exécution. On écrit donc contre ce que la base
+    contient, ce qui rend ce script valable sur les deux schémas.
+    """
+    return {
+        r[0]
+        for r in conn.execute(
+            text("""
+                SELECT column_name FROM information_schema.columns
+                WHERE table_schema = 'public' AND table_name = 'user_roles'
+            """)
+        )
+    }
+
+
+def construire_insert(
+    colonnes: set[str], *, user_id: str, tenant_id: Optional[str] = None
+) -> tuple[str, dict]:
+    """
+    Construit l'INSERT adapté aux colonnes réellement présentes.
+
+    Production : `(user_id, role)` seulement. Base créée depuis les modèles :
+    `id` et `tenant_id` en plus, tous deux NOT NULL. Une seule requête écrite
+    en dur ne peut pas satisfaire les deux — d'où cette construction.
+    """
+    if not {"user_id", "role"} <= colonnes:
+        raise RuntimeError("Table `user_roles` inattendue : colonnes user_id/role absentes.")
+
+    champs = ["user_id", "role"]
+    valeurs = ["CAST(:uid AS uuid)", f"'{ROLE}'"]
+    params: dict = {"uid": str(user_id)}
+
+    if "id" in colonnes:
+        champs.insert(0, "id")
+        valeurs.insert(0, "gen_random_uuid()")
+    if "tenant_id" in colonnes:
+        champs.append("tenant_id")
+        valeurs.append("CAST(:tid AS uuid)")
+        params["tid"] = str(tenant_id) if tenant_id else None
+
+    return (
+        f"INSERT INTO public.user_roles ({', '.join(champs)}) "
+        f"VALUES ({', '.join(valeurs)})",
+        params,
+    )
+
+
 def lister() -> int:
     with engine.begin() as conn:
         _bypass(conn)
+        date_dispo = "created_at" in _colonnes(conn)
+        depuis = "ur.created_at" if date_dispo else "NULL::timestamptz"
         rows = conn.execute(
             text(f"""
-                SELECT u.email, u.full_name, ur.created_at
+                SELECT u.email, u.full_name, {depuis} AS depuis
                 FROM public.user_roles ur
                 JOIN public.users u ON u.id = ur.user_id
-                WHERE ur.role = '{ROLE}'
-                ORDER BY ur.created_at
+                WHERE ur.role::text = '{ROLE}'
+                ORDER BY u.email
             """)
         ).mappings().all()
 
@@ -52,7 +108,8 @@ def lister() -> int:
         return 0
     print(f"{len(rows)} super-administrateur(s) :")
     for r in rows:
-        print(f"  · {r['email']}  ({r['full_name']})  depuis le {r['created_at']:%d/%m/%Y}")
+        quand = f"  depuis le {r['depuis']:%d/%m/%Y}" if r["depuis"] else ""
+        print(f"  · {r['email']}  ({r['full_name']}){quand}")
     return 0
 
 
@@ -72,29 +129,38 @@ def appliquer(email: str, *, revoke: bool) -> int:
         if not row:
             print(f"✗ Aucun compte pour {email}")
             return 1
-        if not row["tenant_id"]:
-            print(f"✗ {email} n'est rattaché à aucun tenant — rattachez-le d'abord.")
-            return 1
 
         if revoke:
             n = conn.execute(
                 text(f"""
                     DELETE FROM public.user_roles
-                    WHERE user_id = CAST(:uid AS uuid) AND role = '{ROLE}'
+                    WHERE user_id = CAST(:uid AS uuid) AND role::text = '{ROLE}'
                 """),
                 {"uid": row["id"]},
             ).rowcount
             print(f"✓ SUPER_ADMIN retiré à {email}" if n else f"= {email} n'était pas SUPER_ADMIN")
-        else:
-            conn.execute(
-                text(f"""
-                    INSERT INTO public.user_roles (id, tenant_id, user_id, role)
-                    VALUES (gen_random_uuid(), CAST(:tid AS uuid), CAST(:uid AS uuid), '{ROLE}')
-                    ON CONFLICT (tenant_id, user_id, role) DO NOTHING
-                """),
-                {"tid": row["tenant_id"], "uid": row["id"]},
-            )
-            print(f"✓ {email} ({row['full_name']}) est super-administrateur.")
+            print("→ Effectif à la prochaine connexion (la revendication est dans le jeton).")
+            return 0
+
+        deja = conn.execute(
+            text(f"""
+                SELECT 1 FROM public.user_roles
+                WHERE user_id = CAST(:uid AS uuid) AND role::text = '{ROLE}'
+            """),
+            {"uid": row["id"]},
+        ).first()
+        if deja:
+            print(f"= {email} est déjà super-administrateur.")
+            return 0
+
+        cols = _colonnes(conn)
+        if "tenant_id" in cols and not row["tenant_id"]:
+            print(f"✗ {email} n'est rattaché à aucun tenant — rattachez-le d'abord.")
+            return 1
+
+        sql, params = construire_insert(cols, user_id=row["id"], tenant_id=row["tenant_id"])
+        conn.execute(text(sql), params)
+        print(f"✓ {email} ({row['full_name']}) est super-administrateur.")
 
     print("→ Effectif à la prochaine connexion (la revendication est dans le jeton).")
     return 0

tests/unit/test_grant_super_admin.py

@@ -0,0 +1,57 @@
+"""
+L'INSERT du script d'attribution doit s'adapter au schéma rencontré.
+
+La production n'a que `(user_id, role)` ; une base créée depuis les modèles a
+en plus `id` et `tenant_id`, tous deux NOT NULL. Une requête écrite en dur ne
+peut pas satisfaire les deux — et comme le schéma de test vient des modèles,
+seul un test sur les colonnes attrape l'écart.
+"""
+from __future__ import annotations
+
+import pytest
+
+from app.scripts.grant_super_admin import construire_insert
+
+pytestmark = pytest.mark.unit
+
+PROD = {"user_id", "role"}
+MODELES = {"id", "tenant_id", "user_id", "role", "created_at"}
+
+
+class TestSchemaProduction:
+    def test_n_insere_que_les_colonnes_existantes(self):
+        sql, params = construire_insert(PROD, user_id="u-1", tenant_id="t-1")
+        assert "user_roles (user_id, role)" in sql
+        assert "id" not in sql.split("VALUES")[0].replace("user_id", "")
+        assert "tenant_id" not in sql
+        assert params == {"uid": "u-1"}
+
+    def test_le_tenant_n_est_pas_transmis(self):
+        _, params = construire_insert(PROD, user_id="u-1", tenant_id="t-1")
+        assert "tid" not in params, "un paramètre non utilisé ferait échouer psycopg"
+
+
+class TestSchemaModeles:
+    def test_ajoute_id_et_tenant(self):
+        sql, params = construire_insert(MODELES, user_id="u-1", tenant_id="t-1")
+        assert "gen_random_uuid()" in sql
+        assert "tenant_id" in sql
+        assert params == {"uid": "u-1", "tid": "t-1"}
+
+    def test_ordre_champs_et_valeurs_coherent(self):
+        sql, _ = construire_insert(MODELES, user_id="u-1", tenant_id="t-1")
+        champs = sql.split("(", 1)[1].split(")", 1)[0].split(", ")
+        valeurs = sql.split("VALUES (", 1)[1].rstrip(")").split(", ")
+        assert len(champs) == len(valeurs)
+        assert champs[0] == "id" and valeurs[0] == "gen_random_uuid()"
+
+
+class TestGardeFou:
+    def test_table_inattendue_refusee(self):
+        with pytest.raises(RuntimeError, match="user_roles"):
+            construire_insert({"machin", "truc"}, user_id="u-1")
+
+    def test_role_toujours_super_admin(self):
+        for cols in (PROD, MODELES):
+            sql, _ = construire_insert(cols, user_id="u-1", tenant_id="t-1")
+            assert "'SUPER_ADMIN'" in sql