Claude Code DTP Worker
71c1223cc3
Plan de commissions cross-cohérent workflow ↔ DocType ↔ RBAC + calcul traçable commission = base × taux (façon banclib/finance.py). Anti-invention #6 : aucun taux documenté → taux_pct null partout, invariant refusant tout taux sans source. Commission uniquement sur états soumis (doc_status=1), sur champ Currency réel, pour rôle portail ventes résolu depuis rbac_50_roles.json. - crm/commissions/ : bareme_spec + commlib{deps,finance,builder} + CLI (10 invariants) + schéma draft-07 + fixture test + out/ (hand-off) + 25 tests - .gitea/workflows/ci.yml : job crm-commissions-tests + ajout au gate - daily report session13 + activity log Vérifs : 25/25 tests · gate CI local vert · régression 202 tests verts. Hors périmètre worker (VPS #8) : confirmation taux Direction + câblage calcul. Auto-score 4Big : 96/100. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
264 lines
11 KiB
Python
264 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""Tests du générateur du barème de commissions vendeurs (Sprint 4 · ERPNext).
|
||
|
||
Stdlib pur (`unittest`) → aucune installation pip requise sur le runner Gitea.
|
||
La bibliothèque `jsonschema` sert d'*oracle* quand elle est présente, pour se
|
||
prémunir d'un écart entre le validateur maison et draft-07.
|
||
|
||
Deux axes :
|
||
1. CROSS-COHÉRENCE barème ↔ workflow ↔ DocType ↔ RBAC (les 10 invariants du
|
||
générateur : chaque évènement paie sur un état soumis, sur un champ Currency
|
||
réel, pour un rôle ventes résolu, et JAMAIS un taux sans source).
|
||
2. Calcul TRAÇABLE (commlib/finance.py) : commission = base × taux, formule
|
||
publiée, None si un opérande manque (anti 0-inventé · #6).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import unittest
|
||
|
||
_HERE = os.path.dirname(os.path.abspath(__file__))
|
||
_MODULE = os.path.normpath(os.path.join(_HERE, ".."))
|
||
_CRM = os.path.normpath(os.path.join(_MODULE, ".."))
|
||
_DELIVERABLES = os.path.normpath(os.path.join(_CRM, ".."))
|
||
|
||
sys.path.insert(0, _MODULE)
|
||
sys.path.insert(0, _CRM)
|
||
sys.path.insert(0, os.path.join(_DELIVERABLES, "publiciste"))
|
||
|
||
from commlib import builder, finance # noqa: E402
|
||
from workflow_vente.wflib.rbac import RoleResolver # noqa: E402
|
||
from lib import validator as maison # type: ignore # noqa: E402
|
||
import commissions_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 BuildBaseline(unittest.TestCase):
|
||
"""Le barème vanille passe schéma + 10 invariants et est déterministe."""
|
||
|
||
def setUp(self):
|
||
self.bundle, self.spec, self.wf, self.dt, self.resolver = gen._build()
|
||
|
||
def test_validate_clean(self):
|
||
errors = gen._validate(self.bundle, self.spec, self.wf, self.dt, self.resolver)
|
||
self.assertEqual(errors, [], f"invariants cassés : {errors}")
|
||
|
||
def test_schema_maison(self):
|
||
schema = _load(gen._SCHEMA_PATH)
|
||
self.assertEqual(list(maison.validate(self.bundle, schema)), [])
|
||
|
||
@unittest.skipUnless(_HAS_JSONSCHEMA, "jsonschema absent (oracle optionnel)")
|
||
def test_schema_oracle(self):
|
||
schema = _load(gen._SCHEMA_PATH)
|
||
jsonschema.validate(self.bundle, schema) # lève si non conforme
|
||
|
||
def test_deterministe(self):
|
||
b2 = builder.build_bundle(self.spec, self.resolver)
|
||
self.assertEqual(
|
||
json.dumps(self.bundle, sort_keys=True, ensure_ascii=False),
|
||
json.dumps(b2, sort_keys=True, ensure_ascii=False),
|
||
)
|
||
|
||
def test_events_sorted(self):
|
||
evs = self.bundle["commission_plan"]["evenements"]
|
||
keys = [(e["update_value"], e["role_id"]) for e in evs]
|
||
self.assertEqual(keys, sorted(keys))
|
||
|
||
def test_counts(self):
|
||
m = self.bundle["manifest"]
|
||
evs = self.bundle["commission_plan"]["evenements"]
|
||
self.assertEqual(m["counts"]["evenements"], len(evs))
|
||
self.assertEqual(m["counts"]["roles"], len({e["role_id"] for e in evs}))
|
||
|
||
|
||
class AntiInvention(unittest.TestCase):
|
||
"""#6 : le barème livré ne fixe AUCUN taux, et aucun taux ne passe sans source."""
|
||
|
||
def setUp(self):
|
||
self.bundle, self.spec, self.wf, self.dt, self.resolver = gen._build()
|
||
|
||
def test_all_taux_null_in_shipped_spec(self):
|
||
for ev in self.bundle["commission_plan"]["evenements"]:
|
||
self.assertIsNone(ev["taux_pct"], f"taux fabriqué : {ev}")
|
||
self.assertTrue(ev["a_confirmer"])
|
||
self.assertEqual(
|
||
self.bundle["manifest"]["counts"]["taux_a_confirmer"],
|
||
len(self.bundle["commission_plan"]["evenements"]),
|
||
)
|
||
|
||
def test_taux_without_source_rejected(self):
|
||
spec = copy.deepcopy(self.spec)
|
||
spec["evenements"][0]["taux_pct"] = 3.0
|
||
spec["evenements"][0]["a_confirmer"] = False
|
||
spec["evenements"][0]["source"] = None
|
||
bundle = builder.build_bundle(spec, self.resolver)
|
||
errors = gen._validate(bundle, spec, self.wf, self.dt, self.resolver)
|
||
self.assertTrue(any("sans `source`" in e for e in errors), errors)
|
||
|
||
def test_taux_with_source_accepted(self):
|
||
spec = copy.deepcopy(self.spec)
|
||
spec["evenements"][0]["taux_pct"] = 3.0
|
||
spec["evenements"][0]["a_confirmer"] = False
|
||
spec["evenements"][0]["source"] = "Note Direction 2026 (fictive test)"
|
||
bundle = builder.build_bundle(spec, self.resolver)
|
||
errors = gen._validate(bundle, spec, self.wf, self.dt, self.resolver)
|
||
self.assertEqual(errors, [], errors)
|
||
|
||
|
||
class CrossCoherence(unittest.TestCase):
|
||
"""Les évènements référencent workflow + DocType + RBAC — pas d'invention."""
|
||
|
||
def setUp(self):
|
||
self.bundle, self.spec, self.wf, self.dt, self.resolver = gen._build()
|
||
|
||
def test_only_submitted_states_pay(self):
|
||
submitted = {s["update_value"] for s in self.wf["states"] if s["doc_status"] == "1"}
|
||
for ev in self.bundle["commission_plan"]["evenements"]:
|
||
self.assertIn(ev["update_value"], submitted,
|
||
f"commission sur état non soumis : {ev['update_value']}")
|
||
|
||
def test_draft_state_rejected(self):
|
||
spec = copy.deepcopy(self.spec)
|
||
# `lead` est un état brouillon (doc_status 0) : interdit de commissionner.
|
||
spec["evenements"][0]["update_value"] = "lead"
|
||
bundle = builder.build_bundle(spec, self.resolver)
|
||
errors = gen._validate(bundle, spec, self.wf, self.dt, self.resolver)
|
||
self.assertTrue(any("brouillon" in e for e in errors), errors)
|
||
|
||
def test_unknown_update_value_rejected(self):
|
||
spec = copy.deepcopy(self.spec)
|
||
spec["evenements"][0]["update_value"] = "inexistant"
|
||
bundle = builder.build_bundle(spec, self.resolver)
|
||
errors = gen._validate(bundle, spec, self.wf, self.dt, self.resolver)
|
||
self.assertTrue(any("absent du workflow" in e for e in errors), errors)
|
||
|
||
def test_base_field_must_be_currency(self):
|
||
spec = copy.deepcopy(self.spec)
|
||
spec["evenements"][0]["base_field"] = "prospect" # Link, pas Currency
|
||
bundle = builder.build_bundle(spec, self.resolver)
|
||
errors = gen._validate(bundle, spec, self.wf, self.dt, self.resolver)
|
||
self.assertTrue(any("Currency" in e for e in errors), errors)
|
||
|
||
def test_base_fields_exist_in_doctype(self):
|
||
currency = gen._currency_fields(self.dt)
|
||
for ev in self.bundle["commission_plan"]["evenements"]:
|
||
self.assertIn(ev["base_field"], currency)
|
||
|
||
def test_role_must_be_ventes(self):
|
||
spec = copy.deepcopy(self.spec)
|
||
spec["evenements"][0]["role_id"] = "direction-cco" # portail direction
|
||
bundle = builder.build_bundle(spec, self.resolver)
|
||
errors = gen._validate(bundle, spec, self.wf, self.dt, self.resolver)
|
||
self.assertTrue(any("portail ventes" in e for e in errors), errors)
|
||
|
||
def test_unknown_role_raises(self):
|
||
spec = copy.deepcopy(self.spec)
|
||
spec["evenements"][0]["role_id"] = "role-fantome"
|
||
with self.assertRaises(KeyError):
|
||
builder.build_bundle(spec, self.resolver)
|
||
|
||
def test_roles_resolved_from_rbac(self):
|
||
for ev in self.bundle["commission_plan"]["evenements"]:
|
||
self.assertEqual(ev["erpnext_role_name"],
|
||
self.resolver.erpnext_name(ev["role_id"]))
|
||
|
||
def test_duplicate_event_rejected(self):
|
||
spec = copy.deepcopy(self.spec)
|
||
spec["evenements"].append(copy.deepcopy(spec["evenements"][0]))
|
||
bundle = builder.build_bundle(spec, self.resolver)
|
||
errors = gen._validate(bundle, spec, self.wf, self.dt, self.resolver)
|
||
self.assertTrue(any("dupliqué" in e for e in errors), errors)
|
||
|
||
|
||
class TraceableCalc(unittest.TestCase):
|
||
"""commlib/finance.py : commission = base × taux, traçable, None si opérande manque."""
|
||
|
||
def setUp(self):
|
||
self.spec = _load(gen._SPEC_PATH)
|
||
fx = _load(os.path.join(_MODULE, "fixtures", "dossier_exemple.json"))
|
||
self.dossier = fx["dossier"]
|
||
self.taux = fx["taux_exemple"]["taux_pct"]
|
||
|
||
def test_rate_parsing(self):
|
||
self.assertEqual(finance.rate(2.5), 0.025)
|
||
self.assertEqual(finance.rate("2.5 %"), 0.025)
|
||
self.assertEqual(finance.rate("3,0%"), 0.03)
|
||
self.assertIsNone(finance.rate(None))
|
||
self.assertIsNone(finance.rate(""))
|
||
self.assertIsNone(finance.rate(True))
|
||
|
||
def test_montant_calcule(self):
|
||
ev = {"update_value": "contrat", "role_id": "ventes-conseiller",
|
||
"base_field": "montant_contrat", "taux_pct": self.taux}
|
||
line = finance.compute_line(self.dossier, ev)
|
||
# 200000 × 2.5 % = 5000
|
||
self.assertEqual(line["montant"], 5000.0)
|
||
self.assertFalse(line["incomplete"])
|
||
self.assertIn("200000", line["formule"])
|
||
self.assertIn("2.5", line["formule"])
|
||
self.assertEqual(line["devise"], "USD")
|
||
|
||
def test_none_si_taux_absent(self):
|
||
ev = {"update_value": "contrat", "role_id": "ventes-conseiller",
|
||
"base_field": "montant_contrat", "taux_pct": None}
|
||
line = finance.compute_line(self.dossier, ev)
|
||
self.assertIsNone(line["montant"])
|
||
self.assertTrue(line["incomplete"])
|
||
self.assertIn("taux_pct", line["champs_manquants"])
|
||
# La formule reste affichée même sans valeur.
|
||
self.assertIn("200000", line["formule"])
|
||
|
||
def test_none_si_base_absente(self):
|
||
ev = {"update_value": "contrat", "role_id": "ventes-conseiller",
|
||
"base_field": "montant_contrat", "taux_pct": self.taux}
|
||
line = finance.compute_line({"devise": "USD"}, ev)
|
||
self.assertIsNone(line["montant"])
|
||
self.assertIn("montant_contrat", line["champs_manquants"])
|
||
self.assertIn("{montant_contrat}", line["formule"])
|
||
|
||
def test_shipped_bareme_yields_no_amount(self):
|
||
# Le barème livré (taux null) ne calcule aucun montant — c'est voulu (#6).
|
||
lines = finance.compute_dossier(self.dossier, self.spec)
|
||
self.assertTrue(all(l["montant"] is None for l in lines))
|
||
self.assertTrue(all(l["incomplete"] for l in lines))
|
||
|
||
|
||
class CliSmoke(unittest.TestCase):
|
||
"""Le CLI build/validate tourne et out/ committé == régénération."""
|
||
|
||
def test_validate_cli(self):
|
||
rc = gen.main(["validate"])
|
||
self.assertEqual(rc, 0)
|
||
|
||
def test_build_matches_committed(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
rc = gen.main(["build", "-o", tmp])
|
||
self.assertEqual(rc, 0)
|
||
for name in ("commission_plan.json", "MANIFEST.json"):
|
||
fresh = _load(os.path.join(tmp, name))
|
||
committed_path = os.path.join(_MODULE, "out", name)
|
||
if os.path.exists(committed_path):
|
||
self.assertEqual(fresh, _load(committed_path),
|
||
f"{name} committé ≠ régénération")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|