[DTP-Worker] Sprint 2 · Générateur plan User Permission (RBAC row-level scope_donnees → Frappe)
Complète le pipeline RBAC (Role + Custom DocPerm déjà livrés) par la dimension row-level. Mapping natif ERPNext v15 des 4 scope_donnees : - entite → User Permission allow=Company (28 templates, user=sentinelle) - own → if_owner (déjà posé par fixtures_gen) - groupe → aucune restriction (vue consolidée) - equipe → pas de dimension native → signalé VPS (jamais mappé, #6) Module userperm_gen/ : permlib/{frappe,builder}, CLI build/validate (refuse d'écrire si invariant KO), userperm.schema.json (validateur maison, zéro pip), 12 tests unittest, README. Job CI rbac-userperm-tests ajouté au gate. Docs SPEC §7 + fixtures_gen README cousues. Régression 72 tests verts, gate OK. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Code DTP Worker
@@ -0,0 +1,234 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Générateur de plan `User Permission` · RBAC 50 rôles → row-level (Sprint 2).
|
||||
|
||||
Complément du générateur `../fixtures_gen` (Role + Custom DocPerm) : il couvre la
|
||||
dimension ROW-LEVEL (`scope_donnees`) que les DocPerm ne portent pas seuls.
|
||||
|
||||
Cible de portage VPS : agent ERPNext Backend (SPEC §7). Ce worker n'écrit JAMAIS
|
||||
sur le VPS — il produit en-repo un PLAN que l'agent applique côté serveur en
|
||||
matérialisant, par utilisateur assigné à un rôle, les `User Permission` du
|
||||
template (champ `user` = sentinelle `__ASSIGN_PER_USER__`).
|
||||
|
||||
Transforme `rbac_50_roles.json` (contrat validé par `rbac.schema.json`) en :
|
||||
- user_permission_plan.json → 1 entrée/rôle : mécanisme d'enforcement + template
|
||||
- MANIFEST.json → traçabilité, comptes par mécanisme, Companies +
|
||||
rôles `equipe` à confirmer VPS
|
||||
|
||||
Sous-commandes :
|
||||
build [-o OUT] → écrit user_permission_plan.json / MANIFEST.json
|
||||
validate [-o OUT] → (re)génère en mémoire, valide vs userperm.schema.json +
|
||||
invariants row-level — sort en erreur si un invariant casse.
|
||||
|
||||
Sortie déterministe (tri stable, aucun horodatage) → diffable + re-générable.
|
||||
Anti-invention (#6) : 100 % des données proviennent du contrat ; aucun
|
||||
utilisateur n'est inventé ; les portées sans mécanisme natif (`equipe`) sont
|
||||
signalées, pas mappées arbitrairement.
|
||||
"""
|
||||
|
||||
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 permlib 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, "userperm.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_plan(_load(_CONTRACT_PATH))
|
||||
|
||||
|
||||
def _validate_bundle(bundle: dict) -> list[str]:
|
||||
"""Valide le plan contre userperm.schema.json + invariants row-level RBAC."""
|
||||
schema = _load(_SCHEMA_PATH)
|
||||
errors = list(maison.validate(bundle, schema))
|
||||
|
||||
manifest = bundle["manifest"]
|
||||
plan = bundle["plan"]
|
||||
contract = _load(_CONTRACT_PATH)
|
||||
scope_by_role = {r["erpnext_role_name"]: r["scope_donnees"] for r in contract["roles"]}
|
||||
entite_by_role = {r["erpnext_role_name"]: r["entite_principale"] for r in contract["roles"]}
|
||||
|
||||
# Invariant 1 · cible non-négociable « RBAC 50 rôles » : 1 entrée par rôle.
|
||||
if manifest["cible_rbac_roles"] != len(plan):
|
||||
errors.append(
|
||||
f"cible_rbac_roles={manifest['cible_rbac_roles']} ≠ plan entries={len(plan)}"
|
||||
)
|
||||
|
||||
# Invariant 2 · couverture exhaustive & unique des 50 rôles du contrat.
|
||||
plan_roles = [e["erpnext_role_name"] for e in plan]
|
||||
if len(plan_roles) != len(set(plan_roles)):
|
||||
errors.append("Rôle dupliqué dans le plan.")
|
||||
if set(plan_roles) != set(scope_by_role):
|
||||
missing = set(scope_by_role) - set(plan_roles)
|
||||
extra = set(plan_roles) - set(scope_by_role)
|
||||
errors.append(f"Plan ≠ contrat (manquants={sorted(missing)}, en trop={sorted(extra)}).")
|
||||
|
||||
seen_companies: set[str] = set()
|
||||
for e in plan:
|
||||
role = e["erpnext_role_name"]
|
||||
scope = e["scope_donnees"]
|
||||
tmpl = e["user_permission_template"]
|
||||
|
||||
# Invariant 3 · fidélité au contrat (scope + entité, aucune dérive).
|
||||
if scope != scope_by_role.get(role):
|
||||
errors.append(f"{role}: scope {scope!r} ≠ contrat {scope_by_role.get(role)!r}.")
|
||||
if e["entite_principale"] != entite_by_role.get(role):
|
||||
errors.append(f"{role}: entite_principale divergente du contrat.")
|
||||
|
||||
# Invariant 4 · mécanisme déterministe = mapping natif du scope.
|
||||
expected_mech = frappe.SCOPE_MECHANISM.get(scope)
|
||||
if e["mechanism"] != expected_mech:
|
||||
errors.append(f"{role}: mechanism {e['mechanism']!r} ≠ attendu {expected_mech!r}.")
|
||||
|
||||
# Invariant 5 · un template existe SSI portée `entite` (aucune
|
||||
# restriction fabriquée pour own/groupe/equipe).
|
||||
if scope == "entite":
|
||||
if tmpl is None:
|
||||
errors.append(f"{role}: portée `entite` sans template User Permission.")
|
||||
else:
|
||||
# Invariant 6 · le template est fidèle (Company + entité du contrat).
|
||||
if tmpl.get("allow") != frappe.ALLOW_COMPANY:
|
||||
errors.append(f"{role}: template allow ≠ Company.")
|
||||
if tmpl.get("for_value") != e["entite_principale"]:
|
||||
errors.append(f"{role}: for_value ≠ entite_principale du contrat.")
|
||||
if tmpl.get("user") != frappe.USER_PLACEHOLDER:
|
||||
errors.append(f"{role}: user template n'est pas la sentinelle (invention ?).")
|
||||
if tmpl.get("apply_to_all_doctypes") != 1:
|
||||
errors.append(f"{role}: apply_to_all_doctypes ≠ 1.")
|
||||
if tmpl["for_value"] in frappe.NON_COMPANY_ENTITES:
|
||||
errors.append(f"{role}: for_value consolidé ({tmpl['for_value']}) ne peut être une Company.")
|
||||
seen_companies.add(tmpl["for_value"])
|
||||
elif tmpl is not None:
|
||||
errors.append(f"{role}: portée {scope!r} ne doit porter aucun template (a un template).")
|
||||
|
||||
# Invariant 7 · le manifeste liste EXACTEMENT les Companies des templates.
|
||||
if set(manifest["companies_a_confirmer"]) != seen_companies:
|
||||
errors.append("companies_a_confirmer ≠ Companies réellement ciblées par les templates.")
|
||||
|
||||
# Invariant 8 · les rôles `equipe` du contrat sont TOUS signalés (jamais mappés en douce).
|
||||
equipe_contract = sorted(r for r, s in scope_by_role.items() if s == "equipe")
|
||||
if manifest["roles_scope_equipe_a_confirmer"] != equipe_contract:
|
||||
errors.append("roles_scope_equipe_a_confirmer ≠ rôles `equipe` du contrat.")
|
||||
|
||||
# Invariant 9 · cohérence des comptages du manifeste.
|
||||
counts = manifest["counts"]
|
||||
if counts["plan_entries"] != len(plan):
|
||||
errors.append("counts.plan_entries incohérent.")
|
||||
tmpl_count = sum(1 for e in plan if e["user_permission_template"] is not None)
|
||||
if counts["user_permission_templates"] != tmpl_count:
|
||||
errors.append("counts.user_permission_templates incohérent avec le plan.")
|
||||
if sum(counts["by_mechanism"].values()) != len(plan):
|
||||
errors.append("counts.by_mechanism ne somme pas à 50.")
|
||||
|
||||
return errors
|
||||
|
||||
|
||||
def cmd_build(args: argparse.Namespace) -> int:
|
||||
bundle = _build_bundle()
|
||||
errors = _validate_bundle(bundle)
|
||||
if errors:
|
||||
_eprint("❌ Plan 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, "user_permission_plan.json"), bundle["plan"])
|
||||
_write_json(os.path.join(out, "MANIFEST.json"), bundle["manifest"])
|
||||
|
||||
m = bundle["manifest"]
|
||||
bm = m["counts"]["by_mechanism"]
|
||||
print(f"✅ Plan User Permission généré dans {out}")
|
||||
print(f" user_permission_plan.json : {m['counts']['plan_entries']} entrées (1/rôle)")
|
||||
print(
|
||||
f" templates Company : {m['counts']['user_permission_templates']} "
|
||||
f"(portée `entite`)"
|
||||
)
|
||||
print(
|
||||
" mécanismes : "
|
||||
f"entite={bm['user_permission_company']} · "
|
||||
f"consolidé={bm['none_consolidated']} · "
|
||||
f"own={bm['docperm_if_owner']} · "
|
||||
f"equipe(à confirmer)={bm['vps_confirm_team']}"
|
||||
)
|
||||
if m["companies_a_confirmer"]:
|
||||
print(
|
||||
" ⚠ Companies à créer/confirmer VPS avant application : "
|
||||
+ ", ".join(m["companies_a_confirmer"])
|
||||
)
|
||||
if m["roles_scope_equipe_a_confirmer"]:
|
||||
print(
|
||||
" ⚠ Rôles `equipe` sans dimension row-level native — mécanisme à "
|
||||
"décider VPS : " + ", ".join(m["roles_scope_equipe_a_confirmer"])
|
||||
)
|
||||
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
|
||||
m = bundle["manifest"]
|
||||
print(
|
||||
f"✅ Validation OK — {m['counts']['plan_entries']} entrées, "
|
||||
f"{m['counts']['user_permission_templates']} templates Company, "
|
||||
f"schéma + invariants row-level verts."
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
p = argparse.ArgumentParser(
|
||||
description="Générateur de plan User Permission depuis le contrat RBAC."
|
||||
)
|
||||
sub = p.add_subparsers(dest="cmd", required=True)
|
||||
|
||||
pb = sub.add_parser("build", help="génère user_permission_plan.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 plan (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