[DTP-Worker] Sprint 5 · Générateur Audit 5D conformité (QA · roadmap L58)
Audit de second niveau : lit les hand-off out/ des livrables (workflow vente, Dossier Vente, commissions, e-CF DGII, CONFOTUR) et vérifie 17 contrôles en 5 dimensions (D1 Traçabilité/ISA 500 · D2 AML-UAF/Ley 155-17 · D3 Fiscal e-CF/Ley 32-23 · D4 Intégrité/IFRS · D5 Gouvernance-SoD/ISA 315). Anti-invention #6 : paramètre réglementaire non confirmé → A_CONFIRMER (open item assigné au métier), jamais fabriqué. Verdict PASS_WITH_OPEN_ITEMS (13 PASS, 0 FAIL, 4 à confirmer). Réutilise validateur Publiciste + RoleResolver CRM + roles_targeting CONFOTUR (zéro duplication). 37 tests · 15 invariants · build déterministe · régression 341 tests verts. Job CI qa-audit-5d-tests + gate. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Code DTP Worker
@@ -0,0 +1,260 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Tests du générateur d'audit 5D de conformité (QA · Sprint 5).
|
||||
|
||||
Stdlib pur (`unittest`) → aucune installation pip requise sur le runner Gitea
|
||||
(CLAUDE.md #2). Couvre : chargement des artefacts, chaque contrôle (positif +
|
||||
injection négative sur copie profonde de l'artefact), synthèse par dimension,
|
||||
verdict, open items, invariants du générateur, déterminisme et schéma.
|
||||
|
||||
Axe central : l'audit LIT les hand-off `out/` des livrables et remonte les
|
||||
paramètres réglementaires non confirmés en A_CONFIRMER — jamais en valeur
|
||||
fabriquée (#6). Un FAIL ne doit jamais survenir sur les livrables réels.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
_HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
_MOD = os.path.normpath(os.path.join(_HERE, ".."))
|
||||
sys.path.insert(0, _MOD)
|
||||
|
||||
from qalib import artifacts, builder, controls # noqa: E402
|
||||
import audit_5d_gen as gen # noqa: E402
|
||||
|
||||
PASS, FAIL, AC = controls.PASS, controls.FAIL, controls.A_CONFIRMER
|
||||
_FN = {c["id"]: c["fn"] for c in controls.CONTROLS}
|
||||
|
||||
|
||||
def _spec():
|
||||
with open(os.path.join(_MOD, "audit_spec.json"), encoding="utf-8") as fh:
|
||||
return json.load(fh)
|
||||
|
||||
|
||||
def _art():
|
||||
return artifacts.load_artifacts()
|
||||
|
||||
|
||||
def _run(cid, art, spec):
|
||||
return _FN[cid](art, spec)
|
||||
|
||||
|
||||
class ArtifactsTest(unittest.TestCase):
|
||||
def test_load_all_present(self):
|
||||
art = _art()
|
||||
self.assertEqual(set(art), artifacts.known_ids())
|
||||
self.assertEqual(art["workflow"]["doctype"], "Workflow") # liste dépliée
|
||||
|
||||
def test_missing_artifact_raises(self):
|
||||
orig = artifacts.ARTIFACT_PATHS.copy()
|
||||
artifacts.ARTIFACT_PATHS["ecf"] = os.path.join("fiscal", "n_existe_pas.json")
|
||||
try:
|
||||
self.assertRaises(FileNotFoundError, artifacts.load_artifacts)
|
||||
finally:
|
||||
artifacts.ARTIFACT_PATHS.clear()
|
||||
artifacts.ARTIFACT_PATHS.update(orig)
|
||||
|
||||
|
||||
class ControlsHappyPathTest(unittest.TestCase):
|
||||
def test_no_fail_on_real_artifacts(self):
|
||||
art, spec = _art(), _spec()
|
||||
for c in controls.CONTROLS:
|
||||
statut, detail = c["fn"](art, spec)
|
||||
self.assertIn(statut, (PASS, AC), f"{c['id']} inattendu {statut}")
|
||||
self.assertTrue(detail, f"{c['id']} sans détail")
|
||||
|
||||
def test_open_items_are_the_four_regulatory_params(self):
|
||||
art, spec = _art(), _spec()
|
||||
ac = {c["id"] for c in controls.CONTROLS if c["fn"](art, spec)[0] == AC}
|
||||
self.assertEqual(ac, {"D1.1", "D1.2", "D1.3", "D2.3"})
|
||||
|
||||
|
||||
class D1TracabiliteTest(unittest.TestCase):
|
||||
def test_d1_1_fabricated_rate_fails(self):
|
||||
art, spec = copy.deepcopy(_art()), _spec()
|
||||
art["commissions"]["evenements"][0]["taux_pct"] = 5.0
|
||||
art["commissions"]["evenements"][0]["a_confirmer"] = False
|
||||
self.assertEqual(_run("D1.1", art, spec)[0], FAIL)
|
||||
|
||||
def test_d1_1_rate_with_source_passes(self):
|
||||
art, spec = copy.deepcopy(_art()), _spec()
|
||||
for e in art["commissions"]["evenements"]:
|
||||
e["taux_pct"], e["source"], e["a_confirmer"] = 3.0, "barème Direction 2026", False
|
||||
self.assertEqual(_run("D1.1", art, spec)[0], PASS)
|
||||
|
||||
def test_d1_2_fabricated_rnc_fails(self):
|
||||
art, spec = copy.deepcopy(_art()), _spec()
|
||||
art["ecf"]["emisor"]["rnc_emisor"] = "131000000"
|
||||
self.assertEqual(_run("D1.2", art, spec)[0], FAIL)
|
||||
|
||||
def test_d1_3_fabricated_itbis_fails(self):
|
||||
art, spec = copy.deepcopy(_art()), _spec()
|
||||
art["ecf"]["taxes"][0]["taux_pct"] = 18.0
|
||||
self.assertEqual(_run("D1.3", art, spec)[0], FAIL)
|
||||
|
||||
def test_d1_4_default_on_data_field_fails(self):
|
||||
art, spec = copy.deepcopy(_art()), _spec()
|
||||
for f in art["confotur"]["fields"]:
|
||||
if f["fieldname"] == "referencia_autoridad":
|
||||
f["default"] = "RES-CONFOTUR-0000"
|
||||
self.assertEqual(_run("D1.4", art, spec)[0], FAIL)
|
||||
|
||||
def test_d1_4_passes_on_real(self):
|
||||
self.assertEqual(_run("D1.4", _art(), _spec())[0], PASS)
|
||||
|
||||
|
||||
class D2AmlUafTest(unittest.TestCase):
|
||||
def test_d2_1_missing_client_fails(self):
|
||||
art, spec = copy.deepcopy(_art()), _spec()
|
||||
art["dossier"]["fields"] = [f for f in art["dossier"]["fields"]
|
||||
if f["fieldname"] != "client"]
|
||||
self.assertEqual(_run("D2.1", art, spec)[0], FAIL)
|
||||
|
||||
def test_d2_2_missing_identity_piece_fails(self):
|
||||
art, spec = copy.deepcopy(_art()), _spec()
|
||||
art["confotur"]["field_order"] = [x for x in art["confotur"]["field_order"]
|
||||
if x != "piece_identidad_cliente"]
|
||||
self.assertEqual(_run("D2.2", art, spec)[0], FAIL)
|
||||
|
||||
def test_d2_3_threshold_is_open_item(self):
|
||||
self.assertEqual(_run("D2.3", _art(), _spec())[0], AC)
|
||||
|
||||
def test_d2_3_fabricated_threshold_fails(self):
|
||||
spec = _spec()
|
||||
spec["uaf"]["seuil_operacion"] = 1000000
|
||||
self.assertEqual(_run("D2.3", _art(), spec)[0], FAIL)
|
||||
|
||||
|
||||
class D3FiscalTest(unittest.TestCase):
|
||||
def test_d3_1_devise_mismatch_fails(self):
|
||||
art, spec = copy.deepcopy(_art()), _spec()
|
||||
art["ecf"]["devise_field"] = "champ_inexistant"
|
||||
self.assertEqual(_run("D3.1", art, spec)[0], FAIL)
|
||||
|
||||
def test_d3_2_emission_on_draft_fails(self):
|
||||
art, spec = copy.deepcopy(_art()), _spec()
|
||||
for s in art["workflow"]["states"]:
|
||||
if s["update_value"] == "contrat":
|
||||
s["doc_status"] = "0"
|
||||
self.assertEqual(_run("D3.2", art, spec)[0], FAIL)
|
||||
|
||||
def test_d3_3_wrong_formapago_fails(self):
|
||||
art, spec = copy.deepcopy(_art()), _spec()
|
||||
art["ecf"]["forma_pago_defaut"]["code"] = "1"
|
||||
self.assertEqual(_run("D3.3", art, spec)[0], FAIL)
|
||||
|
||||
|
||||
class D4IntegriteTest(unittest.TestCase):
|
||||
def test_d4_1_states_diverge_fails(self):
|
||||
art, spec = copy.deepcopy(_art()), _spec()
|
||||
for f in art["dossier"]["fields"]:
|
||||
if f["fieldname"] == "workflow_state":
|
||||
f["options"] = "Lead\nÉtat fantôme"
|
||||
self.assertEqual(_run("D4.1", art, spec)[0], FAIL)
|
||||
|
||||
def test_d4_2_base_not_currency_fails(self):
|
||||
art, spec = copy.deepcopy(_art()), _spec()
|
||||
art["commissions"]["evenements"][0]["base_field"] = "prospect"
|
||||
self.assertEqual(_run("D4.2", art, spec)[0], FAIL)
|
||||
|
||||
def test_d4_3_link_wrong_target_fails(self):
|
||||
art, spec = copy.deepcopy(_art()), _spec()
|
||||
for f in art["confotur"]["fields"]:
|
||||
if f["fieldname"] == "dossier_vente":
|
||||
f["options"] = "Autre DocType"
|
||||
self.assertEqual(_run("D4.3", art, spec)[0], FAIL)
|
||||
|
||||
def test_d4_4_unknown_update_value_fails(self):
|
||||
art, spec = copy.deepcopy(_art()), _spec()
|
||||
art["commissions"]["evenements"][0]["update_value"] = "etat_inconnu"
|
||||
self.assertEqual(_run("D4.4", art, spec)[0], FAIL)
|
||||
|
||||
|
||||
class D5GouvernanceTest(unittest.TestCase):
|
||||
def test_d5_1_unknown_role_fails(self):
|
||||
art, spec = copy.deepcopy(_art()), _spec()
|
||||
art["commissions"]["evenements"][0]["role_id"] = "role-fantome-999"
|
||||
self.assertEqual(_run("D5.1", art, spec)[0], FAIL)
|
||||
|
||||
def test_d5_1_passes_on_real(self):
|
||||
self.assertEqual(_run("D5.1", _art(), _spec())[0], PASS)
|
||||
|
||||
def test_d5_2_segregation_breach_fails(self):
|
||||
art, spec = copy.deepcopy(_art()), _spec()
|
||||
for ev in art["ecf"]["emission_events"]:
|
||||
ev["role_id"] = "ventes-reservations"
|
||||
self.assertEqual(_run("D5.2", art, spec)[0], FAIL)
|
||||
|
||||
def test_d5_3_perms_mismatch_fails(self):
|
||||
art, spec = copy.deepcopy(_art()), _spec()
|
||||
art["confotur"]["permissions"][0]["submit"] = 0
|
||||
self.assertEqual(_run("D5.3", art, spec)[0], FAIL)
|
||||
|
||||
def test_d5_3_passes_on_real(self):
|
||||
self.assertEqual(_run("D5.3", _art(), _spec())[0], PASS)
|
||||
|
||||
|
||||
class BuilderTest(unittest.TestCase):
|
||||
def test_verdict_pass_with_open_items(self):
|
||||
r = builder.build(_spec())
|
||||
self.assertEqual(r["verdict"], "PASS_WITH_OPEN_ITEMS")
|
||||
self.assertEqual(r["totals"], {"controls": 17, "pass": 13, "fail": 0, "a_confirmer": 4})
|
||||
|
||||
def test_deterministic(self):
|
||||
self.assertEqual(builder.build(_spec()), builder.build(_spec()))
|
||||
|
||||
def test_open_items_match_a_confirmer(self):
|
||||
r = builder.build(_spec())
|
||||
ac = {c["id"] for c in r["controls"] if c["statut"] == AC}
|
||||
oi = {o["control"] for o in r["open_items"]}
|
||||
self.assertEqual(ac, oi)
|
||||
self.assertTrue(all(o["owner"] for o in r["open_items"]))
|
||||
|
||||
def test_dimensions_cover_all_controls(self):
|
||||
r = builder.build(_spec())
|
||||
self.assertEqual(sum(d["controls_total"] for d in r["dimensions"]), 17)
|
||||
self.assertEqual(len(r["dimensions"]), 5)
|
||||
|
||||
def test_controls_sorted(self):
|
||||
ids = [c["id"] for c in builder.build(_spec())["controls"]]
|
||||
self.assertEqual(ids, sorted(ids))
|
||||
|
||||
|
||||
class GeneratorTest(unittest.TestCase):
|
||||
def test_validate_clean(self):
|
||||
spec = _spec()
|
||||
self.assertEqual(gen._validate(spec, builder.build(spec)), [])
|
||||
|
||||
def test_spec_code_ids_match(self):
|
||||
self.assertEqual(sorted(c["id"] for c in _spec()["controls"]),
|
||||
controls.control_ids())
|
||||
|
||||
def test_report_validates_schema(self):
|
||||
from qalib.deps import validate as maison
|
||||
with open(os.path.join(_MOD, "audit.schema.json"), encoding="utf-8") as fh:
|
||||
schema = json.load(fh)
|
||||
self.assertEqual(list(maison(builder.build(_spec()), schema)), [])
|
||||
|
||||
def test_validate_detects_injected_fail(self):
|
||||
spec = _spec()
|
||||
r = copy.deepcopy(builder.build(spec))
|
||||
r["controls"][0]["statut"] = FAIL
|
||||
self.assertTrue(any("FAIL" in e for e in gen._validate(spec, r)))
|
||||
|
||||
def test_validate_detects_id_drift(self):
|
||||
spec = _spec()
|
||||
r = builder.build(spec)
|
||||
spec2 = copy.deepcopy(spec)
|
||||
spec2["controls"].pop()
|
||||
self.assertNotEqual(gen._validate(spec2, r), [])
|
||||
|
||||
def test_cli_validate_returns_zero(self):
|
||||
self.assertEqual(gen.main(["validate"]), 0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user