[DTP-Worker] Sprint 4 · Générateur DocType porteur OTO Dossier Vente (complète hand-off workflow vente)
DocType custom cible du Workflow OTO Vente Pipeline. Cross-cohérence workflow↔DocType : nom/champ d'état/valeurs de statut/is_submittable/permissions tous dérivés de workflow_vente_spec.json (source unique, anti-dérive). Rôles résolus via rbac_50_roles.json (#6). CLI build|validate · 12 invariants · 31 tests. Job CI crm-dossier-vente-tests ajouté au gate. Régression 177 tests 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,248 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Tests du générateur du DocType porteur `OTO Dossier Vente` (Sprint 4 · CRM).
|
||||
|
||||
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.
|
||||
|
||||
Axe central : la CROSS-COHÉRENCE workflow ↔ DocType (le DocType doit porter
|
||||
exactement l'état, les valeurs de statut, la soumissibilité et les rôles du
|
||||
pipeline `crm/workflow_vente/`).
|
||||
"""
|
||||
|
||||
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 dvlib import builder, frappe # noqa: E402
|
||||
from dvlib.builder import _UPDATE_FIELD # noqa: E402
|
||||
from workflow_vente.wflib.rbac import RoleResolver # noqa: E402
|
||||
from lib import validator as maison # type: ignore # noqa: E402
|
||||
import doctype_dossier_vente_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 BaseFixture(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls) -> None:
|
||||
cls.spec = _load(gen._SPEC_PATH)
|
||||
cls.wf_spec = _load(gen._WF_SPEC_PATH)
|
||||
cls.schema = _load(gen._SCHEMA_PATH)
|
||||
cls.resolver = RoleResolver.from_path()
|
||||
cls.bundle = builder.build_bundle(cls.spec, cls.wf_spec, cls.resolver)
|
||||
cls.dt = cls.bundle["doctype"]
|
||||
cls.manifest = cls.bundle["manifest"]
|
||||
cls.fields = {f["fieldname"]: f for f in cls.dt["fields"]}
|
||||
|
||||
def _opts(self, fieldname: str) -> list[str]:
|
||||
return [ln for ln in self.fields[fieldname].get("options", "").split("\n") if ln]
|
||||
|
||||
|
||||
class TestSchema(BaseFixture):
|
||||
def test_maison_schema_ok(self):
|
||||
self.assertEqual(maison.validate(self.bundle, self.schema), [])
|
||||
|
||||
@unittest.skipUnless(_HAS_JSONSCHEMA, "jsonschema absent")
|
||||
def test_oracle_jsonschema_ok(self):
|
||||
jsonschema.validate(self.bundle, self.schema)
|
||||
|
||||
def test_validate_returns_empty(self):
|
||||
self.assertEqual(gen._validate(self.bundle, self.wf_spec, self.resolver), [])
|
||||
|
||||
|
||||
class TestCrossCoherence(BaseFixture):
|
||||
def test_name_equals_workflow_document_type(self):
|
||||
self.assertEqual(self.dt["name"], self.wf_spec["document_type"])
|
||||
|
||||
def test_custom_matches_workflow(self):
|
||||
self.assertEqual(bool(self.dt["custom"]), bool(self.wf_spec["document_type_custom"]))
|
||||
|
||||
def test_state_field_options_equal_state_names(self):
|
||||
field = self.wf_spec["workflow_state_field"]
|
||||
names = [s["state"] for s in self.wf_spec["states"]]
|
||||
self.assertEqual(self.fields[field]["fieldtype"], "Select")
|
||||
self.assertEqual(self._opts(field), names)
|
||||
self.assertEqual(self.fields[field]["read_only"], 1)
|
||||
|
||||
def test_pipeline_value_field_options_equal_update_values(self):
|
||||
values = [s["update_value"] for s in self.wf_spec["states"]]
|
||||
self.assertIn(_UPDATE_FIELD, self.fields)
|
||||
self.assertEqual(self._opts(_UPDATE_FIELD), values)
|
||||
|
||||
def test_is_submittable_derived_from_docstatus(self):
|
||||
max_ds = max(s["doc_status"] for s in self.wf_spec["states"])
|
||||
self.assertEqual(self.dt["is_submittable"], 1 if max_ds >= "1" else 0)
|
||||
|
||||
def test_update_field_comes_from_workflow_builder(self):
|
||||
# Anti-dérive : le champ de valeur machine est importé du builder voisin,
|
||||
# pas réécrit en dur ici.
|
||||
self.assertEqual(_UPDATE_FIELD, "statut_pipeline")
|
||||
|
||||
|
||||
class TestPermissions(BaseFixture):
|
||||
def setUp(self):
|
||||
self.perm = {p["role"]: p for p in self.dt["permissions"]}
|
||||
self.docstatus = {s["state"]: s["doc_status"] for s in self.wf_spec["states"]}
|
||||
|
||||
def test_every_workflow_role_present(self):
|
||||
role_ids = {s["role_id"] for s in self.wf_spec["states"]}
|
||||
role_ids |= {t["role_id"] for t in self.wf_spec["transitions"]}
|
||||
for rid in role_ids:
|
||||
self.assertIn(self.resolver.erpnext_name(rid), self.perm)
|
||||
|
||||
def test_allow_edit_roles_have_write(self):
|
||||
for s in self.wf_spec["states"]:
|
||||
p = self.perm[self.resolver.erpnext_name(s["role_id"])]
|
||||
self.assertEqual(p.get("write"), 1, s["state"])
|
||||
if s["doc_status"] == "0":
|
||||
self.assertEqual(p.get("create"), 1, s["state"])
|
||||
|
||||
def test_submit_and_cancel_derivation(self):
|
||||
for t in self.wf_spec["transitions"]:
|
||||
p = self.perm[self.resolver.erpnext_name(t["role_id"])]
|
||||
nxt = self.docstatus[t["next_state"]]
|
||||
if nxt == "1":
|
||||
self.assertEqual(p.get("submit"), 1, t["action"])
|
||||
if nxt == "2":
|
||||
self.assertEqual(p.get("cancel"), 1, t["action"])
|
||||
|
||||
def test_no_role_name_hardcoded_in_spec(self):
|
||||
# #6 : aucun NOM DE RÔLE Frappe en dur dans le contrat métier ; les rôles
|
||||
# ne viennent que du workflow (role_id) résolu via le RBAC.
|
||||
blob = json.dumps(self.spec, ensure_ascii=False)
|
||||
for rid in self.resolver.known_ids():
|
||||
self.assertNotIn(self.resolver.erpnext_name(rid), blob)
|
||||
|
||||
|
||||
class TestFrappeKnowledge(unittest.TestCase):
|
||||
def test_docfield_rejects_unknown_fieldtype(self):
|
||||
with self.assertRaises(ValueError):
|
||||
frappe.docfield("x", "Rocket")
|
||||
|
||||
def test_permission_row_rejects_unknown_perm(self):
|
||||
with self.assertRaises(ValueError):
|
||||
frappe.permission_row("OTO X", {"read", "teleport"})
|
||||
|
||||
def test_docfield_only_emits_set_flags(self):
|
||||
f = frappe.docfield("x", "Data", flags={"reqd": 1})
|
||||
self.assertEqual(f.get("reqd"), 1)
|
||||
self.assertNotIn("hidden", f)
|
||||
|
||||
def test_naming_series_field_is_readonly_select(self):
|
||||
f = frappe.naming_series_field("OTO-DV-.YYYY.-")
|
||||
self.assertEqual(f["fieldtype"], "Select")
|
||||
self.assertEqual(f["read_only"], 1)
|
||||
|
||||
|
||||
class TestInvariantsAndStructure(BaseFixture):
|
||||
def test_field_order_matches_fields(self):
|
||||
self.assertEqual(self.dt["field_order"], [f["fieldname"] for f in self.dt["fields"]])
|
||||
|
||||
def test_fieldnames_unique(self):
|
||||
names = [f["fieldname"] for f in self.dt["fields"]]
|
||||
self.assertEqual(len(names), len(set(names)))
|
||||
|
||||
def test_no_currency_default(self):
|
||||
for f in self.dt["fields"]:
|
||||
if f["fieldtype"] == "Currency":
|
||||
self.assertNotIn("default", f, f["fieldname"])
|
||||
|
||||
def test_currency_options_reference_devise(self):
|
||||
for f in self.dt["fields"]:
|
||||
if f["fieldtype"] == "Currency":
|
||||
self.assertEqual(f.get("options"), "devise")
|
||||
|
||||
def test_links_declared_in_manifest(self):
|
||||
declared = set(self.manifest["doctypes_lies_a_confirmer"])
|
||||
for f in self.dt["fields"]:
|
||||
if f["fieldtype"] == "Link":
|
||||
self.assertIn(f["options"], declared)
|
||||
|
||||
def test_naming_series_present(self):
|
||||
self.assertIn("naming_series", self.fields)
|
||||
self.assertEqual(self.dt["autoname"], "naming_series:")
|
||||
|
||||
def test_manifest_counts_coherent(self):
|
||||
c = self.manifest["counts"]
|
||||
self.assertEqual(c["fields"], len(self.dt["fields"]))
|
||||
self.assertEqual(c["permissions"], len(self.dt["permissions"]))
|
||||
self.assertEqual(c["pipeline_states"], len({s["state"] for s in self.wf_spec["states"]}))
|
||||
|
||||
|
||||
class TestDeterminismAndAntiRegression(BaseFixture):
|
||||
def test_build_is_deterministic(self):
|
||||
again = builder.build_bundle(self.spec, self.wf_spec, self.resolver)
|
||||
self.assertEqual(json.dumps(again, sort_keys=True), json.dumps(self.bundle, sort_keys=True))
|
||||
|
||||
def test_committed_out_matches_regeneration(self):
|
||||
out = os.path.join(_MODULE, "out", "doctype_oto_dossier_vente.json")
|
||||
man = os.path.join(_MODULE, "out", "MANIFEST.json")
|
||||
if not (os.path.exists(out) and os.path.exists(man)):
|
||||
self.skipTest("out/ non commité")
|
||||
self.assertEqual(_load(out), self.dt)
|
||||
self.assertEqual(_load(man), self.manifest)
|
||||
|
||||
|
||||
class TestTamperDetection(BaseFixture):
|
||||
def test_break_is_submittable_detected(self):
|
||||
bad = copy.deepcopy(self.bundle)
|
||||
bad["doctype"]["is_submittable"] = 0
|
||||
self.assertTrue(gen._validate(bad, self.wf_spec, self.resolver))
|
||||
|
||||
def test_drop_state_option_detected(self):
|
||||
bad = copy.deepcopy(self.bundle)
|
||||
for f in bad["doctype"]["fields"]:
|
||||
if f["fieldname"] == self.wf_spec["workflow_state_field"]:
|
||||
f["options"] = "\n".join(f["options"].split("\n")[:-1])
|
||||
self.assertTrue(gen._validate(bad, self.wf_spec, self.resolver))
|
||||
|
||||
def test_rename_doctype_detected(self):
|
||||
bad = copy.deepcopy(self.bundle)
|
||||
bad["doctype"]["name"] = "Autre Chose"
|
||||
self.assertTrue(gen._validate(bad, self.wf_spec, self.resolver))
|
||||
|
||||
def test_inject_currency_default_detected(self):
|
||||
bad = copy.deepcopy(self.bundle)
|
||||
for f in bad["doctype"]["fields"]:
|
||||
if f["fieldtype"] == "Currency":
|
||||
f["default"] = "150000"
|
||||
break
|
||||
self.assertTrue(gen._validate(bad, self.wf_spec, self.resolver))
|
||||
|
||||
|
||||
class TestCLI(unittest.TestCase):
|
||||
def test_cli_validate_exit_zero(self):
|
||||
r = subprocess.run(
|
||||
[sys.executable, os.path.join(_MODULE, "doctype_dossier_vente_gen.py"), "validate"],
|
||||
capture_output=True, text=True,
|
||||
)
|
||||
self.assertEqual(r.returncode, 0, r.stderr)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
Reference in New Issue
Block a user