[DTP-Worker] Sprint 2 · Générateur Role Profile par portail (RBAC 50 rôles → 6 bundles assignables ERPNext v15)
- roleprofile_gen/ : profilelib (frappe Role Profile + Has Role natif v15, builder déterministe), roleprofile.schema.json, CLI build/validate refusant d'écrire si invariant cassé, 11 tests stdlib, README, .gitignore (out/). - 6 profils couvrant les 50 rôles de façon bijective (5 portails métier + console technique plateforme). Aucun DocType custom (que du natif). - Anti-invention #6 : rôles 100 % issus du contrat, nom de profil = convention déterministe dérivée de la clé portail. - CI : job rbac-roleprofile-tests ajouté au gate (.gitea/workflows/ci.yml). - Doc : SPEC §7 ét.5 + encart livré. Daily report session 8. - Régression : 83 tests verts (72 + 11). Gate CI local vert. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Code DTP Worker
@@ -0,0 +1,229 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Générateur de `Role Profile` par portail · RBAC 50 rôles (Sprint 2).
|
||||
|
||||
Troisième et dernier maillon RBAC en-repo, complément de `../fixtures_gen`
|
||||
(Role + Custom DocPerm) et `../userperm_gen` (row-level) : il produit les
|
||||
**bundles assignables** — un `Role Profile` ERPNext v15 natif par portail — qui
|
||||
permettent à l'agent ERPNext d'attribuer à un utilisateur, en un seul geste,
|
||||
tous les rôles de SON portail (champ `User.role_profile_name`).
|
||||
|
||||
Cible de portage VPS : agent ERPNext Backend (SPEC §7). Ce worker n'écrit JAMAIS
|
||||
sur le VPS — il produit en-repo un bundle de fixtures que l'agent dépose dans
|
||||
`fixtures/` de l'app OTO puis applique via `bench migrate`.
|
||||
|
||||
Transforme `rbac_50_roles.json` (contrat validé par `rbac.schema.json`) en :
|
||||
- role_profile.json → 1 fixture `Role Profile` par portail (+ ses `Has Role`)
|
||||
- MANIFEST.json → traçabilité : comptes, portails métier vs technique,
|
||||
mapping portail → profil
|
||||
|
||||
Sous-commandes :
|
||||
build [-o OUT] → écrit role_profile.json / MANIFEST.json
|
||||
validate [-o OUT] → (re)génère en mémoire, valide vs roleprofile.schema.json
|
||||
+ invariants — sort en erreur si un invariant casse.
|
||||
|
||||
Sortie déterministe (tri stable, aucun horodatage) → diffable + re-générable.
|
||||
Anti-invention (#6) : 100 % des rôles proviennent du contrat ; le nom du profil
|
||||
est une convention déterministe dérivée de la clé `portail` (aucun libellé
|
||||
métier fabriqué) ; aucun DocType custom (que du natif v15).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
_HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
_RBAC_DIR = os.path.normpath(os.path.join(_HERE, "..")) # rbac/
|
||||
_DELIVERABLES = os.path.normpath(os.path.join(_RBAC_DIR, "..")) # 05_deliverables_mvp/
|
||||
|
||||
sys.path.insert(0, _HERE)
|
||||
# Réutilisation (workflow #5 : zéro duplication) du validateur maison Publiciste.
|
||||
sys.path.insert(0, os.path.join(_DELIVERABLES, "publiciste"))
|
||||
|
||||
from profilelib import builder, frappe # noqa: E402
|
||||
from lib import validator as maison # type: ignore # noqa: E402
|
||||
|
||||
_CONTRACT_PATH = os.path.join(_RBAC_DIR, "rbac_50_roles.json")
|
||||
_SCHEMA_PATH = os.path.join(_HERE, "roleprofile.schema.json")
|
||||
_DEFAULT_OUT = os.path.join(_HERE, "out")
|
||||
|
||||
|
||||
def _eprint(*args) -> None:
|
||||
print(*args, file=sys.stderr)
|
||||
|
||||
|
||||
def _load(path: str) -> dict:
|
||||
with open(path, encoding="utf-8") as fh:
|
||||
return json.load(fh)
|
||||
|
||||
|
||||
def _write_json(path: str, data) -> None:
|
||||
with open(path, "w", encoding="utf-8") as fh:
|
||||
json.dump(data, fh, ensure_ascii=False, indent=2)
|
||||
fh.write("\n")
|
||||
|
||||
|
||||
def _build_bundle() -> dict:
|
||||
return builder.build_bundle(_load(_CONTRACT_PATH))
|
||||
|
||||
|
||||
def _validate_bundle(bundle: dict) -> list[str]:
|
||||
"""Valide le bundle contre roleprofile.schema.json + invariants RBAC."""
|
||||
schema = _load(_SCHEMA_PATH)
|
||||
errors = list(maison.validate(bundle, schema))
|
||||
|
||||
manifest = bundle["manifest"]
|
||||
profiles = bundle["role_profile"]
|
||||
contract = _load(_CONTRACT_PATH)
|
||||
contract_roles = {r["erpnext_role_name"] for r in contract["roles"]}
|
||||
portail_of_role = {r["erpnext_role_name"]: r["portail"] for r in contract["roles"]}
|
||||
portails_metier = set(contract.get("portails_business", []))
|
||||
|
||||
# Invariant 1 · cible non-négociable « RBAC 50 rôles » : couverture exhaustive.
|
||||
if manifest["cible_rbac_roles"] != len(contract_roles):
|
||||
errors.append(
|
||||
f"cible_rbac_roles={manifest['cible_rbac_roles']} ≠ rôles contrat={len(contract_roles)}"
|
||||
)
|
||||
|
||||
# Invariant 2 · couverture BIJECTIVE : chaque rôle du contrat dans EXACTEMENT
|
||||
# un profil, aucun rôle inventé, aucun manquant.
|
||||
assigned: list[str] = []
|
||||
for prof in profiles:
|
||||
for row in prof["roles"]:
|
||||
assigned.append(row["role"])
|
||||
if len(assigned) != len(set(assigned)):
|
||||
errors.append("Un rôle apparaît dans plusieurs Role Profile (couverture non bijective).")
|
||||
if set(assigned) != contract_roles:
|
||||
missing = sorted(contract_roles - set(assigned))
|
||||
extra = sorted(set(assigned) - contract_roles)
|
||||
errors.append(f"Couverture ≠ contrat (manquants={missing}, en trop={extra}).")
|
||||
|
||||
# Invariant 3 · un profil par portail distinct du contrat, nom déterministe.
|
||||
contract_portails = {r["portail"] for r in contract["roles"]}
|
||||
seen_names: set[str] = set()
|
||||
for prof in profiles:
|
||||
name = prof["name"]
|
||||
if name != prof["role_profile"]:
|
||||
errors.append(f"{name}: name ≠ role_profile.")
|
||||
if name in seen_names:
|
||||
errors.append(f"Role Profile dupliqué : {name}.")
|
||||
seen_names.add(name)
|
||||
# Chaque rôle du profil appartient bien au même portail que le profil.
|
||||
expected = {n: p for n, p in portail_of_role.items()}
|
||||
portails_in_prof = {expected[row["role"]] for row in prof["roles"]}
|
||||
if len(portails_in_prof) != 1:
|
||||
errors.append(f"{name}: mélange de portails {sorted(portails_in_prof)}.")
|
||||
else:
|
||||
portail = portails_in_prof.pop()
|
||||
if name != frappe.role_profile_name(portail):
|
||||
errors.append(f"{name}: nom ≠ convention pour portail {portail!r}.")
|
||||
|
||||
# Invariant 4 · le manifeste liste EXACTEMENT les portails du contrat.
|
||||
meta_portails = {m["portail"] for m in manifest["profiles"]}
|
||||
if meta_portails != contract_portails:
|
||||
errors.append(f"manifest.profiles portails {sorted(meta_portails)} ≠ contrat {sorted(contract_portails)}.")
|
||||
|
||||
# Invariant 5 · métadonnées fidèles (flag métier + comptes par profil).
|
||||
nb_by_portail: dict[str, int] = {}
|
||||
for prof in profiles:
|
||||
p = portail_of_role[prof["roles"][0]["role"]]
|
||||
nb_by_portail[p] = len(prof["roles"])
|
||||
for m in manifest["profiles"]:
|
||||
p = m["portail"]
|
||||
if m["metier"] != (p in portails_metier):
|
||||
errors.append(f"profil {p!r}: flag metier incohérent avec portails_business.")
|
||||
if m["role_profile"] != frappe.role_profile_name(p):
|
||||
errors.append(f"profil {p!r}: role_profile ≠ convention de nommage.")
|
||||
if m["nb_roles"] != nb_by_portail.get(p):
|
||||
errors.append(f"profil {p!r}: nb_roles {m['nb_roles']} ≠ réel {nb_by_portail.get(p)}.")
|
||||
|
||||
# Invariant 6 · cohérence des comptages agrégés du manifeste.
|
||||
counts = manifest["counts"]
|
||||
if counts["role_profiles"] != len(profiles):
|
||||
errors.append("counts.role_profiles incohérent.")
|
||||
if counts["roles_couverts"] != len(assigned):
|
||||
errors.append("counts.roles_couverts incohérent avec les profils.")
|
||||
nb_metier = sum(1 for m in manifest["profiles"] if m["metier"])
|
||||
if counts["portails_metier"] != nb_metier:
|
||||
errors.append("counts.portails_metier incohérent.")
|
||||
if counts["portails_techniques"] != len(profiles) - nb_metier:
|
||||
errors.append("counts.portails_techniques incohérent.")
|
||||
|
||||
# Invariant 7 · déterminisme : profils triés par portail, rôles triés par nom.
|
||||
if [m["portail"] for m in manifest["profiles"]] != sorted(meta_portails):
|
||||
errors.append("manifest.profiles non trié par portail (déterminisme).")
|
||||
for prof in profiles:
|
||||
role_seq = [row["role"] for row in prof["roles"]]
|
||||
if role_seq != sorted(role_seq):
|
||||
errors.append(f"{prof['name']}: Has Role non triés (déterminisme).")
|
||||
|
||||
return errors
|
||||
|
||||
|
||||
def cmd_build(args: argparse.Namespace) -> int:
|
||||
bundle = _build_bundle()
|
||||
errors = _validate_bundle(bundle)
|
||||
if errors:
|
||||
_eprint("❌ Bundle invalide — génération refusée (anti-régression) :")
|
||||
for e in errors:
|
||||
_eprint(f" - {e}")
|
||||
return 1
|
||||
|
||||
out = os.path.abspath(args.out)
|
||||
os.makedirs(out, exist_ok=True)
|
||||
_write_json(os.path.join(out, "role_profile.json"), bundle["role_profile"])
|
||||
_write_json(os.path.join(out, "MANIFEST.json"), bundle["manifest"])
|
||||
|
||||
m = bundle["manifest"]
|
||||
c = m["counts"]
|
||||
print(f"✅ Role Profile générés dans {out}")
|
||||
print(
|
||||
f" role_profile.json : {c['role_profiles']} profils "
|
||||
f"({c['portails_metier']} métier + {c['portails_techniques']} technique) "
|
||||
f"· {c['roles_couverts']}/50 rôles couverts"
|
||||
)
|
||||
for meta in m["profiles"]:
|
||||
tag = "métier" if meta["metier"] else "technique"
|
||||
print(f" - {meta['role_profile']} : {meta['nb_roles']} rôles ({tag})")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_validate(args: argparse.Namespace) -> int:
|
||||
bundle = _build_bundle()
|
||||
errors = _validate_bundle(bundle)
|
||||
if errors:
|
||||
_eprint("❌ Validation KO :")
|
||||
for e in errors:
|
||||
_eprint(f" - {e}")
|
||||
return 1
|
||||
c = bundle["manifest"]["counts"]
|
||||
print(
|
||||
f"✅ Validation OK — {c['role_profiles']} Role Profile, "
|
||||
f"{c['roles_couverts']}/50 rôles couverts (bijectif), "
|
||||
f"schéma + invariants verts."
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
p = argparse.ArgumentParser(
|
||||
description="Générateur de Role Profile (bundles de rôles) depuis le contrat RBAC."
|
||||
)
|
||||
sub = p.add_subparsers(dest="cmd", required=True)
|
||||
|
||||
pb = sub.add_parser("build", help="génère role_profile.json / MANIFEST.json")
|
||||
pb.add_argument("-o", "--out", default=_DEFAULT_OUT, help="dossier de sortie (défaut: ./out)")
|
||||
pb.set_defaults(func=cmd_build)
|
||||
|
||||
pv = sub.add_parser("validate", help="valide le bundle (schéma + invariants) sans écrire")
|
||||
pv.add_argument("-o", "--out", default=_DEFAULT_OUT, help="ignoré (compat)")
|
||||
pv.set_defaults(func=cmd_validate)
|
||||
|
||||
args = p.parse_args(argv)
|
||||
return args.func(args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user