oto-enterprise-os-dtp/05_deliverables_mvp/rbac/apply_plan/rbac_apply_plan.py

#!/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())