oto-enterprise-os-dtp/05_deliverables_mvp/rbac/userperm_gen/userperm_gen.py

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