Claude Code DTP Worker
76fb0e768d
Défaut #1 (bug CI réel) : ci/check_regression.sh exigeait regression_run.json commité comme baseline, mais le fichier était .gitignore → absent en checkout propre → le job Gitea check-regression échouait sur CHAQUE run CI (ne passait qu'en local via un fichier non suivi). L'artefact run est byte-déterministe (aucun horodatage/hôte/durée · path relatif) → committer est licite. Fix : run désormais commité (baseline), .gitignore + README réécrits. Défaut #2 (invariant manquant) : coverage_report prouvait CI→disque mais pas l'inverse. Nouveau disk_test_modules() + orphan_tests_dirs dans INV4 : un module gagnant un tests/ sans job CI fait chuter la couverture (fini la sous-comptée silencieuse de la matrice). Schéma + 2 tests de morsure. Consommateur régénéré : qa/audit_4big/quality_report.json (README 3710→4236 o, 24→26 méthodes). Matrice 534/21/PASS inchangée (harnais self-exclu, INV3). 5 gates verts · 26+34 tests OK · YAML valide.
271 lines
11 KiB
Python
271 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Tests du harnais de matrice de régression exhaustive (QA · Sprint 8).
|
|
|
|
Stdlib pur (`unittest`) → aucune installation pip requise sur le runner Gitea
|
|
(CLAUDE.md #2). Couvre : le parseur PUR de sortie unittest (OK / skipped /
|
|
FAILED / vide), l'exécution réelle d'une suite synthétique en tmpdir (verte +
|
|
rouge + tests/ absent), la découverte des suites depuis le CI + faits de disque,
|
|
la preuve de couverture, l'assemblage du plan, les invariants du générateur
|
|
(dont anti-invention #6 : aucun compteur de résultat dans le plan · recompute des
|
|
comptes depuis le disque), le déterminisme et le schéma.
|
|
|
|
Ce job NE ré-exécute PAS les suites du mandat (le gate le fait déjà job par job) —
|
|
il teste le HARNAIS lui-même.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
_HERE = os.path.dirname(os.path.abspath(__file__))
|
|
_MOD = os.path.normpath(os.path.join(_HERE, ".."))
|
|
sys.path.insert(0, _MOD)
|
|
|
|
from reglib import builder, discovery, runner # noqa: E402
|
|
import regression_gen as gen # noqa: E402
|
|
|
|
|
|
def _spec():
|
|
with open(os.path.join(_MOD, "regression_spec.json"), encoding="utf-8") as fh:
|
|
return json.load(fh)
|
|
|
|
|
|
def _make_suite(root: str, *, methods: int, failing: bool = False,
|
|
with_tests: bool = True) -> str:
|
|
"""Arbre de module synthétique avec une suite unittest ; renvoie son chemin."""
|
|
mod = os.path.join(root, "mod")
|
|
os.makedirs(mod, exist_ok=True)
|
|
if with_tests:
|
|
tdir = os.path.join(mod, "tests")
|
|
os.makedirs(tdir, exist_ok=True)
|
|
body = "".join(
|
|
f" def test_case_{i}(self):\n"
|
|
f" self.assertTrue({'False' if (failing and i == 0) else 'True'})\n"
|
|
for i in range(methods)
|
|
)
|
|
with open(os.path.join(tdir, "test_mod.py"), "w", encoding="utf-8") as fh:
|
|
fh.write("import unittest\nclass T(unittest.TestCase):\n" +
|
|
(body or " pass\n"))
|
|
return mod
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Parseur PUR de la sortie unittest (aucune exécution).
|
|
# --------------------------------------------------------------------------
|
|
class TestParseOutput(unittest.TestCase):
|
|
def test_ok(self):
|
|
out = "....\n----\nRan 4 tests in 0.01s\n\nOK\n"
|
|
r = runner.parse_unittest_output(out, 0)
|
|
self.assertEqual(r["ran"], 4)
|
|
self.assertEqual(r["passed"], 4)
|
|
self.assertEqual((r["failures"], r["errors"], r["skipped"]), (0, 0, 0))
|
|
self.assertTrue(r["ok"])
|
|
|
|
def test_ok_with_skipped(self):
|
|
out = "Ran 5 tests in 0.02s\n\nOK (skipped=2)\n"
|
|
r = runner.parse_unittest_output(out, 0)
|
|
self.assertEqual(r["ran"], 5)
|
|
self.assertEqual(r["skipped"], 2)
|
|
self.assertEqual(r["passed"], 3)
|
|
self.assertTrue(r["ok"])
|
|
|
|
def test_failed(self):
|
|
out = "Ran 10 tests in 0.03s\n\nFAILED (failures=1, errors=2, skipped=3)\n"
|
|
r = runner.parse_unittest_output(out, 1)
|
|
self.assertEqual(r["ran"], 10)
|
|
self.assertEqual((r["failures"], r["errors"], r["skipped"]), (1, 2, 3))
|
|
self.assertEqual(r["passed"], 4)
|
|
self.assertFalse(r["ok"])
|
|
|
|
def test_empty_output_is_not_ok(self):
|
|
r = runner.parse_unittest_output("", 0)
|
|
self.assertEqual(r["ran"], 0)
|
|
self.assertFalse(r["ok"]) # ran==0 → jamais vert (une suite doit tourner)
|
|
|
|
def test_returncode_overrides_ok(self):
|
|
out = "Ran 3 tests in 0.01s\n\nOK\n"
|
|
self.assertFalse(runner.parse_unittest_output(out, 1)["ok"])
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Exécution réelle d'une suite (subprocess).
|
|
# --------------------------------------------------------------------------
|
|
class TestRunSuite(unittest.TestCase):
|
|
def test_green_suite(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
mod = _make_suite(tmp, methods=3)
|
|
r = runner.run_suite(mod)
|
|
self.assertTrue(r["ok"])
|
|
self.assertEqual(r["ran"], 3)
|
|
self.assertEqual(r["failures"], 0)
|
|
|
|
def test_red_suite(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
mod = _make_suite(tmp, methods=3, failing=True)
|
|
r = runner.run_suite(mod)
|
|
self.assertFalse(r["ok"])
|
|
self.assertGreaterEqual(r["failures"], 1)
|
|
|
|
def test_missing_tests_dir(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
mod = _make_suite(tmp, methods=0, with_tests=False)
|
|
r = runner.run_suite(mod)
|
|
self.assertFalse(r["ok"])
|
|
self.assertEqual(r["returncode"], -1)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Découverte + comptage disque.
|
|
# --------------------------------------------------------------------------
|
|
class TestDiscovery(unittest.TestCase):
|
|
def test_count_tests(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
mod = _make_suite(tmp, methods=7)
|
|
files, methods = discovery.count_tests(os.path.join(mod, "tests"))
|
|
self.assertEqual(files, 1)
|
|
self.assertEqual(methods, 7)
|
|
|
|
def test_count_tests_absent(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
self.assertEqual(discovery.count_tests(os.path.join(tmp, "nope")),
|
|
(0, 0))
|
|
|
|
def test_discover_excludes_self_and_is_sorted(self):
|
|
spec = _spec()
|
|
suites = discovery.discover_suites(spec)
|
|
paths = [s["path"] for s in suites]
|
|
self.assertNotIn(spec["self_module"], paths) # séparation des pouvoirs
|
|
self.assertEqual(paths, sorted(paths)) # déterministe
|
|
self.assertGreater(len(suites), 1)
|
|
for s in suites: # faits de disque réels
|
|
self.assertTrue(s["has_tests_dir"], s["path"])
|
|
self.assertGreaterEqual(s["test_methods"], 1, s["path"])
|
|
|
|
def test_coverage_ok_on_repo(self):
|
|
spec = _spec()
|
|
suites = discovery.discover_suites(spec)
|
|
cov = discovery.coverage_report(spec, suites)
|
|
self.assertTrue(cov["ok"], cov)
|
|
self.assertTrue(cov["self_module_gated"]) # le harnais est gated
|
|
self.assertEqual(cov["missing_tests_dir"], [])
|
|
self.assertEqual(cov["not_in_gate"], [])
|
|
self.assertEqual(cov["orphan_tests_dirs"], []) # aucun tests/ hors CI
|
|
|
|
def test_disk_test_modules_subset_of_ci(self):
|
|
# Sens inverse de la couverture : tout `tests/` sur disque est gated.
|
|
disk = set(discovery.disk_test_modules())
|
|
ci = set(discovery.parse_ci()["job_to_path"].values())
|
|
self.assertIn(_spec()["self_module"], disk) # le harnais a des tests
|
|
self.assertEqual(disk - ci, set(), f"orphelins disque→CI : {disk - ci}")
|
|
|
|
def test_orphan_tests_dir_detected(self):
|
|
# Preuve de morsure : un module portant un `tests/` non câblé au CI doit
|
|
# faire chuter la couverture (dérive silencieuse disque→CI interdite).
|
|
spec = _spec()
|
|
suites = discovery.discover_suites(spec)
|
|
real = discovery.disk_test_modules
|
|
try:
|
|
discovery.disk_test_modules = lambda: real() + ["orphan/module"]
|
|
cov = discovery.coverage_report(spec, suites)
|
|
finally:
|
|
discovery.disk_test_modules = real
|
|
self.assertIn("orphan/module", cov["orphan_tests_dirs"])
|
|
self.assertFalse(cov["ok"], cov)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Assemblage du plan + invariants du générateur.
|
|
# --------------------------------------------------------------------------
|
|
class TestPlanAndInvariants(unittest.TestCase):
|
|
def setUp(self):
|
|
self.spec = _spec()
|
|
self.plan = builder.build_plan(self.spec)
|
|
|
|
def test_plan_passes_and_is_clean(self):
|
|
self.assertEqual(self.plan["verdict"], "PASS")
|
|
self.assertEqual(gen.check_invariants(self.plan, self.spec), [])
|
|
|
|
def test_plan_has_no_result_counters(self):
|
|
# #6 : le plan (statique) ne doit contenir AUCUN compteur d'exécution.
|
|
forbidden = {"green", "red", "passed", "ran", "failures", "errors"}
|
|
self.assertEqual(forbidden & set(self.plan["totals"].keys()), set())
|
|
|
|
def test_totals_match_suites(self):
|
|
s = self.plan["suites"]
|
|
self.assertEqual(self.plan["totals"]["test_methods"],
|
|
sum(x["test_methods"] for x in s))
|
|
self.assertEqual(self.plan["totals"]["suites"], len(s))
|
|
|
|
def test_determinism(self):
|
|
self.assertEqual(builder.build_plan(self.spec), self.plan)
|
|
|
|
def test_schema_valid(self):
|
|
with open(os.path.join(_MOD, "regression.schema.json"),
|
|
encoding="utf-8") as fh:
|
|
schema = json.load(fh)
|
|
from reglib.deps import validate as v
|
|
self.assertEqual(v(self.plan, schema), [])
|
|
|
|
def test_inv_self_in_suites(self):
|
|
bad = copy.deepcopy(self.plan)
|
|
bad["suites"].append({
|
|
"id": "self", "path": self.spec["self_module"], "jobs": ["x"],
|
|
"in_gate": True, "has_tests_dir": True, "test_files": 1,
|
|
"test_methods": 9,
|
|
})
|
|
errs = gen.check_invariants(bad, self.spec)
|
|
self.assertTrue(any("INV3" in e for e in errs), errs)
|
|
|
|
def test_inv_forbidden_counter_leak(self):
|
|
bad = copy.deepcopy(self.plan)
|
|
bad["totals"]["green"] = 19
|
|
errs = gen.check_invariants(bad, self.spec)
|
|
self.assertTrue(any("INV2" in e for e in errs), errs)
|
|
|
|
def test_inv_frozen_count_detected(self):
|
|
# Un compte de disque figé/fabriqué (≠ recompute) doit être détecté (#6).
|
|
bad = copy.deepcopy(self.plan)
|
|
bad["suites"][0]["test_methods"] += 100
|
|
bad["totals"]["test_methods"] += 100
|
|
errs = gen.check_invariants(bad, self.spec)
|
|
self.assertTrue(any("INV5" in e for e in errs), errs)
|
|
|
|
def test_inv_totals_tampered(self):
|
|
bad = copy.deepcopy(self.plan)
|
|
bad["totals"]["suites"] += 1
|
|
errs = gen.check_invariants(bad, self.spec)
|
|
self.assertTrue(any("INV7" in e for e in errs), errs)
|
|
|
|
def test_inv_verdict_tampered(self):
|
|
bad = copy.deepcopy(self.plan)
|
|
bad["verdict"] = "FAIL" if self.plan["verdict"] == "PASS" else "PASS"
|
|
errs = gen.check_invariants(bad, self.spec)
|
|
self.assertTrue(any("INV8" in e for e in errs), errs)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# CLI (entrypoints reproductibles).
|
|
# --------------------------------------------------------------------------
|
|
class TestCLI(unittest.TestCase):
|
|
def test_validate_ok(self):
|
|
self.assertEqual(gen.main(["validate"]), 0)
|
|
|
|
def test_build_writes_artifacts(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
self.assertEqual(gen.main(["build", "-o", tmp]), 0)
|
|
for f in ("regression_plan.json", "MANIFEST.json"):
|
|
self.assertTrue(os.path.exists(os.path.join(tmp, f)), f)
|
|
man = json.load(open(os.path.join(tmp, "MANIFEST.json"),
|
|
encoding="utf-8"))
|
|
self.assertEqual(man["verdict"], "PASS")
|
|
self.assertEqual(man["generator"], "regression_gen.py")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|