#!/usr/bin/env python3 """Agrégateur RBAC · run-book d'application VPS unifié (Sprint 2). Quatrième et dernier maillon RBAC en-repo — la **vue d'ensemble**. Les trois générateurs livrés répondent chacun à une question isolée : - `../fixtures_gen` → Role + Custom DocPerm (quels verbes / quel DocType). - `../userperm_gen` → plan User Permission row-level (sur quelles lignes). - `../roleprofile_gen`→ Role Profile par portail (quel bundle assignable). Il manquait l'**ordre d'application** consolidé : ce module produit le run-book (SPEC §7) + un manifeste agrégé qui recoupe la cohérence des trois volets (même contrat, même cible 50 rôles, couverture bijective). Cible de portage VPS : agent ERPNext Backend. Ce worker n'écrit JAMAIS sur le VPS — il produit en-repo un plan ordonné que l'agent suit pas à pas. Produit : - apply_plan.json → run-book ordonné (6 étapes SPEC §7 : responsable, statut, commande worker, artefacts produits, dépendances, confirmations préalables VPS, lien doc). - MANIFEST.json → manifeste agrégé (comptes consolidés des 3 volets, items « à confirmer VPS » fusionnés, recoupement de cohérence). Sous-commandes : build [-o OUT] → écrit apply_plan.json / MANIFEST.json validate [-o OUT] → (re)génère en mémoire, valide vs apply_plan.schema.json + invariants — sort en erreur si un invariant casse. Sortie déterministe (ordre d'étapes fixe, listes triées, aucun horodatage). Anti-invention (#6) : aucun chiffre recalculé — tout compte provient d'un manifeste de builder ; chaque item « à confirmer » est repris tel quel. """ 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/ # Réutilisation (workflow #5 : zéro duplication) des trois builders RBAC + du # validateur maison Publiciste. Paquets aux noms distincts (fixturelib / permlib # / profilelib / applylib / lib) → aucune collision de module. sys.path.insert(0, _HERE) sys.path.insert(0, os.path.join(_RBAC_DIR, "fixtures_gen")) sys.path.insert(0, os.path.join(_RBAC_DIR, "userperm_gen")) sys.path.insert(0, os.path.join(_RBAC_DIR, "roleprofile_gen")) sys.path.insert(0, os.path.join(_DELIVERABLES, "publiciste")) from applylib import aggregator # noqa: E402 from fixturelib import builder as fixtures_builder # type: ignore # noqa: E402 from permlib import builder as userperm_builder # type: ignore # noqa: E402 from profilelib import builder as roleprofile_builder # type: ignore # 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, "apply_plan.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_plan() -> dict: """Appelle les 3 builders sur le contrat, puis agrège en run-book.""" contract = _load(_CONTRACT_PATH) return aggregator.build_apply_plan( contract, fixtures_builder.build_bundle(contract), userperm_builder.build_plan(contract), roleprofile_builder.build_bundle(contract), ) def _validate_plan(plan: dict) -> list[str]: """Valide le plan contre apply_plan.schema.json + invariants RBAC agrégés.""" schema = _load(_SCHEMA_PATH) errors = list(maison.validate(plan, schema)) manifest = plan["manifest"] steps = plan["apply_plan"] # Invariant 1 · le run-book couvre EXACTEMENT les étapes SPEC §7, dans l'ordre. expected_ids = aggregator.step_ids() got_ids = [s["id"] for s in steps] if got_ids != expected_ids: errors.append(f"apply_plan: étapes {got_ids} ≠ SPEC §7 {expected_ids}.") orders = [s["order"] for s in steps] if orders != sorted(orders) or orders != list(range(1, len(steps) + 1)): errors.append(f"apply_plan: champ `order` non séquentiel {orders}.") # Invariant 2 · intégrité du graphe de dépendances (pas de renvoi fantôme, # pas de dépendance en avant → l'agent applique dans l'ordre sans blocage). id_to_order = {s["id"]: s["order"] for s in steps} for s in steps: for dep in s["depends_on"]: if dep not in id_to_order: errors.append(f"{s['id']}: dépendance inconnue {dep!r}.") elif id_to_order[dep] >= s["order"]: errors.append(f"{s['id']}: dépend d'une étape postérieure {dep!r}.") # Invariant 3 · dépendance métier ERPNext critique : Role Profile importé # APRÈS la migration des fixtures Role (Has Role référence des Role existants). rp = next((s for s in steps if s["id"] == "roleprofile-apply"), None) if rp is not None and "fixtures-migrate" not in rp["depends_on"]: errors.append("roleprofile-apply doit dépendre de fixtures-migrate (Role avant Role Profile).") # Invariant 4 · toute confirmation citée par une étape existe dans le manifeste # ET toute confirmation non vide du manifeste est référencée par au moins une # étape (aucun item « à confirmer » orphelin, aucune référence fantôme). conf = manifest["confirmations_vps"] referenced: set[str] = set() for s in steps: for key in s["confirmations"]: if key not in conf: errors.append(f"{s['id']}: confirmation inconnue {key!r}.") referenced.add(key) for key, items in conf.items(): if items and key not in referenced: errors.append(f"confirmations_vps.{key} non vide mais référencée par aucune étape.") # Invariant 5 · cohérence inter-volets : les 3 volets dérivent du même contrat. cons = manifest["consistency"] if not cons["sources_coherentes"]: errors.append("consistency.sources_coherentes=false (volets RBAC incohérents).") if cons["source_version"] != manifest["source_version"]: errors.append("consistency.source_version ≠ manifest.source_version.") if cons["cible_rbac_roles"] != manifest["cible_rbac_roles"]: errors.append("consistency.cible_rbac_roles ≠ manifest.cible_rbac_roles.") # Invariant 6 · couverture bijective des 50 rôles à travers les 3 volets : # fixtures Role == cible == entrées plan UP == rôles couverts par Role Profile. cible = manifest["cible_rbac_roles"] if not (cons["roles_fixtures"] == cible == cons["userperm_plan_entries"] == cons["roleprofile_roles_couverts"]): errors.append( "Couverture non bijective : " f"Role={cons['roles_fixtures']}, cible={cible}, " f"UP={cons['userperm_plan_entries']}, RoleProfile={cons['roleprofile_roles_couverts']}." ) if not cons["couverture_bijective"]: errors.append("consistency.couverture_bijective=false alors que les comptes concordent (ou l'inverse).") if manifest["counts"]["roles"] != cible: errors.append("counts.roles ≠ cible_rbac_roles (les fixtures ne couvrent pas les 50 rôles).") # Invariant 7 · déterminisme : listes de confirmations triées. for key, items in conf.items(): if items != sorted(items): errors.append(f"confirmations_vps.{key} non trié (déterminisme).") return errors def cmd_build(args: argparse.Namespace) -> int: plan = _build_plan() errors = _validate_plan(plan) if errors: _eprint("❌ Run-book 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, "apply_plan.json"), plan["apply_plan"]) _write_json(os.path.join(out, "MANIFEST.json"), plan["manifest"]) m = plan["manifest"] c = m["counts"] print(f"✅ Run-book RBAC généré dans {out}") print( f" apply_plan.json : {len(plan['apply_plan'])} étapes (SPEC §7) · " f"MANIFEST.json : {c['roles']} rôles, {c['custom_docperm']} DocPerm, " f"{c['user_permission_templates']} UP templates, {c['role_profiles']} Role Profile" ) for s in plan["apply_plan"]: cmd = f" → {s['commande']}" if s["commande"] else "" print(f" {s['order']}. [{s['responsable']}] {s['titre']} ({s['statut']}){cmd}") conf = m["confirmations_vps"] print( f" Confirmations VPS : {len(conf['custom_doctypes'])} DocType custom, " f"{len(conf['companies'])} Company, {len(conf['roles_scope_equipe'])} rôles `equipe`." ) return 0 def cmd_validate(args: argparse.Namespace) -> int: plan = _build_plan() errors = _validate_plan(plan) if errors: _eprint("❌ Validation KO :") for e in errors: _eprint(f" - {e}") return 1 cons = plan["manifest"]["consistency"] print( f"✅ Validation OK — {len(plan['apply_plan'])} étapes, " f"couverture bijective {cons['roles_fixtures']}/50 sur les 3 volets, " f"schéma + invariants verts." ) return 0 def main(argv: list[str] | None = None) -> int: p = argparse.ArgumentParser( description="Agrégateur RBAC : run-book d'application VPS unifié (3 volets → 1 plan ordonné)." ) sub = p.add_subparsers(dest="cmd", required=True) pb = sub.add_parser("build", help="génère apply_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())