diff --git a/tests/test_decision_engine_ext.py b/tests/test_decision_engine_ext.py new file mode 100644 index 0000000..90a94aa --- /dev/null +++ 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)