291 lines
12 KiB
Python
291 lines
12 KiB
Python
"""Tests P0 : correctifs bloquants (BUG-1, BUG-2, LOGIC-1)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from src.config import (
|
|
CodeDecision,
|
|
Diagnostic,
|
|
DossierMedical,
|
|
RAGSource,
|
|
Sejour,
|
|
)
|
|
|
|
|
|
# ============================================================
|
|
# P0-1 — BUG-1 : VETO-02 ne doit pas s'appliquer au DP Trackare
|
|
# ============================================================
|
|
|
|
class TestVeto02SkipsTrackareDp:
|
|
"""VETO-02 (DP sans preuve) doit être ignoré quand dp.source == 'trackare'."""
|
|
|
|
@patch("src.quality.veto_engine.rule_enabled", return_value=True)
|
|
@patch("src.quality.veto_engine.rule_force_severity", return_value=None)
|
|
def test_trackare_dp_no_veto02(self, _mock_sev, _mock_rule, dossier_trackare_dp):
|
|
"""Un DP Trackare sans preuve ne doit PAS déclencher VETO-02 HARD."""
|
|
from src.quality.veto_engine import apply_vetos
|
|
|
|
report = apply_vetos(dossier_trackare_dp)
|
|
|
|
veto02_dp = [
|
|
i for i in report.issues
|
|
if i.veto == "VETO-02" and i.where == "diagnostic_principal"
|
|
]
|
|
assert veto02_dp == [], (
|
|
f"VETO-02 déclenché à tort sur DP Trackare : {veto02_dp}"
|
|
)
|
|
|
|
@patch("src.quality.veto_engine.rule_enabled", return_value=True)
|
|
@patch("src.quality.veto_engine.rule_force_severity", return_value=None)
|
|
def test_trackare_dp_verdict_not_fail_from_veto02(self, _mock_sev, _mock_rule, dossier_trackare_dp):
|
|
"""Le verdict global ne doit pas être FAIL à cause du seul VETO-02 DP Trackare."""
|
|
from src.quality.veto_engine import apply_vetos
|
|
|
|
report = apply_vetos(dossier_trackare_dp)
|
|
|
|
# Pas de HARD issue liée à VETO-02 sur le DP
|
|
hard_veto02 = [
|
|
i for i in report.issues
|
|
if i.veto == "VETO-02" and i.where == "diagnostic_principal" and i.severity == "HARD"
|
|
]
|
|
assert hard_veto02 == []
|
|
|
|
@patch("src.quality.veto_engine.rule_enabled", return_value=True)
|
|
@patch("src.quality.veto_engine.rule_force_severity", return_value=None)
|
|
def test_crh_dp_without_evidence_still_triggers_veto02(self, _mock_sev, _mock_rule):
|
|
"""Un DP CRH sans preuve doit toujours déclencher VETO-02 HARD (non-régression)."""
|
|
from src.quality.veto_engine import apply_vetos
|
|
|
|
dossier = DossierMedical(
|
|
document_type="crh",
|
|
sejour=Sejour(sexe="M", age=50),
|
|
diagnostic_principal=Diagnostic(
|
|
texte="Pneumopathie",
|
|
cim10_suggestion="J18.9",
|
|
source="crh",
|
|
# Pas de preuve
|
|
),
|
|
)
|
|
report = apply_vetos(dossier)
|
|
|
|
veto02_dp = [
|
|
i for i in report.issues
|
|
if i.veto == "VETO-02" and i.where == "diagnostic_principal"
|
|
]
|
|
assert len(veto02_dp) == 1, "VETO-02 doit toujours s'appliquer aux DP CRH sans preuve"
|
|
assert veto02_dp[0].severity == "HARD"
|
|
|
|
@patch("src.quality.veto_engine.rule_enabled", return_value=True)
|
|
@patch("src.quality.veto_engine.rule_force_severity", return_value=None)
|
|
def test_trackare_dp_with_source_none_still_triggers_veto02(self, _mock_sev, _mock_rule):
|
|
"""Un DP sans source définie (source=None) doit déclencher VETO-02 normalement."""
|
|
from src.quality.veto_engine import apply_vetos
|
|
|
|
dossier = DossierMedical(
|
|
sejour=Sejour(),
|
|
diagnostic_principal=Diagnostic(
|
|
texte="Test",
|
|
cim10_suggestion="A00.0",
|
|
# source=None (défaut)
|
|
),
|
|
)
|
|
report = apply_vetos(dossier)
|
|
|
|
veto02_dp = [
|
|
i for i in report.issues
|
|
if i.veto == "VETO-02" and i.where == "diagnostic_principal"
|
|
]
|
|
assert len(veto02_dp) == 1, "DP sans source doit déclencher VETO-02"
|
|
|
|
|
|
# ============================================================
|
|
# P0-2 — BUG-2 : sources_rag toujours initialisé (même si RAG vide)
|
|
# ============================================================
|
|
|
|
class TestRagZeroResultsSetsSourcesRag:
|
|
"""enrich_diagnostic() doit toujours initialiser sources_rag, même sans résultat FAISS."""
|
|
|
|
@patch("src.medical.rag_search.search_similar", return_value=[])
|
|
def test_zero_results_sets_empty_list(self, _mock_faiss):
|
|
"""sources_rag doit être [] (pas None) quand FAISS retourne 0 résultat."""
|
|
from src.medical.rag_search import enrich_diagnostic
|
|
|
|
diag = Diagnostic(texte="Test diagnostic", cim10_suggestion="K85.1")
|
|
assert diag.sources_rag == [] # Pydantic default
|
|
|
|
enrich_diagnostic(diag, contexte={}, est_dp=True, cache=None)
|
|
|
|
assert diag.sources_rag == [], (
|
|
f"sources_rag devrait être [] après 0 résultat FAISS, got: {diag.sources_rag}"
|
|
)
|
|
|
|
@patch("src.medical.rag_search.search_similar", return_value=[])
|
|
def test_zero_results_with_cache_hit_applies_cached(self, _mock_faiss):
|
|
"""Avec 0 résultat FAISS mais un cache hit, le résultat LLM doit être appliqué."""
|
|
from src.medical.rag_search import enrich_diagnostic
|
|
|
|
diag = Diagnostic(texte="Pancréatite aiguë", cim10_suggestion="K85.1")
|
|
|
|
mock_cache = MagicMock()
|
|
mock_cache.get.return_value = {
|
|
"code": "K85.1",
|
|
"confidence": "high",
|
|
"justification": "Cached justification",
|
|
}
|
|
|
|
with patch("src.medical.rag_search._apply_llm_result_diagnostic") as mock_apply:
|
|
enrich_diagnostic(diag, contexte={}, est_dp=True, cache=mock_cache)
|
|
|
|
# Le cache hit doit être appliqué malgré 0 résultat FAISS
|
|
mock_apply.assert_called_once()
|
|
assert diag.sources_rag == []
|
|
|
|
@patch("src.medical.rag_search.search_similar", return_value=[
|
|
{"document": "cim10", "page": 42, "code": "K85.1", "extrait": "Pancréatite aigüe biliaire"},
|
|
])
|
|
def test_with_results_sets_sources_rag(self, _mock_faiss):
|
|
"""Avec des résultats FAISS, sources_rag doit être rempli normalement (non-régression)."""
|
|
from src.medical.rag_search import enrich_diagnostic
|
|
|
|
diag = Diagnostic(texte="Pancréatite", cim10_suggestion="K85.1")
|
|
|
|
with patch("src.medical.rag_search._call_ollama", return_value=None):
|
|
enrich_diagnostic(diag, contexte={}, est_dp=True, cache=None)
|
|
|
|
assert len(diag.sources_rag) == 1
|
|
assert diag.sources_rag[0].document == "cim10"
|
|
assert diag.sources_rag[0].code == "K85.1"
|
|
|
|
|
|
# ============================================================
|
|
# P0-3 — LOGIC-1 : promotion DAS→DP doit être tracée
|
|
# ============================================================
|
|
|
|
class TestDasToDpPromotionTraced:
|
|
"""RULE-DAS-TO-DP doit laisser une trace dans alertes_codage."""
|
|
|
|
@patch("src.quality.decision_engine.rule_enabled", return_value=True)
|
|
@patch("src.quality.decision_engine.cim10_validate", return_value=(True, "label"))
|
|
@patch("src.quality.decision_engine.load_reference_ranges", return_value={})
|
|
@patch("src.quality.decision_engine.load_bio_rules", return_value={})
|
|
def test_promotion_adds_alerte(self, _bio, _ref, _valid, _rule):
|
|
"""Quand un DAS est promu DP, alertes_codage doit contenir RULE-DAS-TO-DP."""
|
|
from src.quality.decision_engine import apply_decisions
|
|
|
|
das_candidate = Diagnostic(
|
|
texte="Pancréatite aiguë biliaire",
|
|
cim10_suggestion="K85.1",
|
|
cim10_confidence="high",
|
|
source="crh",
|
|
)
|
|
dossier = DossierMedical(
|
|
sejour=Sejour(),
|
|
diagnostic_principal=None,
|
|
diagnostics_associes=[das_candidate],
|
|
)
|
|
|
|
apply_decisions(dossier)
|
|
|
|
# Le DP doit avoir été promu
|
|
assert dossier.diagnostic_principal is not None
|
|
assert dossier.diagnostic_principal.cim10_final == "K85.1"
|
|
assert dossier.diagnostic_principal.cim10_decision.action == "PROMOTE_DP"
|
|
|
|
# Traçabilité : alerte avec règle et code
|
|
matching_alertes = [
|
|
a for a in dossier.alertes_codage
|
|
if "RULE-DAS-TO-DP" in a and "K85.1" in a
|
|
]
|
|
assert len(matching_alertes) == 1, (
|
|
f"alertes_codage devrait contenir une entrée RULE-DAS-TO-DP, got: {dossier.alertes_codage}"
|
|
)
|
|
|
|
@patch("src.quality.decision_engine.rule_enabled", return_value=True)
|
|
@patch("src.quality.decision_engine.cim10_validate", return_value=(True, "label"))
|
|
@patch("src.quality.decision_engine.load_reference_ranges", return_value={})
|
|
@patch("src.quality.decision_engine.load_bio_rules", return_value={})
|
|
def test_promotion_alerte_contains_diagnosis_text(self, _bio, _ref, _valid, _rule):
|
|
"""L'alerte de promotion doit mentionner le texte du diagnostic promu."""
|
|
from src.quality.decision_engine import apply_decisions
|
|
|
|
dossier = DossierMedical(
|
|
sejour=Sejour(),
|
|
diagnostic_principal=None,
|
|
diagnostics_associes=[
|
|
Diagnostic(
|
|
texte="Embolie pulmonaire",
|
|
cim10_suggestion="I26.9",
|
|
cim10_confidence="high",
|
|
source="crh",
|
|
),
|
|
],
|
|
)
|
|
|
|
apply_decisions(dossier)
|
|
|
|
assert dossier.diagnostic_principal is not None
|
|
alerte = [a for a in dossier.alertes_codage if "RULE-DAS-TO-DP" in a]
|
|
assert len(alerte) == 1
|
|
assert "I26.9" in alerte[0]
|
|
assert "Embolie pulmonaire" in alerte[0]
|
|
|
|
@patch("src.quality.decision_engine.rule_enabled", return_value=True)
|
|
@patch("src.quality.decision_engine.cim10_validate", return_value=(True, "label"))
|
|
@patch("src.quality.decision_engine.load_reference_ranges", return_value={})
|
|
@patch("src.quality.decision_engine.load_bio_rules", return_value={})
|
|
def test_no_promotion_when_dp_exists(self, _bio, _ref, _valid, _rule):
|
|
"""Pas de promotion si un DP existe déjà (non-régression)."""
|
|
from src.quality.decision_engine import apply_decisions
|
|
|
|
dossier = DossierMedical(
|
|
sejour=Sejour(),
|
|
diagnostic_principal=Diagnostic(
|
|
texte="DP existant",
|
|
cim10_suggestion="J18.9",
|
|
source="crh",
|
|
),
|
|
diagnostics_associes=[
|
|
Diagnostic(
|
|
texte="DAS candidat",
|
|
cim10_suggestion="K85.1",
|
|
cim10_confidence="high",
|
|
),
|
|
],
|
|
)
|
|
|
|
apply_decisions(dossier)
|
|
|
|
# DP inchangé
|
|
assert dossier.diagnostic_principal.cim10_suggestion == "J18.9"
|
|
# Pas d'alerte de promotion
|
|
promotion_alertes = [a for a in dossier.alertes_codage if "RULE-DAS-TO-DP" in a]
|
|
assert promotion_alertes == []
|
|
|
|
@patch("src.quality.decision_engine.rule_enabled", return_value=True)
|
|
@patch("src.quality.decision_engine.cim10_validate", return_value=(True, "label"))
|
|
@patch("src.quality.decision_engine.load_reference_ranges", return_value={})
|
|
@patch("src.quality.decision_engine.load_bio_rules", return_value={})
|
|
def test_promotion_removes_das_from_list(self, _bio, _ref, _valid, _rule):
|
|
"""Le DAS promu doit être retiré de diagnostics_associes."""
|
|
from src.quality.decision_engine import apply_decisions
|
|
|
|
das1 = Diagnostic(texte="DAS gardé", cim10_suggestion="R10.4", cim10_confidence="high")
|
|
das2 = Diagnostic(texte="Pancréatite", cim10_suggestion="K85.1", cim10_confidence="high")
|
|
dossier = DossierMedical(
|
|
sejour=Sejour(),
|
|
diagnostic_principal=None,
|
|
diagnostics_associes=[das1, das2],
|
|
)
|
|
|
|
apply_decisions(dossier)
|
|
|
|
# K85.1 promu (score pathologie > symptôme R)
|
|
assert dossier.diagnostic_principal is not None
|
|
assert dossier.diagnostic_principal.cim10_final == "K85.1"
|
|
# Le DAS promu ne doit plus être dans la liste
|
|
remaining_codes = [d.cim10_suggestion for d in dossier.diagnostics_associes]
|
|
assert "K85.1" not in remaining_codes
|