[DTP-Worker] Sprint 4 · Générateur configuration e-CF DGII (Compupar) (ERPNext Backend · roadmap L51)
Facturation électronique dominicaine cross-cohérente workflow↔DocType↔RBAC : émission sur état soumis, base Currency réelle, rôle compta-fiscaliste-ecf. Anti-invention (#6) : RNC/ITBIS/TipoCambio/endpoints Compupar null (a_confirmer, jamais sans source) ; seules les données de référence DGII encodées avec source. Composeur e-NCF traçable (E+tipo(2)+seq(10)). 39 tests · 12 invariants · gate CI (job fiscal-ecf-tests) · 241 tests de régression verts. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Code DTP Worker
@@ -0,0 +1,290 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Tests du generateur de configuration e-CF DGII (Compupar) · Sprint 4.
|
||||
|
||||
Stdlib pur (`unittest`) → aucune installation pip requise sur le runner Gitea.
|
||||
La bibliotheque `jsonschema` sert d'*oracle* quand elle est presente, pour se
|
||||
premunir d'un ecart entre le validateur maison et draft-07.
|
||||
|
||||
Trois axes :
|
||||
1. CROSS-COHERENCE e-CF ↔ workflow ↔ DocType ↔ RBAC (les 12 invariants : chaque
|
||||
emission se declenche sur un etat SOUMIS, sur un champ Currency reel, par un
|
||||
role Compta resolu, avec un tipo_ecf en perimetre — ou null a confirmer).
|
||||
2. Composition TRACABLE de l'e-NCF (ecflib/ncf.py) : E+tipo(2)+seq(10)=13, None
|
||||
si un operande manque (anti 0-invente · #6), re-parse fidele.
|
||||
3. ANTI-INVENTION (#6) : RNC / ITBIS / TipoCambio jamais fixes sans source.
|
||||
"""
|
||||
|
||||
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, ".."))
|
||||
_DELIVERABLES = os.path.normpath(os.path.join(_MODULE, "..", ".."))
|
||||
_CRM = os.path.join(_DELIVERABLES, "crm")
|
||||
|
||||
sys.path.insert(0, _MODULE)
|
||||
sys.path.insert(0, _CRM)
|
||||
sys.path.insert(0, os.path.join(_DELIVERABLES, "publiciste"))
|
||||
|
||||
from ecflib import builder, ncf # noqa: E402
|
||||
from workflow_vente.wflib.rbac import RoleResolver # noqa: E402
|
||||
from lib import validator as maison # type: ignore # noqa: E402
|
||||
import ecf_dgii_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 contrat vanille passe schema + 12 invariants et est deterministe."""
|
||||
|
||||
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 casses : {errors}")
|
||||
|
||||
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_provider_compupar(self):
|
||||
self.assertEqual(self.bundle["ecf_plan"]["provider"], "Compupar")
|
||||
self.assertEqual(self.bundle["manifest"]["provider"], "Compupar")
|
||||
|
||||
def test_schema_maison(self):
|
||||
schema = _load(gen._SCHEMA_PATH)
|
||||
self.assertEqual(maison.validate(self.bundle, schema), [])
|
||||
|
||||
@unittest.skipUnless(_HAS_JSONSCHEMA, "jsonschema non installe (oracle optionnel)")
|
||||
def test_schema_oracle_jsonschema(self):
|
||||
schema = _load(gen._SCHEMA_PATH)
|
||||
jsonschema.validate(self.bundle, schema) # ne doit pas lever
|
||||
|
||||
def test_counts_coherents(self):
|
||||
m = self.bundle["manifest"]["counts"]
|
||||
plan = self.bundle["ecf_plan"]
|
||||
self.assertEqual(m["tipos_ecf"], len(plan["tipos_ecf"]))
|
||||
self.assertEqual(m["tipos_en_scope"], len(plan["tipos_en_scope"]))
|
||||
self.assertEqual(m["emission_events"], len(plan["emission_events"]))
|
||||
|
||||
|
||||
class CrossCoherence(unittest.TestCase):
|
||||
"""Chaque invariant attrape la mutation correspondante."""
|
||||
|
||||
def setUp(self):
|
||||
self.bundle, self.spec, self.wf, self.dt, self.resolver = gen._build()
|
||||
|
||||
def _errs(self, mutate):
|
||||
b = copy.deepcopy(self.bundle)
|
||||
mutate(b)
|
||||
return gen._validate(b, self.spec, self.wf, self.dt, self.resolver)
|
||||
|
||||
def test_update_value_inconnu(self):
|
||||
errs = self._errs(lambda b: b["ecf_plan"]["emission_events"][0].__setitem__("update_value", "zzz"))
|
||||
self.assertTrue(any("absent du workflow" in e for e in errs), errs)
|
||||
|
||||
def test_etat_non_soumis_refuse(self):
|
||||
# 'lead' existe mais doc_status=0 → pas d'e-CF sur brouillon.
|
||||
errs = self._errs(lambda b: b["ecf_plan"]["emission_events"][0].__setitem__("update_value", "lead"))
|
||||
self.assertTrue(any("pas d'e-CF sur brouillon" in e for e in errs), errs)
|
||||
|
||||
def test_base_field_non_currency(self):
|
||||
errs = self._errs(lambda b: b["ecf_plan"]["emission_events"][0].__setitem__("base_field", "projet"))
|
||||
self.assertTrue(any("Currency" in e for e in errs), errs)
|
||||
|
||||
def test_base_field_inexistant(self):
|
||||
errs = self._errs(lambda b: b["ecf_plan"]["emission_events"][0].__setitem__("base_field", "inexistant"))
|
||||
self.assertTrue(any("Currency" in e for e in errs), errs)
|
||||
|
||||
def test_role_hors_compta(self):
|
||||
# ventes-conseiller existe dans RBAC mais portail ventes, pas compta.
|
||||
def mut(b):
|
||||
ev = b["ecf_plan"]["emission_events"][0]
|
||||
ev["role_id"] = "ventes-conseiller"
|
||||
ev["erpnext_role_name"] = self.resolver.erpnext_name("ventes-conseiller")
|
||||
errs = self._errs(mut)
|
||||
self.assertTrue(any("hors portail compta" in e for e in errs), errs)
|
||||
|
||||
def test_role_name_incoherent(self):
|
||||
errs = self._errs(lambda b: b["ecf_plan"]["emission_events"][0].__setitem__("erpnext_role_name", "Faux Role"))
|
||||
self.assertTrue(any("incoherent avec RBAC" in e for e in errs), errs)
|
||||
|
||||
def test_tipo_null_exige_a_confirmer(self):
|
||||
def mut(b):
|
||||
ev = b["ecf_plan"]["emission_events"][0]
|
||||
ev["tipo_ecf"] = None
|
||||
ev["a_confirmer"] = False
|
||||
errs = self._errs(mut)
|
||||
self.assertTrue(any("a_confirmer=false" in e for e in errs), errs)
|
||||
|
||||
def test_tipo_hors_catalogue(self):
|
||||
errs = self._errs(lambda b: b["ecf_plan"]["emission_events"][0].__setitem__("tipo_ecf", "99"))
|
||||
self.assertTrue(any("absent du catalogue" in e for e in errs), errs)
|
||||
|
||||
def test_tipo_hors_scope(self):
|
||||
# 41 est un type reel du catalogue mais en_scope=false (achats).
|
||||
errs = self._errs(lambda b: b["ecf_plan"]["emission_events"][0].__setitem__("tipo_ecf", "41"))
|
||||
self.assertTrue(any("hors perimetre" in e for e in errs), errs)
|
||||
|
||||
def test_tipo_en_scope_valide(self):
|
||||
# 31 est en_scope → aucune erreur specifique de type.
|
||||
errs = self._errs(lambda b: b["ecf_plan"]["emission_events"][0].__setitem__("tipo_ecf", "31"))
|
||||
self.assertFalse(any("catalogue" in e or "perimetre" in e for e in errs), errs)
|
||||
|
||||
def test_evenement_duplique(self):
|
||||
def mut(b):
|
||||
evs = b["ecf_plan"]["emission_events"]
|
||||
evs.append(copy.deepcopy(evs[0]))
|
||||
errs = self._errs(mut)
|
||||
self.assertTrue(any("duplique" in e for e in errs), errs)
|
||||
|
||||
def test_forma_pago_defaut_hors_table(self):
|
||||
errs = self._errs(lambda b: b["ecf_plan"]["forma_pago_defaut"].__setitem__("code", "99"))
|
||||
self.assertTrue(any("FormaPago" in e for e in errs), errs)
|
||||
|
||||
def test_forma_pago_source_sans_cardnet(self):
|
||||
errs = self._errs(lambda b: b["ecf_plan"]["forma_pago_defaut"].__setitem__("source", "DGII seul"))
|
||||
self.assertTrue(any("Cardnet" in e for e in errs), errs)
|
||||
|
||||
def test_moneda_options_non_usd_dop(self):
|
||||
errs = self._errs(lambda b: b["ecf_plan"]["moneda"].__setitem__("options", ["EUR"]))
|
||||
self.assertTrue(any("USD/DOP" in e for e in errs), errs)
|
||||
|
||||
def test_field_map_champ_inexistant(self):
|
||||
errs = self._errs(lambda b: b["ecf_plan"]["field_map"][0].__setitem__("dossier_field", "inexistant"))
|
||||
self.assertTrue(any("n'est pas un champ du Dossier Vente" in e for e in errs), errs)
|
||||
|
||||
def test_tipos_en_scope_desaligne(self):
|
||||
errs = self._errs(lambda b: b["ecf_plan"]["tipos_en_scope"].append("99"))
|
||||
self.assertTrue(any("tipos_en_scope" in e for e in errs), errs)
|
||||
|
||||
|
||||
class AntiInvention(unittest.TestCase):
|
||||
"""#6 : aucun chiffre OTO fixe sans source."""
|
||||
|
||||
def setUp(self):
|
||||
self.bundle, self.spec, self.wf, self.dt, self.resolver = gen._build()
|
||||
|
||||
def _errs(self, mutate):
|
||||
b = copy.deepcopy(self.bundle)
|
||||
mutate(b)
|
||||
return gen._validate(b, self.spec, self.wf, self.dt, self.resolver)
|
||||
|
||||
def test_baseline_tout_a_confirmer(self):
|
||||
plan = self.bundle["ecf_plan"]
|
||||
self.assertIsNone(plan["emisor"]["rnc_emisor"])
|
||||
self.assertIsNone(plan["moneda"]["tipo_cambio"])
|
||||
self.assertTrue(all(t["taux_pct"] is None for t in plan["taxes"]))
|
||||
self.assertTrue(all(e["tipo_ecf"] is None for e in plan["emission_events"]))
|
||||
|
||||
def test_rnc_sans_source_refuse(self):
|
||||
errs = self._errs(lambda b: b["ecf_plan"]["emisor"].__setitem__("rnc_emisor", "101010101"))
|
||||
self.assertTrue(any("emisor renseigne sans source" in e for e in errs), errs)
|
||||
|
||||
def test_rnc_avec_source_ok(self):
|
||||
def mut(b):
|
||||
b["ecf_plan"]["emisor"]["rnc_emisor"] = "101010101"
|
||||
b["ecf_plan"]["emisor"]["source"] = "RNC officiel WA SRL (registre DGII)"
|
||||
errs = self._errs(mut)
|
||||
self.assertFalse(any("emisor renseigne sans source" in e for e in errs), errs)
|
||||
|
||||
def test_itbis_sans_source_refuse(self):
|
||||
errs = self._errs(lambda b: b["ecf_plan"]["taxes"][0].__setitem__("taux_pct", 18))
|
||||
self.assertTrue(any("taux_pct fixe sans source" in e for e in errs), errs)
|
||||
|
||||
def test_tipo_cambio_sans_source_refuse(self):
|
||||
errs = self._errs(lambda b: b["ecf_plan"]["moneda"].__setitem__("tipo_cambio", 60))
|
||||
self.assertTrue(any("tipo_cambio fixe sans source" in e for e in errs), errs)
|
||||
|
||||
def test_valeurs_a_confirmer_comptees(self):
|
||||
# emisor + provider + moneda + 1 tax + 2 events = 6.
|
||||
self.assertEqual(self.bundle["manifest"]["counts"]["valeurs_a_confirmer"], 6)
|
||||
|
||||
|
||||
class NcfTracable(unittest.TestCase):
|
||||
"""Composition e-NCF tracable (E+tipo(2)+seq(10)) · anti 0-invente (#6)."""
|
||||
|
||||
def test_compose_complet(self):
|
||||
r = ncf.compose_encf("31", "34")
|
||||
self.assertEqual(r["e_ncf"], "E310000000034")
|
||||
self.assertEqual(len(r["e_ncf"]), 13)
|
||||
self.assertFalse(r["incomplete"])
|
||||
self.assertTrue(ncf.is_valid_encf(r["e_ncf"]))
|
||||
|
||||
def test_compose_sans_secuencia_reste_none(self):
|
||||
r = ncf.compose_encf("31", None)
|
||||
self.assertIsNone(r["e_ncf"])
|
||||
self.assertTrue(r["incomplete"])
|
||||
self.assertIn("secuencia", r["champs_manquants"])
|
||||
|
||||
def test_compose_sans_tipo_reste_none(self):
|
||||
r = ncf.compose_encf(None, "34")
|
||||
self.assertIsNone(r["e_ncf"])
|
||||
self.assertIn("tipo", r["champs_manquants"])
|
||||
|
||||
def test_formule_toujours_publiee(self):
|
||||
self.assertEqual(ncf.compose_encf(None, None)["formule"], "E + tipoeCF(2) + secuencia(10)")
|
||||
|
||||
def test_tipo_mal_forme_rejete(self):
|
||||
self.assertIsNone(ncf.compose_encf("3", "34")["e_ncf"]) # 1 chiffre
|
||||
self.assertIsNone(ncf.compose_encf("abc", "34")["e_ncf"])
|
||||
|
||||
def test_secuencia_trop_longue_rejetee(self):
|
||||
self.assertIsNone(ncf.compose_encf("31", "12345678901")["e_ncf"]) # 11 chiffres
|
||||
|
||||
def test_parse_roundtrip(self):
|
||||
r = ncf.compose_encf("46", "999")
|
||||
p = ncf.parse_encf(r["e_ncf"])
|
||||
self.assertEqual(p, {"prefix": "E", "tipo": "46", "secuencia": "0000000999"})
|
||||
|
||||
def test_is_valid_rejette_ncf_physique(self):
|
||||
# NCF physique (B01...) n'est pas un e-NCF.
|
||||
self.assertFalse(ncf.is_valid_encf("B0100000001"))
|
||||
self.assertFalse(ncf.is_valid_encf(None))
|
||||
|
||||
def test_fixture_exemple(self):
|
||||
fx = _load(os.path.join(_MODULE, "fixtures", "dossier_exemple.json"))
|
||||
ex = fx["encf_exemple"]
|
||||
r = ncf.compose_encf(ex["tipo"], ex["secuencia"])
|
||||
self.assertEqual(r["e_ncf"], "E310000000034")
|
||||
self.assertIn("FICTIF", ex["source"])
|
||||
|
||||
|
||||
class CliSmoke(unittest.TestCase):
|
||||
"""Le CLI build/validate tourne et l'output re-valide (idempotence)."""
|
||||
|
||||
def test_validate_cli(self):
|
||||
r = subprocess.run([sys.executable, "ecf_dgii_gen.py", "validate"],
|
||||
cwd=_MODULE, capture_output=True, text=True)
|
||||
self.assertEqual(r.returncode, 0, r.stderr)
|
||||
self.assertIn("12 invariants verts", r.stdout)
|
||||
|
||||
def test_build_output_revalide(self):
|
||||
plan = _load(os.path.join(_MODULE, "out", "ecf_plan.json"))
|
||||
manifest = _load(os.path.join(_MODULE, "out", "MANIFEST.json"))
|
||||
schema = _load(gen._SCHEMA_PATH)
|
||||
self.assertEqual(maison.validate({"manifest": manifest, "ecf_plan": plan}, schema), [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
Reference in New Issue
Block a user