[DTP-Worker 20260813_020316] Auto exec · session 20260813_020316

05_deliverables_mvp/faisabilite/v18_master_intake/tests/test_master_intake.py

@@ -0,0 +1,290 @@
+"""Suite Phase 1 · Master Project Intake / Master Data Model V18.
+
+Vérifie le CONTRAT (pas un moteur métier — il n'y en a pas encore, audit §11) :
+statuts/catégories canoniques · DataPoint traçable · garde canonique #9/#10 ·
+intégrité de l'Evidence Register · rétro-compat STRICTE avec brief.json V12
+(D-08) · cohérence schéma ↔ code. stdlib pur ; l'oracle jsonschema est utilisé
+s'il est présent, sinon ignoré (comme les 17 skips « par design » du dépôt).
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+import unittest
+
+_HERE = os.path.dirname(os.path.abspath(__file__))
+_MODULE = os.path.dirname(_HERE)
+for p in (_MODULE,):
+    if p not in sys.path:
+        sys.path.insert(0, p)
+
+from mdmlib import (  # noqa: E402
+    CANONICAL,
+    CATEGORIES,
+    STATUSES,
+    DataPoint,
+    Document,
+    DocumentRegister,
+    EvidenceRegister,
+    MasterDataError,
+    assert_not_canonical,
+    is_present,
+)
+import master_intake_gen as cli  # noqa: E402
+
+SCHEMA_PATH = os.path.join(_MODULE, "master_intake.schema.json")
+BRIEF_SCHEMA_PATH = os.path.join(_MODULE, "..", "generator", "brief.schema.json")
+FIXTURE = os.path.join(_MODULE, "fixtures", "intake_P01_coralis.json")
+GEN_CANONICAL_PATH = os.path.join(
+    _MODULE, "..", "generator", "genlib", "model.py"
+)
+
+
+def _load(path: str) -> dict:
+    with open(path, encoding="utf-8") as fh:
+        return json.load(fh)
+
+
+class TestStatusesCategories(unittest.TestCase):
+    def test_exactly_11_statuses_in_canonical_order(self):
+        self.assertEqual(len(STATUSES), 11)
+        self.assertEqual(
+            STATUSES,
+            (
+                "VERIFIED", "CONFIRMED", "SOURCE_BASED", "CALCULATED",
+                "ESTIMATED", "ASSUMPTION", "TARGET", "BANK_REQUIREMENT",
+                "PENDING", "MISSING", "NOT_APPLICABLE",
+            ),
+        )
+
+    def test_exactly_5_categories(self):
+        self.assertEqual(
+            CATEGORIES,
+            ("INPUT", "TARGET", "ASSUMPTION", "BANK_REQUIREMENT", "CALCULATED"),
+        )
+
+    def test_is_present_only_false_for_pending_and_missing(self):
+        absent = {s for s in STATUSES if not is_present(s)}
+        self.assertEqual(absent, {"PENDING", "MISSING"})
+
+
+class TestSchemaMirrorsCode(unittest.TestCase):
+    """Le schéma machine et le code Python doivent porter les MÊMES enums (pas
+    de dérive silencieuse entre les deux surfaces du contrat)."""
+
+    def setUp(self):
+        self.schema = _load(SCHEMA_PATH)
+
+    def test_schema_is_draft07(self):
+        self.assertEqual(
+            self.schema["$schema"], "http://json-schema.org/draft-07/schema#"
+        )
+
+    def test_status_enum_matches_code(self):
+        enum = self.schema["definitions"]["data_status"]["enum"]
+        self.assertEqual(tuple(enum), STATUSES)
+
+    def test_category_enum_matches_code(self):
+        enum = self.schema["definitions"]["data_category"]["enum"]
+        self.assertEqual(tuple(enum), CATEGORIES)
+
+    def test_dataset_version_enum(self):
+        self.assertEqual(
+            self.schema["properties"]["dataset_version"]["enum"], ["V12", "V18"]
+        )
+
+
+class TestRetroCompatV12(unittest.TestCase):
+    """D-08 : master_intake = SUR-ENSEMBLE STRICT du brief.json V12."""
+
+    def setUp(self):
+        self.master = _load(SCHEMA_PATH)
+        self.brief = _load(BRIEF_SCHEMA_PATH)
+
+    def test_master_includes_brief_by_ref(self):
+        refs = [a.get("$ref") for a in self.master.get("allOf", [])]
+        self.assertIn(self.brief["$id"], refs,
+                      "brief.schema.json V12 doit être inclus via allOf/$ref (D-08 règle 2)")
+
+    def test_no_v12_key_shadowed_with_conflicting_type(self):
+        # Toute clé V12 réintroduite dans les properties V18 doit garder un type
+        # compatible (rétro-compat parser Publiciste · D-08 règle 1). Ici V18
+        # n'en réintroduit AUCUNE (extension pure) — on le prouve.
+        v12_keys = set(self.brief.get("properties", {}))
+        v18_keys = set(self.master.get("properties", {}))
+        self.assertEqual(
+            v12_keys & v18_keys, set(),
+            "les champs V18 sont ajoutés en EXTENSION, jamais en remplacement (D-08 règle 3)",
+        )
+
+    def test_dataset_version_field_present(self):
+        self.assertIn("dataset_version", self.master["properties"])
+
+
+class TestDataPoint(unittest.TestCase):
+    def test_valid_point(self):
+        dp = DataPoint(key="k", value=1, category="INPUT", status="VERIFIED")
+        self.assertTrue(dp.present)
+        self.assertEqual(dp.as_dict()["key"], "k")
+
+    def test_unknown_status_rejected(self):
+        with self.assertRaises(MasterDataError):
+            DataPoint(key="k", value=1, category="INPUT", status="NOPE")
+
+    def test_unknown_category_rejected(self):
+        with self.assertRaises(MasterDataError):
+            DataPoint(key="k", value=1, category="NOPE", status="VERIFIED")
+
+    def test_empty_key_rejected(self):
+        with self.assertRaises(MasterDataError):
+            DataPoint(key="  ", value=1, category="INPUT", status="VERIFIED")
+
+    def test_confidence_out_of_range_rejected(self):
+        with self.assertRaises(MasterDataError):
+            DataPoint(key="k", value=1, category="INPUT", status="VERIFIED",
+                      confidence=1.5)
+
+    def test_confidence_bool_rejected(self):
+        with self.assertRaises(MasterDataError):
+            DataPoint(key="k", value=1, category="INPUT", status="VERIFIED",
+                      confidence=True)
+
+    def test_source_based_constructs_without_inline_doc(self):
+        # La provenance d'un SOURCE_BASED peut venir d'un lien Evidence : le
+        # point seul se construit (la complétude est vérifiée au niveau dataset,
+        # cf. TestEvidenceRegister.test_dangling_evidence_detected).
+        dp = DataPoint(key="k", value="x", category="INPUT", status="SOURCE_BASED")
+        self.assertTrue(dp.present)
+        DataPoint(key="k", value="x", category="INPUT", status="SOURCE_BASED",
+                  source_document="DOC-1")
+
+    def test_absent_status_cannot_carry_value(self):
+        with self.assertRaises(MasterDataError):
+            DataPoint(key="k", value=42, category="INPUT", status="PENDING")
+        # None est OK pour une lacune
+        DataPoint(key="k", value=None, category="INPUT", status="MISSING")
+
+
+class TestCanonicalGuard(unittest.TestCase):
+    """Les 6 canoniques #9/#10 ne peuvent jamais devenir un intake saisissable."""
+
+    def test_each_canonical_key_rejected(self):
+        for key in CANONICAL:
+            with self.assertRaises(MasterDataError):
+                assert_not_canonical(key)
+            with self.assertRaises(MasterDataError):
+                DataPoint(key=key, value="x", category="INPUT", status="VERIFIED")
+
+    def test_canonical_values_match_generator_v12(self):
+        # Single source of truth : les valeurs canoniques V18 sont IDENTIQUES à
+        # celles du générateur V12 (audit §8 « réimposer identiques »). On les
+        # relit dans generator/genlib/model.py sans l'importer (évite tout effet
+        # de bord d'import du module de prod) et on exige l'égalité stricte.
+        ns: dict = {}
+        with open(GEN_CANONICAL_PATH, encoding="utf-8") as fh:
+            src = fh.read()
+        # extrait le littéral CANONICAL = {...}
+        start = src.index("CANONICAL = {")
+        end = src.index("}", start) + 1
+        exec(src[start:end], ns)  # noqa: S102 — littéral de constantes contrôlé
+        self.assertEqual(CANONICAL, ns["CANONICAL"])
+
+
+class TestEvidenceRegister(unittest.TestCase):
+    def setUp(self):
+        self.docs = DocumentRegister()
+        self.docs.add(Document(doc_id="D1", title="Pièce 1"))
+
+    def test_duplicate_doc_id_rejected(self):
+        with self.assertRaises(MasterDataError):
+            self.docs.add(Document(doc_id="D1", title="doublon"))
+
+    def test_link_to_unknown_doc_rejected(self):
+        ev = EvidenceRegister(self.docs)
+        with self.assertRaises(MasterDataError):
+            ev.link("k", "GHOST")
+
+    def test_dangling_evidence_detected(self):
+        ev = EvidenceRegister(self.docs)
+        # SOURCE_BASED dont la pièce n'est pas cataloguée → pendouillant
+        dp = DataPoint(key="k", value="x", category="INPUT",
+                       status="SOURCE_BASED", source_document="GHOST")
+        self.assertEqual(ev.dangling_evidence([dp]), ["k"])
+
+    def test_resolved_evidence_not_dangling(self):
+        ev = EvidenceRegister(self.docs)
+        dp = DataPoint(key="k", value="x", category="INPUT",
+                       status="SOURCE_BASED", source_document="D1")
+        self.assertEqual(ev.dangling_evidence([dp]), [])
+
+    def test_link_resolves_evidence(self):
+        ev = EvidenceRegister(self.docs)
+        dp = DataPoint(key="k", value="x", category="INPUT",
+                       status="SOURCE_BASED", source_document=None)
+        # pas de source_document mais un lien Evidence vers D1 → résout
+        ev.link("k", "D1")
+        self.assertEqual(ev.dangling_evidence([dp]), [])
+
+
+class TestPilotFixture(unittest.TestCase):
+    """Le fixture pilote P01 respecte le contrat et est un vrai sur-ensemble V12."""
+
+    def setUp(self):
+        self.intake = _load(FIXTURE)
+
+    def test_is_superset_of_v12_brief(self):
+        # champ requis V12 présent + version V18 + champs V18 d'extension
+        self.assertEqual(self.intake["projet"], "P01")
+        self.assertEqual(self.intake["dataset_version"], "V18")
+        self.assertIn("data_register", self.intake)
+
+    def test_synthetique_flagged(self):
+        self.assertTrue(self.intake.get("synthetique"),
+                        "un fixture de test doit être marqué synthetique (jamais publiable)")
+
+    def test_contract_holds_and_no_dangling(self):
+        points, ev = cli.build_registers(self.intake)
+        self.assertEqual(ev.dangling_evidence(points), [])
+        # honnêteté : la majorité des données sont des lacunes au démarrage
+        present = [p for p in points if p.present]
+        self.assertTrue(len(present) >= 1)
+
+    def test_cli_validate_returns_zero(self):
+        self.assertEqual(cli.cmd_validate(FIXTURE), 0)
+
+    def test_cli_maturity_returns_zero(self):
+        self.assertEqual(cli.cmd_maturity(FIXTURE), 0)
+
+    def test_jsonschema_oracle_if_present(self):
+        try:
+            import jsonschema  # noqa: F401
+        except Exception:
+            self.skipTest("jsonschema absent — oracle structurel ignoré (par design)")
+        errors = cli._schema_validate(self.intake)
+        self.assertEqual(errors, [], f"fixture non conforme : {errors}")
+
+
+class TestCliContractViolations(unittest.TestCase):
+    def test_bad_status_fixture_fails_validate(self):
+        import tempfile
+        bad = {
+            "projet": "P02",
+            "dataset_version": "V18",
+            "data_register": [
+                {"key": "x", "value": 1, "category": "INPUT", "status": "BOGUS"}
+            ],
+        }
+        with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False,
+                                         encoding="utf-8") as fh:
+            json.dump(bad, fh)
+            path = fh.name
+        try:
+            self.assertEqual(cli.cmd_validate(path), 1)
+        finally:
+            os.unlink(path)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)