feat: tests étendus pour src/quality/decision_engine.py (53 tests)

Couvre les règles non couvertes dans le fichier existant : - Fonctions internes (_norm, _first_float, _parse_normal_range, etc.) - Nettoyage hiérarchique (.9 vs code spécifique) - D50 sans preuve martiale → downgrade D64.9 - D69.6 thrombopénie vs plaquettes normales → RULED_OUT - decision_summaries pour toutes les actions (DOWNGRADE, REMOVE, RULED_OUT, NEED_INFO) - Détection analytes (sodium, potassium) - _age_band et cas limites de scoring Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-08 11:59:57 +01:00
parent 5ba3903569
commit caaa6deb14
1 changed files with 493 additions and 0 deletions
--- a/tests/test_decision_engine_ext.py
+++ b/tests/test_decision_engine_ext.py
@@ -0,0 +1,493 @@
+"""Tests étendus pour src/quality/decision_engine.py.
+
+Couvre les règles non couvertes dans test_decision_engine.py :
+- Nettoyage hiérarchique (.9 vs spécifique)
+- D50 sans preuve martiale → downgrade D64.9
+- D69.6 thrombopénie vs plaquettes normales
+- Bio rules génériques (YAML-driven)
+- Fonctions internes (_norm, _first_float, _parse_normal_range, etc.)
+- decision_summaries (DOWNGRADE, REMOVE, RULED_OUT, NEED_INFO)
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pytest
+
+from src.config import (
+    BiologieCle,
+    CodeDecision,
+    Diagnostic,
+    DossierMedical,
+    PreuveClinique,
+    Sejour,
+    Traitement,
+    VetoReport,
+    VetoIssue,
+)
+from src.quality.decision_engine import (
+    _norm,
+    _first_float,
+    _parse_normal_range,
+    _parse_float,
+    _is_sodium_test,
+    _is_potassium_test,
+    _das_promotion_score,
+    _age_band,
+    apply_decisions,
+    decision_summaries,
+)
+
+
+# ── Helpers ────────────────────────────────────────────────────────────
+
+
+def _dossier(
+    dp_code: str | None = None,
+    das: list[Diagnostic] | None = None,
+    bio: list[BiologieCle] | None = None,
+    traitements: list[Traitement] | None = None,
+    age: int | None = None,
+    sexe: str | None = None,
+) -> DossierMedical:
+    dp = None
+    if dp_code:
+        dp = Diagnostic(texte="DP", cim10_suggestion=dp_code)
+    return DossierMedical(
+        sejour=Sejour(age=age, sexe=sexe),
+        diagnostic_principal=dp,
+        diagnostics_associes=das or [],
+        biologie_cle=bio or [],
+        traitements_sortie=traitements or [],
+    )
+
+
+def _diag(
+    texte: str,
+    code: str,
+    confidence: str = "high",
+    source: str = "trackare",
+    status: str | None = None,
+    source_excerpt: str | None = None,
+    preuves: list[PreuveClinique] | None = None,
+) -> Diagnostic:
+    return Diagnostic(
+        texte=texte,
+        cim10_suggestion=code,
+        cim10_confidence=confidence,
+        source=source,
+        status=status,
+        source_excerpt=source_excerpt,
+        preuves_cliniques=preuves or [],
+    )
+
+
+def _bio(test: str, valeur: str | None = None, valeur_num: float | None = None,
+         quality: str | None = None) -> BiologieCle:
+    return BiologieCle(test=test, valeur=valeur, valeur_num=valeur_num, quality=quality)
+
+
+# ===================================================================
+# Fonctions internes
+# ===================================================================
+
+
+class TestNorm:
+    def test_basic_normalization(self):
+        assert _norm("  Hello   World  ") == "hello world"
+
+    def test_accent_removal(self):
+        assert _norm("Hémoglobine élevée") == "hemoglobine elevee"
+
+    def test_apostrophe_normalization(self):
+        assert _norm("l\u2019anémie") == "l'anemie"
+
+
+class TestFirstFloat:
+    @pytest.mark.parametrize("text,expected", [
+        ("CRP 180 mg/L", 180.0),
+        ("Hb 7,5 g/dL", 7.5),
+        ("-3.2 mmol/L", -3.2),
+        ("pas de nombre", None),
+        ("", None),
+    ])
+    def test_extraction(self, text, expected):
+        assert _first_float(text) == expected
+
+
+class TestParseNormalRange:
+    def test_standard_range(self):
+        lo, hi = _parse_normal_range("Plaquettes 250 G/L [N: 150-450]")
+        assert lo == 150.0
+        assert hi == 450.0
+
+    def test_decimal_range(self):
+        lo, hi = _parse_normal_range("K+ 4.2 mmol/L [N: 3,5 - 5,0]")
+        assert lo == 3.5
+        assert hi == 5.0
+
+    def test_no_range(self):
+        lo, hi = _parse_normal_range("CRP 180 mg/L")
+        assert lo is None
+        assert hi is None
+
+
+class TestParseFloat:
+    @pytest.mark.parametrize("value,expected", [
+        ("4.5", 4.5),
+        ("4,5", 4.5),
+        ("  7.2  ", 7.2),
+        ("abc", None),
+        (None, None),
+        ("", None),
+        ("-3.1", -3.1),
+    ])
+    def test_parsing(self, value, expected):
+        assert _parse_float(value) == expected
+
+
+class TestIsSodiumTest:
+    @pytest.mark.parametrize("test,expected", [
+        ("Sodium", True),
+        ("sodium", True),
+        ("Natrémie", True),
+        ("Na+", True),
+        ("Na", True),
+        ("Potassium", False),
+        ("CRP", False),
+    ])
+    def test_detection(self, test, expected):
+        assert _is_sodium_test(test) == expected
+
+
+class TestIsPotassiumTest:
+    @pytest.mark.parametrize("test,expected", [
+        ("Potassium", True),
+        ("potassium", True),
+        ("Kaliémie", True),
+        ("K+", True),
+        ("K", True),
+        ("Sodium", False),
+        ("CRP", False),
+    ])
+    def test_detection(self, test, expected):
+        assert _is_potassium_test(test) == expected
+
+
+class TestAgeBand:
+    def test_adult(self):
+        d = _dossier(age=65)
+        assert _age_band(d, {"age_bands": {"adult_min_years": 18}}) == "adult"
+
+    def test_child(self):
+        d = _dossier(age=5)
+        assert _age_band(d, {"age_bands": {"adult_min_years": 18}}) == "child"
+
+    def test_unknown_age(self):
+        d = _dossier()
+        assert _age_band(d, {}) == "unknown"
+
+    def test_boundary_18(self):
+        d = _dossier(age=18)
+        assert _age_band(d, {"age_bands": {"adult_min_years": 18}}) == "adult"
+
+
+# ===================================================================
+# Nettoyage hiérarchique (.9 vs spécifique)
+# ===================================================================
+
+
+@patch("src.quality.decision_engine.load_reference_ranges", return_value={})
+@patch("src.quality.decision_engine.load_bio_rules", return_value={})
+@patch("src.quality.decision_engine.rule_enabled", return_value=True)
+@patch("src.quality.decision_engine.cim10_validate", return_value=(True, "label"))
+class TestHierarchyCleanup:
+
+    def test_generic_removed_when_specific_exists(self, mock_val, mock_rule, mock_bio, mock_ref):
+        """K81.9 (générique) retiré quand K81.0 (spécifique) est présent."""
+        dp = Diagnostic(texte="Cholécystite", cim10_suggestion="K81.0")
+        das1 = _diag("Cholécystite SAI", "K81.9")
+        das2 = _diag("HTA", "I10")
+        dossier = _dossier(dp_code=None, das=[das1, das2])
+        dossier.diagnostic_principal = dp
+
+        apply_decisions(dossier)
+
+        # K81.9 doit être supprimé
+        k81_9 = [d for d in dossier.diagnostics_associes if d.cim10_suggestion == "K81.9"][0]
+        assert k81_9.cim10_decision is not None
+        assert k81_9.cim10_decision.action == "REMOVE"
+        assert "RULE-HIERARCHY-CLEANUP" in k81_9.cim10_decision.applied_rules
+        assert k81_9.cim10_final is None
+
+    def test_no_cleanup_when_no_specific(self, mock_val, mock_rule, mock_bio, mock_ref):
+        """K81.9 seul (pas de K81.x spécifique) → conservé."""
+        das1 = _diag("Cholécystite SAI", "K81.9")
+        das2 = _diag("HTA", "I10")
+        dp = Diagnostic(texte="DP", cim10_suggestion="J18.9")
+        dossier = _dossier(das=[das1, das2])
+        dossier.diagnostic_principal = dp
+
+        apply_decisions(dossier)
+
+        k81_9 = [d for d in dossier.diagnostics_associes if d.cim10_suggestion == "K81.9"][0]
+        # Pas de décision REMOVE
+        assert k81_9.cim10_decision is None or k81_9.cim10_decision.action != "REMOVE"
+
+
+# ===================================================================
+# D50 sans preuve martiale → downgrade D64.9
+# ===================================================================
+
+
+@patch("src.quality.decision_engine.load_reference_ranges", return_value={})
+@patch("src.quality.decision_engine.load_bio_rules", return_value={})
+class TestD50NeedsIron:
+
+    @patch("src.quality.decision_engine.rule_enabled", return_value=True)
+    @patch("src.quality.decision_engine.cim10_validate", return_value=(True, "label"))
+    def test_d50_downgraded_without_iron_evidence(self, mock_val, mock_rule, mock_bio, mock_ref):
+        """D50 sans preuve ferritine/martial → downgrade D64.9."""
+        das = _diag(
+            "Anémie ferriprive",
+            "D50",
+            preuves=[
+                PreuveClinique(type="biologie", element="Hb 9.5 g/dL", interpretation="Anémie confirmée"),
+            ],
+        )
+        dossier = _dossier(dp_code="K85.9", das=[das], age=65)
+
+        apply_decisions(dossier)
+
+        d50 = [d for d in dossier.diagnostics_associes if d.cim10_suggestion == "D50"][0]
+        assert d50.cim10_final == "D64.9"
+        assert d50.cim10_decision.action == "DOWNGRADE"
+        assert "RULE-D50-NEEDS-IRON" in d50.cim10_decision.applied_rules
+
+    @patch("src.quality.decision_engine.rule_enabled", return_value=True)
+    @patch("src.quality.decision_engine.cim10_validate", return_value=(True, "label"))
+    def test_d50_kept_with_ferritin_evidence(self, mock_val, mock_rule, mock_bio, mock_ref):
+        """D50 avec preuve de ferritine → conservé."""
+        das = _diag(
+            "Anémie ferriprive",
+            "D50",
+            preuves=[
+                PreuveClinique(type="biologie", element="Hb 9.5 g/dL", interpretation="Anémie confirmée"),
+                PreuveClinique(type="biologie", element="Ferritine 8 ng/mL", interpretation="Carence martiale"),
+            ],
+        )
+        dossier = _dossier(dp_code="K85.9", das=[das], age=65)
+
+        apply_decisions(dossier)
+
+        d50 = [d for d in dossier.diagnostics_associes if d.cim10_suggestion == "D50"][0]
+        assert d50.cim10_decision is None or d50.cim10_decision.action != "DOWNGRADE"
+
+    @patch("src.quality.decision_engine.rule_enabled", return_value=True)
+    @patch("src.quality.decision_engine.cim10_validate", return_value=(True, "label"))
+    def test_d50_kept_with_martial_treatment(self, mock_val, mock_rule, mock_bio, mock_ref):
+        """D50 avec traitement martial → conservé (preuve par le traitement)."""
+        das = _diag(
+            "Anémie ferriprive",
+            "D50",
+            preuves=[
+                PreuveClinique(type="biologie", element="Hb 8.0 g/dL", interpretation="Anémie confirmée"),
+            ],
+        )
+        dossier = _dossier(
+            dp_code="K85.9",
+            das=[das],
+            age=65,
+            traitements=[Traitement(medicament="Fer intraveineux", posologie="200mg")],
+        )
+
+        apply_decisions(dossier)
+
+        d50 = [d for d in dossier.diagnostics_associes if d.cim10_suggestion == "D50"][0]
+        assert d50.cim10_decision is None or d50.cim10_decision.action != "DOWNGRADE"
+
+    def test_d50_rule_disabled(self, mock_bio, mock_ref):
+        """RULE-D50-NEEDS-IRON désactivée → D50 conservé."""
+        def _rule_selective(rule_id):
+            return rule_id != "RULE-D50-NEEDS-IRON"
+
+        with patch("src.quality.decision_engine.rule_enabled", side_effect=_rule_selective):
+            with patch("src.quality.decision_engine.cim10_validate", return_value=(True, "label")):
+                das = _diag(
+                    "Anémie ferriprive",
+                    "D50",
+                    preuves=[
+                        PreuveClinique(type="biologie", element="Hb 9.5", interpretation="Anémie confirmée"),
+                    ],
+                )
+                dossier = _dossier(dp_code="K85.9", das=[das], age=65)
+                apply_decisions(dossier)
+                d50 = [d for d in dossier.diagnostics_associes if d.cim10_suggestion == "D50"][0]
+                assert d50.cim10_decision is None or d50.cim10_decision.action != "DOWNGRADE"
+
+
+# ===================================================================
+# D69.6 thrombopénie vs plaquettes normales
+# ===================================================================
+
+
+@patch("src.quality.decision_engine.load_bio_rules", return_value={})
+class TestD696PltNormal:
+
+    @patch("src.quality.decision_engine.load_reference_ranges", return_value={
+        "age_bands": {"adult_min_years": 18},
+        "fallback_ranges": {"adult": {"platelets": {"low": 150, "high": 450}}},
+        "safe_zones_unknown_age": {"platelets_ruled_out_low": 170},
+    })
+    @patch("src.quality.decision_engine.rule_enabled", return_value=True)
+    @patch("src.quality.decision_engine.cim10_validate", return_value=(True, "label"))
+    def test_d696_ruled_out_with_normal_platelets(self, mock_val, mock_rule, mock_ref, mock_bio):
+        """D69.6 avec plaquettes normales → RULED_OUT."""
+        das = _diag("Thrombopénie", "D69.6")
+        bio = [_bio("Plaquettes", "250 G/L [N: 150-450]", valeur_num=250.0)]
+        dossier = _dossier(dp_code="K85.9", das=[das], bio=bio, age=65)
+
+        apply_decisions(dossier)
+
+        d696 = [d for d in dossier.diagnostics_associes if d.cim10_suggestion == "D69.6"][0]
+        assert d696.cim10_decision is not None
+        assert d696.cim10_decision.action == "RULED_OUT"
+        assert d696.status == "ruled_out"
+        assert d696.cim10_final is None
+
+    @patch("src.quality.decision_engine.load_reference_ranges", return_value={
+        "age_bands": {"adult_min_years": 18},
+        "fallback_ranges": {"adult": {"platelets": {"low": 150, "high": 450}}},
+        "safe_zones_unknown_age": {"platelets_ruled_out_low": 170},
+    })
+    @patch("src.quality.decision_engine.rule_enabled", return_value=True)
+    @patch("src.quality.decision_engine.cim10_validate", return_value=(True, "label"))
+    def test_d696_kept_with_low_platelets(self, mock_val, mock_rule, mock_ref, mock_bio):
+        """D69.6 avec plaquettes basses → conservé."""
+        das = _diag("Thrombopénie", "D69.6")
+        bio = [_bio("Plaquettes", "80 G/L", valeur_num=80.0)]
+        dossier = _dossier(dp_code="K85.9", das=[das], bio=bio, age=65)
+
+        apply_decisions(dossier)
+
+        d696 = [d for d in dossier.diagnostics_associes if d.cim10_suggestion == "D69.6"][0]
+        assert d696.cim10_decision is None or d696.cim10_decision.action != "RULED_OUT"
+
+
+# ===================================================================
+# _das_promotion_score — cas supplémentaires
+# ===================================================================
+
+
+class TestDasPromotionScoreExtra:
+    def test_empty_code(self):
+        """Un diagnostic sans code final a un score minimal."""
+        diag = Diagnostic(texte="Test", cim10_final="")
+        score = _das_promotion_score(diag)
+        # pas de lettre → pertinence=2, mais specificite=0
+        assert score[2] == 0
+
+    def test_none_confidence(self):
+        """Confiance None → score minimal."""
+        diag = Diagnostic(texte="Test", cim10_final="K85.9", cim10_confidence=None)
+        score = _das_promotion_score(diag)
+        assert score[1] == 1  # default low
+
+
+# ===================================================================
+# decision_summaries — toutes les actions
+# ===================================================================
+
+
+class TestDecisionSummaries:
+
+    def test_keep_no_output(self):
+        """KEEP ne produit pas de résumé."""
+        das = Diagnostic(texte="HTA", cim10_suggestion="I10", cim10_decision=CodeDecision(action="KEEP"))
+        dossier = DossierMedical(diagnostics_associes=[das])
+        lines = decision_summaries(dossier)
+        assert len(lines) == 0
+
+    def test_downgrade_summary(self):
+        das = Diagnostic(
+            texte="Anémie",
+            cim10_suggestion="D50",
+            cim10_decision=CodeDecision(
+                action="DOWNGRADE",
+                final_code="D64.9",
+                downgraded_from="D50",
+                applied_rules=["RULE-D50-NEEDS-IRON"],
+                needs_info=["Bilan martial disponible ?"],
+            ),
+        )
+        dossier = DossierMedical(diagnostics_associes=[das])
+        lines = decision_summaries(dossier)
+        assert any("D50" in l and "D64.9" in l for l in lines)
+        assert any("besoin_info" in l for l in lines)
+
+    def test_remove_summary(self):
+        das = Diagnostic(
+            texte="Cholécystite SAI",
+            cim10_suggestion="K81.9",
+            cim10_decision=CodeDecision(
+                action="REMOVE",
+                applied_rules=["RULE-HIERARCHY-CLEANUP"],
+            ),
+        )
+        dossier = DossierMedical(diagnostics_associes=[das])
+        lines = decision_summaries(dossier)
+        assert any("K81.9" in l and "supprimé" in l for l in lines)
+
+    def test_ruled_out_summary(self):
+        das = Diagnostic(
+            texte="Thrombopénie",
+            cim10_suggestion="D69.6",
+            cim10_decision=CodeDecision(
+                action="RULED_OUT",
+                applied_rules=["RULE-D69.6-PLT-NORMAL"],
+                reason="Plaquettes normales",
+            ),
+        )
+        dossier = DossierMedical(diagnostics_associes=[das])
+        lines = decision_summaries(dossier)
+        assert any("D69.6" in l and "écarté" in l for l in lines)
+        assert any("raison" in l for l in lines)
+
+    def test_need_info_summary(self):
+        das = Diagnostic(
+            texte="Hyponatrémie",
+            cim10_suggestion="E87.1",
+            cim10_decision=CodeDecision(
+                action="NEED_INFO",
+                applied_rules=["RULE-HYPONATREMIA-MISSING"],
+                reason="Sodium non extrait",
+                needs_info=["Valeurs de sodium ?"],
+            ),
+        )
+        dossier = DossierMedical(diagnostics_associes=[das])
+        lines = decision_summaries(dossier)
+        assert any("NEED_INFO" in l for l in lines)
+        assert any("besoin_info" in l for l in lines)
+
+    def test_no_diagnostics(self):
+        """Dossier sans diagnostics → liste vide."""
+        dossier = DossierMedical()
+        lines = decision_summaries(dossier)
+        assert lines == []
+
+    def test_dp_decision_included(self):
+        """Décision sur le DP est incluse dans le résumé."""
+        dp = Diagnostic(
+            texte="DP",
+            cim10_suggestion="D50",
+            cim10_decision=CodeDecision(
+                action="DOWNGRADE",
+                final_code="D64.9",
+                downgraded_from="D50",
+                applied_rules=["RULE-D50-NEEDS-IRON"],
+            ),
+        )
+        dossier = DossierMedical(diagnostic_principal=dp)
+        lines = decision_summaries(dossier)
+        assert any("diagnostic_principal" in l for l in lines)