Claude Code DTP Worker
4b0752906d
Harnais méta-niveau : agrège l'exécution de toutes les suites gated en une matrice + verdict PASS/FAIL et fournit le compte agrégé faisant autorité (N tests verts). Périmètre dérivé du CI (réutilise q4lib/registry.parse_ci · zéro duplication) ; anti-invention (#6) : le plan ne contient aucun compteur de résultat, recomputé à la validation. Enregistré dans l'audit 4Big (18→19 modules · PASS 19/19). run exhaustif : 19/19 suites vertes · 474 tests passés. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
93 lines
3.5 KiB
Python
93 lines
3.5 KiB
Python
"""Exécution réelle des suites + parseur PUR de la sortie unittest.
|
|
|
|
C'est le cœur de la régression EXÉCUTÉE (mode `run`) : on lance chaque suite dans
|
|
un sous-processus isolé (`python -m unittest discover -s tests`, exactement comme
|
|
le CI Gitea), on parse sa sortie réelle, et on agrège les compteurs. Aucun chiffre
|
|
n'est fabriqué — tout provient de la sortie de unittest (#6 anti-invention).
|
|
|
|
Isolation par subprocess (et non import in-process) : les modules du mandat ont
|
|
des fichiers `tests/test_*.py` aux noms qui se recouvrent → un discover global
|
|
in-process provoquerait des collisions de modules. Le subprocess reproduit
|
|
fidèlement l'environnement du CI.
|
|
|
|
`parse_unittest_output` est PUR (str → dict) donc testable sans exécuter quoi que
|
|
ce soit — on lui injecte des sorties unittest synthétiques (OK / FAILED / skipped).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
_RAN_RE = None # compilé paresseusement pour rester importable sans `re` en tête
|
|
|
|
|
|
def _res():
|
|
import re
|
|
return (
|
|
re.compile(r"Ran (\d+) tests? in"),
|
|
re.compile(r"failures=(\d+)"),
|
|
re.compile(r"errors=(\d+)"),
|
|
re.compile(r"skipped=(\d+)"),
|
|
)
|
|
|
|
|
|
def parse_unittest_output(text: str, returncode: int) -> dict:
|
|
"""Extrait les compteurs de la sortie texte de unittest.
|
|
|
|
unittest écrit son résumé sur stderr :
|
|
« Ran 12 tests in 0.03s » puis « OK » | « OK (skipped=2) »
|
|
| « FAILED (failures=1, errors=2, skipped=3) ».
|
|
`ok` reflète le code de retour du processus (source de vérité), pas un parse
|
|
fragile du mot « OK »."""
|
|
ran_re, fail_re, err_re, skip_re = _res()
|
|
ran_m = ran_re.search(text)
|
|
fail_m = fail_re.search(text)
|
|
err_m = err_re.search(text)
|
|
skip_m = skip_re.search(text)
|
|
ran = int(ran_m.group(1)) if ran_m else 0
|
|
failures = int(fail_m.group(1)) if fail_m else 0
|
|
errors = int(err_m.group(1)) if err_m else 0
|
|
skipped = int(skip_m.group(1)) if skip_m else 0
|
|
passed = max(ran - failures - errors - skipped, 0)
|
|
return {
|
|
"ran": ran,
|
|
"passed": passed,
|
|
"failures": failures,
|
|
"errors": errors,
|
|
"skipped": skipped,
|
|
"ok": returncode == 0 and ran > 0,
|
|
"returncode": returncode,
|
|
}
|
|
|
|
|
|
def run_suite(abs_path: str, *, python: str | None = None,
|
|
discover_cmd: list[str] | None = None,
|
|
timeout: int = 600) -> dict:
|
|
"""Exécute une suite dans son propre répertoire et renvoie ses compteurs.
|
|
|
|
`abs_path` = répertoire du module (contenant `tests/`). Aucune sortie n'est
|
|
imprimée ici — l'agrégateur décide de l'affichage."""
|
|
py = python or sys.executable
|
|
cmd = list(discover_cmd or ["-m", "unittest", "discover", "-s", "tests"])
|
|
tests_dir = os.path.join(abs_path, "tests")
|
|
if not os.path.isdir(tests_dir):
|
|
return {
|
|
"ran": 0, "passed": 0, "failures": 0, "errors": 0, "skipped": 0,
|
|
"ok": False, "returncode": -1, "note": "tests/ absent",
|
|
}
|
|
try:
|
|
proc = subprocess.run(
|
|
[py, *cmd], cwd=abs_path, capture_output=True, text=True,
|
|
timeout=timeout,
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
return {
|
|
"ran": 0, "passed": 0, "failures": 0, "errors": 0, "skipped": 0,
|
|
"ok": False, "returncode": -2, "note": f"timeout {timeout}s",
|
|
}
|
|
# unittest écrit son résumé sur stderr ; on garde stdout en secours.
|
|
return parse_unittest_output((proc.stderr or "") + (proc.stdout or ""),
|
|
proc.returncode)
|