oto-enterprise-os-dtp/05_deliverables_mvp/qa/audit_5d/audit_5d_gen.py

#!/usr/bin/env python3
"""Générateur de l'audit 5D de conformité · Sprint 5 · QA.

Roadmap Sprint 5 · QA : « Audit UAF + normes ISA/IFRS 5D ». Ce générateur est un
audit de SECOND NIVEAU : il lit les hand-off `out/` déjà commités par les
générateurs amont (workflow vente, DocType Dossier Vente, barème commissions,
plan e-CF DGII, DocType CONFOTUR) et vérifie 17 contrôles répartis en 5
dimensions (Traçabilité ISA 500 · AML/UAF Ley 155-17 · Fiscal e-CF Ley 32-23 ·
Intégrité IFRS · Gouvernance ISA 315).

ANTI-INVENTION (#6) : l'audit ne fabrique AUCUN paramètre réglementaire. Un
paramètre réel non confirmé (taux commission, RNC/ITBIS/TipoCambio, seuil UAF)
produit un statut A_CONFIRMER — un « open item » remonté au métier propriétaire —
jamais une valeur inventée « pour faire PASS ». Un FAIL signale une incohérence
inter-livrables ou une valeur chiffrée présente SANS `source`.

Sous-commandes :
  build    [-o OUT]  → écrit audit_report.json + MANIFEST.json
  validate           → (re)génère en mémoire, valide schéma + 15 invariants ;
                       sort en erreur si un contrôle FAIL ou un invariant casse.

Sortie déterministe (tri stable, aucun horodatage) → diffable + re-générable.
"""

from __future__ import annotations

import argparse
import json
import os
import sys

_HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, _HERE)

from qalib import artifacts, builder, controls  # noqa: E402
from qalib.deps import validate as maison_validate  # noqa: E402

