diff --git a/app/models/user_role.py b/app/models/user_role.py index 83fdaf6..0136c10 100644 --- a/app/models/user_role.py +++ b/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) diff --git a/app/scripts/grant_super_admin.py b/app/scripts/grant_super_admin.py index fb0fdbe..97066d9 100644 --- a/app/scripts/grant_super_admin.py +++ b/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 diff --git a/tests/unit/test_grant_super_admin.py b/tests/unit/test_grant_super_admin.py new file mode 100644 index 0000000..0f1c2bb --- /dev/null +++ b/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