07c267539c
- test_extraction: +21 tests (sections diag_sortie/diag_principal/synthese, variantes titres, terminaisons, faux positifs mid-sentence, biosynthèse) - test_dp_selector: +55 tests (flags, candidates, scoring, hardening DIM, bonus +4/+2, evidence excerpt, cas 74 D50→I25.1 corrigé) - test_fusion: +39 tests (propagation dp_selection evidence/reason/verdict, source 2e dossier, pas de crash si aucun DP) - fixtures: case_74_min.json + 3 fixtures DP existantes Aucun mock utilisé — données synthétiques uniquement. Le test cas 74 passe : I25.1 gagne sur D50 grâce au bonus diag_sortie +4. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
812 lines
33 KiB
Python
812 lines
33 KiB
Python
"""Tests NUKE-3 — Sélecteur DP type DIM.
|
|
|
|
Sans mocks : mini-fixtures JSON + mode LLM désactivé.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from src.config import (
|
|
Diagnostic,
|
|
DossierMedical,
|
|
DPCandidate,
|
|
DPSelection,
|
|
Sejour,
|
|
)
|
|
from src.medical.dp_selector import (
|
|
COMORBIDITY_PREFIXES,
|
|
Z_CODE_DP_WHITELIST,
|
|
_is_act_only,
|
|
_is_comorbidity_like,
|
|
_is_symptom_like,
|
|
build_candidates,
|
|
score_candidates,
|
|
select_dp,
|
|
)
|
|
|
|
RESOURCES = Path(__file__).parent / "resources"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _load_fixture(name: str) -> dict:
|
|
"""Charge un mini-dossier JSON depuis tests/resources/."""
|
|
path = RESOURCES / name
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def _build_dossier(data: dict) -> DossierMedical:
|
|
"""Construit un DossierMedical depuis le dict de fixture."""
|
|
d = data["dossier"]
|
|
dp = None
|
|
if d.get("diagnostic_principal"):
|
|
dp = Diagnostic(**d["diagnostic_principal"])
|
|
das = [Diagnostic(**x) for x in d.get("diagnostics_associes", [])]
|
|
return DossierMedical(
|
|
document_type=d.get("document_type", "crh"),
|
|
sejour=Sejour(**d.get("sejour", {})),
|
|
diagnostic_principal=dp,
|
|
diagnostics_associes=das,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests : flags de classification
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestDPFlags:
|
|
"""Détection comorbidité / symptôme / acte-seul."""
|
|
|
|
def test_comorbidity_hta(self):
|
|
assert _is_comorbidity_like("I10") is True
|
|
|
|
def test_comorbidity_diabetes(self):
|
|
assert _is_comorbidity_like("E11.9") is True
|
|
|
|
def test_not_comorbidity_pancreatitis(self):
|
|
assert _is_comorbidity_like("K85.1") is False
|
|
|
|
def test_symptom_r_code(self):
|
|
assert _is_symptom_like("R10.4") is True
|
|
|
|
def test_not_symptom_k_code(self):
|
|
assert _is_symptom_like("K85.1") is False
|
|
|
|
def test_symptom_none(self):
|
|
assert _is_symptom_like(None) is False
|
|
|
|
def test_act_only_cholecystectomie(self):
|
|
assert _is_act_only("Cholécystectomie") is True
|
|
|
|
def test_not_act_long_text(self):
|
|
assert _is_act_only("Cholécystectomie pour cholécystite aiguë lithiasique avec iléus réflexe") is False
|
|
|
|
def test_not_act_diagnostic(self):
|
|
assert _is_act_only("Pancréatite aiguë biliaire") is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests : construction des candidats
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestBuildCandidates:
|
|
"""Construction du pool de candidats depuis le dossier."""
|
|
|
|
def test_dp_and_das(self):
|
|
dossier = DossierMedical(
|
|
sejour=Sejour(),
|
|
diagnostic_principal=Diagnostic(texte="DP", cim10_suggestion="K85.1", source="regex"),
|
|
diagnostics_associes=[
|
|
Diagnostic(texte="DAS1", cim10_suggestion="R10.4", source="edsnlp"),
|
|
Diagnostic(texte="DAS2", cim10_suggestion="I10", source="edsnlp"),
|
|
],
|
|
)
|
|
candidates = build_candidates(dossier)
|
|
assert len(candidates) == 3
|
|
assert candidates[0].term == "DP"
|
|
assert candidates[0].index == 0
|
|
|
|
def test_no_dp_das_only(self):
|
|
dossier = DossierMedical(
|
|
sejour=Sejour(),
|
|
diagnostics_associes=[
|
|
Diagnostic(texte="DAS1", cim10_suggestion="K85.1"),
|
|
],
|
|
)
|
|
candidates = build_candidates(dossier)
|
|
assert len(candidates) == 1
|
|
|
|
def test_ruled_out_excluded(self):
|
|
dossier = DossierMedical(
|
|
sejour=Sejour(),
|
|
diagnostics_associes=[
|
|
Diagnostic(texte="Exclu", cim10_suggestion="R10.4", status="ruled_out"),
|
|
Diagnostic(texte="Gardé", cim10_suggestion="K85.1"),
|
|
],
|
|
)
|
|
candidates = build_candidates(dossier)
|
|
assert len(candidates) == 1
|
|
assert candidates[0].term == "Gardé"
|
|
|
|
def test_no_code_excluded(self):
|
|
dossier = DossierMedical(
|
|
sejour=Sejour(),
|
|
diagnostics_associes=[
|
|
Diagnostic(texte="Sans code"),
|
|
Diagnostic(texte="Avec code", cim10_suggestion="K85.1"),
|
|
],
|
|
)
|
|
candidates = build_candidates(dossier)
|
|
assert len(candidates) == 1
|
|
|
|
def test_empty_dossier(self):
|
|
dossier = DossierMedical(sejour=Sejour())
|
|
candidates = build_candidates(dossier)
|
|
assert candidates == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests : scoring déterministe
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestScoring:
|
|
"""Vérification du scoring déterministe."""
|
|
|
|
def test_high_confidence_beats_medium(self):
|
|
candidates = [
|
|
DPCandidate(index=0, term="A", code="K85.1", confidence="high", section_strength=1),
|
|
DPCandidate(index=1, term="B", code="J18.9", confidence="medium", section_strength=1),
|
|
]
|
|
scored = score_candidates(candidates, {})
|
|
assert scored[0].code == "K85.1"
|
|
assert scored[0].score > scored[1].score
|
|
|
|
def test_comorbidity_malus(self):
|
|
candidates = [
|
|
DPCandidate(index=0, term="HTA", code="I10", confidence="high",
|
|
is_comorbidity_like=True, section_strength=2),
|
|
DPCandidate(index=1, term="EP", code="I26.9", confidence="high",
|
|
section_strength=1),
|
|
]
|
|
scored = score_candidates(candidates, {})
|
|
# I26.9 doit gagner malgré section_strength plus faible (malus comorbidité -3)
|
|
assert scored[0].code == "I26.9"
|
|
|
|
def test_symptom_malus(self):
|
|
candidates = [
|
|
DPCandidate(index=0, term="Douleur", code="R10.4", confidence="medium",
|
|
is_symptom_like=True, section_strength=2),
|
|
DPCandidate(index=1, term="Pancréatite", code="K85.1", confidence="high",
|
|
section_strength=1),
|
|
]
|
|
scored = score_candidates(candidates, {})
|
|
assert scored[0].code == "K85.1"
|
|
|
|
def test_motif_alignment_bonus(self):
|
|
candidates = [
|
|
DPCandidate(index=0, term="Pancréatite", code="K85.1", confidence="high", section_strength=1),
|
|
DPCandidate(index=1, term="Cholécystite", code="K81.0", confidence="high", section_strength=1),
|
|
]
|
|
synthese = {"motif": "Pancréatite aiguë biliaire"}
|
|
scored = score_candidates(candidates, synthese)
|
|
# "Pancréatite" est dans le motif → bonus +2
|
|
assert scored[0].code == "K85.1"
|
|
assert scored[0].score_details.get("motif_align") == 2
|
|
|
|
def test_z_code_malus_non_whitelisted(self):
|
|
candidates = [
|
|
DPCandidate(index=0, term="Suivi Z", code="Z80.0", confidence="high", section_strength=1),
|
|
DPCandidate(index=1, term="Pathologie", code="K85.1", confidence="high", section_strength=1),
|
|
]
|
|
scored = score_candidates(candidates, {})
|
|
# À confiance et section égales, Z80 non whitelisté perd (malus -2)
|
|
assert scored[0].code == "K85.1"
|
|
|
|
def test_z_code_whitelisted_no_malus(self):
|
|
candidates = [
|
|
DPCandidate(index=0, term="Chimio", code="Z51.1", confidence="high", section_strength=3),
|
|
]
|
|
scored = score_candidates(candidates, {})
|
|
assert "z_code_malus" not in scored[0].score_details
|
|
|
|
def test_diag_sortie_bonus(self):
|
|
"""Un candidat mentionné dans 'diag_sortie' reçoit +4."""
|
|
candidates = [
|
|
DPCandidate(index=0, term="Anémie", code="D50", confidence="medium", section_strength=3),
|
|
DPCandidate(index=1, term="SCA", code="I25.1", confidence="high", section_strength=1),
|
|
]
|
|
synthese = {"diag_sortie": "SCA ST+ antérieur I25.1"}
|
|
scored = score_candidates(candidates, synthese)
|
|
# SCA doit gagner grâce au bonus diag_sortie +4
|
|
assert scored[0].code == "I25.1"
|
|
assert scored[0].score_details.get("diag_section_bonus") == 4
|
|
|
|
def test_diag_principal_bonus(self):
|
|
"""Un candidat mentionné dans 'diag_principal' reçoit +4."""
|
|
candidates = [
|
|
DPCandidate(index=0, term="Pneumopathie", code="J18.9", confidence="medium", section_strength=1),
|
|
DPCandidate(index=1, term="Anémie", code="D50", confidence="medium", section_strength=3),
|
|
]
|
|
synthese = {"diag_principal": "Pneumopathie J18.9"}
|
|
scored = score_candidates(candidates, synthese)
|
|
assert scored[0].code == "J18.9"
|
|
assert scored[0].score_details.get("diag_section_bonus") == 4
|
|
|
|
def test_synthese_section_bonus(self):
|
|
"""Un candidat mentionné dans 'synthese' reçoit +2."""
|
|
candidates = [
|
|
DPCandidate(index=0, term="Embolie pulmonaire", code="I26.9", confidence="medium", section_strength=1),
|
|
]
|
|
synthese = {"synthese": "Patient admis pour embolie pulmonaire massive."}
|
|
scored = score_candidates(candidates, synthese)
|
|
assert scored[0].score_details.get("diag_section_bonus") == 2
|
|
|
|
def test_diag_sortie_code_match(self):
|
|
"""Le matching fonctionne aussi par code CIM-10."""
|
|
candidates = [
|
|
DPCandidate(index=0, term="Syndrome coronarien aigu", code="I25.1",
|
|
confidence="medium", section_strength=1),
|
|
]
|
|
synthese = {"diag_sortie": "I25.1 — cardiopathie ischémique"}
|
|
scored = score_candidates(candidates, synthese)
|
|
assert scored[0].score_details.get("diag_section_bonus") == 4
|
|
|
|
def test_no_diag_section_no_bonus(self):
|
|
"""Sans sections diagnostiques, pas de bonus."""
|
|
candidates = [
|
|
DPCandidate(index=0, term="Test", code="K85.1", confidence="medium", section_strength=1),
|
|
]
|
|
scored = score_candidates(candidates, {"motif": "Douleur abdominale"})
|
|
assert "diag_section_bonus" not in scored[0].score_details
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests : select_dp (pré-ranker, LLM off)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestSelectDP:
|
|
"""Tests de bout en bout du sélecteur DP (LLM désactivé)."""
|
|
|
|
def test_etiology_beats_symptom(self):
|
|
"""Fixture : étiologie > symptôme."""
|
|
fixture = _load_fixture("dp_etiology_vs_symptom.json")
|
|
dossier = _build_dossier(fixture)
|
|
synthese = fixture["synthese_nuke1"]
|
|
|
|
selection = select_dp(dossier, synthese, config={"llm_enabled": False})
|
|
|
|
assert selection.chosen_code == fixture["expected"]["chosen_code"]
|
|
assert selection.verdict == fixture["expected"]["verdict"]
|
|
|
|
def test_acute_beats_comorbidity(self):
|
|
"""Fixture : aigu > comorbidité."""
|
|
fixture = _load_fixture("dp_acute_vs_comorbidity.json")
|
|
dossier = _build_dossier(fixture)
|
|
synthese = fixture["synthese_nuke1"]
|
|
|
|
selection = select_dp(dossier, synthese, config={"llm_enabled": False})
|
|
|
|
assert selection.chosen_code == fixture["expected"]["chosen_code"]
|
|
assert selection.verdict == fixture["expected"]["verdict"]
|
|
|
|
def test_ambiguous_returns_review(self):
|
|
"""Fixture : scores proches + LLM off → REVIEW."""
|
|
fixture = _load_fixture("dp_ambiguous.json")
|
|
dossier = _build_dossier(fixture)
|
|
synthese = fixture["synthese_nuke1"]
|
|
|
|
selection = select_dp(dossier, synthese, config={"llm_enabled": False})
|
|
|
|
assert selection.verdict == fixture["expected"]["verdict"]
|
|
assert selection.chosen_code is not None # suggestion quand même
|
|
assert len(selection.candidates) == 2
|
|
|
|
def test_trackare_bypasses_selection(self):
|
|
"""Trackare DP → CONFIRMED sans scoring."""
|
|
dossier = DossierMedical(
|
|
document_type="trackare",
|
|
sejour=Sejour(sexe="F", age=72),
|
|
diagnostic_principal=Diagnostic(
|
|
texte="Pancréatite", cim10_suggestion="K85.1", source="trackare",
|
|
),
|
|
)
|
|
selection = select_dp(dossier, {}, config={"llm_enabled": False})
|
|
|
|
assert selection.verdict == "CONFIRMED"
|
|
assert selection.chosen_code == "K85.1"
|
|
assert "Trackare" in (selection.reason or "")
|
|
assert selection.candidates == [] # pas de scoring
|
|
|
|
def test_no_candidates_returns_review(self):
|
|
"""Aucun candidat → REVIEW."""
|
|
dossier = DossierMedical(
|
|
document_type="crh",
|
|
sejour=Sejour(),
|
|
)
|
|
selection = select_dp(dossier, {}, config={"llm_enabled": False})
|
|
|
|
assert selection.verdict == "REVIEW"
|
|
assert "Aucun candidat" in (selection.reason or "")
|
|
|
|
def test_single_candidate_confirmed(self):
|
|
"""Candidat unique → CONFIRMED."""
|
|
dossier = DossierMedical(
|
|
document_type="crh",
|
|
sejour=Sejour(),
|
|
diagnostics_associes=[
|
|
Diagnostic(texte="Pneumopathie", cim10_suggestion="J18.9", cim10_confidence="high"),
|
|
],
|
|
)
|
|
selection = select_dp(dossier, {}, config={"llm_enabled": False})
|
|
|
|
assert selection.verdict == "CONFIRMED"
|
|
assert selection.chosen_code == "J18.9"
|
|
assert "unique" in (selection.reason or "").lower()
|
|
|
|
def test_textual_beats_act_only(self):
|
|
"""Diagnostic textuel > acte seul."""
|
|
dossier = DossierMedical(
|
|
document_type="crh",
|
|
sejour=Sejour(),
|
|
diagnostics_associes=[
|
|
Diagnostic(texte="Cholécystectomie", cim10_suggestion="K80.2",
|
|
cim10_confidence="high", source="edsnlp"),
|
|
Diagnostic(texte="Cholécystite aiguë lithiasique avec angiocholite",
|
|
cim10_suggestion="K81.0", cim10_confidence="high", source="llm_das"),
|
|
],
|
|
)
|
|
selection = select_dp(dossier, {}, config={"llm_enabled": False})
|
|
|
|
# K81.0 (diagnostic) doit battre K80.2 (décrit par un acte seul)
|
|
assert selection.chosen_code == "K81.0"
|
|
|
|
def test_dp_selection_has_candidates(self):
|
|
"""Le résultat contient la liste des candidats scorés."""
|
|
dossier = DossierMedical(
|
|
document_type="crh",
|
|
sejour=Sejour(),
|
|
diagnostics_associes=[
|
|
Diagnostic(texte="A", cim10_suggestion="R10.4", cim10_confidence="medium"),
|
|
Diagnostic(texte="B", cim10_suggestion="K85.1", cim10_confidence="high"),
|
|
],
|
|
)
|
|
selection = select_dp(dossier, {}, config={"llm_enabled": False})
|
|
|
|
assert len(selection.candidates) >= 1
|
|
assert all(isinstance(c, DPCandidate) for c in selection.candidates)
|
|
assert all(c.score_details for c in selection.candidates)
|
|
|
|
def test_dp_selection_serializable(self):
|
|
"""DPSelection est sérialisable en JSON (Pydantic)."""
|
|
dossier = DossierMedical(
|
|
document_type="crh",
|
|
sejour=Sejour(),
|
|
diagnostics_associes=[
|
|
Diagnostic(texte="Test", cim10_suggestion="K85.1", cim10_confidence="high"),
|
|
],
|
|
)
|
|
selection = select_dp(dossier, {}, config={"llm_enabled": False})
|
|
data = selection.model_dump()
|
|
|
|
assert isinstance(data, dict)
|
|
assert "verdict" in data
|
|
assert "candidates" in data
|
|
# Roundtrip
|
|
restored = DPSelection(**data)
|
|
assert restored.verdict == selection.verdict
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests : hardening DIM (A1/A2/A3)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestDPHardening:
|
|
"""Tests des gardes-fous DIM : evidence, mono-candidat fragile, confidence cap."""
|
|
|
|
def test_confirmed_requires_evidence(self):
|
|
"""A1 — CONFIRMED avec evidence vide → downgrade REVIEW."""
|
|
# Construire une DPSelection CONFIRMED sans evidence
|
|
from src.medical.dp_selector import _enforce_confirmed_rules
|
|
|
|
selection = DPSelection(
|
|
chosen_index=0,
|
|
chosen_term="Test",
|
|
chosen_code="K85.1",
|
|
confidence="high",
|
|
verdict="CONFIRMED",
|
|
evidence=[], # vide !
|
|
reason="Test",
|
|
candidates=[DPCandidate(index=0, term="Test", code="K85.1",
|
|
section_strength=3, confidence="high")],
|
|
)
|
|
result = _enforce_confirmed_rules(selection, {})
|
|
assert result.verdict == "REVIEW"
|
|
assert result.confidence != "high"
|
|
assert "preuve" in (result.reason or "").lower()
|
|
|
|
def test_confirmed_with_evidence_stays_confirmed(self):
|
|
"""A1 — CONFIRMED avec evidence non vide reste CONFIRMED."""
|
|
from src.medical.dp_selector import _enforce_confirmed_rules
|
|
|
|
selection = DPSelection(
|
|
chosen_index=0,
|
|
chosen_term="Pancréatite",
|
|
chosen_code="K85.1",
|
|
confidence="high",
|
|
verdict="CONFIRMED",
|
|
evidence=["Score 6.0 — source: regex (section forte)"],
|
|
reason="Écart net",
|
|
candidates=[DPCandidate(index=0, term="Pancréatite", code="K85.1",
|
|
section_strength=3, confidence="high")],
|
|
)
|
|
result = _enforce_confirmed_rules(selection, {})
|
|
assert result.verdict == "CONFIRMED"
|
|
assert result.confidence == "high"
|
|
|
|
def test_mono_candidate_fragile_returns_review(self):
|
|
"""A2 — Mono-candidat comorbidité → REVIEW."""
|
|
dossier = DossierMedical(
|
|
document_type="crh",
|
|
sejour=Sejour(),
|
|
diagnostics_associes=[
|
|
Diagnostic(texte="Hypertension artérielle",
|
|
cim10_suggestion="I10", cim10_confidence="high",
|
|
source="edsnlp"),
|
|
],
|
|
)
|
|
selection = select_dp(dossier, {}, config={"llm_enabled": False})
|
|
|
|
assert selection.verdict == "REVIEW"
|
|
assert "fragile" in (selection.reason or "").lower()
|
|
|
|
def test_mono_candidate_symptom_returns_review(self):
|
|
"""A2 — Mono-candidat symptôme (R-code) → REVIEW."""
|
|
dossier = DossierMedical(
|
|
document_type="crh",
|
|
sejour=Sejour(),
|
|
diagnostics_associes=[
|
|
Diagnostic(texte="Douleur abdominale",
|
|
cim10_suggestion="R10.4", cim10_confidence="medium",
|
|
source="edsnlp"),
|
|
],
|
|
)
|
|
selection = select_dp(dossier, {}, config={"llm_enabled": False})
|
|
|
|
assert selection.verdict == "REVIEW"
|
|
assert "fragile" in (selection.reason or "").lower()
|
|
|
|
def test_mono_candidate_strong_stays_confirmed(self):
|
|
"""A2 — Mono-candidat non-fragile avec evidence → CONFIRMED."""
|
|
dossier = DossierMedical(
|
|
document_type="crh",
|
|
sejour=Sejour(),
|
|
diagnostics_associes=[
|
|
Diagnostic(texte="Pneumopathie infectieuse",
|
|
cim10_suggestion="J18.9", cim10_confidence="high",
|
|
source="regex"),
|
|
],
|
|
)
|
|
synthese = {"motif": "Pneumopathie infectieuse fébrile"}
|
|
selection = select_dp(dossier, synthese, config={"llm_enabled": False})
|
|
|
|
assert selection.verdict == "CONFIRMED"
|
|
assert len(selection.evidence) >= 1
|
|
|
|
def test_confidence_high_not_allowed_on_review(self):
|
|
"""A3 — confidence='high' impossible si verdict='REVIEW'."""
|
|
dossier = DossierMedical(
|
|
document_type="crh",
|
|
sejour=Sejour(),
|
|
diagnostics_associes=[
|
|
Diagnostic(texte="Pneumopathie", cim10_suggestion="J18.9",
|
|
cim10_confidence="high", source="llm_das"),
|
|
Diagnostic(texte="Insuffisance cardiaque", cim10_suggestion="I50.1",
|
|
cim10_confidence="high", source="llm_das"),
|
|
],
|
|
)
|
|
selection = select_dp(dossier, {}, config={"llm_enabled": False})
|
|
|
|
# Scores proches → REVIEW
|
|
assert selection.verdict == "REVIEW"
|
|
# confidence ne peut PAS être "high" en REVIEW
|
|
assert selection.confidence != "high"
|
|
|
|
def test_confidence_high_requires_strong_section(self):
|
|
"""A3 — confidence='high' requiert section_strength >= 2."""
|
|
from src.medical.dp_selector import _enforce_confirmed_rules
|
|
|
|
# Candidat avec section_strength=1 (llm_das) mais confidence high
|
|
selection = DPSelection(
|
|
chosen_index=0,
|
|
chosen_term="Test",
|
|
chosen_code="K85.1",
|
|
confidence="high",
|
|
verdict="CONFIRMED",
|
|
evidence=["Score 4.0 — source: llm_das"],
|
|
reason="Candidat unique",
|
|
candidates=[DPCandidate(index=0, term="Test", code="K85.1",
|
|
section_strength=1, confidence="high")],
|
|
)
|
|
result = _enforce_confirmed_rules(selection, {})
|
|
# section_strength=1 < 2 → confidence downgrade
|
|
assert result.confidence != "high"
|
|
|
|
def test_env_var_t2a_dp_ranker_llm_controls_flag(self):
|
|
"""B — T2A_DP_RANKER_LLM contrôle get_dp_ranker_llm_enabled()."""
|
|
import os
|
|
from src.config import get_dp_ranker_llm_enabled
|
|
|
|
old = os.environ.get("T2A_DP_RANKER_LLM")
|
|
try:
|
|
os.environ["T2A_DP_RANKER_LLM"] = "0"
|
|
assert get_dp_ranker_llm_enabled() is False
|
|
|
|
os.environ["T2A_DP_RANKER_LLM"] = "1"
|
|
assert get_dp_ranker_llm_enabled() is True
|
|
|
|
os.environ["T2A_DP_RANKER_LLM"] = "false"
|
|
assert get_dp_ranker_llm_enabled() is False
|
|
|
|
os.environ["T2A_DP_RANKER_LLM"] = "true"
|
|
assert get_dp_ranker_llm_enabled() is True
|
|
finally:
|
|
if old is None:
|
|
os.environ.pop("T2A_DP_RANKER_LLM", None)
|
|
else:
|
|
os.environ["T2A_DP_RANKER_LLM"] = old
|
|
|
|
def test_legacy_env_var_accepted(self):
|
|
"""B — DP_RANKER_LLM_ENABLED (ancien nom) fonctionne en fallback."""
|
|
import os
|
|
from src.config import get_dp_ranker_llm_enabled
|
|
|
|
old_canonical = os.environ.pop("T2A_DP_RANKER_LLM", None)
|
|
old_legacy = os.environ.get("DP_RANKER_LLM_ENABLED")
|
|
try:
|
|
os.environ["DP_RANKER_LLM_ENABLED"] = "0"
|
|
assert get_dp_ranker_llm_enabled() is False
|
|
finally:
|
|
os.environ.pop("DP_RANKER_LLM_ENABLED", None)
|
|
if old_canonical is not None:
|
|
os.environ["T2A_DP_RANKER_LLM"] = old_canonical
|
|
if old_legacy is not None:
|
|
os.environ["DP_RANKER_LLM_ENABLED"] = old_legacy
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests : scoring bonus sections diagnostiques fortes (B)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestDiagSectionScoring:
|
|
"""Validation du bonus +4/+2 quand un candidat apparaît dans diag_sortie/diag_principal/synthese."""
|
|
|
|
def test_bonus_diag_sortie_beats_bio_noise(self):
|
|
"""I25.1 dans diag_sortie bat D50 qui n'est PAS dans la section diagnostique."""
|
|
candidates = [
|
|
DPCandidate(index=0, term="Anémie", code="D50",
|
|
confidence="medium", section_strength=3, source="regex"),
|
|
DPCandidate(index=1, term="SCA", code="I25.1",
|
|
confidence="high", section_strength=1, source="llm_das"),
|
|
]
|
|
synthese = {
|
|
# Seul SCA est dans le diagnostic de sortie (Anémie = biologie, pas diagnostic)
|
|
"diag_sortie": "SCA ST+ antérieur traité par angioplastie — I25.1",
|
|
"motif": "Douleur thoracique",
|
|
}
|
|
scored = score_candidates(candidates, synthese)
|
|
|
|
# I25.1 : section=1 + confidence=3 + diag_sortie=4 = 8
|
|
# D50 : section=3 + confidence=1 + pas de bonus = 4
|
|
assert scored[0].code == "I25.1"
|
|
assert scored[0].score_details.get("diag_section_bonus") == 4
|
|
assert "diag_section_bonus" not in scored[1].score_details
|
|
# Delta >= 3 (écart suffisant pour CONFIRMED)
|
|
assert scored[0].score - scored[1].score >= 3
|
|
|
|
def test_bonus_diag_principal_match_code(self):
|
|
"""Bonus +4 via match code dans diag_principal."""
|
|
candidates = [
|
|
DPCandidate(index=0, term="Embolie pulmonaire", code="I26.9",
|
|
confidence="medium", section_strength=1, source="edsnlp"),
|
|
DPCandidate(index=1, term="Thrombose veineuse", code="I80.2",
|
|
confidence="medium", section_strength=1, source="edsnlp"),
|
|
]
|
|
synthese = {"diag_principal": "I26.9 Embolie pulmonaire bilatérale"}
|
|
scored = score_candidates(candidates, synthese)
|
|
|
|
assert scored[0].code == "I26.9"
|
|
assert scored[0].score_details.get("diag_section_bonus") == 4
|
|
# I80.2 ne doit PAS avoir le bonus
|
|
i802 = next(c for c in scored if c.code == "I80.2")
|
|
assert "diag_section_bonus" not in i802.score_details
|
|
|
|
def test_bonus_diag_principal_match_term(self):
|
|
"""Bonus +4 via match terme (sans code CIM-10 dans la section)."""
|
|
candidates = [
|
|
DPCandidate(index=0, term="Pancréatite aiguë", code="K85.1",
|
|
confidence="medium", section_strength=1),
|
|
]
|
|
synthese = {"diag_principal": "Pancréatite aiguë biliaire"}
|
|
scored = score_candidates(candidates, synthese)
|
|
|
|
assert scored[0].score_details.get("diag_section_bonus") == 4
|
|
|
|
def test_bonus_synthese_is_weaker_than_diag_sortie(self):
|
|
"""Candidat A en synthese (+2) perd contre candidat B en diag_sortie (+4)."""
|
|
candidates = [
|
|
DPCandidate(index=0, term="Pneumopathie", code="J18.9",
|
|
confidence="medium", section_strength=1, source="edsnlp"),
|
|
DPCandidate(index=1, term="Sepsis", code="A41.9",
|
|
confidence="medium", section_strength=1, source="edsnlp"),
|
|
]
|
|
synthese = {
|
|
"synthese": "Pneumopathie traitée, évolution favorable",
|
|
"diag_sortie": "Sepsis à point de départ pulmonaire A41.9",
|
|
}
|
|
scored = score_candidates(candidates, synthese)
|
|
|
|
# A41.9 diag_sortie +4, J18.9 synthese +2 → A41.9 gagne
|
|
assert scored[0].code == "A41.9"
|
|
assert scored[0].score_details.get("diag_section_bonus") == 4
|
|
j189 = next(c for c in scored if c.code == "J18.9")
|
|
assert j189.score_details.get("diag_section_bonus") == 2
|
|
|
|
def test_no_bonus_when_sections_absent(self):
|
|
"""Aucune section diagnostique → aucun bonus diag_section_bonus."""
|
|
candidates = [
|
|
DPCandidate(index=0, term="Test", code="K85.1",
|
|
confidence="high", section_strength=2),
|
|
DPCandidate(index=1, term="Autre", code="J18.9",
|
|
confidence="medium", section_strength=1),
|
|
]
|
|
synthese = {"motif": "Douleur abdominale", "conclusion": "Pancréatite aiguë"}
|
|
scored = score_candidates(candidates, synthese)
|
|
|
|
for c in scored:
|
|
assert "diag_section_bonus" not in c.score_details
|
|
|
|
def test_evidence_includes_diag_sortie_excerpt(self):
|
|
"""_collect_evidence() cite un extrait de diag_sortie si le gagnant y apparaît."""
|
|
from src.medical.dp_selector import _collect_evidence
|
|
|
|
winner = DPCandidate(
|
|
index=0, term="SCA", code="I25.1", confidence="high",
|
|
section_strength=1, source="llm_das", score=8.0,
|
|
)
|
|
runner = DPCandidate(
|
|
index=1, term="Anémie", code="D50", confidence="medium",
|
|
section_strength=3, source="regex", score=4.0,
|
|
)
|
|
synthese = {
|
|
"diag_sortie": "SCA ST+ antérieur traité par angioplastie coronaire percutanée",
|
|
"motif": "Douleur thoracique",
|
|
}
|
|
evidence = _collect_evidence(winner, [winner, runner], synthese)
|
|
|
|
# L'evidence doit contenir un excerpt de diag_sortie
|
|
assert len(evidence) >= 1
|
|
diag_ev = [e for e in evidence if "Diagnostic de sortie" in e]
|
|
assert len(diag_ev) == 1, f"Attendu 1 evidence 'Diagnostic de sortie', got: {evidence}"
|
|
# L'excerpt doit contenir le texte réel, pas juste le label
|
|
assert "SCA" in diag_ev[0]
|
|
assert len(diag_ev[0]) >= 20
|
|
|
|
def test_evidence_includes_diag_principal_excerpt(self):
|
|
"""_collect_evidence() cite diag_principal si présent et gagnant y figure."""
|
|
from src.medical.dp_selector import _collect_evidence
|
|
|
|
winner = DPCandidate(
|
|
index=0, term="Embolie pulmonaire", code="I26.9", confidence="high",
|
|
section_strength=2, source="edsnlp", score=9.0,
|
|
)
|
|
synthese = {
|
|
"diag_principal": "Embolie pulmonaire massive bilatérale avec défaillance VD",
|
|
}
|
|
evidence = _collect_evidence(winner, [winner], synthese)
|
|
|
|
diag_ev = [e for e in evidence if "Diagnostic principal" in e]
|
|
assert len(diag_ev) == 1
|
|
assert "Embolie pulmonaire" in diag_ev[0]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests : cas 74 — régression ciblée D50 vs I25.1 (C)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCase74Regression:
|
|
"""Prouve que le patch diag_sortie corrige le bug du cas 74.
|
|
|
|
Bug original : D50 (Anémie, section_strength=3 regex) et I25.1 (SCA,
|
|
section_strength=1 llm_das + confidence=3) avaient un score ex-aequo de 4.0.
|
|
Sans section diagnostique, D50 gagnait par position.
|
|
|
|
Correction : la section 'Diagnostic de sortie' contient 'SCA' et 'I25.1',
|
|
ce qui donne +4 à I25.1 → I25.1 gagne.
|
|
"""
|
|
|
|
def test_case74_i25_beats_d50(self):
|
|
"""select_dp() choisit I25.1 grâce au bonus diag_sortie."""
|
|
fixture = _load_fixture("case_74_min.json")
|
|
dossier = _build_dossier(fixture)
|
|
synthese = fixture["synthese_nuke1"]
|
|
|
|
selection = select_dp(dossier, synthese, config={"llm_enabled": False})
|
|
|
|
assert selection.chosen_code == "I25.1", (
|
|
f"Attendu I25.1, obtenu {selection.chosen_code}. "
|
|
f"Scores: {[(c.code, c.score) for c in selection.candidates]}"
|
|
)
|
|
|
|
def test_case74_verdict_confirmed(self):
|
|
"""Avec le bonus +4, le delta doit être suffisant pour CONFIRMED."""
|
|
fixture = _load_fixture("case_74_min.json")
|
|
dossier = _build_dossier(fixture)
|
|
synthese = fixture["synthese_nuke1"]
|
|
|
|
selection = select_dp(dossier, synthese, config={"llm_enabled": False})
|
|
|
|
assert selection.verdict == "CONFIRMED", (
|
|
f"Attendu CONFIRMED, obtenu {selection.verdict}. "
|
|
f"Reason: {selection.reason}"
|
|
)
|
|
# Règle A1 : CONFIRMED ⇒ evidence non vide
|
|
assert len(selection.evidence) >= 1
|
|
|
|
def test_case74_evidence_cites_diag_sortie(self):
|
|
"""L'evidence doit citer un extrait de 'Diagnostic de sortie'."""
|
|
fixture = _load_fixture("case_74_min.json")
|
|
dossier = _build_dossier(fixture)
|
|
synthese = fixture["synthese_nuke1"]
|
|
|
|
selection = select_dp(dossier, synthese, config={"llm_enabled": False})
|
|
|
|
diag_ev = [e for e in selection.evidence if "Diagnostic de sortie" in e]
|
|
assert len(diag_ev) >= 1, (
|
|
f"Evidence ne cite pas 'Diagnostic de sortie': {selection.evidence}"
|
|
)
|
|
assert "SCA" in diag_ev[0]
|
|
|
|
def test_case74_score_details(self):
|
|
"""Vérification fine des scores pour traçabilité."""
|
|
fixture = _load_fixture("case_74_min.json")
|
|
dossier = _build_dossier(fixture)
|
|
synthese = fixture["synthese_nuke1"]
|
|
|
|
selection = select_dp(dossier, synthese, config={"llm_enabled": False})
|
|
|
|
# Trouver I25.1 et D50 dans les candidats
|
|
i25 = next((c for c in selection.candidates if c.code == "I25.1"), None)
|
|
d50 = next((c for c in selection.candidates if c.code == "D50"), None)
|
|
|
|
assert i25 is not None, "I25.1 absent des candidats"
|
|
assert d50 is not None, "D50 absent des candidats"
|
|
|
|
# I25.1 : section=1 + confidence=3(high) + diag_sortie=4 = 8
|
|
assert i25.score_details.get("diag_section_bonus") == 4
|
|
assert i25.score_details.get("confidence") == 3
|
|
|
|
# D50 : section=3(regex) + confidence=1(medium) + diag_sortie=4 = 8
|
|
# MAIS I25.1 a +2 de plus via confidence (3 vs 1)
|
|
assert i25.score > d50.score, (
|
|
f"I25.1 ({i25.score}) doit battre D50 ({d50.score})"
|
|
)
|
|
|
|
def test_case74_z95_not_dp(self):
|
|
"""Z95.5 (stent) ne doit pas être choisi comme DP (malus Z-code)."""
|
|
fixture = _load_fixture("case_74_min.json")
|
|
dossier = _build_dossier(fixture)
|
|
synthese = fixture["synthese_nuke1"]
|
|
|
|
selection = select_dp(dossier, synthese, config={"llm_enabled": False})
|
|
|
|
assert selection.chosen_code != "Z95.5"
|
|
z95 = next((c for c in selection.candidates if c.code == "Z95.5"), None)
|
|
if z95:
|
|
assert "z_code_malus" in z95.score_details
|