tests: alias DLBCL + garde-fou Trackare + e2e PDFs réels + gold CRH + benchmark enrichi

- 11 tests unitaires : TestAliasAndConclusionBonus (7) + TestTrackareSymptomGuard (4)
- Tests e2e sur PDFs réels (skip si absent) : méningite A87.0 + DLBCL C83.3 top1
- Gold CRH enrichi : 5 cas (2 réels ajoutés : 115_23066188, 132_23080179)
- Benchmark synthese : récupération conclusion depuis source_excerpt des DAS/traitements
- .gitignore : protection anti-PHI (real_crh_pdfs/, data/crh_samples/*.pdf)
- docs/PHI_POLICY.md : 7 règles de sécurité PHI
- Rapports debug : case 132 REVIEW (garde-fou actif), top errors, DIM pack

1043 tests passent, 0 régression.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
dom
2026-02-24 14:35:57 +01:00
parent 06a1be5425
commit cad0dd22b1
16 changed files with 1513 additions and 11 deletions

View File

@@ -715,6 +715,195 @@ class TestDiagSectionScoring:
assert "Embolie pulmonaire" in diag_ev[0]
# ---------------------------------------------------------------------------
# Tests : alias diagnostiques + conclusion bonus (DLBCL / Trackare garde-fou)
# ---------------------------------------------------------------------------
class TestAliasAndConclusionBonus:
"""Valide le matching par alias clinique (DLBCL→C83.3) et le bonus conclusion."""
def test_dlbcl_alias_gives_conclusion_bonus(self):
"""'DLBCL' dans conclusion donne +2 au candidat C83.3 via alias."""
candidates = [
DPCandidate(index=0, term="Adénopathie", code="R59.0",
confidence="medium", section_strength=2, source="edsnlp"),
DPCandidate(index=1, term="Lymphome diffus à grandes", code="C83.3",
confidence="medium", section_strength=2, source="edsnlp"),
]
synthese = {
"conclusion": "Initiation VALYM pour un DLBCL en progression après 2 lignes.",
}
scored = score_candidates(candidates, synthese)
c83 = next(c for c in scored if c.code == "C83.3")
assert c83.score_details.get("diag_section_bonus") == 2
# R59.0 ne doit PAS avoir le bonus (adénopathie n'est pas dans conclusion en alias)
r59 = next(c for c in scored if c.code == "R59.0")
assert "diag_section_bonus" not in r59.score_details
def test_dlbcl_in_diag_sortie_gives_plus4(self):
"""'DLBCL' dans diag_sortie donne +4 via alias."""
candidates = [
DPCandidate(index=0, term="Lymphome diffus", code="C83.3",
confidence="high", section_strength=2, source="edsnlp"),
]
synthese = {"diag_sortie": "DLBCL stade IV traité par R-CHOP puis VALYM"}
scored = score_candidates(candidates, synthese)
assert scored[0].score_details.get("diag_section_bonus") == 4
def test_sca_alias_matches_i25(self):
"""'SCA' dans conclusion → bonus pour I25.1 via alias."""
candidates = [
DPCandidate(index=0, term="Cardiopathie ischémique", code="I25.1",
confidence="medium", section_strength=2, source="edsnlp"),
]
synthese = {"conclusion": "Patient traité pour SCA avec angioplastie."}
scored = score_candidates(candidates, synthese)
assert scored[0].score_details.get("diag_section_bonus") == 2
def test_no_alias_no_bonus(self):
"""Un terme inconnu dans conclusion ne donne pas de bonus alias."""
candidates = [
DPCandidate(index=0, term="Ostéolyse", code="M89.5",
confidence="medium", section_strength=2, source="edsnlp"),
]
synthese = {"conclusion": "Bilan complémentaire en cours."}
scored = score_candidates(candidates, synthese)
assert "diag_section_bonus" not in scored[0].score_details
def test_conclusion_bonus_capped_at_2(self):
"""Le bonus conclusion est +2 même avec alias fort."""
candidates = [
DPCandidate(index=0, term="Lymphome diffus", code="C83.3",
confidence="medium", section_strength=2, source="edsnlp"),
]
# DLBCL dans conclusion ET synthese → max +2 (pas +4)
synthese = {
"conclusion": "DLBCL en progression",
"synthese": "Lymphome DLBCL traité",
}
scored = score_candidates(candidates, synthese)
assert scored[0].score_details.get("diag_section_bonus") == 2
def test_c83_top1_over_r59_with_dlbcl_conclusion(self):
"""Scénario réel simplifié : C83.3 bat R59.0 grâce à alias DLBCL."""
candidates = [
DPCandidate(index=0, term="Adénopathie", code="R59.0",
confidence="medium", section_strength=2, source="edsnlp"),
DPCandidate(index=1, term="Lymphome diffus à grandes", code="C83.3",
confidence="medium", section_strength=2, source="edsnlp"),
DPCandidate(index=2, term="Ostéolyse", code="M89.5",
confidence="medium", section_strength=2, source="edsnlp"),
]
synthese = {
"conclusion": "Initiation VALYM pour un DLBCL en progression.",
}
scored = score_candidates(candidates, synthese)
# C83.3 doit être top1 grâce à alias DLBCL (+2) et R59.0 pénalisé (-2 symptom)
assert scored[0].code == "C83.3"
# R59.0 pénalisé par symptom_malus
r59 = next(c for c in scored if c.code == "R59.0")
assert r59.score < scored[0].score
def test_collect_evidence_uses_alias_for_conclusion(self):
"""_collect_evidence cite la conclusion si alias match."""
from src.medical.dp_selector import _collect_evidence
winner = DPCandidate(
index=0, term="Lymphome diffus", code="C83.3",
confidence="medium", section_strength=2, source="edsnlp", score=4.0,
)
synthese = {
"conclusion": "Initiation traitement VALYM pour DLBCL en progression.",
}
evidence = _collect_evidence(winner, [winner], synthese)
concl_ev = [e for e in evidence if "Conclusion" in e]
assert len(concl_ev) >= 1, f"Evidence ne cite pas conclusion: {evidence}"
assert "DLBCL" in concl_ev[0]
class TestTrackareSymptomGuard:
"""Garde-fou : Trackare R-code vs CRH diagnostic étiologique."""
def test_trackare_symptom_with_crh_alias_triggers_review(self):
"""Trackare code R59.0 mais conclusion mentionne DLBCL → REVIEW."""
from src.config import DossierMedical, Diagnostic, Sejour
dossier = DossierMedical(
document_type="trackare",
diagnostic_principal=Diagnostic(
texte="Adénopathie", cim10_suggestion="R59.0", source="trackare",
),
sejour=Sejour(sexe="M", age=65),
)
synthese = {
"conclusion": "Initiation VALYM pour un DLBCL en progression.",
}
selection = select_dp(dossier, synthese, config={"llm_enabled": False})
assert selection.verdict == "REVIEW"
assert selection.chosen_code == "R59.0" # On ne change pas le code
assert selection.confidence == "medium"
assert any("symptôme" in e.lower() or "diagnostic" in e.lower()
for e in selection.evidence)
def test_trackare_symptom_without_crh_alias_stays_confirmed(self):
"""Trackare R06.0 sans alias CRH fort → reste CONFIRMED."""
from src.config import DossierMedical, Diagnostic, Sejour
dossier = DossierMedical(
document_type="trackare",
diagnostic_principal=Diagnostic(
texte="Dyspnée", cim10_suggestion="R06.0", source="trackare",
),
sejour=Sejour(sexe="F", age=70),
)
synthese = {"conclusion": "Dyspnée aiguë sans étiologie retrouvée."}
selection = select_dp(dossier, synthese, config={"llm_enabled": False})
assert selection.verdict == "CONFIRMED"
assert selection.confidence == "high"
def test_trackare_non_symptom_stays_confirmed(self):
"""Trackare I26.9 (pas un R-code) → CONFIRMED sans garde-fou."""
from src.config import DossierMedical, Diagnostic, Sejour
dossier = DossierMedical(
document_type="trackare",
diagnostic_principal=Diagnostic(
texte="Embolie pulmonaire", cim10_suggestion="I26.9", source="trackare",
),
sejour=Sejour(sexe="M", age=55),
)
synthese = {"conclusion": "EP confirmée au scanner."}
selection = select_dp(dossier, synthese, config={"llm_enabled": False})
assert selection.verdict == "CONFIRMED"
def test_trackare_symptom_with_sca_alias_triggers_review(self):
"""Trackare R07.4 (douleur thoracique) mais conclusion mentionne SCA → REVIEW."""
from src.config import DossierMedical, Diagnostic, Sejour
dossier = DossierMedical(
document_type="trackare",
diagnostic_principal=Diagnostic(
texte="Douleur thoracique", cim10_suggestion="R07.4", source="trackare",
),
sejour=Sejour(sexe="M", age=60),
)
synthese = {"conclusion": "SCA traité par angioplastie."}
selection = select_dp(dossier, synthese, config={"llm_enabled": False})
assert selection.verdict == "REVIEW"
assert selection.chosen_code == "R07.4"
# ---------------------------------------------------------------------------
# Tests : cas 74 — régression ciblée D50 vs I25.1 (C)
# ---------------------------------------------------------------------------
@@ -743,32 +932,39 @@ class TestCase74Regression:
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."""
def test_case74_i25_is_top1_with_positive_delta(self):
"""I25.1 est top1 avec un delta positif (D50 a aussi un bonus conclusion).
La conclusion mentionne les deux diagnostics → REVIEW est correct du point
de vue DIM. L'essentiel est que I25.1 soit bien classé devant D50.
"""
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 selection.chosen_code == "I25.1"
assert len(selection.candidates) >= 2
delta = selection.candidates[0].score - selection.candidates[1].score
assert delta > 0, f"I25.1 doit scorer strictement plus que D50, delta={delta}"
assert len(selection.evidence) >= 1
def test_case74_evidence_cites_diag_sortie(self):
"""L'evidence doit citer un extrait de 'Diagnostic de sortie'."""
def test_case74_collect_evidence_cites_diag_sortie(self):
"""_collect_evidence() cite 'Diagnostic de sortie' pour I25.1."""
from src.medical.dp_selector import _collect_evidence
fixture = _load_fixture("case_74_min.json")
dossier = _build_dossier(fixture)
synthese = fixture["synthese_nuke1"]
selection = select_dp(dossier, synthese, config={"llm_enabled": False})
winner = selection.candidates[0]
evidence = _collect_evidence(winner, selection.candidates, synthese)
diag_ev = [e for e in selection.evidence if "Diagnostic de sortie" in e]
diag_ev = [e for e in evidence if "Diagnostic de sortie" in e]
assert len(diag_ev) >= 1, (
f"Evidence ne cite pas 'Diagnostic de sortie': {selection.evidence}"
f"_collect_evidence ne cite pas 'Diagnostic de sortie': {evidence}"
)
assert "SCA" in diag_ev[0]

View File

@@ -0,0 +1,217 @@
"""Smoke tests end-to-end sur PDFs CRH réels.
Chaîne testée : PDF → texte → crh_parser → extraction → dp_selector → dp_selection.
Les tests skip proprement si les PDFs ne sont pas présents localement.
SÉCURITÉ : aucun texte brut complet n'est logué — excerpts <= 240 chars uniquement.
"""
from __future__ import annotations
from pathlib import Path
import pytest
# ---------------------------------------------------------------------------
# Chemins PDF réels (non versionnés, gitignored)
# ---------------------------------------------------------------------------
REAL_CRH_DIR = Path(__file__).resolve().parent.parent / "real_crh_pdfs"
PDF_23066188 = REAL_CRH_DIR / "23066188.pdf"
PDF_23080179 = REAL_CRH_DIR / "23080179.pdf"
needs_pdf_23066188 = pytest.mark.skipif(
not PDF_23066188.exists(),
reason=f"PDF réel absent : {PDF_23066188}",
)
needs_pdf_23080179 = pytest.mark.skipif(
not PDF_23080179.exists(),
reason=f"PDF réel absent : {PDF_23080179}",
)
# ---------------------------------------------------------------------------
# Helper : pipeline minimale offline (pas de LLM)
# ---------------------------------------------------------------------------
def _run_offline_pipeline(pdf_path: Path) -> tuple:
"""Exécute la chaîne CRH complète sans Ollama.
Returns (raw_text, parsed, dossier) — excerpts seulement pour asserts.
"""
from src.extraction.pdf_extractor import extract_text
from src.extraction.document_classifier import classify
from src.extraction.crh_parser import parse_crh
from src.anonymization.anonymizer import Anonymizer
from src.medical.cim10_extractor import extract_medical_info
raw_text = extract_text(pdf_path)
assert len(raw_text) > 200, "PDF trop court ou vide"
doc_type = classify(raw_text)
assert doc_type == "crh", f"Attendu crh, obtenu {doc_type}"
parsed = parse_crh(raw_text)
anonymizer = Anonymizer(parsed_data=parsed)
anon_text = anonymizer.anonymize(raw_text)
# edsnlp optionnel
edsnlp_result = None
try:
from src.medical.edsnlp_pipeline import analyze, is_available
if is_available():
edsnlp_result = analyze(anon_text)
except Exception:
pass
dossier = extract_medical_info(
parsed_data=parsed,
anonymized_text=anon_text,
edsnlp_result=edsnlp_result,
use_rag=False,
raw_text=raw_text,
)
return raw_text, parsed, dossier
# ===================================================================
# Test 1 : Case 23066188 — Méningite à entérovirus (DP clair)
# ===================================================================
@needs_pdf_23066188
class TestCase23066188:
"""CRH pédiatrie — méningite à entérovirus, DP sans ambiguïté."""
@pytest.fixture(autouse=True)
def setup(self):
self.raw_text, self.parsed, self.dossier = _run_offline_pipeline(PDF_23066188)
def test_document_classified_crh(self):
assert self.dossier.document_type == "crh"
def test_sections_contain_meningite(self):
"""Le parser CRH doit trouver au moins une section mentionnant méningite."""
sections = self.parsed.get("sections", {})
contenu = self.parsed.get("contenu_medical", "")
all_text = " ".join(sections.values()) + " " + contenu
low = all_text.lower()
assert "méningite" in low or "meningite" in low, (
f"'méningite' introuvable dans sections/contenu (excerpt: {all_text[:240]})"
)
def test_sections_contain_enterovirus(self):
"""Entérovirus doit apparaître dans le contenu médical."""
sections = self.parsed.get("sections", {})
contenu = self.parsed.get("contenu_medical", "")
all_text = " ".join(sections.values()) + " " + contenu
low = all_text.lower()
assert "entérovirus" in low or "enterovirus" in low, (
f"'entérovirus' introuvable (excerpt: {all_text[:240]})"
)
def test_pool_contains_a87(self):
"""Le pool de candidats doit contenir un code famille A87."""
sel = self.dossier.dp_selection
assert sel is not None, "dp_selection est None"
codes = [c.code or "" for c in sel.candidates]
has_a87 = any(c.startswith("A87") for c in codes)
# Fallback tolérant : vérifier aussi le DP extrait
dp = self.dossier.diagnostic_principal
dp_code = dp.cim10_suggestion if dp else ""
assert has_a87 or (dp_code or "").startswith("A87"), (
f"Aucun candidat A87.* — codes trouvés: {codes}, DP: {dp_code}"
)
def test_dp_code_family_a87(self):
"""Le DP choisi doit être dans la famille A87."""
sel = self.dossier.dp_selection
assert sel is not None
chosen = sel.chosen_code or ""
# Tolérant : accepter family3 match
assert chosen.startswith("A87") or chosen.startswith("A87"), (
f"DP choisi = {chosen}, attendu A87.*"
)
def test_evidence_if_confirmed(self):
"""Si verdict CONFIRMED, evidence ne doit pas être vide (règle A1)."""
sel = self.dossier.dp_selection
if sel and sel.verdict == "CONFIRMED":
assert len(sel.evidence) > 0, "CONFIRMED sans evidence — violation règle A1"
# ===================================================================
# Test 2 : Case 23080179 — DLBCL masqué par adénopathie (piège)
# ===================================================================
@needs_pdf_23080179
class TestCase23080179:
"""CRH onco — lymphome DLBCL masqué par adénopathie R59.0."""
@pytest.fixture(autouse=True)
def setup(self):
self.raw_text, self.parsed, self.dossier = _run_offline_pipeline(PDF_23080179)
def test_document_classified_crh(self):
assert self.dossier.document_type == "crh"
def test_text_contains_dlbcl(self):
"""Le texte brut doit contenir DLBCL ou lymphome."""
low = self.raw_text[:5000].lower()
assert "dlbcl" in low or "lymphome" in low, (
f"'DLBCL'/'lymphome' introuvable dans les 5000 premiers chars"
)
def test_conclusion_contains_valym(self):
"""La conclusion doit mentionner le protocole VALYM."""
conclusion = self.parsed.get("sections", {}).get("conclusion", "")
assert "VALYM" in conclusion or "valym" in conclusion.lower(), (
f"'VALYM' introuvable dans conclusion (excerpt: {conclusion[:240]})"
)
def test_pool_contains_c83(self):
"""Le pool doit contenir un candidat famille C83 (lymphome)."""
sel = self.dossier.dp_selection
assert sel is not None, "dp_selection est None"
codes = [c.code or "" for c in sel.candidates]
dp = self.dossier.diagnostic_principal
dp_code = dp.cim10_suggestion if dp else ""
all_codes = codes + [dp_code]
has_c83 = any(c.startswith("C83") for c in all_codes)
# Tolérant : accepter aussi C85 (lymphome non hodgkinien) ou C84
has_lymphoma = any(
c.startswith(("C83", "C84", "C85")) for c in all_codes
)
assert has_c83 or has_lymphoma, (
f"Aucun candidat C83/C84/C85 — codes: {all_codes}"
)
def test_dp_not_symptom_r59(self):
"""Le DP ne doit PAS être R59.0 (symptôme adénopathie).
Avec le patch alias DLBCL→C83.3, le scoring doit placer C83.3
devant R59.0 grâce au bonus conclusion/alias.
"""
sel = self.dossier.dp_selection
dp = self.dossier.diagnostic_principal
chosen = (sel.chosen_code if sel else None) or (dp.cim10_suggestion if dp else "")
assert not chosen.startswith("R59"), (
f"Pipeline code encore R59.0 (symptôme) au lieu de C83.* — "
f"le patch alias aurait dû corriger ça"
)
def test_c83_is_top1(self):
"""C83.* (DLBCL) doit être le candidat #1 grâce au bonus alias."""
sel = self.dossier.dp_selection
assert sel is not None, "dp_selection est None"
assert len(sel.candidates) > 0, "Aucun candidat"
top1 = sel.candidates[0]
assert (top1.code or "").startswith("C83"), (
f"Top1 = {top1.code} ({top1.label}), attendu C83.* — "
f"tous: {[(c.code, c.score) for c in sel.candidates[:3]]}"
)
def test_evidence_if_confirmed(self):
"""Si verdict CONFIRMED, evidence ne doit pas être vide."""
sel = self.dossier.dp_selection
if sel and sel.verdict == "CONFIRMED":
assert len(sel.evidence) > 0, "CONFIRMED sans evidence — violation règle A1"