[DTP-Worker] Sprint 8 · Générateur Matrice de régression exhaustive (19 suites · 474 tests · gate méta-niveau) (QA · roadmap L74)
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>
This commit is contained in:
Claude Code DTP Worker
@@ -0,0 +1,248 @@
|
||||
#!/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"], [])
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 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()
|
||||
Reference in New Issue
Block a user