[DTP-Worker 20260803_090713] Auto exec · session 20260803_090713

05_deliverables_mvp/crm/financement_bancaire/tests/test_financement_bancaire.py

@@ -0,0 +1,312 @@
+"""Tests du module Financement Bancaire — cœur métier (gate) + générateur.
+
+Deux familles :
+  1. Cœur métier `finlib.gate` — calcul d'apport 20/30 + gate check 4 conditions.
+     Fonctions pures → scénarios exhaustifs (chaque condition qui bloque, cas
+     nominal qui passe, frontière apport).
+  2. Générateur — validate/build, déterminisme (build == build), schéma +
+     12 invariants, résolution RBAC réelle.
+"""
+
+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 finlib import gate  # noqa: E402
+from finlib.builder import build_bundle  # noqa: E402
+from finlib.rbac import RoleResolver  # noqa: E402
+
+_SPEC_PATH = os.path.join(_MODULE, "financement_spec.json")
+_GEN = os.path.join(_MODULE, "financement_bancaire_gen.py")
+
+
+def _load_spec():
+    with open(_SPEC_PATH, encoding="utf-8") as fh:
+        return json.load(fh)
+
+
+class ApportTest(unittest.TestCase):
+    def setUp(self):
+        self.cfg = gate.GateConfig.from_spec(_load_spec())
+
+    def test_taux_resident_20_etranger_30(self):
+        self.assertEqual(self.cfg.taux_apport_pct["resident_rd"], 20.0)
+        self.assertEqual(self.cfg.taux_apport_pct["etranger"], 30.0)
+
+    def test_apport_requis_resident(self):
+        self.assertAlmostEqual(gate.apport_requis_usd(100000, "resident_rd", self.cfg), 20000.0)
+
+    def test_apport_requis_etranger(self):
+        self.assertAlmostEqual(gate.apport_requis_usd(100000, "etranger", self.cfg), 30000.0)
+
+    def test_apport_requis_type_inconnu_leve(self):
+        with self.assertRaises(KeyError):
+            gate.apport_requis_usd(100000, "martien", self.cfg)
+
+    def test_apport_verse_somme_paiements(self):
+        dossier = {"paiements": [{"montant_verse_usd": 5000}, {"montant_verse_usd": 2500.5}]}
+        self.assertAlmostEqual(gate.apport_verse_usd(dossier), 7500.5)
+
+    def test_apport_verse_aucun_paiement(self):
+        self.assertEqual(gate.apport_verse_usd({}), 0.0)
+
+
+class RequiredDocumentsTest(unittest.TestCase):
+    def setUp(self):
+        self.cfg = gate.GateConfig.from_spec(_load_spec())
+
+    def test_etranger_exige_plus_que_resident(self):
+        res = set(gate.required_document_ids("resident_rd", self.cfg))
+        etr = set(gate.required_document_ids("etranger", self.cfg))
+        self.assertTrue(res < etr)  # sous-ensemble strict
+
+    def test_resident_ninclut_pas_docs_etrangers(self):
+        res = gate.required_document_ids("resident_rd", self.cfg)
+        self.assertNotIn("lettre_origine_fonds", res)
+        self.assertNotIn("passeport_2cotes", res)
+
+    def test_etranger_inclut_docs_etrangers(self):
+        etr = gate.required_document_ids("etranger", self.cfg)
+        self.assertIn("lettre_origine_fonds", etr)
+        self.assertIn("refs_bancaires_intl", etr)
+
+    def test_document_optionnel_exclu(self):
+        # titre_definitif is_required=false → jamais dans la liste des exigés.
+        self.assertNotIn("titre_definitif", gate.required_document_ids("etranger", self.cfg))
+
+
+def _dossier_complet(cfg, spec):
+    """Un dossier qui satisfait les 4 conditions (résident RD, 100 000 USD)."""
+    required = gate.required_document_ids("resident_rd", cfg)
+    return {
+        "id": "T-OK",
+        "prix_usd": 100000,
+        "residence_type": "resident_rd",
+        "banque_choisie": "banreservas",
+        "paiements": [{"montant_verse_usd": 20000}],  # 20 % pile
+        "documents": [{"id": d, "statut": "Validé WAG"} for d in required],
+        "autorisations": [{"id": a["id"], "signature_date": "2026-08-01"} for a in spec["autorisations"]],
+        "wag_validated_by": "ventes-conseiller:amelie",
+    }
+
+
+class GateCheckTest(unittest.TestCase):
+    def setUp(self):
+        self.spec = _load_spec()
+        self.cfg = gate.GateConfig.from_spec(self.spec)
+
+    def test_dossier_complet_soumissible(self):
+        ok, reasons = gate.can_submit_dossier(_dossier_complet(self.cfg, self.spec), self.cfg)
+        self.assertTrue(ok)
+        self.assertEqual(reasons, [])
+
+    def test_apport_pile_au_seuil_passe(self):
+        # Exactement 20 % → apport complet (frontière incluse).
+        d = _dossier_complet(self.cfg, self.spec)
+        status = gate.gate_status(d, self.cfg)
+        self.assertTrue(status["conditions"]["apport_initial_complet"]["ok"])
+        self.assertEqual(status["conditions"]["apport_initial_complet"]["percent"], 100.0)
+
+    def test_apport_manquant_bloque(self):
+        d = _dossier_complet(self.cfg, self.spec)
+        d["paiements"] = [{"montant_verse_usd": 19999}]
+        ok, reasons = gate.can_submit_dossier(d, self.cfg)
+        self.assertFalse(ok)
+        self.assertTrue(any("Apport initial incomplet" in r for r in reasons))
+        c = gate.gate_status(d, self.cfg)["conditions"]["apport_initial_complet"]
+        self.assertAlmostEqual(c["remaining"], 1.0)
+
+    def test_document_non_valide_bloque(self):
+        d = _dossier_complet(self.cfg, self.spec)
+        d["documents"][0]["statut"] = "Déposé"  # déposé mais pas validé WAG
+        ok, reasons = gate.can_submit_dossier(d, self.cfg)
+        self.assertFalse(ok)
+        self.assertTrue(any("Documents exigés incomplets" in r for r in reasons))
+
+    def test_document_envoye_banque_est_recevable(self):
+        d = _dossier_complet(self.cfg, self.spec)
+        d["documents"][0]["statut"] = "Envoyé banque"
+        ok, _ = gate.can_submit_dossier(d, self.cfg)
+        self.assertTrue(ok)
+
+    def test_document_manquant_compte_comme_non_depose(self):
+        d = _dossier_complet(self.cfg, self.spec)
+        d["documents"] = d["documents"][:-1]  # retire un requis
+        c = gate.gate_status(d, self.cfg)["conditions"]["documents_exiges"]
+        self.assertFalse(c["ok"])
+        self.assertEqual(c["validated"], c["total"] - 1)
+
+    def test_autorisation_non_signee_bloque(self):
+        d = _dossier_complet(self.cfg, self.spec)
+        d["autorisations"][0]["signature_date"] = None
+        ok, reasons = gate.can_submit_dossier(d, self.cfg)
+        self.assertFalse(ok)
+        self.assertTrue(any("Autorisations non signées" in r for r in reasons))
+
+    def test_validation_wag_manquante_bloque(self):
+        d = _dossier_complet(self.cfg, self.spec)
+        d["wag_validated_by"] = None
+        ok, reasons = gate.can_submit_dossier(d, self.cfg)
+        self.assertFalse(ok)
+        self.assertTrue(any("Validation WAG manquante" in r for r in reasons))
+
+    def test_quatre_conditions_toutes_ko(self):
+        # Dossier vide → 4 raisons, 0 condition complétée.
+        d = {"id": "T-VIDE", "prix_usd": 100000, "residence_type": "etranger"}
+        status = gate.gate_status(d, self.cfg)
+        self.assertFalse(status["can_submit"])
+        self.assertEqual(status["conditions_completed"], 0)
+        ok, reasons = gate.can_submit_dossier(d, self.cfg)
+        self.assertFalse(ok)
+        self.assertEqual(len(reasons), 4)
+
+    def test_gate_status_forme_endpoint(self):
+        # Forme conforme au contrat de l'endpoint gate-status.
+        d = _dossier_complet(self.cfg, self.spec)
+        status = gate.gate_status(d, self.cfg)
+        self.assertIn("can_submit", status)
+        self.assertEqual(set(status["conditions"]), set(gate.CONDITION_KEYS))
+        for k in gate.CONDITION_KEYS:
+            self.assertIn("percent", status["conditions"][k])
+
+    def test_reasons_ordre_canonique(self):
+        d = {"id": "T", "prix_usd": 100000, "residence_type": "etranger"}
+        _, reasons = gate.can_submit_dossier(d, self.cfg)
+        self.assertIn("Apport", reasons[0])
+        self.assertIn("Documents", reasons[1])
+        self.assertIn("Autorisations", reasons[2])
+        self.assertIn("Validation WAG", reasons[3])
+
+
+class BuilderTest(unittest.TestCase):
+    def setUp(self):
+        self.spec = _load_spec()
+        self.resolver = RoleResolver.from_path()
+        self.bundle = build_bundle(self.spec, self.resolver)
+
+    def test_bundle_cles(self):
+        for k in ("manifest", "banques", "sections", "documents", "autorisations", "gate_spec", "gate_status_example"):
+            self.assertIn(k, self.bundle)
+
+    def test_role_validation_resolu(self):
+        rv = self.bundle["gate_spec"]["role_validation"]
+        self.assertEqual(rv["role_id"], "ventes-conseiller")
+        self.assertEqual(rv["erpnext_role_name"], "OTO Ventes Conseiller")
+
+    def test_role_inconnu_leve(self):
+        bad = copy.deepcopy(self.spec)
+        bad["roles"]["validation_dossier"] = "role-inexistant"
+        with self.assertRaises(KeyError):
+            build_bundle(bad, self.resolver)
+
+    def test_sections_triees_par_ordre(self):
+        ordres = [s["ordre"] for s in self.bundle["sections"]]
+        self.assertEqual(ordres, sorted(ordres))
+        self.assertEqual(self.bundle["sections"][0]["id"], "apport_initial")
+
+    def test_exemple_est_bloque(self):
+        # Le dossier d'exemple DOIT démontrer un blocage (valeur pédagogique).
+        self.assertFalse(self.bundle["gate_status_example"]["can_submit"])
+        self.assertTrue(self.bundle["gate_status_example"]["reasons"])
+
+    def test_deterministe(self):
+        again = build_bundle(_load_spec(), RoleResolver.from_path())
+        self.assertEqual(
+            json.dumps(self.bundle, sort_keys=True, ensure_ascii=False),
+            json.dumps(again, sort_keys=True, ensure_ascii=False),
+        )
+
+
+class GeneratorCliTest(unittest.TestCase):
+    def _run(self, *argv):
+        return subprocess.run(
+            [sys.executable, _GEN, *argv],
+            capture_output=True, text=True, cwd=_MODULE,
+        )
+
+    def test_validate_ok(self):
+        r = self._run("validate")
+        self.assertEqual(r.returncode, 0, r.stderr)
+        self.assertIn("12 invariants", r.stdout)
+
+    @staticmethod
+    def _read(path):
+        with open(path, encoding="utf-8") as fh:
+            return fh.read()
+
+    def test_build_reproductible(self):
+        with tempfile.TemporaryDirectory() as d:
+            r1 = self._run("build", "-o", d)
+            self.assertEqual(r1.returncode, 0, r1.stderr)
+            first = {f: self._read(os.path.join(d, f)) for f in os.listdir(d)}
+            r2 = self._run("build", "-o", d)
+            self.assertEqual(r2.returncode, 0, r2.stderr)
+            second = {f: self._read(os.path.join(d, f)) for f in os.listdir(d)}
+        self.assertEqual(first, second)
+
+    def test_build_egale_commite(self):
+        # L'artefact commité out/ doit être identique à un build frais (miroir
+        # local du gate ci/check_artifacts.sh).
+        with tempfile.TemporaryDirectory() as d:
+            self.assertEqual(self._run("build", "-o", d).returncode, 0)
+            for fname in ("banques.json", "sections.json", "documents.json",
+                          "autorisations.json", "gate_spec.json",
+                          "gate_status_example.json", "MANIFEST.json"):
+                fresh = self._read(os.path.join(d, fname))
+                committed_path = os.path.join(_MODULE, "out", fname)
+                if os.path.exists(committed_path):
+                    self.assertEqual(fresh, self._read(committed_path),
+                                     f"{fname} a dérivé de sa source")
+
+
+class InvariantGuardTest(unittest.TestCase):
+    """Chaque invariant du générateur DOIT mordre une mutation (anti-régression)."""
+
+    def setUp(self):
+        sys.path.insert(0, _MODULE)
+        import financement_bancaire_gen as gen  # noqa
+        self.gen = gen
+        self.spec = _load_spec()
+        self.resolver = RoleResolver.from_path()
+
+    def _errors(self, spec):
+        bundle = build_bundle(spec, self.resolver)
+        return self.gen._validate_bundle(spec, bundle)
+
+    def test_baseline_clean(self):
+        self.assertEqual(self._errors(self.spec), [])
+
+    def test_taux_ordre_mordu(self):
+        bad = copy.deepcopy(self.spec)
+        bad["apport"]["residence_types"][0]["taux_apport_pct"] = 40  # résident > étranger
+        self.assertTrue(any("Taux d'apport" in e for e in self._errors(bad)))
+
+    def test_section_gating_deplacee_mordue(self):
+        bad = copy.deepcopy(self.spec)
+        bad["sections"][0]["gating"] = False  # plus aucune section gating
+        self.assertTrue(any("gating" in e for e in self._errors(bad)))
+
+    def test_id_duplique_mordu(self):
+        bad = copy.deepcopy(self.spec)
+        bad["banques"].append(dict(bad["banques"][0]))
+        self.assertTrue(any("dupliqué" in e for e in self._errors(bad)))
+
+    def test_ltv_incoherent_mordu(self):
+        bad = copy.deepcopy(self.spec)
+        bad["banques"][1]["ltv_min_pct"] = 99  # min > max (85)
+        self.assertTrue(any("LTV" in e for e in self._errors(bad)))
+
+
+if __name__ == "__main__":
+    unittest.main()