[DTP-Worker] Sprint 7 · Générateur Audit 4Big qualité (95+/100 sur 100% deliverables) (QA · roadmap L69)
Audit de méta-niveau + gate : note la qualité 4Big de 100% des livrables gated et bloque (FAIL) si un module < 95/100 (CLAUDE.md #5). Couverture PROUVÉE par recoupement bijectif registre ↔ working-directory du CI (moins l'auditeur · SoD ISA 315). 5 critères déterministes (DOC/CONTRAT/TESTS/CLI/HANDOFF) renormalisés par archétype. Anti-invention (#6) : chaque note est recalculée depuis des faits du dépôt, jamais saisie ; un invariant recompute chaque note. Résultat : PASS · 17/17 modules à 100/100. Régression 442 tests verts (+34). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,12 @@
|
||||
"""q4lib · briques de l'audit 4Big qualité (Sprint 7 · QA).
|
||||
|
||||
Audit de MÉTA-NIVEAU : il note la qualité 4Big de 100% des livrables du mandat
|
||||
à partir de FAITS du dépôt (documentation, contrat de sortie, tests, CLI,
|
||||
intégrité du hand-off), et exige ≥ 95/100 partout (CLAUDE.md #5).
|
||||
|
||||
registry → liste des modules audités + preuve de couverture 100% vs CI Gitea
|
||||
criteria → 5 critères 4Big déterministes (lecture du système de fichiers)
|
||||
scoring → note par module (renormalisation par archétype) + verdict
|
||||
builder → assemblage du rapport quality_report.json + MANIFEST
|
||||
deps → réutilisation du validateur maison Publiciste (zéro pip · #5)
|
||||
"""
|
||||
@@ -0,0 +1,52 @@
|
||||
"""Assemblage déterministe du rapport d'audit 4Big.
|
||||
|
||||
Verdict global = PASS ssi (a) la couverture est bijective vs le CI (100% des
|
||||
livrables gated, hors auditeur) ET (b) les 17 modules atteignent ≥ 95/100.
|
||||
Sinon FAIL — l'audit est un GATE, pas un rapport indicatif.
|
||||
|
||||
Aucune date/horodatage → build reproductible + diffable (gate CI stable).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from . import registry, scoring
|
||||
|
||||
REPORT_NAME = "OTO QA · Audit 4Big Qualité"
|
||||
|
||||
|
||||
def build(spec: dict[str, Any]) -> dict[str, Any]:
|
||||
coverage = registry.coverage_report(spec)
|
||||
modules = scoring.score_all(spec)
|
||||
|
||||
scores = [m["score"] for m in modules]
|
||||
below = [m["id"] for m in modules if m["verdict"] != "PASS"]
|
||||
totals = {
|
||||
"modules": len(modules),
|
||||
"pass": sum(1 for m in modules if m["verdict"] == "PASS"),
|
||||
"fail": len(below),
|
||||
"min_score": min(scores) if scores else 0,
|
||||
"max_score": max(scores) if scores else 0,
|
||||
"pass_rate_pct": scoring._round_half_up(
|
||||
100.0 * sum(1 for m in modules if m["verdict"] == "PASS") / len(modules)
|
||||
) if modules else 0,
|
||||
}
|
||||
|
||||
verdict = "PASS" if (coverage["ok"] and not below) else "FAIL"
|
||||
|
||||
return {
|
||||
"audit": REPORT_NAME,
|
||||
"version": spec["version"],
|
||||
"reference_cadre": spec["reference_cadre"],
|
||||
"pass_score": spec["thresholds"]["pass_score"],
|
||||
"coverage": coverage,
|
||||
"criteria": [
|
||||
{"id": c["id"], "label": c["label"], "weight": c["weight"]}
|
||||
for c in spec["criteria"]
|
||||
],
|
||||
"modules": modules,
|
||||
"totals": totals,
|
||||
"verdict": verdict,
|
||||
"notes": spec.get("notes", []),
|
||||
}
|
||||
@@ -0,0 +1,122 @@
|
||||
"""Les 5 critères 4Big — purs, déterministes, calculés depuis le dépôt.
|
||||
|
||||
Chaque critère lit UNIQUEMENT le système de fichiers (jamais une valeur saisie)
|
||||
et renvoie {passed: bool, evidence: str}. C'est le cœur anti-invention (#6) de
|
||||
l'audit : une note ne peut pas être « écrite pour faire 95 », elle est recomputée
|
||||
à partir de faits vérifiables (présence d'un fichier, taille, comptage AST léger,
|
||||
intégrité JSON).
|
||||
|
||||
Aucune dépendance pip ; stdlib pure (le runner Gitea tourne sans réseau · #2).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
|
||||
_TEST_DEF_RE = re.compile(r"^\s*def (test_[A-Za-z0-9_]+)\s*\(", re.MULTILINE)
|
||||
_ADD_PARSER_RE = re.compile(r"add_parser\(|add_subparsers\(")
|
||||
|
||||
|
||||
def _direct_files(mod_dir: str, suffix: str) -> list[str]:
|
||||
"""Fichiers se terminant par `suffix` directement dans mod_dir (non récursif)."""
|
||||
if not os.path.isdir(mod_dir):
|
||||
return []
|
||||
return sorted(
|
||||
os.path.join(mod_dir, n)
|
||||
for n in os.listdir(mod_dir)
|
||||
if n.endswith(suffix) and os.path.isfile(os.path.join(mod_dir, n))
|
||||
)
|
||||
|
||||
|
||||
def crit_doc(mod_dir: str, min_bytes: int) -> dict:
|
||||
"""DOC — au moins un Markdown (README/SPEC) d'au moins `min_bytes` octets."""
|
||||
for md in _direct_files(mod_dir, ".md"):
|
||||
size = os.path.getsize(md)
|
||||
if size >= min_bytes:
|
||||
return {"passed": True,
|
||||
"evidence": f"{os.path.basename(md)} ({size} octets)"}
|
||||
return {"passed": False,
|
||||
"evidence": f"aucun *.md ≥ {min_bytes} octets à la racine du module"}
|
||||
|
||||
|
||||
def crit_contrat(mod_dir: str) -> dict:
|
||||
"""CONTRAT — au moins un contrat de sortie *.schema.json."""
|
||||
schemas = _direct_files(mod_dir, ".schema.json")
|
||||
if schemas:
|
||||
return {"passed": True,
|
||||
"evidence": ", ".join(os.path.basename(s) for s in schemas)}
|
||||
return {"passed": False, "evidence": "aucun *.schema.json à la racine du module"}
|
||||
|
||||
|
||||
def crit_tests(mod_dir: str, min_methods: int) -> dict:
|
||||
"""TESTS — ≥ min_methods méthodes `def test_*` dans tests/."""
|
||||
tests_dir = os.path.join(mod_dir, "tests")
|
||||
count = 0
|
||||
files = 0
|
||||
if os.path.isdir(tests_dir):
|
||||
for root, _dirs, names in os.walk(tests_dir):
|
||||
if "__pycache__" in root:
|
||||
continue
|
||||
for n in names:
|
||||
if n.startswith("test_") and n.endswith(".py"):
|
||||
files += 1
|
||||
with open(os.path.join(root, n), encoding="utf-8") as fh:
|
||||
count += len(_TEST_DEF_RE.findall(fh.read()))
|
||||
passed = count >= min_methods
|
||||
return {"passed": passed,
|
||||
"evidence": f"{count} méthodes test_* dans {files} fichier(s) "
|
||||
f"(seuil {min_methods})"}
|
||||
|
||||
|
||||
def crit_cli(mod_dir: str) -> dict:
|
||||
"""CLI — un entrypoint __main__ avec sous-commandes argparse à la racine."""
|
||||
for py in _direct_files(mod_dir, ".py"):
|
||||
with open(py, encoding="utf-8") as fh:
|
||||
src = fh.read()
|
||||
if '__name__ == "__main__"' in src and _ADD_PARSER_RE.search(src):
|
||||
return {"passed": True,
|
||||
"evidence": f"{os.path.basename(py)} (argparse + __main__)"}
|
||||
return {"passed": False,
|
||||
"evidence": "aucun entrypoint __main__ avec add_parser/add_subparsers"}
|
||||
|
||||
|
||||
def crit_handoff(mod_dir: str) -> dict:
|
||||
"""HANDOFF — out/MANIFEST.json + ≥1 autre artefact, tous JSON non vides."""
|
||||
out_dir = os.path.join(mod_dir, "out")
|
||||
manifest = os.path.join(out_dir, "MANIFEST.json")
|
||||
if not os.path.isfile(manifest):
|
||||
return {"passed": False, "evidence": "out/MANIFEST.json absent"}
|
||||
others = [os.path.join(out_dir, n) for n in sorted(os.listdir(out_dir))
|
||||
if n.endswith(".json") and n != "MANIFEST.json"]
|
||||
if not others:
|
||||
return {"passed": False,
|
||||
"evidence": "out/ ne contient que MANIFEST.json (aucun artefact)"}
|
||||
for art in [manifest] + others:
|
||||
try:
|
||||
with open(art, encoding="utf-8") as fh:
|
||||
data = json.load(fh)
|
||||
except (json.JSONDecodeError, OSError) as exc:
|
||||
return {"passed": False,
|
||||
"evidence": f"{os.path.basename(art)} illisible : {exc}"}
|
||||
if data in (None, {}, [], ""):
|
||||
return {"passed": False,
|
||||
"evidence": f"{os.path.basename(art)} JSON vide"}
|
||||
return {"passed": True,
|
||||
"evidence": f"MANIFEST + {len(others)} artefact(s) JSON valides"}
|
||||
|
||||
|
||||
# id critère → (fonction, kwargs supplémentaires tirés des thresholds)
|
||||
def evaluate(mod_dir: str, criterion_id: str, thresholds: dict) -> dict:
|
||||
if criterion_id == "DOC":
|
||||
return crit_doc(mod_dir, thresholds["min_doc_bytes"])
|
||||
if criterion_id == "CONTRAT":
|
||||
return crit_contrat(mod_dir)
|
||||
if criterion_id == "TESTS":
|
||||
return crit_tests(mod_dir, thresholds["min_test_methods"])
|
||||
if criterion_id == "CLI":
|
||||
return crit_cli(mod_dir)
|
||||
if criterion_id == "HANDOFF":
|
||||
return crit_handoff(mod_dir)
|
||||
raise KeyError(f"Critère inconnu : {criterion_id}")
|
||||
@@ -0,0 +1,31 @@
|
||||
"""Réutilisation des briques déjà livrées (workflow #5 · zéro duplication).
|
||||
|
||||
L'audit 4Big ne redéfinit rien qui existe ailleurs. On importe — jamais on ne
|
||||
duplique — le validateur JSON-Schema maison du Publiciste (draft-07, sous-
|
||||
ensemble) pour valider le rapport de sortie SANS pip : le gate CI Gitea Actions
|
||||
tourne sans réseau (CLAUDE.md #2).
|
||||
|
||||
Import par `sys.path` (idiome des modules voisins) — une seule source de vérité.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
_HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
# qa/audit_4big/q4lib → 05_deliverables_mvp
|
||||
_DELIVERABLES = os.path.normpath(os.path.join(_HERE, "..", "..", ".."))
|
||||
_PUB = os.path.join(_DELIVERABLES, "publiciste")
|
||||
|
||||
if _PUB not in sys.path:
|
||||
sys.path.insert(0, _PUB)
|
||||
|
||||
from lib import validator # type: ignore # noqa: E402
|
||||
|
||||
validate = validator.validate
|
||||
|
||||
# Racine des livrables, exposée aux autres modules q4lib.
|
||||
DELIVERABLES_ROOT = _DELIVERABLES
|
||||
|
||||
__all__ = ["validate", "DELIVERABLES_ROOT"]
|
||||
@@ -0,0 +1,107 @@
|
||||
"""Registre des modules audités + PREUVE de couverture 100% vs le CI Gitea.
|
||||
|
||||
Le périmètre revendiqué par cet audit — « 100% des livrables » — n'est crédible
|
||||
que s'il est PROUVÉ, pas déclaré. On lit donc `.gitea/workflows/ci.yml` (la seule
|
||||
autorité sur ce qui est réellement gated) et on vérifie :
|
||||
|
||||
1. l'ensemble des modules du registre == l'ensemble des `working-directory:`
|
||||
sous `05_deliverables_mvp/` du CI, PRIVÉ du module auditeur lui-même
|
||||
(séparation des pouvoirs · l'auditeur ne s'auto-note pas · ISA 315) ;
|
||||
2. chaque module gated alimente bien le job d'agrégat `gate` (via `needs`).
|
||||
|
||||
Ainsi, tout futur livrable ajouté au CI sans mise à jour du registre casse la
|
||||
couverture → le gate rougit. Aucune omission silencieuse possible.
|
||||
|
||||
Parsing YAML volontairement minimal (stdlib pure · aucune dépendance pip · #2).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
|
||||
from .deps import DELIVERABLES_ROOT
|
||||
|
||||
_REPO_ROOT = os.path.normpath(os.path.join(DELIVERABLES_ROOT, ".."))
|
||||
_CI_PATH = os.path.join(_REPO_ROOT, ".gitea", "workflows", "ci.yml")
|
||||
|
||||
_PREFIX = "05_deliverables_mvp/"
|
||||
|
||||
_JOB_RE = re.compile(r"^ ([A-Za-z0-9_-]+):\s*$")
|
||||
_WD_RE = re.compile(r"^\s*working-directory:\s*(\S+)\s*$")
|
||||
_NEEDS_RE = re.compile(r"^\s*needs:\s*\[(.*)\]\s*$")
|
||||
|
||||
|
||||
def _read_ci() -> str:
|
||||
with open(_CI_PATH, encoding="utf-8") as fh:
|
||||
return fh.read()
|
||||
|
||||
|
||||
def parse_ci() -> dict:
|
||||
"""Retourne {job_id -> chemin module} pour les jobs sous 05_deliverables_mvp
|
||||
et la liste `needs` du job `gate`."""
|
||||
lines = _read_ci().splitlines()
|
||||
current_job = None
|
||||
job_to_path: dict[str, str] = {}
|
||||
gate_needs: list[str] = []
|
||||
in_gate = False
|
||||
|
||||
for line in lines:
|
||||
m = _JOB_RE.match(line)
|
||||
if m:
|
||||
current_job = m.group(1)
|
||||
in_gate = current_job == "gate"
|
||||
continue
|
||||
wd = _WD_RE.match(line)
|
||||
if wd and current_job:
|
||||
val = wd.group(1)
|
||||
if val.startswith(_PREFIX):
|
||||
job_to_path[current_job] = val[len(_PREFIX):]
|
||||
continue
|
||||
if in_gate:
|
||||
nm = _NEEDS_RE.match(line)
|
||||
if nm:
|
||||
gate_needs = [j.strip() for j in nm.group(1).split(",") if j.strip()]
|
||||
|
||||
return {"job_to_path": job_to_path, "gate_needs": gate_needs}
|
||||
|
||||
|
||||
def ci_module_paths(exclude: str | None = None) -> set[str]:
|
||||
"""Chemins de modules gated par le CI (hors `exclude`, ex. l'auditeur)."""
|
||||
paths = set(parse_ci()["job_to_path"].values())
|
||||
if exclude is not None:
|
||||
paths.discard(exclude)
|
||||
return paths
|
||||
|
||||
|
||||
def coverage_report(spec: dict) -> dict:
|
||||
"""Compare le registre du spec à la réalité du CI. Retourne un diagnostic
|
||||
sérialisable ; `ok=True` ssi couverture bijective + tous gated."""
|
||||
self_module = spec["self_module"]
|
||||
registry_paths = {m["path"] for m in spec["modules"]}
|
||||
ci = parse_ci()
|
||||
job_to_path = ci["job_to_path"]
|
||||
ci_paths = {p for p in job_to_path.values() if p != self_module}
|
||||
|
||||
missing_in_registry = sorted(ci_paths - registry_paths) # gated mais non audité
|
||||
missing_in_ci = sorted(registry_paths - ci_paths) # audité mais non gated
|
||||
|
||||
# Chaque module audité doit alimenter le gate d'agrégat.
|
||||
path_to_jobs: dict[str, list[str]] = {}
|
||||
for job, path in job_to_path.items():
|
||||
path_to_jobs.setdefault(path, []).append(job)
|
||||
not_in_gate = sorted(
|
||||
p for p in registry_paths
|
||||
if not any(j in ci["gate_needs"] for j in path_to_jobs.get(p, []))
|
||||
)
|
||||
|
||||
ok = not missing_in_registry and not missing_in_ci and not not_in_gate
|
||||
return {
|
||||
"ok": ok,
|
||||
"ci_modules_count": len(ci_paths),
|
||||
"registry_modules_count": len(registry_paths),
|
||||
"missing_in_registry": missing_in_registry,
|
||||
"missing_in_ci": missing_in_ci,
|
||||
"not_in_gate": not_in_gate,
|
||||
"self_module_excluded": self_module,
|
||||
}
|
||||
@@ -0,0 +1,83 @@
|
||||
"""Notation d'un module : renormalisation par archétype + verdict.
|
||||
|
||||
Principe : seuls les critères APPLICABLES à l'archétype du module comptent. Le
|
||||
dénominateur est la somme de leurs poids (jamais 100 en dur) ; ainsi un module
|
||||
de contrat (sans hand-off) n'est pas pénalisé pour un artefact out/ qu'il n'a
|
||||
jamais vocation à produire. La note est renormalisée sur 100 puis arrondie.
|
||||
|
||||
note = round(100 * Σ poids(critères applicables PASS) / Σ poids(applicables))
|
||||
|
||||
Un module PASS ssi note ≥ pass_score (95 · CLAUDE.md #5).
|
||||
|
||||
Déterministe : arrondi « demi vers le pair » interdit (dépend de la plateforme) →
|
||||
on utilise un arrondi explicite demi-supérieur, stable et reproductible.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
from . import criteria
|
||||
from .deps import DELIVERABLES_ROOT
|
||||
|
||||
|
||||
def _round_half_up(x: float) -> int:
|
||||
return int(x + 0.5)
|
||||
|
||||
|
||||
def _weight_of(spec: dict, cid: str) -> int:
|
||||
for c in spec["criteria"]:
|
||||
if c["id"] == cid:
|
||||
return c["weight"]
|
||||
raise KeyError(f"Poids introuvable pour le critère {cid}")
|
||||
|
||||
|
||||
def score_module(spec: dict, module: dict) -> dict:
|
||||
thresholds = spec["thresholds"]
|
||||
archetype = module["archetype"]
|
||||
applicable = spec["archetypes"][archetype]["applicable"]
|
||||
mod_dir = os.path.join(DELIVERABLES_ROOT, module["path"])
|
||||
|
||||
if not os.path.isdir(mod_dir):
|
||||
raise FileNotFoundError(
|
||||
f"Module déclaré introuvable sur disque : {module['path']}")
|
||||
|
||||
checks = []
|
||||
earned = 0
|
||||
total = 0
|
||||
for cid in applicable:
|
||||
weight = _weight_of(spec, cid)
|
||||
res = criteria.evaluate(mod_dir, cid, thresholds)
|
||||
total += weight
|
||||
if res["passed"]:
|
||||
earned += weight
|
||||
checks.append({
|
||||
"criterion": cid,
|
||||
"weight": weight,
|
||||
"passed": res["passed"],
|
||||
"evidence": res["evidence"],
|
||||
})
|
||||
|
||||
if total == 0:
|
||||
raise ValueError(
|
||||
f"Archétype {archetype} sans critère applicable — spec incohérent.")
|
||||
|
||||
score = _round_half_up(100.0 * earned / total)
|
||||
verdict = "PASS" if score >= thresholds["pass_score"] else "FAIL"
|
||||
return {
|
||||
"id": module["id"],
|
||||
"path": module["path"],
|
||||
"sprint": module["sprint"],
|
||||
"archetype": archetype,
|
||||
"applicable_weight": total,
|
||||
"earned_weight": earned,
|
||||
"score": score,
|
||||
"verdict": verdict,
|
||||
"checks": checks,
|
||||
}
|
||||
|
||||
|
||||
def score_all(spec: dict) -> list[dict]:
|
||||
# Ordre stable : par sprint puis par chemin (déterminisme du hand-off).
|
||||
mods = sorted(spec["modules"], key=lambda m: (m["sprint"], m["path"]))
|
||||
return [score_module(spec, m) for m in mods]
|
||||
Reference in New Issue
Block a user