[DTP-Worker] Sprint 8 · Générateur Run-book de déploiement VPS unifié (7 phases · 20 modules gated · couverture bijective vs CI) (DevOps · roadmap L73)
Agrégateur de méta-niveau au-dessus du run-book RBAC : ordonne le déploiement VPS de TOUS les livrables gated en 7 phases (Prérequis → DocTypes → RBAC → Workflow/métier → Frontend → Contenu → Vérification QA), avec graphe de dépendances inter-phases acyclique et confirmations préalables sourcées. Anti-invention (#6) : périmètre dérivé du CI (parse_ci réutilisé), couverture bijective module→phase (un module gated non planifié OU un module planifié non gated → refus), SoD (auto-exclusion), zéro chiffre métier (confirmations sourcées via audit_5d D1.1/D1.2/D1.3/D2.3 + endpoint OTOIA). Vérifs : 29/29 tests module (dont 14 injections négatives) · audit_4big PASS 20/20 à 100 · régression run 20/20 suites · 503 tests · 0 échec · gate CI local vert. CI job devops-deploy-runbook-tests + gate ; enregistrement audit_4big (19→20) ; plan régression régénéré (19→20 suites). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Code DTP Worker
@@ -0,0 +1,259 @@
|
||||
"""Tests du run-book de déploiement VPS unifié (stdlib pur · zéro pip).
|
||||
|
||||
Couvre : le build réel (couverture bijective vs CI Gitea), la conformité au
|
||||
schéma, les 11 familles d'invariants, le déterminisme, et une batterie
|
||||
d'injections négatives (chaque invariant DOIT rougir quand on le viole).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
_HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
_MOD = os.path.normpath(os.path.join(_HERE, ".."))
|
||||
sys.path.insert(0, _MOD)
|
||||
|
||||
import deploy_runbook_gen as gen # noqa: E402
|
||||
from deploylib import builder, deps # noqa: E402
|
||||
|
||||
|
||||
def _spec() -> dict:
|
||||
return gen._load(gen._SPEC_PATH)
|
||||
|
||||
|
||||
def _ci() -> dict:
|
||||
return deps.parse_ci()
|
||||
|
||||
|
||||
def _synthetic_ci(paths, self_module="devops/deploy_runbook") -> dict:
|
||||
"""CI factice : un job par module + le job du run-book lui-même."""
|
||||
j2p = {f"{p.replace('/', '-')}-tests": p for p in paths}
|
||||
j2p["devops-deploy-runbook-tests"] = self_module
|
||||
return {"job_to_path": j2p, "gate_needs": list(j2p)}
|
||||
|
||||
|
||||
class TestRealBuild(unittest.TestCase):
|
||||
"""Build réel contre le CI et le spec du dépôt."""
|
||||
|
||||
def setUp(self):
|
||||
self.spec = _spec()
|
||||
self.ci = _ci()
|
||||
self.bundle = builder.build_runbook(self.spec, self.ci)
|
||||
|
||||
def test_build_validates_clean(self):
|
||||
errors = gen.validate_bundle(self.bundle, self.spec, self.ci)
|
||||
self.assertEqual(errors, [], f"invariants rouges: {errors}")
|
||||
|
||||
def test_coverage_bijective(self):
|
||||
cov = self.bundle["manifest"]["coverage"]
|
||||
self.assertTrue(cov["bijective"])
|
||||
self.assertEqual(cov["missing_in_map"], [])
|
||||
self.assertEqual(cov["extra_in_map"], [])
|
||||
self.assertEqual(cov["unknown_phase"], [])
|
||||
|
||||
def test_every_gated_module_assigned_once(self):
|
||||
gated = set(builder.gated_modules(self.ci, self.spec["self_module"]))
|
||||
assigned = [m["module"] for p in self.bundle["runbook"] for m in p["modules"]]
|
||||
self.assertEqual(sorted(assigned), sorted(gated))
|
||||
self.assertEqual(len(assigned), len(set(assigned)), "un module dans 2 phases")
|
||||
|
||||
def test_self_module_excluded(self):
|
||||
self_mod = self.spec["self_module"]
|
||||
self.assertNotIn(self_mod, self.spec["module_phase"])
|
||||
allmods = [m["module"] for p in self.bundle["runbook"] for m in p["modules"]]
|
||||
self.assertNotIn(self_mod, allmods)
|
||||
|
||||
def test_phases_sequential(self):
|
||||
orders = [p["order"] for p in self.bundle["runbook"]]
|
||||
self.assertEqual(orders, list(range(1, len(orders) + 1)))
|
||||
|
||||
def test_dependencies_point_backward(self):
|
||||
order_of = {p["id"]: p["order"] for p in self.bundle["runbook"]}
|
||||
for p in self.bundle["runbook"]:
|
||||
for dep in p["depends_on"]:
|
||||
self.assertIn(dep, order_of)
|
||||
self.assertLess(order_of[dep], p["order"])
|
||||
|
||||
def test_every_module_has_ci_job(self):
|
||||
for p in self.bundle["runbook"]:
|
||||
for m in p["modules"]:
|
||||
self.assertTrue(m["ci_job"], f"{m['module']} sans ci_job")
|
||||
|
||||
def test_confirmations_sourced(self):
|
||||
for cid, entry in self.spec["confirmations"].items():
|
||||
self.assertTrue(entry.get("owner"), f"{cid} sans owner")
|
||||
self.assertTrue(entry.get("source"), f"{cid} sans source")
|
||||
|
||||
def test_no_orphan_confirmation(self):
|
||||
self.assertEqual(self.bundle["manifest"]["graph"]["orphan_confirmations"], [])
|
||||
self.assertEqual(self.bundle["manifest"]["graph"]["unknown_confirmations"], [])
|
||||
|
||||
def test_spec_has_no_business_number(self):
|
||||
# Anti-invention (#6) : aucun champ de valeur chiffrée dans le catalogue.
|
||||
forbidden = {"valeur", "value", "montant", "taux", "amount", "rate"}
|
||||
for entry in self.spec["confirmations"].values():
|
||||
self.assertEqual(forbidden & set(entry), set())
|
||||
|
||||
def test_counts_consistent(self):
|
||||
c = self.bundle["manifest"]["counts"]
|
||||
gated = builder.gated_modules(self.ci, self.spec["self_module"])
|
||||
self.assertEqual(c["phases"], len(self.bundle["runbook"]))
|
||||
self.assertEqual(c["modules"], len(gated))
|
||||
self.assertEqual(c["confirmations"], len(self.spec["confirmations"]))
|
||||
|
||||
def test_deterministic(self):
|
||||
again = builder.build_runbook(self.spec, self.ci)
|
||||
self.assertEqual(
|
||||
json.dumps(self.bundle, sort_keys=True),
|
||||
json.dumps(again, sort_keys=True),
|
||||
)
|
||||
|
||||
def test_modules_sorted_in_each_phase(self):
|
||||
for p in self.bundle["runbook"]:
|
||||
names = [m["module"] for m in p["modules"]]
|
||||
self.assertEqual(names, sorted(names))
|
||||
|
||||
|
||||
class TestSchema(unittest.TestCase):
|
||||
def test_output_matches_schema(self):
|
||||
bundle = builder.build_runbook(_spec(), _ci())
|
||||
schema = gen._load(gen._SCHEMA_PATH)
|
||||
self.assertEqual(list(deps.validate(bundle, schema)), [])
|
||||
|
||||
def test_written_artifacts_are_valid_json(self):
|
||||
# Le hand-off commité doit être lisible (critère HANDOFF de l'audit 4Big).
|
||||
out = os.path.join(_MOD, "out")
|
||||
for name in ("deploy_runbook.json", "MANIFEST.json"):
|
||||
path = os.path.join(out, name)
|
||||
if os.path.isfile(path):
|
||||
with open(path, encoding="utf-8") as fh:
|
||||
self.assertNotIn(json.load(fh), (None, {}, [], ""))
|
||||
|
||||
|
||||
class TestNegativeInjections(unittest.TestCase):
|
||||
"""Chaque invariant doit rougir quand on viole sa condition."""
|
||||
|
||||
def _errors(self, spec, ci):
|
||||
bundle = builder.build_runbook(spec, ci)
|
||||
return gen.validate_bundle(bundle, spec, ci)
|
||||
|
||||
def _base(self):
|
||||
"""Spec minimal cohérent + CI synthétique assorti (2 modules, 1 phase)."""
|
||||
spec = {
|
||||
"version": "1.0",
|
||||
"reference_cadre": "test",
|
||||
"cible_portage": "test",
|
||||
"self_module": "devops/deploy_runbook",
|
||||
"phases": [
|
||||
{"order": 1, "id": "p1", "titre": "P1", "responsable": "vps",
|
||||
"rationale": "r", "depends_on": [], "confirmations": ["c1"]},
|
||||
],
|
||||
"module_phase": {"a/one": "p1", "b/two": "p1"},
|
||||
"confirmations": {
|
||||
"c1": {"libelle": "L", "owner": "O", "source": "S"},
|
||||
},
|
||||
}
|
||||
ci = _synthetic_ci(["a/one", "b/two"])
|
||||
return spec, ci
|
||||
|
||||
def test_base_is_clean(self):
|
||||
spec, ci = self._base()
|
||||
self.assertEqual(self._errors(spec, ci), [])
|
||||
|
||||
def test_missing_module_flagged(self):
|
||||
spec, ci = self._base()
|
||||
del spec["module_phase"]["b/two"] # gated mais non planifié
|
||||
errs = self._errors(spec, ci)
|
||||
self.assertTrue(any("non planifiés" in e for e in errs), errs)
|
||||
|
||||
def test_extra_module_flagged(self):
|
||||
spec, ci = self._base()
|
||||
spec["module_phase"]["z/fake"] = "p1" # planifié mais non gated
|
||||
errs = self._errors(spec, ci)
|
||||
self.assertTrue(any("non gated" in e for e in errs), errs)
|
||||
|
||||
def test_self_in_map_flagged(self):
|
||||
spec, ci = self._base()
|
||||
spec["module_phase"]["devops/deploy_runbook"] = "p1"
|
||||
errs = self._errors(spec, ci)
|
||||
self.assertTrue(any("SoD" in e for e in errs), errs)
|
||||
|
||||
def test_unknown_phase_flagged(self):
|
||||
spec, ci = self._base()
|
||||
spec["module_phase"]["a/one"] = "ghost"
|
||||
errs = self._errors(spec, ci)
|
||||
self.assertTrue(any("phase" in e.lower() for e in errs), errs)
|
||||
|
||||
def test_forward_dependency_flagged(self):
|
||||
spec, ci = self._base()
|
||||
spec["phases"].append(
|
||||
{"order": 2, "id": "p2", "titre": "P2", "responsable": "vps",
|
||||
"rationale": "r", "depends_on": [], "confirmations": []}
|
||||
)
|
||||
# p1 dépend de p2 (postérieure) → renvoi en avant
|
||||
spec["phases"][0]["depends_on"] = ["p2"]
|
||||
spec["module_phase"]["b/two"] = "p2"
|
||||
errs = self._errors(spec, ci)
|
||||
self.assertTrue(any("postérieure" in e for e in errs), errs)
|
||||
|
||||
def test_unknown_dependency_flagged(self):
|
||||
spec, ci = self._base()
|
||||
spec["phases"][0]["depends_on"] = ["nope"]
|
||||
errs = self._errors(spec, ci)
|
||||
self.assertTrue(any("dépendance inconnue" in e for e in errs), errs)
|
||||
|
||||
def test_confirmation_without_source_flagged(self):
|
||||
spec, ci = self._base()
|
||||
spec["confirmations"]["c1"]["source"] = ""
|
||||
errs = self._errors(spec, ci)
|
||||
self.assertTrue(any("source" in e.lower() for e in errs), errs)
|
||||
|
||||
def test_confirmation_without_owner_flagged(self):
|
||||
spec, ci = self._base()
|
||||
spec["confirmations"]["c1"]["owner"] = ""
|
||||
errs = self._errors(spec, ci)
|
||||
self.assertTrue(any("owner" in e for e in errs), errs)
|
||||
|
||||
def test_orphan_confirmation_flagged(self):
|
||||
spec, ci = self._base()
|
||||
spec["confirmations"]["c2"] = {"libelle": "L", "owner": "O", "source": "S"}
|
||||
errs = self._errors(spec, ci)
|
||||
self.assertTrue(any("orphelin" in e for e in errs), errs)
|
||||
|
||||
def test_unknown_confirmation_ref_flagged(self):
|
||||
spec, ci = self._base()
|
||||
spec["phases"][0]["confirmations"] = ["c1", "cX"]
|
||||
errs = self._errors(spec, ci)
|
||||
self.assertTrue(any("cX" in e or "inconnue" in e for e in errs), errs)
|
||||
|
||||
def test_numeric_value_in_confirmation_flagged(self):
|
||||
spec, ci = self._base()
|
||||
spec["confirmations"]["c1"]["taux"] = 3.0 # chiffre fabriqué interdit
|
||||
errs = self._errors(spec, ci)
|
||||
self.assertTrue(any("chiffrée interdite" in e for e in errs), errs)
|
||||
|
||||
def test_duplicate_ci_job_flagged(self):
|
||||
spec, ci = self._base()
|
||||
ci["job_to_path"]["a-one-bis-tests"] = "a/one" # 2 jobs → 1 module
|
||||
errs = self._errors(spec, ci)
|
||||
self.assertTrue(any("job(s) CI" in e for e in errs), errs)
|
||||
|
||||
def test_empty_phase_when_module_removed(self):
|
||||
# Retirer les modules d'une phase la rend vide → le schéma (minItems=1)
|
||||
# ou la partition doit rougir.
|
||||
spec, ci = self._base()
|
||||
spec["phases"].append(
|
||||
{"order": 2, "id": "p2", "titre": "P2", "responsable": "vps",
|
||||
"rationale": "r", "depends_on": ["p1"], "confirmations": []}
|
||||
)
|
||||
# p2 n'a aucun module assigné → phase vide
|
||||
errs = self._errors(spec, ci)
|
||||
self.assertTrue(errs, "une phase vide doit être signalée")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
Reference in New Issue
Block a user