"""Tests pour le module de scoring DP (Diagnostic Principal).""" import pytest from src.config import ( ActeCCAM, DossierMedical, Diagnostic, DPCandidate, DPPoolCandidate, DPSelection, DP_SCORING_WEIGHTS, DP_REVIEW_THRESHOLD, PreuveSynthese, SynthesePMSI, Sejour, ) from src.medical.dp_scoring import ( build_dp_shortlist, build_dp_candidate_pool, score_candidates, select_dp, generate_synthese_pmsi, llm_dp_pool_rank, _format_pool_for_prompt, _build_clinical_context, _get_context_window, _is_z_code_whitelisted, _is_comorbidity_code, _has_explicit_pec_proof, _dedup_by_code, _dedup_pool, _is_pool_excluded, _normalize_evidence_section, _is_comorbidite_banale, _has_pec_marker, _has_acute_problem, _postprocess_synthese, _build_motif, ) # --- Helpers --- def _make_parsed(sections: dict | None = None, diagnostics: list | None = None) -> dict: return { "type": "crh", "patient": {"sexe": "M"}, "sejour": {}, "diagnostics": diagnostics or [], "sections": sections or {}, } def _make_candidate( code: str = "K85.1", label: str = "Pancréatite aiguë biliaire", source_section: str = "diag_sortie", **kwargs, ) -> DPCandidate: return DPCandidate(code=code, label=label, source_section=source_section, **kwargs) # === Tests build_dp_shortlist === class TestBuildDPShortlist: def test_from_diag_sortie_with_cim10_code(self): parsed = _make_parsed(sections={ "diag_sortie": "Pancréatite aiguë biliaire K85.1", }) dossier = DossierMedical() candidates = build_dp_shortlist(parsed, "", None, dossier) codes = [c.code for c in candidates] assert "K85.1" in codes def test_from_diag_principal_section(self): parsed = _make_parsed(sections={ "diag_principal": "Embolie pulmonaire I26.9", }) dossier = DossierMedical() candidates = build_dp_shortlist(parsed, "", None, dossier) codes = [c.code for c in candidates] assert "I26.9" in codes def test_from_conclusion_via_cim10_map(self): parsed = _make_parsed(sections={ "conclusion": "pancréatite aiguë biliaire, bonne évolution", }) dossier = DossierMedical() candidates = build_dp_shortlist(parsed, "", None, dossier) codes = [c.code for c in candidates] assert "K85.1" in codes def test_from_regex_fallback(self): parsed = _make_parsed(sections={}) text = "Au total : pancréatite aiguë biliaire.\nDevenir : retour." dossier = DossierMedical() candidates = build_dp_shortlist(parsed, text, None, dossier) codes = [c.code for c in candidates] assert "K85.1" in codes def test_from_edsnlp(self): from src.medical.edsnlp_pipeline import EdsnlpResult, CIM10Entity parsed = _make_parsed(sections={}) edsnlp = EdsnlpResult(cim10_entities=[ CIM10Entity(texte="douleur abdominale", code="R10.4", negation=False), ]) dossier = DossierMedical() candidates = build_dp_shortlist(parsed, "", edsnlp, dossier) codes = [c.code for c in candidates] assert "R10.4" in codes def test_edsnlp_negated_excluded(self): from src.medical.edsnlp_pipeline import EdsnlpResult, CIM10Entity parsed = _make_parsed(sections={}) edsnlp = EdsnlpResult(cim10_entities=[ CIM10Entity(texte="fièvre", code="R50.9", negation=True), ]) dossier = DossierMedical() candidates = build_dp_shortlist(parsed, "", edsnlp, dossier) codes = [c.code for c in candidates] assert "R50.9" not in codes def test_dedup_keeps_strongest_section(self): """Si le même code vient de diag_sortie et conclusion, garder diag_sortie.""" parsed = _make_parsed(sections={ "diag_sortie": "Pancréatite K85.1", "conclusion": "pancréatite K85.1 bonne évolution", }) dossier = DossierMedical() candidates = build_dp_shortlist(parsed, "", None, dossier) k85_candidates = [c for c in candidates if c.code == "K85.1"] assert len(k85_candidates) == 1 assert k85_candidates[0].source_section == "diag_sortie" def test_empty_sections_returns_empty(self): parsed = _make_parsed(sections={}) dossier = DossierMedical() candidates = build_dp_shortlist(parsed, "Patient en bon état.", None, dossier) assert candidates == [] # === Tests score_candidates === class TestScoreCandidates: def test_section_bonus_diag_sortie(self): c = _make_candidate(source_section="diag_sortie") scored = score_candidates([c], DossierMedical()) assert scored[0].score_details.get("section") == DP_SCORING_WEIGHTS["section_diag_sortie"] def test_section_bonus_conclusion(self): c = _make_candidate(source_section="conclusion") scored = score_candidates([c], DossierMedical()) assert scored[0].score_details.get("section") == DP_SCORING_WEIGHTS["section_conclusion"] def test_section_bonus_edsnlp(self): c = _make_candidate(source_section="edsnlp") scored = score_candidates([c], DossierMedical()) assert scored[0].score_details.get("section") == DP_SCORING_WEIGHTS["section_edsnlp"] def test_proof_excerpt_bonus(self): c = _make_candidate(source_excerpt="Pancréatite aiguë biliaire confirmée au scanner") scored = score_candidates([c], DossierMedical()) assert scored[0].score_details.get("proof_excerpt") == DP_SCORING_WEIGHTS["proof_excerpt"] def test_no_proof_bonus_without_excerpt(self): c = _make_candidate(source_excerpt=None) scored = score_candidates([c], DossierMedical()) assert "proof_excerpt" not in scored[0].score_details def test_negation_penalty(self): c = _make_candidate(label="Fièvre") text = "Pas de fièvre constatée." scored = score_candidates([c], DossierMedical(), full_text=text) assert scored[0].is_negated is True assert scored[0].score_details.get("negation") == DP_SCORING_WEIGHTS["negation"] def test_conditional_penalty(self): c = _make_candidate(label="Embolie pulmonaire", code="I26.9") text = "Embolie pulmonaire suspectée, à confirmer par angioscanner." scored = score_candidates([c], DossierMedical(), full_text=text) assert scored[0].is_conditional is True assert scored[0].score_details.get("conditional") == DP_SCORING_WEIGHTS["conditional"] def test_z_code_penalty(self): c = _make_candidate(code="Z76.0", label="Bilan de santé", source_section="conclusion") scored = score_candidates([c], DossierMedical()) assert scored[0].score_details.get("z_code_dp") == DP_SCORING_WEIGHTS["z_code_dp"] def test_z_code_whitelist_no_penalty(self): c = _make_candidate(code="Z51.1", label="Chimiothérapie", source_section="conclusion") scored = score_candidates([c], DossierMedical()) assert "z_code_dp" not in scored[0].score_details def test_r_code_penalty(self): c = _make_candidate(code="R10.4", label="Douleur abdominale", source_section="edsnlp") scored = score_candidates([c], DossierMedical()) assert scored[0].score_details.get("r_code_dp") == DP_SCORING_WEIGHTS["r_code_dp"] def test_sort_by_score_descending(self): c1 = _make_candidate(code="K85.1", source_section="diag_sortie") c2 = _make_candidate(code="R10.4", label="Douleur", source_section="edsnlp") scored = score_candidates([c2, c1], DossierMedical()) assert scored[0].code == "K85.1" # diag_sortie score > edsnlp def test_combined_scoring(self): """Score = section bonus + proof - negation penalties.""" c = _make_candidate( code="K85.1", source_section="diag_sortie", source_excerpt="Pancréatite aiguë", ) scored = score_candidates([c], DossierMedical()) expected = DP_SCORING_WEIGHTS["section_diag_sortie"] + DP_SCORING_WEIGHTS["proof_excerpt"] assert scored[0].score == expected # === Tests select_dp === class TestSelectDP: def test_no_candidates_returns_review(self): sel = select_dp([], DossierMedical()) assert sel.verdict == "review" def test_single_candidate_confirmed(self): c = _make_candidate() c.score = 6 sel = select_dp([c], DossierMedical()) assert sel.verdict == "confirmed" assert sel.winner_reason == "candidat unique" def test_clear_winner_confirmed(self): c1 = _make_candidate(code="K85.1") c1.score = 6 c2 = _make_candidate(code="R10.4", label="Douleur", source_section="edsnlp") c2.score = 1 sel = select_dp([c1, c2], DossierMedical()) assert sel.verdict == "confirmed" assert "delta" in sel.winner_reason def test_close_scores_returns_review(self): c1 = _make_candidate(code="K85.1") c1.score = 3 c2 = _make_candidate(code="K80.5", label="Lithiase", source_section="conclusion") c2.score = 2 sel = select_dp([c1, c2], DossierMedical()) assert sel.verdict == "review" def test_review_returns_top3(self): candidates = [ _make_candidate(code=f"K8{i}.{i}", label=f"Diag {i}") for i in range(5) ] for i, c in enumerate(candidates): c.score = 5 - i # delta between top1 and top2 = 1, < DP_REVIEW_THRESHOLD sel = select_dp(candidates, DossierMedical()) assert sel.verdict == "review" assert len(sel.candidates) <= 3 # === Tests utilitaires === class TestContextWindow: def test_finds_label_in_text(self): text = "Patient admis pour pancréatite aiguë biliaire confirmée." window = _get_context_window(text, "pancréatite aiguë", radius=50) assert "pancréatite" in window.lower() def test_returns_empty_when_not_found(self): text = "Patient en bon état." window = _get_context_window(text, "embolie pulmonaire") assert window == "" class TestZCodeWhitelist: def test_z51_1_whitelisted(self): assert _is_z_code_whitelisted("Z51.1") is True def test_z45_prefix_whitelisted(self): assert _is_z_code_whitelisted("Z45.80") is True def test_z76_not_whitelisted(self): assert _is_z_code_whitelisted("Z76.0") is False class TestDedupByCode: def test_dedup_same_code_keeps_strongest(self): c1 = _make_candidate(code="K85.1", source_section="conclusion") c2 = _make_candidate(code="K85.1", source_section="diag_sortie") priority = ["diag_sortie", "diag_principal", "motif_hospitalisation", "conclusion", "synthese"] result = _dedup_by_code([c1, c2], priority) assert len(result) == 1 assert result[0].source_section == "diag_sortie" def test_dedup_different_codes_kept(self): c1 = _make_candidate(code="K85.1") c2 = _make_candidate(code="K80.5", label="Lithiase") priority = ["diag_sortie"] result = _dedup_by_code([c1, c2], priority) assert len(result) == 2 # === Tests intégration légère === class TestDPScoringIntegration: def test_crh_with_diag_sortie_section(self): """Un CRH avec section 'Diagnostic de sortie' produit un dp_selection.""" from src.medical.cim10_extractor import extract_medical_info parsed = { "type": "crh", "patient": {"sexe": "M"}, "sejour": {}, "diagnostics": [], "sections": { "diag_sortie": "Pancréatite aiguë biliaire K85.1", }, } text = "Diagnostic de sortie :\nPancréatite aiguë biliaire K85.1\n\nTraitement de sortie :\nParacétamol" dossier = extract_medical_info(parsed, text) assert dossier.diagnostic_principal is not None assert dossier.diagnostic_principal.cim10_suggestion == "K85.1" assert dossier.dp_selection is not None assert dossier.dp_selection.verdict == "confirmed" def test_llm_fallback_confirmed_high_strong_section(self): """LLM one-shot CONFIRMED : high confidence + section forte.""" from unittest.mock import patch from src.medical.cim10_extractor import extract_medical_info parsed = { "type": "crh", "patient": {"sexe": "M"}, "sejour": {}, "diagnostics": [], "sections": { "conclusion": "Pancréatite aiguë biliaire avec HTA connue.", }, } text = "Conclusion : Pancréatite aiguë biliaire avec HTA connue." mock_result = { "dp_code": "K85.1", "dp_label": "Pancréatite aiguë biliaire", "evidence_section": "conclusion", "evidence_excerpt": "Pancréatite aiguë biliaire", "confidence": "high", } with patch("src.medical.ollama_client.call_ollama", return_value=mock_result): dossier = extract_medical_info(parsed, text, use_rag=True) assert dossier.dp_selection is not None assert dossier.dp_selection.verdict == "confirmed" assert dossier.diagnostic_principal is not None assert dossier.diagnostic_principal.cim10_suggestion == "K85.1" def test_llm_fallback_confirmed_conclusion_section(self): """LLM one-shot CONFIRMED : conclusion est section forte.""" from unittest.mock import patch from src.medical.cim10_extractor import extract_medical_info parsed = { "type": "crh", "patient": {"sexe": "M"}, "sejour": {}, "diagnostics": [], "sections": {"conclusion": "Pneumopathie avec insuffisance rénale aiguë."}, } text = "Conclusion : Pneumopathie avec insuffisance rénale aiguë." mock_result = { "dp_code": "J18.9", "dp_label": "Pneumopathie, sans précision", "evidence_section": "conclusion", "evidence_excerpt": "Pneumopathie avec insuffisance rénale aiguë", "confidence": "high", } with patch("src.medical.ollama_client.call_ollama", return_value=mock_result): dossier = extract_medical_info(parsed, text, use_rag=True) assert dossier.dp_selection is not None assert dossier.dp_selection.verdict == "confirmed" assert dossier.diagnostic_principal is not None def test_llm_fallback_review_weak_section(self): """LLM one-shot REVIEW : evidence de histoire_maladie (section faible) → guardrail.""" from unittest.mock import patch from src.medical.dp_scoring import llm_dp_fallback from src.config import DossierMedical, DPCandidate parsed = {"type": "crh", "sections": {"histoire_maladie": "Dyspnée aiguë."}} text = "Histoire de la maladie : Dyspnée aiguë." dossier = DossierMedical() dp_candidates = [DPCandidate(code="R06.0", label="Dyspnée", source_section="edsnlp")] mock_result = { "dp_code": "R06.0", "dp_label": "Dyspnée", "evidence_section": "histoire_maladie", "evidence_excerpt": "Dyspnée aiguë", "confidence": "high", } with patch("src.medical.ollama_client.call_ollama", return_value=mock_result): selection = llm_dp_fallback(parsed, text, dossier, dp_candidates=dp_candidates) assert selection.verdict == "review" assert len(selection.candidates) >= 1 def test_llm_fallback_review_low_confidence(self): """LLM one-shot REVIEW : confidence=medium → guardrail.""" from unittest.mock import patch from src.medical.dp_scoring import llm_dp_fallback from src.config import DossierMedical, DPCandidate parsed = {"type": "crh", "sections": {"conclusion": "HTA connue, diabète équilibré."}} text = "Conclusion : HTA connue, diabète équilibré." dossier = DossierMedical() dp_candidates = [DPCandidate(code="I10", label="HTA", source_section="edsnlp")] mock_result = { "dp_code": "I10", "dp_label": "Hypertension essentielle", "evidence_section": "conclusion", "evidence_excerpt": "HTA connue", "confidence": "medium", } with patch("src.medical.ollama_client.call_ollama", return_value=mock_result): selection = llm_dp_fallback(parsed, text, dossier, dp_candidates=dp_candidates) assert selection.verdict == "review" assert "confidence medium" in selection.winner_reason def test_llm_fallback_guardrail_no_evidence(self): """Garde-fou : LLM renvoie evidence vide → REVIEW.""" from unittest.mock import patch from src.medical.dp_scoring import llm_dp_fallback from src.config import DossierMedical, DPCandidate parsed = {"type": "crh", "sections": {"conclusion": "Pancréatite."}} text = "Conclusion : Pancréatite." dossier = DossierMedical() dp_candidates = [DPCandidate(code="K85.9", label="Pancréatite", source_section="edsnlp")] mock_result = { "dp_code": "K85.9", "dp_label": "Pancréatite aiguë", "evidence_section": "conclusion", "evidence_excerpt": "", "confidence": "high", } with patch("src.medical.ollama_client.call_ollama", return_value=mock_result): selection = llm_dp_fallback(parsed, text, dossier, dp_candidates=dp_candidates) assert selection.verdict == "review" def test_llm_fallback_guardrail_comorbidity_weak_section(self): """Garde-fou : HTA en section non-forte → REVIEW.""" from unittest.mock import patch from src.medical.dp_scoring import llm_dp_fallback from src.config import DossierMedical, DPCandidate parsed = {"type": "crh", "sections": {"histoire_maladie": "Patient hypertendu."}} text = "Histoire de la maladie : Patient hypertendu." dossier = DossierMedical() dp_candidates = [DPCandidate(code="I10", label="HTA", source_section="edsnlp")] mock_result = { "dp_code": "I10", "dp_label": "Hypertension essentielle", "evidence_section": "histoire_maladie", "evidence_excerpt": "Patient hypertendu", "confidence": "high", } with patch("src.medical.ollama_client.call_ollama", return_value=mock_result): selection = llm_dp_fallback(parsed, text, dossier, dp_candidates=dp_candidates) assert selection.verdict == "review" def test_llm_fallback_comorbidity_in_strong_section(self): """I10 en section forte + high confidence → CONFIRMED (garde-fou GF-2 ne bloque pas).""" from unittest.mock import patch from src.medical.dp_scoring import llm_dp_fallback from src.config import DossierMedical, DPCandidate parsed = {"type": "crh", "sections": {"motif_hospitalisation": "HTA maligne."}} text = "Motif d'hospitalisation : HTA maligne." dossier = DossierMedical() dp_candidates = [DPCandidate(code="I10", label="HTA", source_section="edsnlp")] mock_result = { "dp_code": "I10", "dp_label": "Hypertension essentielle", "evidence_section": "motif_hospitalisation", "evidence_excerpt": "HTA maligne", "confidence": "high", } with patch("src.medical.ollama_client.call_ollama", return_value=mock_result): selection = llm_dp_fallback(parsed, text, dossier, dp_candidates=dp_candidates) assert selection.verdict == "confirmed" assert selection.candidates[0].code == "I10" def test_no_llm_fallback_without_use_rag(self): """Sans use_rag, le fallback LLM ne se déclenche PAS.""" from src.medical.cim10_extractor import extract_medical_info parsed = { "type": "crh", "patient": {"sexe": "M"}, "sejour": {}, "diagnostics": [], "sections": {"conclusion": "Bonne évolution."}, } text = "Conclusion : Bonne évolution." dossier = extract_medical_info(parsed, text, use_rag=False) # Sans use_rag → pas de fallback LLM → verdict review assert dossier.dp_selection is not None assert dossier.dp_selection.verdict == "review" def test_trackare_dp_bypasses_scoring(self): """Un Trackare avec DP codé ne déclenche PAS le scoring.""" from src.medical.cim10_extractor import extract_medical_info parsed = { "type": "trackare", "patient": {"sexe": "F"}, "sejour": {"date_entree": "01/01/2024", "date_sortie": "05/01/2024"}, "diagnostics": [ {"type": "Principal", "code_cim10": "K80.5", "libelle": "Calcul des canaux biliaires"}, ], } text = "Calcul des canaux biliaires." dossier = extract_medical_info(parsed, text) assert dossier.diagnostic_principal is not None assert dossier.diagnostic_principal.cim10_suggestion == "K80.5" assert dossier.dp_selection is None # Trackare DP, pas de scoring # === Tests comorbidité-banale DP === class TestComorbidityGuard: """Règle comorbidité-banale : I10/E66.x/E78.x/E11.x/D64.9 en DP → REVIEW sauf preuve explicite de PEC principale.""" def test_is_comorbidity_expanded(self): """La liste élargie couvre I10, E66.*, E78.*, E11.*, D64.9.""" assert _is_comorbidity_code("I10") is True assert _is_comorbidity_code("E66.0") is True assert _is_comorbidity_code("E66.9") is True assert _is_comorbidity_code("E78.0") is True assert _is_comorbidity_code("E11.9") is True assert _is_comorbidity_code("E11.0") is True assert _is_comorbidity_code("D64.9") is True # Pas comorbidité assert _is_comorbidity_code("D64.0") is False assert _is_comorbidity_code("E10.9") is False assert _is_comorbidity_code("K85.1") is False def test_sole_comorbidity_review(self): """Candidat unique comorbidité → REVIEW (même section forte).""" c = _make_candidate(code="E66.0", label="Obésité", source_section="conclusion") c.score = 4 c.score_details = {"section": 2, "proof_excerpt": 2, "comorbidity_weak": -3} sel = select_dp([c], DossierMedical()) assert sel.verdict == "review" assert "comorbidité banale" in sel.winner_reason def test_comorbidity_top1_multi_review(self): """Comorbidité top1 parmi plusieurs → REVIEW.""" c1 = _make_candidate(code="I10", label="Hta", source_section="motif_hospitalisation") c1.score = 3 c1.score_details = {"section": 3, "comorbidity_weak": -3} c2 = _make_candidate(code="K85.1", label="Pancréatite", source_section="edsnlp") c2.score = 1 sel = select_dp([c1, c2], DossierMedical()) assert sel.verdict == "review" assert "comorbidité banale" in sel.winner_reason def test_comorbidity_with_pec_proof_confirmed(self): """Comorbidité + preuve PEC → CONFIRMED.""" c = _make_candidate(code="I10", label="Hta", source_section="motif_hospitalisation") c.score = 3 c.score_details = {"section": 3, "comorbidity_weak": -3, "comorbidity_pec_proof": 3} sel = select_dp([c], DossierMedical()) assert sel.verdict == "confirmed" assert sel.winner_reason == "candidat unique" def test_non_comorbidity_sole_confirmed(self): """Candidat unique non-comorbidité → CONFIRMED (pas affecté).""" c = _make_candidate(code="K85.1", label="Pancréatite", source_section="conclusion") c.score = 4 sel = select_dp([c], DossierMedical()) assert sel.verdict == "confirmed" def test_score_comorbidity_penalty_strong_section(self): """Comorbidité pénalisée même en section forte (conclusion).""" c = _make_candidate(code="E66.0", label="Obésité", source_section="conclusion") scored = score_candidates([c], DossierMedical()) assert "comorbidity_weak" in scored[0].score_details assert scored[0].score_details["comorbidity_weak"] == DP_SCORING_WEIGHTS["comorbidity_weak"] def test_score_comorbidity_penalty_motif(self): """Comorbidité pénalisée en motif_hospitalisation.""" c = _make_candidate(code="I10", label="Hta", source_section="motif_hospitalisation") scored = score_candidates([c], DossierMedical()) assert "comorbidity_weak" in scored[0].score_details def test_pec_proof_detected(self): """PEC proof détectée dans le texte → bonus dans score_details.""" c = _make_candidate(code="I10", label="Hta", source_section="motif_hospitalisation") text = "Patient hospitalisé pour hta maligne résistante au traitement." scored = score_candidates([c], DossierMedical(), full_text=text) assert "comorbidity_pec_proof" in scored[0].score_details assert scored[0].score_details["comorbidity_pec_proof"] > 0 def test_pec_proof_not_found(self): """Pas de PEC proof → pas de bonus.""" c = _make_candidate(code="E66.0", label="Obésité", source_section="conclusion") text = "Patient obèse, pneumopathie communautaire." scored = score_candidates([c], DossierMedical(), full_text=text) assert "comorbidity_pec_proof" not in scored[0].score_details def test_has_explicit_pec_proof_hospitalized(self): """Détection 'hospitalisé pour' + label.""" assert _has_explicit_pec_proof("hta", "Patient hospitalisé pour HTA maligne.") is True def test_has_explicit_pec_proof_prise_en_charge(self): """Détection 'prise en charge' + label.""" assert _has_explicit_pec_proof("obésité", "Prise en charge de l'obésité morbide.") is True def test_has_explicit_pec_proof_absent(self): """Pas de PEC proof pour un label non mentionné.""" assert _has_explicit_pec_proof("hta", "Patient admis pour douleur thoracique.") is False def test_has_explicit_pec_proof_admission(self): """Détection 'admission pour' + label.""" assert _has_explicit_pec_proof("diabète", "Admission pour diabète déséquilibré.") is True class TestSectionNormalization: """Tests pour _normalize_evidence_section — normalisation robuste.""" # --- Correspondances exactes existantes --- def test_exact_conclusion(self): assert _normalize_evidence_section("conclusion") == "conclusion" def test_exact_synthese(self): assert _normalize_evidence_section("synthèse") == "synthese" def test_exact_motif_hospitalisation(self): assert _normalize_evidence_section("motif_hospitalisation") == "motif_hospitalisation" # --- Nouveaux alias exacts --- def test_synthese_du_sejour(self): assert _normalize_evidence_section("synthèse du séjour") == "synthese" def test_synthese_du_sejour_ascii(self): assert _normalize_evidence_section("synthese du sejour") == "synthese" def test_conclusions_pluriel(self): assert _normalize_evidence_section("conclusions") == "conclusion" def test_secretariat_to_autres(self): assert _normalize_evidence_section("secrétariat") == "autres" def test_medecine_interne_to_autres(self): assert _normalize_evidence_section("médecine interne") == "autres" def test_sections_cliniques_to_autres(self): assert _normalize_evidence_section("sections cliniques") == "autres" # --- Nettoyage crochets/guillemets --- def test_brackets_conclusion(self): assert _normalize_evidence_section("[conclusion]") == "conclusion" def test_brackets_motif(self): assert _normalize_evidence_section("[motif_hospitalisation]") == "motif_hospitalisation" def test_colon_conclusion(self): assert _normalize_evidence_section("conclusion:") == "conclusion" def test_quotes_synthese(self): assert _normalize_evidence_section('"synthèse"') == "synthese" # --- Fallback par mots-clés --- def test_keyword_conclusion_du_sejour(self): assert _normalize_evidence_section("conclusion du séjour") == "conclusion" def test_keyword_synthese_medicale(self): assert _normalize_evidence_section("synthèse médicale du dossier") == "synthese" def test_keyword_diagnostic_de_sortie_variant(self): assert _normalize_evidence_section("diagnostic(s) de sortie") == "diag_sortie" def test_keyword_diagnostic_retenu_variant(self): assert _normalize_evidence_section("diagnostics retenus à la sortie") == "diagnostics_retenus" def test_keyword_motif_admission(self): assert _normalize_evidence_section("motif d'admission aux urgences") == "motif_hospitalisation" # --- Cas limites --- def test_empty_string(self): assert _normalize_evidence_section("") == "" def test_none_like_empty(self): assert _normalize_evidence_section(" ") == "" def test_unknown_section_passthrough(self): """Section inconnue sans mot-clé → passthrough nettoyé.""" result = _normalize_evidence_section("biologie") assert result == "biologie" def test_sections_fortes_du_dossier(self): """Alias administratif observé en benchmark.""" assert _normalize_evidence_section("sections fortes du dossier") == "autres" class TestSynthesePMSI: """Tests pour generate_synthese_pmsi().""" def test_returns_synthese_on_valid_response(self, monkeypatch): """Réponse LLM valide → SynthesePMSI complète.""" mock_response = { "motif_admission": "Douleur abdominale aiguë", "probleme_pris_en_charge": "Pancréatite aiguë biliaire", "diagnostic_retenu": "Pancréatite aiguë d'origine biliaire", "actes_ou_traitements_majeurs": ["Scanner abdominal", "Mise à jeun"], "complications": [], "terrain_comorbidites": ["HTA traitée", "Diabète type 2"], "preuves": [ {"section": "motif_hospitalisation", "excerpt": "douleur abdominale intense"}, {"section": "conclusion", "excerpt": "pancréatite aiguë biliaire confirmée"}, ], } def mock_call_ollama(prompt, **kwargs): return mock_response import src.medical.dp_scoring as mod monkeypatch.setattr(mod, "call_ollama", mock_call_ollama, raising=False) # Forcer l'import inline à utiliser notre mock import src.medical.ollama_client as oc_mod monkeypatch.setattr(oc_mod, "call_ollama", mock_call_ollama) parsed = _make_parsed(sections={"conclusion": "pancréatite aiguë biliaire confirmée"}) dossier = DossierMedical() result = generate_synthese_pmsi(parsed, "texte complet", dossier) assert result is not None assert isinstance(result, SynthesePMSI) assert result.probleme_pris_en_charge == "Pancréatite aiguë biliaire" assert result.motif_admission == "Douleur abdominale aiguë" assert "Scanner abdominal" in result.actes_ou_traitements_majeurs assert len(result.terrain_comorbidites) == 2 assert result.complications == [] assert len(result.preuves) == 2 assert result.preuves[0].section == "motif_hospitalisation" def test_returns_none_on_invalid_response(self, monkeypatch): """Réponse LLM non-dict → None.""" def mock_call_ollama(prompt, **kwargs): return "texte brut" import src.medical.ollama_client as oc_mod monkeypatch.setattr(oc_mod, "call_ollama", mock_call_ollama) parsed = _make_parsed() dossier = DossierMedical() result = generate_synthese_pmsi(parsed, "texte", dossier) assert result is None def test_returns_none_on_exception(self, monkeypatch): """Exception LLM → None.""" def mock_call_ollama(prompt, **kwargs): raise ConnectionError("Ollama down") import src.medical.ollama_client as oc_mod monkeypatch.setattr(oc_mod, "call_ollama", mock_call_ollama) parsed = _make_parsed() dossier = DossierMedical() result = generate_synthese_pmsi(parsed, "texte", dossier) assert result is None def test_robust_to_string_lists(self, monkeypatch): """Le LLM renvoie des strings au lieu de listes → toléré.""" mock_response = { "motif_admission": "Fièvre", "probleme_pris_en_charge": "Pneumopathie", "diagnostic_retenu": "Pneumopathie bactérienne", "actes_ou_traitements_majeurs": "Antibiothérapie IV", # string "complications": "Insuffisance respiratoire", # string "terrain_comorbidites": "BPCO", # string "preuves": [], } def mock_call_ollama(prompt, **kwargs): return mock_response import src.medical.ollama_client as oc_mod monkeypatch.setattr(oc_mod, "call_ollama", mock_call_ollama) parsed = _make_parsed() dossier = DossierMedical() result = generate_synthese_pmsi(parsed, "texte", dossier) assert result is not None assert result.actes_ou_traitements_majeurs == ["Antibiothérapie IV"] assert result.complications == ["Insuffisance respiratoire"] assert result.terrain_comorbidites == ["BPCO"] def test_preuves_malformed_skipped(self, monkeypatch): """Preuves sans section/excerpt → ignorées.""" mock_response = { "motif_admission": "Test", "probleme_pris_en_charge": "Test", "diagnostic_retenu": "Test", "preuves": [ {"section": "conclusion", "excerpt": "valide"}, {"section": "", "excerpt": "section vide"}, {"no_section": True}, "pas un dict", ], } def mock_call_ollama(prompt, **kwargs): return mock_response import src.medical.ollama_client as oc_mod monkeypatch.setattr(oc_mod, "call_ollama", mock_call_ollama) parsed = _make_parsed() dossier = DossierMedical() result = generate_synthese_pmsi(parsed, "texte", dossier) assert result is not None assert len(result.preuves) == 1 assert result.preuves[0].section == "conclusion" def test_serialization_round_trip(self): """SynthesePMSI se sérialise/désérialise correctement.""" syn = SynthesePMSI( motif_admission="Douleur thoracique", probleme_pris_en_charge="Infarctus du myocarde", diagnostic_retenu="IDM ST+ antérieur", actes_ou_traitements_majeurs=["Coronarographie", "Angioplastie"], complications=["Insuffisance cardiaque"], terrain_comorbidites=["HTA", "Tabagisme"], preuves=[PreuveSynthese(section="conclusion", excerpt="IDM confirmé")], ) data = syn.model_dump() restored = SynthesePMSI(**data) assert restored.probleme_pris_en_charge == "Infarctus du myocarde" assert len(restored.preuves) == 1 assert restored.preuves[0].section == "conclusion" def test_dossier_medical_field(self): """Le champ synthese_pmsi est disponible sur DossierMedical.""" dossier = DossierMedical() assert dossier.synthese_pmsi is None dossier.synthese_pmsi = SynthesePMSI( probleme_pris_en_charge="Test", ) assert dossier.synthese_pmsi.probleme_pris_en_charge == "Test" data = dossier.model_dump(exclude_none=True) assert "synthese_pmsi" in data # =========================================================================== # DP Candidate Pool # =========================================================================== class TestDPPoolCandidate: """Tests du modèle DPPoolCandidate.""" def test_basic_creation(self): c = DPPoolCandidate(terme="Pancréatite aiguë", section="conclusion") assert c.terme == "Pancréatite aiguë" assert c.section == "conclusion" assert c.score_initial == 0.0 assert c.preuve == "" def test_serialization(self): c = DPPoolCandidate( terme="Cholécystite aiguë", section="diag_sortie", preuve="cholécystite aiguë lithiasique", score_initial=0.9, ) data = c.model_dump() restored = DPPoolCandidate(**data) assert restored.terme == "Cholécystite aiguë" assert restored.score_initial == 0.9 class TestIsPoolExcluded: """Tests du filtrage des candidats pool.""" def test_bio_value_excluded(self): assert _is_pool_excluded("CRP 180 mg/L") is True def test_bio_term_with_number_excluded(self): assert _is_pool_excluded("Hémoglobine 7.2 g/dL") is True def test_vague_symptom_excluded(self): assert _is_pool_excluded("douleur") is True assert _is_pool_excluded("fièvre") is True def test_vague_symptom_with_context_kept(self): """Symptôme qualifié (multi-mots) → conservé.""" assert _is_pool_excluded("douleur abdominale aiguë") is False def test_medical_diagnosis_kept(self): assert _is_pool_excluded("Pancréatite aiguë biliaire") is False def test_numeric_value_excluded(self): assert _is_pool_excluded("12.5 g/dL") is True class TestDedupPool: """Tests de la déduplication du pool.""" def test_dedup_keeps_highest_score(self): candidates = [ DPPoolCandidate(terme="Pancréatite aiguë", section="conclusion", score_initial=0.7), DPPoolCandidate(terme="Pancréatite aiguë", section="diag_sortie", score_initial=1.0), ] result = _dedup_pool(candidates) assert len(result) == 1 assert result[0].score_initial == 1.0 assert result[0].section == "diag_sortie" def test_dedup_normalizes_text(self): """Variantes d'accents/espaces → même clé.""" candidates = [ DPPoolCandidate(terme="Pancréatite aiguë", section="a", score_initial=0.5), DPPoolCandidate(terme="pancreatite aigue", section="b", score_initial=0.8), ] result = _dedup_pool(candidates) assert len(result) == 1 def test_distinct_terms_kept(self): candidates = [ DPPoolCandidate(terme="Pancréatite aiguë", section="a", score_initial=0.7), DPPoolCandidate(terme="Cholécystite aiguë", section="b", score_initial=0.9), ] result = _dedup_pool(candidates) assert len(result) == 2 class TestBuildDPCandidatePool: """Tests d'intégration de build_dp_candidate_pool().""" def test_indicative_phrase_extraction(self): """Les phrases indicatives sont extraites du texte.""" text = "Le patient a été hospitalisé pour pancréatite aiguë biliaire. Suivi habituel." parsed = _make_parsed(sections={"conclusion": "Pancréatite aiguë biliaire confirmée."}) dossier = DossierMedical() pool = build_dp_candidate_pool(parsed, text, None, dossier) termes = [c.terme.lower() for c in pool] assert any("pancréatite" in t or "pancreatite" in t for t in termes) def test_sections_fortes_extraction(self): """Les diagnostics des sections fortes apparaissent dans le pool.""" parsed = _make_parsed(sections={ "diag_sortie": "Cholécystite aiguë lithiasique", "conclusion": "Évolution favorable après cholécystectomie", }) dossier = DossierMedical() pool = build_dp_candidate_pool(parsed, "texte complet", None, dossier) termes = [c.terme.lower() for c in pool] assert any("cholécystite" in t or "cholecystite" in t for t in termes) def test_edsnlp_entities_included(self): """Les entités edsnlp non-niées apparaissent dans le pool.""" from dataclasses import dataclass @dataclass class MockEntity: texte: str code: str negation: bool = False hypothese: bool = False @dataclass class MockResult: cim10_entities: list edsnlp = MockResult(cim10_entities=[ MockEntity(texte="pneumopathie", code="J18.9"), MockEntity(texte="HTA", code="I10", negation=True), # exclu ]) parsed = _make_parsed() dossier = DossierMedical() pool = build_dp_candidate_pool(parsed, "texte", edsnlp, dossier) termes = [c.terme.lower() for c in pool] assert any("pneumopathie" in t for t in termes) # HTA niée ne doit pas apparaître assert not any(t == "hta" for t in termes) def test_actes_included(self): """Les actes CCAM du dossier apparaissent comme candidats.""" parsed = _make_parsed() dossier = DossierMedical() dossier.actes_ccam = [ ActeCCAM(texte="Cholécystectomie", code_ccam_suggestion="HMFC004"), ] pool = build_dp_candidate_pool(parsed, "texte", None, dossier) termes = [c.terme.lower() for c in pool] assert any("cholécystectomie" in t or "cholecystectomie" in t for t in termes) def test_cim10_map_matches(self): """Les termes CIM10_MAP matchés dans les sections fortes sont inclus.""" parsed = _make_parsed(sections={ "conclusion": "Patient avec pancréatite aiguë biliaire sévère.", }) dossier = DossierMedical() pool = build_dp_candidate_pool(parsed, "texte", None, dossier) sections = [c.section for c in pool] assert "cim10_map" in sections def test_bio_values_excluded(self): """Les valeurs biologiques ne polluent pas le pool.""" parsed = _make_parsed(sections={ "conclusion": "CRP 180 mg/L. Hémoglobine 7.2 g/dL. Pancréatite aiguë.", }) dossier = DossierMedical() pool = build_dp_candidate_pool(parsed, "texte", None, dossier) termes = [c.terme.lower() for c in pool] assert not any("crp" in t and "mg" in t for t in termes) def test_dedup_across_sources(self): """Un même terme de 2 sources → 1 seule entrée (meilleur score).""" parsed = _make_parsed(sections={ "conclusion": "Pancréatite aiguë biliaire confirmée.", "motif_hospitalisation": "Pancréatite aiguë biliaire.", }) dossier = DossierMedical() pool = build_dp_candidate_pool(parsed, "texte", None, dossier) # Compter les variantes "pancréatite aiguë biliaire" from src.medical.cim10_dict import normalize_text keys = [normalize_text(c.terme) for c in pool] pancreatite_keys = [k for k in keys if "pancreatite" in k and "biliaire" in k] # Après dedup, devrait être au plus 1-2 (phrase complète vs segment) assert len(pancreatite_keys) <= 2 def test_cap_at_30(self): """Le pool est plafonné à 30 candidats.""" # Créer un texte avec beaucoup de diagnostics diagnostics = [f"diagnostic numéro {i}" for i in range(50)] section_text = ". ".join(diagnostics) + "." parsed = _make_parsed(sections={"conclusion": section_text}) dossier = DossierMedical() pool = build_dp_candidate_pool(parsed, section_text, None, dossier) assert len(pool) <= 30 def test_empty_input(self): """Entrée vide → pool vide.""" parsed = _make_parsed() dossier = DossierMedical() pool = build_dp_candidate_pool(parsed, "", None, dossier) assert isinstance(pool, list) def test_score_ordering(self): """Le pool est trié par score_initial décroissant.""" parsed = _make_parsed(sections={ "diag_sortie": "Cholécystite aiguë", "conclusion": "Angiocholite associée", }) dossier = DossierMedical() pool = build_dp_candidate_pool(parsed, "texte", None, dossier) if len(pool) >= 2: scores = [c.score_initial for c in pool] assert scores == sorted(scores, reverse=True) # =========================================================================== # Anti-comorbidité SynthesePMSI # =========================================================================== class TestIsComorbBanale: """Tests de détection des comorbidités banales.""" def test_hta(self): assert _is_comorbidite_banale("HTA") == "HTA" assert _is_comorbidite_banale("hypertension artérielle") == "HTA" def test_diabete_stable(self): assert _is_comorbidite_banale("diabète de type 2") == "diabète" assert _is_comorbidite_banale("diabète") == "diabète" def test_diabete_decompense_not_banal(self): """Diabète déséquilibré ne doit PAS être considéré banal.""" assert _is_comorbidite_banale("diabète déséquilibré") is None assert _is_comorbidite_banale("diabète décompensé") is None assert _is_comorbidite_banale("acidocétose diabétique") is None def test_obesite(self): assert _is_comorbidite_banale("obésité") == "obésité" def test_anemie_chronique(self): assert _is_comorbidite_banale("anémie chronique") == "anémie" def test_anemie_severe_not_banal(self): """Anémie sévère ne doit PAS être banale.""" assert _is_comorbidite_banale("anémie sévère") is None assert _is_comorbidite_banale("anémie aiguë") is None def test_bpco_stable(self): assert _is_comorbidite_banale("BPCO") == "BPCO" def test_bpco_exacerbation_not_banal(self): """BPCO exacerbée ne doit PAS être banale.""" assert _is_comorbidite_banale("BPCO exacerbée") is None assert _is_comorbidite_banale("BPCO surinfectée") is None def test_non_comorbidite(self): assert _is_comorbidite_banale("pneumothorax") is None assert _is_comorbidite_banale("cholécystite aiguë") is None assert _is_comorbidite_banale("méningite à entérovirus") is None class TestHasPecMarker: """Tests des marqueurs de PEC principale.""" def test_hospitalise_pour(self): assert _has_pec_marker("diabète", "hospitalisé pour diabète déséquilibré") is True def test_desequilibre(self): assert _has_pec_marker("diabète déséquilibré", "") is True def test_acidocetose(self): assert _has_pec_marker("", "acidocétose diabétique") is True def test_transfusion(self): assert _has_pec_marker("anémie", "transfusion de 2 CGR") is True def test_no_marker(self): assert _has_pec_marker("diabète", "diabète type 2 équilibré") is False def test_hta_maligne(self): assert _has_pec_marker("HTA maligne", "") is True class TestHasAcuteProblem: """Tests de détection de problème aigu.""" def test_with_diagnostic_retenu(self): result = { "diagnostic_retenu": "pneumothorax spontané", "complications": [], "actes_ou_traitements_majeurs": [], } assert _has_acute_problem(result) is True def test_with_complications(self): result = { "diagnostic_retenu": "", "complications": ["insuffisance rénale aiguë"], "actes_ou_traitements_majeurs": [], } assert _has_acute_problem(result) is True def test_with_surgical_acte(self): result = { "diagnostic_retenu": "", "complications": [], "actes_ou_traitements_majeurs": ["cholécystectomie"], } assert _has_acute_problem(result) is True def test_only_surveillance(self): result = { "diagnostic_retenu": "", "complications": [], "actes_ou_traitements_majeurs": ["surveillance", "bilan biologique"], } assert _has_acute_problem(result) is False def test_diag_retenu_is_comorb(self): """Si diagnostic_retenu est aussi une comorbidité banale, pas de problème aigu via ce champ.""" result = { "diagnostic_retenu": "diabète", "complications": [], "actes_ou_traitements_majeurs": [], } assert _has_acute_problem(result) is False class TestPostprocessSynthese: """Tests du post-traitement anti-comorbidité.""" def test_non_comorbidite_untouched(self): """Un problème non-banal ne doit pas être modifié.""" result = { "probleme_pris_en_charge": "pneumothorax spontané", "diagnostic_retenu": "pneumothorax spontané", "terrain_comorbidites": ["HTA"], "complications": [], "actes_ou_traitements_majeurs": ["drainage"], "preuves": [], } processed = _postprocess_synthese(result, "texte") assert processed["probleme_pris_en_charge"] == "pneumothorax spontané" def test_comorbidite_with_acute_problem_promoted(self): """Comorbidité banale + problème aigu → diagnostic retenu promu.""" result = { "probleme_pris_en_charge": "diabète", "diagnostic_retenu": "décompensation cardiaque globale", "terrain_comorbidites": [], "complications": [], "actes_ou_traitements_majeurs": ["diurétiques IV"], "preuves": [], } processed = _postprocess_synthese(result, "texte complet") assert processed["probleme_pris_en_charge"] == "décompensation cardiaque globale" assert "diabète" in processed["terrain_comorbidites"] def test_comorbidite_with_pec_marker_kept(self): """Comorbidité banale avec marqueur PEC → conservée.""" result = { "probleme_pris_en_charge": "diabète", "diagnostic_retenu": "diabète déséquilibré", "terrain_comorbidites": [], "complications": [], "actes_ou_traitements_majeurs": [], "preuves": [], } context = "hospitalisé pour diabète déséquilibré avec insulinothérapie IV" processed = _postprocess_synthese(result, context) # Marqueur "hospitalisé pour" + "déséquilibré" trouvé → conservé assert processed["probleme_pris_en_charge"] == "diabète" def test_comorbidite_no_acute_indeterminate(self): """Comorbidité banale sans aigu ni marqueur → indéterminé.""" result = { "probleme_pris_en_charge": "HTA", "diagnostic_retenu": "", "terrain_comorbidites": [], "complications": [], "actes_ou_traitements_majeurs": [], "preuves": [], } processed = _postprocess_synthese(result, "texte") assert "indéterminé" in processed["probleme_pris_en_charge"] assert "HTA" in processed["terrain_comorbidites"] def test_proof_added_on_correction(self): """Une preuve de post-traitement est ajoutée lors de correction.""" result = { "probleme_pris_en_charge": "diabète", "diagnostic_retenu": "pneumopathie bactérienne", "terrain_comorbidites": [], "complications": [], "actes_ou_traitements_majeurs": ["antibiothérapie IV"], "preuves": [], } processed = _postprocess_synthese(result, "texte") sections = [p["section"] for p in processed["preuves"]] assert "post-traitement" in sections class TestBuildMotifFallback: """Tests du fallback motif admission.""" def test_mode_entree_priority(self): """Le mode_entree du séjour a priorité.""" parsed = _make_parsed() dossier = DossierMedical() dossier.sejour = Sejour(mode_entree="Urgences") assert _build_motif(parsed, dossier) == "Urgences" def test_section_motif_hospitalisation(self): """Section motif_hospitalisation utilisée si mode_entree vide.""" parsed = _make_parsed(sections={"motif_hospitalisation": "Douleur thoracique"}) dossier = DossierMedical() assert _build_motif(parsed, dossier) == "Douleur thoracique" def test_fallback_lexical_conclusion(self): """Fallback lexical sur la conclusion.""" parsed = _make_parsed(sections={ "conclusion": "Patient hospitalisé pour pneumothorax spontané." }) dossier = DossierMedical() result = _build_motif(parsed, dossier) assert "pneumothorax" in result.lower() def test_fallback_lexical_full_text(self): """Fallback lexical sur le texte complet.""" parsed = _make_parsed() dossier = DossierMedical() text = "Compte-rendu\nMotif d'hospitalisation : ictère choléstatique.\nExamen..." result = _build_motif(parsed, dossier, full_text=text) assert "ictère" in result.lower() def test_non_renseigne_when_nothing(self): """Pas de motif trouvé → 'Non renseigné'.""" parsed = _make_parsed() dossier = DossierMedical() assert _build_motif(parsed, dossier) == "Non renseigné" # =================================================================== # Tests DP Pool Rank # =================================================================== class TestFormatPoolForPrompt: """Tests pour _format_pool_for_prompt().""" def test_basic_formatting(self): """Vérifie le format des candidats pour le prompt.""" pool = [ DPPoolCandidate(terme="Pneumopathie", section="conclusion", preuve="Au total : pneumopathie", score_initial=0.7), DPPoolCandidate(terme="Embolie pulmonaire", section="diag_sortie", preuve="Diagnostic de sortie", score_initial=1.0), ] text = _format_pool_for_prompt(pool) assert "[0]" in text assert "[1]" in text assert "Pneumopathie" in text assert "Embolie pulmonaire" in text assert "conclusion" in text assert "diag_sortie" in text def test_max_items_cap(self): """Vérifie que max_items est respecté.""" pool = [ DPPoolCandidate(terme=f"Diag_{i}", section="conclusion", score_initial=0.5) for i in range(10) ] text = _format_pool_for_prompt(pool, max_items=3) assert "[0]" in text assert "[2]" in text assert "[3]" not in text def test_empty_pool(self): """Pool vide → texte vide.""" assert _format_pool_for_prompt([]) == "" class TestBuildClinicalContext: """Tests pour _build_clinical_context().""" def test_with_synthese(self): """Avec SynthesePMSI disponible.""" synthese = SynthesePMSI( motif_admission="Douleur thoracique", probleme_pris_en_charge="Embolie pulmonaire", diagnostic_retenu="Embolie pulmonaire bilatérale", ) parsed = _make_parsed() dossier = DossierMedical() ctx = _build_clinical_context(parsed, dossier, "", synthese) assert "Embolie pulmonaire" in ctx assert "Douleur thoracique" in ctx def test_without_synthese(self): """Sans SynthesePMSI → fallback sections fortes.""" parsed = _make_parsed(sections={"conclusion": "Pneumopathie traitée"}) dossier = DossierMedical() ctx = _build_clinical_context(parsed, dossier, "texte complet", None) assert "Pneumopathie traitée" in ctx assert "Motif" in ctx class TestLlmDpPoolRank: """Tests unitaires pour llm_dp_pool_rank() — sans appel LLM réel.""" def test_empty_pool_fallback_off(self): """Pool vide + fallback OFF → REVIEW.""" parsed = _make_parsed() dossier = DossierMedical() selection = llm_dp_pool_rank( parsed, "texte", dossier, pool_candidates=[], fallback_oneshot=False, ) assert selection.verdict == "review" assert "pool vide" in selection.winner_reason def test_empty_pool_fallback_on(self, monkeypatch): """Pool vide + fallback ON → tente llm_dp_fallback.""" # Mock llm_dp_fallback pour retourner un résultat connu from src.medical import dp_scoring mock_selection = DPSelection( verdict="review", winner_reason="fallback activé", ) monkeypatch.setattr(dp_scoring, "llm_dp_fallback", lambda *a, **kw: mock_selection) parsed = _make_parsed() dossier = DossierMedical() selection = llm_dp_pool_rank( parsed, "texte", dossier, pool_candidates=[], fallback_oneshot=True, ) assert selection.verdict == "review" assert "fallback" in selection.winner_reason def test_valid_llm_response_high_confidence(self, monkeypatch): """Réponse LLM valide avec confidence high → CONFIRMED.""" pool = [ DPPoolCandidate(terme="Embolie pulmonaire", section="conclusion", preuve="Au total : embolie pulmonaire", score_initial=0.7), DPPoolCandidate(terme="HTA", section="conclusion", preuve="terrain HTA", score_initial=0.3), ] # Mock call_ollama def mock_call_ollama(prompt, **kwargs): return { "chosen_index": 0, "chosen_terme": "Embolie pulmonaire", "evidence_section": "conclusion", "evidence_excerpt": "Au total : embolie pulmonaire", "confidence": "high", "reason": "pathologie aiguë traitée", } from src.medical import dp_scoring monkeypatch.setattr("src.medical.ollama_client.call_ollama", mock_call_ollama) parsed = _make_parsed(sections={"conclusion": "Au total : embolie pulmonaire"}) dossier = DossierMedical() selection = llm_dp_pool_rank( parsed, "texte", dossier, pool_candidates=pool, fallback_oneshot=False, ) assert selection.verdict == "confirmed" assert len(selection.candidates) == 1 assert selection.candidates[0].label == "Embolie pulmonaire" assert selection.candidates[0].source_section == "llm_pool_rank (conclusion)" assert selection.candidates[0].code is None # pas de code CIM-10, sera codé en aval def test_valid_llm_response_medium_confidence(self, monkeypatch): """Réponse LLM avec confidence medium → REVIEW.""" pool = [ DPPoolCandidate(terme="Insuffisance cardiaque", section="conclusion", preuve="insuffisance cardiaque", score_initial=0.7), ] def mock_call_ollama(prompt, **kwargs): return { "chosen_index": 0, "chosen_terme": "Insuffisance cardiaque", "evidence_section": "conclusion", "evidence_excerpt": "insuffisance cardiaque globale", "confidence": "medium", "reason": "diagnostic probable", } monkeypatch.setattr("src.medical.ollama_client.call_ollama", mock_call_ollama) parsed = _make_parsed() dossier = DossierMedical() selection = llm_dp_pool_rank( parsed, "texte", dossier, pool_candidates=pool, fallback_oneshot=False, ) assert selection.verdict == "review" assert "confidence medium" in selection.winner_reason def test_chosen_index_minus_one_fallback_off(self, monkeypatch): """chosen_index=-1 + fallback OFF → REVIEW.""" pool = [ DPPoolCandidate(terme="HTA", section="conclusion", preuve="HTA", score_initial=0.3), ] def mock_call_ollama(prompt, **kwargs): return { "chosen_index": -1, "chosen_terme": "", "confidence": "low", "reason": "aucun candidat solide", } monkeypatch.setattr("src.medical.ollama_client.call_ollama", mock_call_ollama) parsed = _make_parsed() dossier = DossierMedical() selection = llm_dp_pool_rank( parsed, "texte", dossier, pool_candidates=pool, fallback_oneshot=False, ) assert selection.verdict == "review" assert "aucun candidat retenu" in selection.winner_reason def test_index_out_of_range_fallback_off(self, monkeypatch): """Index hors plage → REVIEW.""" pool = [ DPPoolCandidate(terme="Pneumopathie", section="conclusion", preuve="...", score_initial=0.7), ] def mock_call_ollama(prompt, **kwargs): return { "chosen_index": 5, "chosen_terme": "Fantôme", "confidence": "high", } monkeypatch.setattr("src.medical.ollama_client.call_ollama", mock_call_ollama) parsed = _make_parsed() dossier = DossierMedical() selection = llm_dp_pool_rank( parsed, "texte", dossier, pool_candidates=pool, fallback_oneshot=False, ) assert selection.verdict == "review" def test_score_details_contain_pool_info(self, monkeypatch): """Les score_details du candidat contiennent les infos pool.""" pool = [ DPPoolCandidate(terme="Cholécystite aiguë", section="diag_sortie", preuve="cholécystite aiguë lithiasique", score_initial=0.9), ] def mock_call_ollama(prompt, **kwargs): return { "chosen_index": 0, "chosen_terme": "Cholécystite aiguë", "evidence_section": "diag_sortie", "evidence_excerpt": "cholécystite aiguë lithiasique", "confidence": "high", "reason": "diagnostic chirurgical aigu", } monkeypatch.setattr("src.medical.ollama_client.call_ollama", mock_call_ollama) parsed = _make_parsed() dossier = DossierMedical() selection = llm_dp_pool_rank( parsed, "texte", dossier, pool_candidates=pool, fallback_oneshot=False, ) assert selection.verdict == "confirmed" details = selection.candidates[0].score_details assert "pool_score" in details assert "pool_index" in details assert details["pool_index"] == 0