#!/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)