[DTP-Worker] Sprint 2 · Agrégateur RBAC : run-book d'application VPS unifié (3 volets → 1 plan ordonné SPEC §7)
Clôt le volet RBAC en-repo : recoud fixtures Role+DocPerm, plan User Permission et Role Profile en un run-book ordonné + manifeste agrégé. Zéro chiffre recalculé (tout vient d'un manifeste source, #6), graphe de dépendances validé (Role avant Role Profile), cohérence inter-volets + couverture bijective 50/50. 16 tests + job CI rbac-applyplan-tests · 99 tests de régression au total. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Code DTP Worker
@@ -0,0 +1,182 @@
|
||||
"""Tests de l'agrégateur RBAC (run-book d'application VPS unifié).
|
||||
|
||||
Stdlib pur (`unittest`) → aucune install pip requise sur le runner Gitea. La
|
||||
bibliothèque `jsonschema`, si présente, sert d'oracle draft-07 en complément du
|
||||
validateur maison Publiciste (parité). Sinon elle est ignorée.
|
||||
|
||||
Couverture : schéma (maison + oracle), fidélité des comptes au manifeste de
|
||||
chaque builder source (anti-invention #6), fusion des confirmations VPS, ordre
|
||||
et graphe de dépendances du run-book (SPEC §7), garde-fou de cohérence
|
||||
inter-volets, déterminisme.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
_HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
_MODULE_DIR = os.path.normpath(os.path.join(_HERE, "..")) # apply_plan/
|
||||
_RBAC_DIR = os.path.normpath(os.path.join(_MODULE_DIR, "..")) # rbac/
|
||||
_DELIVERABLES = os.path.normpath(os.path.join(_RBAC_DIR, "..")) # 05_deliverables_mvp/
|
||||
|
||||
for _p in (
|
||||
_MODULE_DIR,
|
||||
os.path.join(_RBAC_DIR, "fixtures_gen"),
|
||||
os.path.join(_RBAC_DIR, "userperm_gen"),
|
||||
os.path.join(_RBAC_DIR, "roleprofile_gen"),
|
||||
os.path.join(_DELIVERABLES, "publiciste"),
|
||||
):
|
||||
if _p not in sys.path:
|
||||
sys.path.insert(0, _p)
|
||||
|
||||
import rbac_apply_plan as cli # noqa: E402
|
||||
from applylib import aggregator # noqa: E402
|
||||
from fixturelib import builder as fixtures_builder # noqa: E402
|
||||
from permlib import builder as userperm_builder # noqa: E402
|
||||
from profilelib import builder as roleprofile_builder # noqa: E402
|
||||
from lib import validator as maison # noqa: E402
|
||||
|
||||
try:
|
||||
import jsonschema # type: ignore
|
||||
|
||||
_HAS_JSONSCHEMA = True
|
||||
except ImportError: # pragma: no cover
|
||||
_HAS_JSONSCHEMA = False
|
||||
|
||||
|
||||
def _load(path: str) -> dict:
|
||||
with open(path, encoding="utf-8") as fh:
|
||||
return json.load(fh)
|
||||
|
||||
|
||||
class ApplyPlanTest(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls) -> None:
|
||||
cls.contract = _load(cli._CONTRACT_PATH)
|
||||
cls.fixtures = fixtures_builder.build_bundle(cls.contract)
|
||||
cls.userperm = userperm_builder.build_plan(cls.contract)
|
||||
cls.roleprofile = roleprofile_builder.build_bundle(cls.contract)
|
||||
cls.plan = aggregator.build_apply_plan(
|
||||
cls.contract, cls.fixtures, cls.userperm, cls.roleprofile
|
||||
)
|
||||
cls.schema = _load(cli._SCHEMA_PATH)
|
||||
|
||||
# --- schéma ------------------------------------------------------------- #
|
||||
def test_schema_maison(self) -> None:
|
||||
errors = maison.validate(self.plan, self.schema)
|
||||
self.assertEqual(errors, [], f"validateur maison: {errors}")
|
||||
|
||||
@unittest.skipUnless(_HAS_JSONSCHEMA, "jsonschema absent (oracle optionnel)")
|
||||
def test_schema_oracle_jsonschema(self) -> None:
|
||||
jsonschema.validate(self.plan, self.schema)
|
||||
|
||||
def test_cli_validate_no_errors(self) -> None:
|
||||
self.assertEqual(cli._validate_plan(self.plan), [])
|
||||
|
||||
# --- fidélité des comptes (anti-invention #6) --------------------------- #
|
||||
def test_counts_come_from_source_manifests(self) -> None:
|
||||
c = self.plan["manifest"]["counts"]
|
||||
self.assertEqual(c["roles"], self.fixtures["manifest"]["counts"]["role"])
|
||||
self.assertEqual(c["custom_docperm"], self.fixtures["manifest"]["counts"]["custom_docperm"])
|
||||
self.assertEqual(c["doctypes_uniques"], self.fixtures["manifest"]["counts"]["doctypes_uniques"])
|
||||
self.assertEqual(
|
||||
c["user_permission_templates"],
|
||||
self.userperm["manifest"]["counts"]["user_permission_templates"],
|
||||
)
|
||||
self.assertEqual(c["role_profiles"], self.roleprofile["manifest"]["counts"]["role_profiles"])
|
||||
|
||||
def test_confirmations_merged_from_sources(self) -> None:
|
||||
conf = self.plan["manifest"]["confirmations_vps"]
|
||||
self.assertEqual(
|
||||
conf["custom_doctypes"],
|
||||
sorted(self.fixtures["manifest"]["custom_doctypes_a_confirmer"]),
|
||||
)
|
||||
self.assertEqual(
|
||||
conf["companies"], sorted(self.userperm["manifest"]["companies_a_confirmer"])
|
||||
)
|
||||
self.assertEqual(
|
||||
conf["roles_scope_equipe"],
|
||||
sorted(self.userperm["manifest"]["roles_scope_equipe_a_confirmer"]),
|
||||
)
|
||||
|
||||
# --- run-book (SPEC §7) ------------------------------------------------- #
|
||||
def test_steps_match_spec7_order(self) -> None:
|
||||
steps = self.plan["apply_plan"]
|
||||
self.assertEqual([s["id"] for s in steps], aggregator.step_ids())
|
||||
self.assertEqual([s["order"] for s in steps], list(range(1, len(steps) + 1)))
|
||||
|
||||
def test_dependency_graph_no_forward_edges(self) -> None:
|
||||
steps = self.plan["apply_plan"]
|
||||
order = {s["id"]: s["order"] for s in steps}
|
||||
for s in steps:
|
||||
for dep in s["depends_on"]:
|
||||
self.assertIn(dep, order)
|
||||
self.assertLess(order[dep], s["order"], f"{s['id']} dépend en avant de {dep}")
|
||||
|
||||
def test_roleprofile_after_role_fixtures(self) -> None:
|
||||
rp = next(s for s in self.plan["apply_plan"] if s["id"] == "roleprofile-apply")
|
||||
self.assertIn("fixtures-migrate", rp["depends_on"])
|
||||
|
||||
def test_every_confirmation_referenced(self) -> None:
|
||||
conf = self.plan["manifest"]["confirmations_vps"]
|
||||
referenced = {k for s in self.plan["apply_plan"] for k in s["confirmations"]}
|
||||
for key, items in conf.items():
|
||||
if items:
|
||||
self.assertIn(key, referenced, f"{key} non vide mais orpheline")
|
||||
|
||||
# --- cohérence inter-volets --------------------------------------------- #
|
||||
def test_consistency_bijective(self) -> None:
|
||||
cons = self.plan["manifest"]["consistency"]
|
||||
self.assertTrue(cons["sources_coherentes"])
|
||||
self.assertTrue(cons["couverture_bijective"])
|
||||
self.assertEqual(cons["roles_fixtures"], 50)
|
||||
self.assertEqual(cons["userperm_plan_entries"], 50)
|
||||
self.assertEqual(cons["roleprofile_roles_couverts"], 50)
|
||||
|
||||
def test_rejects_version_mismatch(self) -> None:
|
||||
"""Un volet dérivé d'un contrat de version différente → refus."""
|
||||
forged = json.loads(json.dumps(self.userperm))
|
||||
forged["manifest"]["source_version"] = "9.9.9"
|
||||
with self.assertRaises(ValueError):
|
||||
aggregator.build_apply_plan(self.contract, self.fixtures, forged, self.roleprofile)
|
||||
|
||||
def test_rejects_cible_mismatch(self) -> None:
|
||||
forged = json.loads(json.dumps(self.roleprofile))
|
||||
forged["manifest"]["cible_rbac_roles"] = 49
|
||||
with self.assertRaises(ValueError):
|
||||
aggregator.build_apply_plan(self.contract, self.fixtures, self.userperm, forged)
|
||||
|
||||
# --- déterminisme ------------------------------------------------------- #
|
||||
def test_deterministic(self) -> None:
|
||||
again = aggregator.build_apply_plan(
|
||||
self.contract, self.fixtures, self.userperm, self.roleprofile
|
||||
)
|
||||
self.assertEqual(
|
||||
json.dumps(self.plan, sort_keys=True, ensure_ascii=False),
|
||||
json.dumps(again, sort_keys=True, ensure_ascii=False),
|
||||
)
|
||||
|
||||
def test_confirmations_sorted(self) -> None:
|
||||
conf = self.plan["manifest"]["confirmations_vps"]
|
||||
for key, items in conf.items():
|
||||
self.assertEqual(items, sorted(items), f"{key} non trié")
|
||||
|
||||
# --- invariant-breaking detection --------------------------------------- #
|
||||
def test_validate_catches_broken_order(self) -> None:
|
||||
broken = json.loads(json.dumps(self.plan))
|
||||
broken["apply_plan"][0]["order"] = 99
|
||||
self.assertTrue(cli._validate_plan(broken))
|
||||
|
||||
def test_validate_catches_orphan_confirmation(self) -> None:
|
||||
broken = json.loads(json.dumps(self.plan))
|
||||
# Vide toutes les références de confirmation des étapes → items orphelins.
|
||||
for s in broken["apply_plan"]:
|
||||
s["confirmations"] = []
|
||||
self.assertTrue(cli._validate_plan(broken))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user