553 lines
17 KiB
Python
553 lines
17 KiB
Python
"""
|
|
Tests unitaires pour le QuestionGenerator.
|
|
|
|
Ces tests vérifient:
|
|
- La génération de questions priorisées (max 5)
|
|
- La détection d'incohérences codes/faits
|
|
- La priorisation des questions par impact
|
|
"""
|
|
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
|
|
from pipeline_mco_pmsi.models.clinical import (
|
|
ClinicalDocument,
|
|
ClinicalFact,
|
|
Evidence,
|
|
Qualifier,
|
|
Span,
|
|
StructuredStay,
|
|
)
|
|
from pipeline_mco_pmsi.models.coding import Code, CodingProposal
|
|
from pipeline_mco_pmsi.models.metadata import ModelVersion
|
|
from pipeline_mco_pmsi.models.validation import ValidationIssue
|
|
from pipeline_mco_pmsi.validators.question_generator import QuestionGenerator
|
|
|
|
|
|
@pytest.fixture
|
|
def question_generator():
|
|
"""Crée une instance de QuestionGenerator."""
|
|
return QuestionGenerator()
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_document():
|
|
"""Crée un document clinique de test."""
|
|
return ClinicalDocument(
|
|
document_id="doc_001",
|
|
document_type="cr_medical",
|
|
content="Patient présente une gastrite aiguë confirmée par endoscopie.",
|
|
creation_date=datetime(2024, 1, 15, 10, 30),
|
|
author="Dr. Martin",
|
|
priority=2,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_evidence():
|
|
"""Crée une preuve de test."""
|
|
return Evidence(
|
|
document_id="doc_001",
|
|
span=Span(start=20, end=35),
|
|
text="gastrite aiguë",
|
|
context="Patient présente une gastrite aiguë confirmée",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_code(sample_evidence):
|
|
"""Crée un code de test."""
|
|
return Code(
|
|
code="K29.1",
|
|
label="Gastrite aiguë",
|
|
type="dp",
|
|
evidence=[sample_evidence],
|
|
confidence=0.85,
|
|
reasoning="Diagnostic principal confirmé par endoscopie",
|
|
referentiel_version="2026",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_proposal(sample_code):
|
|
"""Crée une proposition de codage de test."""
|
|
return CodingProposal(
|
|
stay_id="stay_001",
|
|
dp=sample_code,
|
|
dr=None,
|
|
das=[],
|
|
ccam=[],
|
|
reasoning="Séjour pour gastrite aiguë",
|
|
model_version=ModelVersion(
|
|
model_name="test-model",
|
|
model_tag="v1.0",
|
|
model_digest="a" * 64,
|
|
),
|
|
prompt_version="v1.0",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_fact(sample_evidence):
|
|
"""Crée un fait clinique de test."""
|
|
return ClinicalFact(
|
|
fact_id="fact_001",
|
|
type="diagnostic",
|
|
text="gastrite aiguë",
|
|
qualifier=Qualifier(
|
|
certainty="affirmé",
|
|
markers=[],
|
|
confidence=0.9,
|
|
),
|
|
temporality="actuel",
|
|
evidence=sample_evidence,
|
|
confidence=0.9,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_stay(sample_document, sample_fact):
|
|
"""Crée un séjour structuré de test."""
|
|
return StructuredStay(
|
|
stay_id="stay_001",
|
|
documents=[sample_document],
|
|
sections=[],
|
|
facts=[sample_fact],
|
|
)
|
|
|
|
|
|
class TestQuestionGeneratorBasic:
|
|
"""Tests de base pour le QuestionGenerator."""
|
|
|
|
def test_initialization(self):
|
|
"""Test l'initialisation du générateur."""
|
|
generator = QuestionGenerator()
|
|
assert generator.MAX_QUESTIONS == 5
|
|
|
|
def test_generate_questions_returns_list(
|
|
self, question_generator, sample_proposal, sample_stay
|
|
):
|
|
"""Test que generate_questions retourne une liste."""
|
|
questions = question_generator.generate_questions(
|
|
sample_proposal, sample_stay, []
|
|
)
|
|
assert isinstance(questions, list)
|
|
|
|
def test_generate_questions_respects_max_limit(
|
|
self, question_generator, sample_proposal, sample_stay
|
|
):
|
|
"""Test que le nombre de questions ne dépasse pas MAX_QUESTIONS."""
|
|
# Créer beaucoup de problèmes de validation
|
|
many_issues = [
|
|
ValidationIssue(
|
|
issue_id=f"i{i}",
|
|
severity="bloquant",
|
|
category="missing_info",
|
|
message=f"Problème {i}",
|
|
affected_codes=[],
|
|
suggested_action=f"Action {i}",
|
|
)
|
|
for i in range(10)
|
|
]
|
|
|
|
questions = question_generator.generate_questions(
|
|
sample_proposal, sample_stay, many_issues
|
|
)
|
|
|
|
assert len(questions) <= QuestionGenerator.MAX_QUESTIONS
|
|
|
|
|
|
class TestQuestionGeneration:
|
|
"""Tests pour la génération de questions."""
|
|
|
|
def test_generate_from_blocking_issues(
|
|
self, question_generator, sample_proposal, sample_stay
|
|
):
|
|
"""Test la génération de questions depuis des problèmes bloquants."""
|
|
blocking_issue = ValidationIssue(
|
|
issue_id="i1",
|
|
severity="bloquant",
|
|
category="missing_info",
|
|
message="DP manquant",
|
|
affected_codes=[],
|
|
suggested_action="Ajouter un DP",
|
|
)
|
|
|
|
questions = question_generator.generate_questions(
|
|
sample_proposal, sample_stay, [blocking_issue]
|
|
)
|
|
|
|
# Vérifier qu'une question est générée
|
|
assert len(questions) > 0
|
|
# Vérifier que la question a une priorité haute (1)
|
|
assert any(q.priority == 1 for q in questions)
|
|
|
|
def test_generate_from_review_issues(
|
|
self, question_generator, sample_proposal, sample_stay
|
|
):
|
|
"""Test la génération de questions depuis des problèmes à revoir."""
|
|
review_issue = ValidationIssue(
|
|
issue_id="i1",
|
|
severity="a_revoir",
|
|
category="contradiction",
|
|
message="Incohérence détectée",
|
|
affected_codes=["K29.1"],
|
|
suggested_action="Vérifier le code",
|
|
)
|
|
|
|
questions = question_generator.generate_questions(
|
|
sample_proposal, sample_stay, [review_issue]
|
|
)
|
|
|
|
# Vérifier qu'une question est générée
|
|
assert len(questions) > 0
|
|
# Vérifier que la question a une priorité moyenne (2)
|
|
assert any(q.priority == 2 for q in questions)
|
|
|
|
def test_no_questions_from_info_issues(
|
|
self, question_generator, sample_proposal, sample_stay
|
|
):
|
|
"""Test qu'aucune question n'est générée depuis des problèmes info."""
|
|
info_issue = ValidationIssue(
|
|
issue_id="i1",
|
|
severity="info",
|
|
category="other",
|
|
message="Information",
|
|
affected_codes=[],
|
|
suggested_action="Aucune",
|
|
)
|
|
|
|
questions = question_generator.generate_questions(
|
|
sample_proposal, sample_stay, [info_issue]
|
|
)
|
|
|
|
# Les questions info ne génèrent pas de questions
|
|
# (mais d'autres sources peuvent en générer)
|
|
# Donc on vérifie juste que ça ne plante pas
|
|
assert isinstance(questions, list)
|
|
|
|
def test_generate_for_suspected_facts(
|
|
self, question_generator, sample_proposal, sample_document
|
|
):
|
|
"""Test la génération de questions pour faits suspectés."""
|
|
suspected_fact = ClinicalFact(
|
|
fact_id="fact_001",
|
|
type="diagnostic",
|
|
text="gastrite",
|
|
qualifier=Qualifier(
|
|
certainty="suspecté",
|
|
markers=["possible"],
|
|
confidence=0.7,
|
|
),
|
|
temporality="actuel",
|
|
evidence=Evidence(
|
|
document_id="doc_001",
|
|
span=Span(start=20, end=35),
|
|
text="possible gastrite",
|
|
context="Patient présente une possible gastrite",
|
|
),
|
|
confidence=0.7,
|
|
)
|
|
|
|
stay = StructuredStay(
|
|
stay_id="stay_001",
|
|
documents=[sample_document],
|
|
sections=[],
|
|
facts=[suspected_fact],
|
|
)
|
|
|
|
questions = question_generator.generate_questions(
|
|
sample_proposal, stay, []
|
|
)
|
|
|
|
# Vérifier qu'une question est générée pour le fait suspecté
|
|
suspected_questions = [
|
|
q for q in questions
|
|
if "suspecté" in q.text.lower() or "confirmé" in q.text.lower()
|
|
]
|
|
assert len(suspected_questions) > 0
|
|
assert suspected_questions[0].category == "clarification"
|
|
|
|
def test_generate_for_low_confidence_codes(
|
|
self, question_generator, sample_stay
|
|
):
|
|
"""Test la génération de questions pour codes à faible confiance."""
|
|
low_confidence_code = Code(
|
|
code="K29.1",
|
|
label="Gastrite",
|
|
type="dp",
|
|
evidence=[sample_stay.facts[0].evidence],
|
|
confidence=0.5, # Confiance faible
|
|
reasoning="Diagnostic incertain",
|
|
referentiel_version="2026",
|
|
)
|
|
|
|
proposal = CodingProposal(
|
|
stay_id="stay_001",
|
|
dp=low_confidence_code,
|
|
dr=None,
|
|
das=[],
|
|
ccam=[],
|
|
reasoning="Test",
|
|
model_version=ModelVersion(
|
|
model_name="test", model_tag="v1", model_digest="a" * 64
|
|
),
|
|
prompt_version="v1",
|
|
)
|
|
|
|
questions = question_generator.generate_questions(
|
|
proposal, sample_stay, []
|
|
)
|
|
|
|
# Vérifier qu'une question est générée pour la faible confiance
|
|
confidence_questions = [
|
|
q for q in questions
|
|
if "confiance" in q.text.lower() or "confirmer" in q.text.lower()
|
|
]
|
|
assert len(confidence_questions) > 0
|
|
|
|
|
|
class TestInconsistencyDetection:
|
|
"""Tests pour la détection d'incohérences."""
|
|
|
|
def test_detect_negated_fact_with_code(
|
|
self, question_generator, sample_document
|
|
):
|
|
"""Test la détection d'un fait nié avec code proposé."""
|
|
negated_fact = ClinicalFact(
|
|
fact_id="fact_001",
|
|
type="diagnostic",
|
|
text="gastrite",
|
|
qualifier=Qualifier(
|
|
certainty="nié",
|
|
markers=["pas de"],
|
|
confidence=0.9,
|
|
),
|
|
temporality="actuel",
|
|
evidence=Evidence(
|
|
document_id="doc_001",
|
|
span=Span(start=20, end=35),
|
|
text="pas de gastrite",
|
|
context="Patient ne présente pas de gastrite",
|
|
),
|
|
confidence=0.9,
|
|
)
|
|
|
|
stay = StructuredStay(
|
|
stay_id="stay_001",
|
|
documents=[sample_document],
|
|
sections=[],
|
|
facts=[negated_fact],
|
|
)
|
|
|
|
# Code pour le diagnostic nié
|
|
code = Code(
|
|
code="K29.1",
|
|
label="Gastrite",
|
|
type="dp",
|
|
evidence=[negated_fact.evidence],
|
|
confidence=0.8,
|
|
reasoning="Test",
|
|
referentiel_version="2026",
|
|
)
|
|
|
|
proposal = CodingProposal(
|
|
stay_id="stay_001",
|
|
dp=code,
|
|
dr=None,
|
|
das=[],
|
|
ccam=[],
|
|
reasoning="Test",
|
|
model_version=ModelVersion(
|
|
model_name="test", model_tag="v1", model_digest="a" * 64
|
|
),
|
|
prompt_version="v1",
|
|
)
|
|
|
|
questions = question_generator.generate_questions(
|
|
proposal, stay, []
|
|
)
|
|
|
|
# Vérifier qu'une question d'incohérence est générée
|
|
inconsistency_questions = [
|
|
q for q in questions
|
|
if q.category == "contradiction" and "nié" in q.text.lower()
|
|
]
|
|
assert len(inconsistency_questions) > 0
|
|
# Vérifier que la priorité est haute (1)
|
|
assert inconsistency_questions[0].priority == 1
|
|
|
|
def test_detect_document_contradictions(
|
|
self, question_generator, sample_proposal
|
|
):
|
|
"""Test la détection de contradictions entre documents."""
|
|
# Créer deux faits contradictoires dans différents documents
|
|
fact1 = ClinicalFact(
|
|
fact_id="fact_001",
|
|
type="diagnostic",
|
|
text="gastrite",
|
|
qualifier=Qualifier(
|
|
certainty="affirmé",
|
|
markers=[],
|
|
confidence=0.9,
|
|
),
|
|
temporality="actuel",
|
|
evidence=Evidence(
|
|
document_id="doc_001",
|
|
span=Span(start=20, end=35),
|
|
text="gastrite confirmée",
|
|
context="Patient présente une gastrite confirmée",
|
|
),
|
|
confidence=0.9,
|
|
)
|
|
|
|
fact2 = ClinicalFact(
|
|
fact_id="fact_002",
|
|
type="diagnostic",
|
|
text="gastrite",
|
|
qualifier=Qualifier(
|
|
certainty="nié",
|
|
markers=["pas de"],
|
|
confidence=0.9,
|
|
),
|
|
temporality="actuel",
|
|
evidence=Evidence(
|
|
document_id="doc_002",
|
|
span=Span(start=10, end=25),
|
|
text="pas de gastrite",
|
|
context="Examen ne montre pas de gastrite",
|
|
),
|
|
confidence=0.9,
|
|
)
|
|
|
|
doc1 = ClinicalDocument(
|
|
document_id="doc_001",
|
|
document_type="cr_medical",
|
|
content="Test",
|
|
creation_date=datetime(2024, 1, 15),
|
|
priority=2,
|
|
)
|
|
|
|
doc2 = ClinicalDocument(
|
|
document_id="doc_002",
|
|
document_type="imagerie",
|
|
content="Test",
|
|
creation_date=datetime(2024, 1, 16),
|
|
priority=3,
|
|
)
|
|
|
|
stay = StructuredStay(
|
|
stay_id="stay_001",
|
|
documents=[doc1, doc2],
|
|
sections=[],
|
|
facts=[fact1, fact2],
|
|
)
|
|
|
|
questions = question_generator.generate_questions(
|
|
sample_proposal, stay, []
|
|
)
|
|
|
|
# Vérifier qu'une question de contradiction est générée
|
|
contradiction_questions = [
|
|
q for q in questions
|
|
if q.category == "contradiction" and "contradiction" in q.text.lower()
|
|
]
|
|
assert len(contradiction_questions) > 0
|
|
|
|
|
|
class TestQuestionPrioritization:
|
|
"""Tests pour la priorisation des questions."""
|
|
|
|
def test_questions_sorted_by_priority(self, question_generator):
|
|
"""Test que les questions sont triées par priorité."""
|
|
# Créer des questions avec différentes priorités
|
|
from pipeline_mco_pmsi.models.validation import Question
|
|
|
|
questions = [
|
|
Question(
|
|
question_id="q1",
|
|
text="Question priorité 3",
|
|
priority=3,
|
|
category="confirmation",
|
|
context="Test",
|
|
suggested_answers=[],
|
|
),
|
|
Question(
|
|
question_id="q2",
|
|
text="Question priorité 1",
|
|
priority=1,
|
|
category="contradiction",
|
|
context="Test",
|
|
suggested_answers=[],
|
|
),
|
|
Question(
|
|
question_id="q3",
|
|
text="Question priorité 2",
|
|
priority=2,
|
|
category="clarification",
|
|
context="Test",
|
|
suggested_answers=[],
|
|
),
|
|
]
|
|
|
|
# Utiliser la méthode de priorisation
|
|
sorted_questions = question_generator._prioritize_and_limit(questions)
|
|
|
|
# Vérifier que les questions sont triées par priorité
|
|
priorities = [q.priority for q in sorted_questions]
|
|
assert priorities == sorted(priorities)
|
|
assert sorted_questions[0].priority == 1
|
|
|
|
def test_contradiction_category_prioritized(self, question_generator):
|
|
"""Test que les contradictions sont priorisées."""
|
|
from pipeline_mco_pmsi.models.validation import Question
|
|
|
|
questions = [
|
|
Question(
|
|
question_id="q1",
|
|
text="Question confirmation",
|
|
priority=2,
|
|
category="confirmation",
|
|
context="Test",
|
|
suggested_answers=[],
|
|
),
|
|
Question(
|
|
question_id="q2",
|
|
text="Question contradiction",
|
|
priority=2,
|
|
category="contradiction",
|
|
context="Test",
|
|
suggested_answers=[],
|
|
),
|
|
]
|
|
|
|
sorted_questions = question_generator._prioritize_and_limit(questions)
|
|
|
|
# Avec la même priorité, la contradiction doit venir en premier
|
|
assert sorted_questions[0].category == "contradiction"
|
|
|
|
def test_max_questions_limit_enforced(self, question_generator):
|
|
"""Test que la limite MAX_QUESTIONS est respectée."""
|
|
from pipeline_mco_pmsi.models.validation import Question
|
|
|
|
# Créer plus de MAX_QUESTIONS questions
|
|
many_questions = [
|
|
Question(
|
|
question_id=f"q{i}",
|
|
text=f"Question {i}",
|
|
priority=i % 5 + 1,
|
|
category="confirmation",
|
|
context="Test",
|
|
suggested_answers=[],
|
|
)
|
|
for i in range(10)
|
|
]
|
|
|
|
limited_questions = question_generator._prioritize_and_limit(many_questions)
|
|
|
|
# Vérifier que seulement MAX_QUESTIONS sont retournées
|
|
assert len(limited_questions) == QuestionGenerator.MAX_QUESTIONS
|
|
# Vérifier que ce sont les plus prioritaires
|
|
assert all(q.priority <= 3 for q in limited_questions)
|