_SPEC_PATH = os.path.join(_HERE, "audit_spec.json")
_SCHEMA_PATH = os.path.join(_HERE, "audit.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 _manifest(spec: dict, report: dict) -> dict:
    return {
        "generated_from": "audit_spec.json",
        "audited_artifacts": sorted(artifacts.ARTIFACT_PATHS.values()),
        "dimensions": [d["id"] for d in report["dimensions"]],
        "controls_total": report["totals"]["controls"],
        "verdict": report["verdict"],
        "open_items": [oi["control"] for oi in report["open_items"]],
        "note_anti_invention": (
            "Audit de second niveau : lit les hand-off out/ des livrables, ne "
            "fabrique aucun paramètre réglementaire (#6). A_CONFIRMER = open "
            "item à confirmer par le métier, jamais une valeur inventée."
        ),
        "hand_off_vps": (
            "Confirmer les paramètres réglementaires (taux, RNC/ITBIS/"
            "TipoCambio, seuil UAF) + exécuter les tests E2E Playwright sur le "
            "desk réel → agents Direction/Compta/ONAPI-Legal (#8)."
        ),
    }


def _validate(spec: dict, report: dict) -> list[str]:
    """Schéma de sortie + 15 invariants de cohérence audit/contrôles."""
    errors = list(maison_validate(report, _load(_SCHEMA_PATH)))

    code_ids = controls.control_ids()
    spec_ids = sorted(c["id"] for c in spec["controls"])
    report_ids = sorted(c["id"] for c in report["controls"])
    code_dims = [d["id"] for d in controls.DIMENSIONS]
    spec_dims = [d["id"] for d in spec["dimensions"]]

    spec_by_id = {c["id"]: c for c in spec["controls"]}
    code_by_id = {c["id"]: c for c in controls.CONTROLS}
    known_art = set(artifacts.ARTIFACT_PATHS)

    # 1 · parité des id de contrôle : code == spec == rapport (aucune dérive).
    if not (code_ids == spec_ids == report_ids):
        errors.append(f"id de contrôle désalignés — code={code_ids} spec={spec_ids} report={report_ids}")

    # 2 · parité des dimensions : code == spec, ordre identique.
    if code_dims != spec_dims:
        errors.append(f"dimensions désalignées — code={code_dims} spec={spec_dims}")

    # 3 · chaque contrôle vise une dimension déclarée.
    for c in report["controls"]:
        if c["dimension"] not in code_dims:
            errors.append(f"contrôle {c['id']} → dimension inconnue {c['dimension']!r}")

    # 4 · artefacts déclarés (spec) == artefacts du code, tous connus.
    for cid in code_ids:
        sart = spec_by_id.get(cid, {}).get("artifacts")
        cart = code_by_id[cid]["artifacts"]
        if sart != cart:
            errors.append(f"contrôle {cid} : artefacts spec {sart} ≠ code {cart}")
        for a in cart:
            if a not in known_art:
                errors.append(f"contrôle {cid} : artefact inconnu {a!r}")

    # 5 · parité des propriétaires (owner) code ↔ spec.
    for cid in code_ids:
        if code_by_id[cid]["owner"] != spec_by_id.get(cid, {}).get("owner"):
            errors.append(f"contrôle {cid} : owner spec ≠ code")

    # 6 · aucun id de contrôle dupliqué dans le rapport.
    if len(report_ids) != len(set(report_ids)):
        errors.append("id de contrôle dupliqué dans le rapport")

    # 7 · totaux = somme des statuts, cohérents avec le nombre de contrôles.
    t = report["totals"]
    if t["pass"] + t["fail"] + t["a_confirmer"] != t["controls"]:
        errors.append("totals: pass+fail+a_confirmer != controls")
    if t["controls"] != len(report["controls"]):
        errors.append("totals.controls != nombre de contrôles")

    # 8 · totaux = somme des synthèses par dimension.
    for key in ("pass", "fail", "a_confirmer"):
        s = sum(d[key] for d in report["dimensions"])
        if s != t[key]:
            errors.append(f"somme dimension.{key} ({s}) != totals.{key} ({t[key]})")

    # 9 · verdict cohérent avec les totaux.
    expected = ("FAIL" if t["fail"] else
                "PASS_WITH_OPEN_ITEMS" if t["a_confirmer"] else "PASS")
    if report["verdict"] != expected:
        errors.append(f"verdict {report['verdict']!r} != attendu {expected!r}")

    # 10 · chaque contrôle A_CONFIRMER a un owner non nul et un open_item.
    oi_by_ctrl = {oi["control"]: oi for oi in report["open_items"]}
    for c in report["controls"]:
        if c["statut"] == controls.A_CONFIRMER:
            if not c["owner"]:
                errors.append(f"contrôle A_CONFIRMER {c['id']} sans owner")
            oi = oi_by_ctrl.get(c["id"])
            if oi is None:
                errors.append(f"contrôle A_CONFIRMER {c['id']} absent des open_items")
            elif oi["owner"] != c["owner"]:
                errors.append(f"open_item {c['id']} : owner incohérent")

    # 11 · tout open_item correspond à un contrôle A_CONFIRMER (pas d'orphelin).
    ac_ids = {c["id"] for c in report["controls"] if c["statut"] == controls.A_CONFIRMER}
    for oi in report["open_items"]:
        if oi["control"] not in ac_ids:
            errors.append(f"open_item orphelin : {oi['control']}")

    # 12 · statut de dimension cohérent avec ses compteurs.
    for d in report["dimensions"]:
        exp = ("FAIL" if d["fail"] else
               "PASS_WITH_OPEN_ITEMS" if d["a_confirmer"] else "PASS")
        if d["statut"] != exp:
            errors.append(f"dimension {d['id']} statut {d['statut']!r} != {exp!r}")
        if d["controls_total"] != d["pass"] + d["fail"] + d["a_confirmer"]:
            errors.append(f"dimension {d['id']} : controls_total incohérent")
        if d["controls_total"] < 1:
            errors.append(f"dimension {d['id']} : aucun contrôle")

    # 13 · les 5 dimensions sont présentes.
    if len(report["dimensions"]) != 5:
        errors.append(f"{len(report['dimensions'])} dimensions (attendu 5)")

    # 14 · AUCUN contrôle FAIL sur les livrables courants (cohérence structurelle).
    fails = [c["id"] for c in report["controls"] if c["statut"] == controls.FAIL]
    if fails:
        errors.append(f"contrôle(s) FAIL sur les livrables : {fails}")

    # 15 · déterminisme : re-générer donne un rapport identique.
    if builder.build(spec) != report:
        errors.append("build non déterministe (deux exécutions divergent)")

    return errors


def cmd_build(args) -> int:
    spec = _load(_SPEC_PATH)
    report = builder.build(spec)
    errors = _validate(spec, report)
    if errors:
        _eprint("ÉCHEC validation — build refusé :")
        for e in errors:
            _eprint("  -", e)
        return 1
    out = args.out or _DEFAULT_OUT
    os.makedirs(out, exist_ok=True)
    _write_json(os.path.join(out, "audit_report.json"), report)
    _write_json(os.path.join(out, "MANIFEST.json"), _manifest(spec, report))
    print(f"OK · audit 5D → {out}")
    print(f"   verdict={report['verdict']} · {report['totals']['controls']} contrôles "
          f"({report['totals']['pass']} PASS / {report['totals']['fail']} FAIL / "
          f"{report['totals']['a_confirmer']} À CONFIRMER)")
    return 0


def cmd_validate(args) -> int:
    spec = _load(_SPEC_PATH)
    report = builder.build(spec)
    errors = _validate(spec, report)
    if errors:
        _eprint("ÉCHEC validation :")
        for e in errors:
            _eprint("  -", e)
        return 1
    print(f"OK · {report['totals']['controls']} contrôles, 15 invariants — "
          f"verdict {report['verdict']}")
    return 0


def main(argv=None) -> int:
    p = argparse.ArgumentParser(description="Audit 5D de conformité (QA · Sprint 5).")
    sub = p.add_subparsers(dest="cmd", required=True)
    b = sub.add_parser("build", help="écrit audit_report.json + MANIFEST.json")
    b.add_argument("-o", "--out", help="dossier de sortie (défaut : ./out)")
    b.set_defaults(func=cmd_build)
    v = sub.add_parser("validate", help="valide schéma + 15 invariants")
    v.set_defaults(func=cmd_validate)
    args = p.parse_args(argv)
    return args.func(args)


if __name__ == "__main__":
    raise SystemExit(main())