Files
oto-enterprise-os-dtp/05_deliverables_mvp/rbac/apply_plan/applylib/aggregator.py
T
Claude Code DTP Worker 75a3b0a471 [DTP-Worker] Sprint 2 · Agrégateur RBAC : run-book d'application VPS unifié (3 volets → 1 plan ordonné SPEC §7)
Clôt le volet RBAC en-repo : recoud fixtures Role+DocPerm, plan User Permission
et Role Profile en un run-book ordonné + manifeste agrégé. Zéro chiffre
recalculé (tout vient d'un manifeste source, #6), graphe de dépendances validé
(Role avant Role Profile), cohérence inter-volets + couverture bijective 50/50.
16 tests + job CI rbac-applyplan-tests · 99 tests de régression au total.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-30 04:33:36 +00:00

195 lines
7.6 KiB
Python

"""Agrégation des trois volets RBAC en un plan d'application VPS unique.
Pourquoi cet agrégateur
-----------------------
Trois générateurs RBAC sont livrés en-repo, chacun répondant à UNE question :
- `fixtures_gen` → « QUELS verbes sur QUEL DocType » (Role + Custom DocPerm).
- `userperm_gen` → « SUR QUELLES LIGNES » (scope_donnees → User Permission).
- `roleprofile_gen`→ « QUEL bundle assignable » (Role Profile par portail).
Il manquait la **vue d'ensemble** : dans quel ORDRE l'agent ERPNext applique-t-il
ces trois sorties sur le VPS, avec quelles dépendances et quelles confirmations
préalables ? C'est le rôle de ce module : produire le **run-book** (SPEC §7)
consolidé, plus un manifeste agrégé qui recoupe la cohérence des trois volets.
Ce module NE ré-implémente RIEN : il APPELLE les trois builders et lit leurs
manifestes. Il n'invente aucun chiffre (#6) — tout compte vient d'un builder ;
tout item « à confirmer VPS » est repris tel quel du manifeste d'origine.
Déterminisme : ordre des étapes fixe (SPEC §7), listes de confirmations triées,
aucun horodatage → sortie diffable et re-générable bit-à-bit à contrat constant.
"""
from __future__ import annotations
from typing import Any
# Étapes du run-book (SPEC §7). L'ordre et les dépendances sont du SAVOIR
# procédural documenté (pas des chiffres inventés) : ils encodent les contraintes
# d'application ERPNext v15 (ex. les `Has Role` d'un Role Profile référencent des
# `Role` qui doivent exister → import Role AVANT Role Profile).
# `responsable` : worker (livré en-repo) | vps (agent ERPNext) | worker+vps.
_STEPS: list[dict[str, Any]] = [
{
"order": 1,
"id": "fixtures-generate",
"titre": "Générer les fixtures Role + Custom DocPerm",
"responsable": "worker",
"statut": "livré",
"commande": "python3 fixtures_gen/rbac_fixtures_gen.py build",
"produces": ["role.json", "custom_docperm.json"],
"depends_on": [],
"confirmations": [],
"doc": "../fixtures_gen/README.md",
},
{
"order": 2,
"id": "custom-doctypes-create",
"titre": "Créer les DocTypes DTP custom manquants (après confirmation)",
"responsable": "vps",
"statut": "à faire VPS",
"commande": None,
"produces": [],
"depends_on": ["fixtures-generate"],
"confirmations": ["custom_doctypes"],
"doc": "../RBAC_50_ROLES_SPEC.md",
},
{
"order": 3,
"id": "fixtures-migrate",
"titre": "Déposer role.json + custom_docperm.json dans fixtures/ puis bench migrate",
"responsable": "vps",
"statut": "à faire VPS",
"commande": None,
"produces": [],
"depends_on": ["fixtures-generate", "custom-doctypes-create"],
"confirmations": [],
"doc": "../fixtures_gen/README.md",
},
{
"order": 4,
"id": "userperm-apply",
"titre": "Matérialiser les User Permission row-level par utilisateur",
"responsable": "worker+vps",
"statut": "plan livré · matérialisation VPS",
"commande": "python3 userperm_gen/userperm_gen.py build",
"produces": ["user_permission_plan.json"],
"depends_on": ["fixtures-migrate"],
"confirmations": ["companies", "roles_scope_equipe"],
"doc": "../userperm_gen/README.md",
},
{
"order": 5,
"id": "roleprofile-apply",
"titre": "Importer les Role Profile (après les Role) puis affecter User.role_profile_name",
"responsable": "worker+vps",
"statut": "bundle livré · affectation VPS",
"commande": "python3 roleprofile_gen/roleprofile_gen.py build",
"produces": ["role_profile.json"],
# Dépend de fixtures-migrate : les Has Role référencent des Role existants.
"depends_on": ["fixtures-migrate"],
"confirmations": [],
"doc": "../roleprofile_gen/README.md",
},
{
"order": 6,
"id": "verify-http-qa",
"titre": "Vérification HTTP post-déploiement + audit QA 4Big",
"responsable": "vps",
"statut": "à faire VPS",
"commande": None,
"produces": [],
"depends_on": ["userperm-apply", "roleprofile-apply"],
"confirmations": [],
"doc": "../RBAC_50_ROLES_SPEC.md",
},
]
def _require_same(label: str, values: list[Any]) -> Any:
"""Renvoie l'unique valeur partagée par les trois volets, ou lève."""
uniques = sorted({repr(v) for v in values})
if len(uniques) != 1:
raise ValueError(
f"Volets RBAC incohérents sur {label} : {uniques}"
"les trois générateurs doivent dériver du MÊME contrat."
)
return values[0]
def build_apply_plan(
contract: dict,
fixtures_bundle: dict,
userperm_plan: dict,
roleprofile_bundle: dict,
) -> dict[str, Any]:
"""Consolide les trois volets RBAC en run-book + manifeste agrégé.
Aucun chiffre n'est recalculé : chaque compte provient du manifeste du
builder correspondant (source unique). La cohérence inter-volets (même
version de contrat, même cible 50 rôles) est vérifiée et refuse un mélange.
"""
fm = fixtures_bundle["manifest"]
um = userperm_plan["manifest"]
pm = roleprofile_bundle["manifest"]
# Cohérence : les trois volets doivent dériver du même contrat (version + cible).
source_version = _require_same(
"source_version", [fm["source_version"], um["source_version"], pm["source_version"]]
)
cible = _require_same(
"cible_rbac_roles",
[fm["cible_rbac_roles"], um["cible_rbac_roles"], pm["cible_rbac_roles"]],
)
generated_from = _require_same(
"generated_from", [fm["generated_from"], um["generated_from"], pm["generated_from"]]
)
# Confirmations VPS consolidées (reprises telles quelles des manifestes source).
confirmations_vps = {
"custom_doctypes": sorted(fm.get("custom_doctypes_a_confirmer", [])),
"companies": sorted(um.get("companies_a_confirmer", [])),
"roles_scope_equipe": sorted(um.get("roles_scope_equipe_a_confirmer", [])),
}
# Comptes agrégés : chaque valeur vient d'un manifeste de builder (pas d'invention).
counts = {
"roles": fm["counts"]["role"],
"custom_docperm": fm["counts"]["custom_docperm"],
"doctypes_uniques": fm["counts"]["doctypes_uniques"],
"user_permission_templates": um["counts"]["user_permission_templates"],
"role_profiles": pm["counts"]["role_profiles"],
}
# Cohérence bijective : le nombre de rôles Role == cible == rôles couverts par
# les Role Profile == entrées du plan User Permission (un plan par rôle).
up_entries = um["counts"]["plan_entries"]
rp_covered = pm["counts"]["roles_couverts"]
consistency = {
"sources_coherentes": True, # garanti par _require_same ci-dessus
"source_version": source_version,
"cible_rbac_roles": cible,
"roles_fixtures": counts["roles"],
"userperm_plan_entries": up_entries,
"roleprofile_roles_couverts": rp_covered,
"couverture_bijective": counts["roles"] == cible == up_entries == rp_covered,
}
manifest = {
"generated_from": generated_from,
"source_version": source_version,
"cible_rbac_roles": cible,
"counts": counts,
"confirmations_vps": confirmations_vps,
"consistency": consistency,
}
return {
"manifest": manifest,
"apply_plan": [dict(step) for step in _STEPS],
}
def step_ids() -> list[str]:
"""IDs des étapes du run-book (pour les invariants du CLI/tests)."""
return [s["id"] for s in _STEPS]