[DTP-Worker] Sprint 4 · Générateur workflow vente ERPNext (lead → CONFOTUR)
Contrat pipeline commercial CRM natif (lead → visite → devis → réservation → contrat → CONFOTUR) → fixtures Frappe/ERPNext v15 : Workflow (9 états / 11 transitions) + Workflow State + Workflow Action Master + MANIFEST. Rôles gardant états/transitions résolus depuis rbac_50_roles.json (réutilisation, zéro duplication · #6) : le contrat ne cite qu'un id de rôle, jamais un nom Frappe en dur. CLI build/validate refuse d'écrire si l'un des 9 invariants de graphe casse (monotonie doc_status, atteignabilité, séparation des pouvoirs sur les étapes engageant de l'argent / clôturant). 25 tests (stdlib pur + oracle jsonschema) · job CI crm-workflow-vente-tests ajouté au gate · 146 tests de régression verts au total. Application VPS (DocType porteur OTO Dossier Vente + bench migrate) = agent ERPNext Backend, hors périmètre worker. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Code DTP Worker
@@ -0,0 +1,272 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Tests du générateur de fixtures ERPNext Workflow (pipeline vente · Sprint 4).
|
||||
|
||||
Stdlib pur (`unittest`) → aucune installation pip requise sur le runner Gitea.
|
||||
La bibliothèque `jsonschema` est utilisée comme *oracle* quand elle est présente,
|
||||
pour se prémunir d'un écart entre le validateur maison et draft-07.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
_HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
_MODULE = os.path.normpath(os.path.join(_HERE, ".."))
|
||||
_DELIVERABLES = os.path.normpath(os.path.join(_MODULE, "..", ".."))
|
||||
|
||||
sys.path.insert(0, _MODULE)
|
||||
sys.path.insert(0, os.path.join(_DELIVERABLES, "publiciste"))
|
||||
|
||||
from wflib import builder, erpnext # noqa: E402
|
||||
from wflib.rbac import RoleResolver # noqa: E402
|
||||
from lib import validator as maison # type: ignore # noqa: E402
|
||||
import workflow_vente_gen as gen # noqa: E402
|
||||
|
||||
try:
|
||||
import jsonschema # type: ignore
|
||||
|
||||
_HAS_JSONSCHEMA = True
|
||||
except Exception: # pragma: no cover
|
||||
_HAS_JSONSCHEMA = False
|
||||
|
||||
|
||||
def _load(path: str) -> dict:
|
||||
with open(path, encoding="utf-8") as fh:
|
||||
return json.load(fh)
|
||||
|
||||
|
||||
class BaseFixture(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls) -> None:
|
||||
cls.spec = _load(gen._SPEC_PATH)
|
||||
cls.schema = _load(gen._SCHEMA_PATH)
|
||||
cls.resolver = RoleResolver.from_path()
|
||||
cls.bundle = builder.build_bundle(cls.spec, cls.resolver)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Résolution des rôles RBAC (réutilisation · zéro invention · #6)
|
||||
# --------------------------------------------------------------------------- #
|
||||
class TestRoleResolver(BaseFixture):
|
||||
def test_resolves_known_id(self) -> None:
|
||||
self.assertEqual(
|
||||
self.resolver.erpnext_name("ventes-conseiller"), "OTO Ventes Conseiller"
|
||||
)
|
||||
|
||||
def test_unknown_id_raises(self) -> None:
|
||||
with self.assertRaises(KeyError):
|
||||
self.resolver.erpnext_name("role-inexistant")
|
||||
|
||||
def test_every_spec_role_id_exists_in_rbac(self) -> None:
|
||||
ids = {s["role_id"] for s in self.spec["states"]}
|
||||
ids |= {t["role_id"] for t in self.spec["transitions"]}
|
||||
for rid in ids:
|
||||
self.assertIn(rid, self.resolver.known_ids(), rid)
|
||||
|
||||
def test_all_resolved_roles_are_oto_prefixed(self) -> None:
|
||||
for entry in self.bundle["manifest"]["roles_rbac_utilises"]:
|
||||
self.assertTrue(entry["erpnext_role_name"].startswith("OTO "))
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Structure du bundle / conformité Frappe
|
||||
# --------------------------------------------------------------------------- #
|
||||
class TestBundleStructure(BaseFixture):
|
||||
def test_workflow_targets_pipeline_doctype(self) -> None:
|
||||
wf = self.bundle["workflow"]
|
||||
self.assertEqual(wf["doctype"], "Workflow")
|
||||
self.assertEqual(wf["document_type"], "OTO Dossier Vente")
|
||||
self.assertEqual(wf["workflow_state_field"], "workflow_state")
|
||||
|
||||
def test_pipeline_covers_the_six_roadmap_stages(self) -> None:
|
||||
# roadmap Sprint 4 : lead -> visite -> devis -> reservation -> contrat -> CONFOTUR
|
||||
states = {s["state"] for s in self.bundle["workflow"]["states"]}
|
||||
for expected in [
|
||||
"Lead",
|
||||
"Visite planifiée",
|
||||
"Devis émis",
|
||||
"Réservation confirmée",
|
||||
"Contrat signé",
|
||||
"CONFOTUR déposé",
|
||||
]:
|
||||
self.assertIn(expected, states, expected)
|
||||
|
||||
def test_masters_cover_states_and_actions_exactly(self) -> None:
|
||||
states = {s["state"] for s in self.bundle["workflow"]["states"]}
|
||||
masters = {m["name"] for m in self.bundle["workflow_state"]}
|
||||
self.assertEqual(states, masters)
|
||||
actions = {t["action"] for t in self.bundle["workflow"]["transitions"]}
|
||||
act_masters = {m["name"] for m in self.bundle["workflow_action_master"]}
|
||||
self.assertEqual(actions, act_masters)
|
||||
|
||||
def test_styles_are_native_frappe(self) -> None:
|
||||
for m in self.bundle["workflow_state"]:
|
||||
self.assertIn(m["style"], erpnext.VALID_STYLES)
|
||||
|
||||
def test_manifest_flags_custom_carrier_doctype(self) -> None:
|
||||
self.assertEqual(
|
||||
self.bundle["manifest"]["custom_doctypes_a_confirmer"], ["OTO Dossier Vente"]
|
||||
)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Invariants métier du graphe (le CLI refuse d'écrire s'ils cassent)
|
||||
# --------------------------------------------------------------------------- #
|
||||
class TestGraphInvariants(BaseFixture):
|
||||
def test_clean_bundle_has_no_errors(self) -> None:
|
||||
self.assertEqual(gen._validate_bundle(self.spec, self.bundle), [])
|
||||
|
||||
def test_docstatus_monotonic_no_illegal_jump(self) -> None:
|
||||
docstatus = {s["state"]: s["doc_status"] for s in self.bundle["workflow"]["states"]}
|
||||
for t in self.bundle["workflow"]["transitions"]:
|
||||
pair = (docstatus[t["state"]], docstatus[t["next_state"]])
|
||||
self.assertIn(pair, gen._ALLOWED_DOCSTATUS_STEPS, f"{t['state']}→{t['next_state']}")
|
||||
|
||||
def test_detects_unreachable_state(self) -> None:
|
||||
spec = copy.deepcopy(self.spec)
|
||||
spec["states"].append(
|
||||
{
|
||||
"state": "Orphelin",
|
||||
"doc_status": "0",
|
||||
"role_id": "ventes-conseiller",
|
||||
"style": "Info",
|
||||
"update_value": "orphelin",
|
||||
"is_terminal": True,
|
||||
"message": "État jamais atteint.",
|
||||
}
|
||||
)
|
||||
bundle = builder.build_bundle(spec, self.resolver)
|
||||
errors = gen._validate_bundle(spec, bundle)
|
||||
self.assertTrue(any("inatteignable" in e for e in errors), errors)
|
||||
|
||||
def test_detects_illegal_docstatus_jump(self) -> None:
|
||||
# Lead(0) -> Perdu(2) : saut 0→2 interdit.
|
||||
spec = copy.deepcopy(self.spec)
|
||||
spec["transitions"].append(
|
||||
{
|
||||
"state": "Lead",
|
||||
"action": "Sauter à perdu",
|
||||
"next_state": "Perdu",
|
||||
"role_id": "ventes-directeur",
|
||||
"allow_self_approval": 1,
|
||||
}
|
||||
)
|
||||
bundle = builder.build_bundle(spec, self.resolver)
|
||||
errors = gen._validate_bundle(spec, bundle)
|
||||
self.assertTrue(any("doc_status illégal" in e for e in errors), errors)
|
||||
|
||||
def test_detects_duplicate_action_from_same_state(self) -> None:
|
||||
spec = copy.deepcopy(self.spec)
|
||||
spec["transitions"].append(
|
||||
{
|
||||
"state": "Lead",
|
||||
"action": "Planifier visite", # doublon (état, action)
|
||||
"next_state": "Abandonné",
|
||||
"role_id": "ventes-conseiller",
|
||||
"allow_self_approval": 1,
|
||||
}
|
||||
)
|
||||
bundle = builder.build_bundle(spec, self.resolver)
|
||||
errors = gen._validate_bundle(spec, bundle)
|
||||
self.assertTrue(any("dupliqué" in e for e in errors), errors)
|
||||
|
||||
def test_detects_separation_of_duties_violation(self) -> None:
|
||||
spec = copy.deepcopy(self.spec)
|
||||
for t in spec["transitions"]:
|
||||
if t.get("separation_of_duties"):
|
||||
t["allow_self_approval"] = 1 # viole la règle
|
||||
break
|
||||
bundle = builder.build_bundle(spec, self.resolver)
|
||||
errors = gen._validate_bundle(spec, bundle)
|
||||
self.assertTrue(any("Séparation des pouvoirs" in e for e in errors), errors)
|
||||
|
||||
def test_reservation_confirm_forbids_self_approval(self) -> None:
|
||||
# Étape argent (devis → réservation) : quatre-yeux obligatoire.
|
||||
tr = next(
|
||||
t for t in self.bundle["workflow"]["transitions"]
|
||||
if t["action"] == "Confirmer réservation"
|
||||
)
|
||||
self.assertEqual(tr["allow_self_approval"], 0)
|
||||
|
||||
def test_confotur_approval_forbids_self_approval(self) -> None:
|
||||
tr = next(
|
||||
t for t in self.bundle["workflow"]["transitions"]
|
||||
if t["action"] == "Approuver CONFOTUR"
|
||||
)
|
||||
self.assertEqual(tr["allow_self_approval"], 0)
|
||||
|
||||
def test_has_submitted_terminal_success_state(self) -> None:
|
||||
outgoing = {t["state"] for t in self.bundle["workflow"]["transitions"]}
|
||||
terminal_success = [
|
||||
s["state"] for s in self.bundle["workflow"]["states"]
|
||||
if s["doc_status"] == "1" and s["state"] not in outgoing
|
||||
]
|
||||
self.assertIn("CONFOTUR approuvé", terminal_success)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Schéma de sortie (validateur maison + oracle jsonschema)
|
||||
# --------------------------------------------------------------------------- #
|
||||
class TestSchema(BaseFixture):
|
||||
def test_bundle_matches_schema_maison(self) -> None:
|
||||
self.assertEqual(list(maison.validate(self.bundle, self.schema)), [])
|
||||
|
||||
@unittest.skipUnless(_HAS_JSONSCHEMA, "jsonschema non installé (oracle facultatif)")
|
||||
def test_bundle_matches_schema_oracle(self) -> None:
|
||||
jsonschema.validate(self.bundle, self.schema) # type: ignore
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Déterminisme + CLI (build/validate)
|
||||
# --------------------------------------------------------------------------- #
|
||||
class TestDeterminismAndCLI(BaseFixture):
|
||||
def test_build_is_deterministic(self) -> None:
|
||||
a = builder.build_bundle(self.spec, self.resolver)
|
||||
b = builder.build_bundle(self.spec, self.resolver)
|
||||
self.assertEqual(
|
||||
json.dumps(a, ensure_ascii=False, sort_keys=True),
|
||||
json.dumps(b, ensure_ascii=False, sort_keys=True),
|
||||
)
|
||||
|
||||
def test_transitions_sorted_stable(self) -> None:
|
||||
trans = self.bundle["workflow"]["transitions"]
|
||||
keys = [(t["state"], t["action"]) for t in trans]
|
||||
self.assertEqual(keys, sorted(keys))
|
||||
|
||||
def test_cli_validate_exit_zero(self) -> None:
|
||||
rc = gen.main(["validate"])
|
||||
self.assertEqual(rc, 0)
|
||||
|
||||
def test_cli_build_writes_four_files(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
rc = gen.main(["build", "-o", tmp])
|
||||
self.assertEqual(rc, 0)
|
||||
for f in [
|
||||
"workflow.json",
|
||||
"workflow_state.json",
|
||||
"workflow_action_master.json",
|
||||
"MANIFEST.json",
|
||||
]:
|
||||
self.assertTrue(os.path.exists(os.path.join(tmp, f)), f)
|
||||
wf = _load(os.path.join(tmp, "workflow.json"))
|
||||
self.assertEqual(wf[0]["doctype"], "Workflow")
|
||||
|
||||
def test_committed_out_matches_regeneration(self) -> None:
|
||||
# La sortie commitée dans out/ doit correspondre bit-à-bit à une
|
||||
# régénération (garantie « diffable + re-générable » · anti-dérive).
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
gen.main(["build", "-o", tmp])
|
||||
for f in ["workflow.json", "workflow_state.json", "workflow_action_master.json", "MANIFEST.json"]:
|
||||
committed = os.path.join(_MODULE, "out", f)
|
||||
if os.path.exists(committed):
|
||||
self.assertEqual(_load(committed), _load(os.path.join(tmp, f)), f)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
Reference in New Issue
Block a user