789 lines
24 KiB
Python
789 lines
24 KiB
Python
"""
|
|
Tests pour le Pipeline principal.
|
|
|
|
Ce module teste l'orchestration complète du pipeline de codage MCO PMSI.
|
|
"""
|
|
|
|
import pytest
|
|
from datetime import datetime, timedelta
|
|
|
|
from pipeline_mco_pmsi.models.clinical import ClinicalDocument
|
|
from pipeline_mco_pmsi.models.metadata import StayMetadata
|
|
from pipeline_mco_pmsi.pipeline import Pipeline, PipelineResult
|
|
from pipeline_mco_pmsi.database.models import StayDB
|
|
|
|
|
|
def create_stay_in_db(db_session, stay_metadata: StayMetadata) -> None:
|
|
"""Helper pour créer un Stay dans la base de données."""
|
|
stay_db = StayDB(
|
|
stay_id=stay_metadata.stay_id,
|
|
admission_date=stay_metadata.admission_date,
|
|
discharge_date=stay_metadata.discharge_date,
|
|
specialty=stay_metadata.specialty,
|
|
unit=stay_metadata.unit,
|
|
age=stay_metadata.age,
|
|
sex=stay_metadata.sex,
|
|
)
|
|
db_session.add(stay_db)
|
|
db_session.commit()
|
|
|
|
|
|
def test_pipeline_initialization(db_session, rag_engine):
|
|
"""Test l'initialisation du pipeline avec tous ses composants."""
|
|
pipeline = Pipeline(
|
|
db_session=db_session,
|
|
rag_engine=rag_engine,
|
|
model_name="mock-llm",
|
|
model_version="1.0.0",
|
|
codeur_prompt_version="codeur-1.0.0",
|
|
verificateur_prompt_version="verificateur-1.0.0",
|
|
groupage_version="2026",
|
|
rules_version="1.0.0",
|
|
conservative_mode=True,
|
|
)
|
|
|
|
assert pipeline is not None
|
|
assert pipeline.document_processor is not None
|
|
assert pipeline.pii_protector is not None
|
|
assert pipeline.clinical_facts_extractor is not None
|
|
assert pipeline.codeur is not None
|
|
assert pipeline.verificateur is not None
|
|
assert pipeline.groupage_validator is not None
|
|
assert pipeline.pmsi_validator is not None
|
|
assert pipeline.question_generator is not None
|
|
assert pipeline.audit_logger is not None
|
|
|
|
|
|
def test_pipeline_process_stay_simple(db_session, rag_engine, sample_stay):
|
|
"""Test le traitement complet d'un séjour simple."""
|
|
pipeline = Pipeline(
|
|
db_session=db_session,
|
|
rag_engine=rag_engine,
|
|
)
|
|
|
|
# Créer un document clinique simple
|
|
documents = [
|
|
ClinicalDocument(
|
|
document_id="doc_001",
|
|
document_type="cr_medical",
|
|
content="""
|
|
Compte Rendu Médical
|
|
|
|
Diagnostic: Gastrite aiguë
|
|
|
|
Le patient présente une gastrite aiguë confirmée par endoscopie.
|
|
Traitement par IPP prescrit.
|
|
""",
|
|
creation_date=datetime.now(),
|
|
author="Dr. Martin",
|
|
priority=2,
|
|
)
|
|
]
|
|
|
|
# Métadonnées du séjour
|
|
stay_metadata = StayMetadata(
|
|
stay_id="stay_001",
|
|
admission_date=datetime.now() - timedelta(days=2),
|
|
discharge_date=datetime.now(),
|
|
specialty="gastro-enterologie",
|
|
unit="medecine",
|
|
age=45,
|
|
sex="M",
|
|
)
|
|
|
|
# Créer le Stay dans la base de données
|
|
create_stay_in_db(db_session, stay_metadata)
|
|
|
|
# Traiter le séjour
|
|
result = pipeline.process_stay(documents, stay_metadata)
|
|
|
|
# Vérifications
|
|
assert result is not None
|
|
assert result.success is True
|
|
assert result.stay_id == "stay_001"
|
|
assert result.structured_stay is not None
|
|
assert len(result.structured_stay.documents) == 1
|
|
assert len(result.structured_stay.sections) > 0
|
|
assert len(result.structured_stay.facts) > 0
|
|
assert result.coding_proposal is not None
|
|
assert result.verification_result is not None
|
|
assert result.groupage_result is not None
|
|
assert result.versions is not None
|
|
|
|
|
|
def test_pipeline_process_stay_with_negation(db_session, rag_engine):
|
|
"""Test le traitement d'un séjour avec diagnostic nié."""
|
|
pipeline = Pipeline(
|
|
db_session=db_session,
|
|
rag_engine=rag_engine,
|
|
conservative_mode=True,
|
|
)
|
|
|
|
# Document avec diagnostic nié
|
|
documents = [
|
|
ClinicalDocument(
|
|
document_id="doc_002",
|
|
document_type="cr_medical",
|
|
content="""
|
|
Compte Rendu Médical
|
|
|
|
Diagnostic: Pas de gastrite
|
|
|
|
Le patient ne présente pas de gastrite à l'endoscopie.
|
|
Absence de lésion muqueuse.
|
|
""",
|
|
creation_date=datetime.now(),
|
|
author="Dr. Dupont",
|
|
priority=2,
|
|
)
|
|
]
|
|
|
|
stay_metadata = StayMetadata(
|
|
stay_id="stay_002",
|
|
admission_date=datetime.now() - timedelta(days=1),
|
|
discharge_date=datetime.now(),
|
|
specialty="gastro-enterologie",
|
|
unit="medecine",
|
|
age=50,
|
|
sex="F",
|
|
)
|
|
|
|
# Créer le Stay dans la base de données
|
|
create_stay_in_db(db_session, stay_metadata)
|
|
|
|
# Traiter le séjour
|
|
result = pipeline.process_stay(documents, stay_metadata)
|
|
|
|
# Vérifications
|
|
assert result.success is True
|
|
|
|
# En mode conservateur, les diagnostics niés ne doivent pas être codés
|
|
if result.coding_proposal and result.coding_proposal.dp:
|
|
# Si un DP est proposé, il ne doit pas être basé sur le fait nié
|
|
for evidence in result.coding_proposal.dp.evidence:
|
|
assert "pas de gastrite" not in evidence.text.lower()
|
|
|
|
|
|
def test_pipeline_process_stay_multi_documents(db_session, rag_engine):
|
|
"""Test le traitement d'un séjour avec plusieurs documents."""
|
|
pipeline = Pipeline(
|
|
db_session=db_session,
|
|
rag_engine=rag_engine,
|
|
)
|
|
|
|
# Plusieurs documents avec priorités différentes
|
|
documents = [
|
|
ClinicalDocument(
|
|
document_id="doc_003",
|
|
document_type="courrier",
|
|
content="Courrier de sortie: Patient traité pour gastrite.",
|
|
creation_date=datetime.now(),
|
|
author="Dr. Martin",
|
|
priority=5, # Basse priorité
|
|
),
|
|
ClinicalDocument(
|
|
document_id="doc_004",
|
|
document_type="cr_medical",
|
|
content="""
|
|
Compte Rendu Médical
|
|
|
|
Diagnostic: Gastrite aiguë hémorragique
|
|
|
|
Le patient présente une gastrite aiguë hémorragique.
|
|
Endoscopie réalisée.
|
|
""",
|
|
creation_date=datetime.now(),
|
|
author="Dr. Dupont",
|
|
priority=2, # Haute priorité
|
|
),
|
|
]
|
|
|
|
stay_metadata = StayMetadata(
|
|
stay_id="stay_003",
|
|
admission_date=datetime.now() - timedelta(days=3),
|
|
discharge_date=datetime.now(),
|
|
specialty="gastro-enterologie",
|
|
unit="medecine",
|
|
age=55,
|
|
sex="M",
|
|
)
|
|
|
|
# Créer le Stay dans la base de données
|
|
create_stay_in_db(db_session, stay_metadata)
|
|
|
|
# Traiter le séjour
|
|
result = pipeline.process_stay(documents, stay_metadata)
|
|
|
|
# Vérifications
|
|
assert result.success is True
|
|
assert len(result.structured_stay.documents) == 2
|
|
|
|
# Vérifier que les documents sont triés par priorité
|
|
assert result.structured_stay.documents[0].priority <= result.structured_stay.documents[1].priority
|
|
|
|
|
|
def test_pipeline_result_can_auto_validate(db_session, rag_engine):
|
|
"""Test la détermination de la possibilité de validation automatique."""
|
|
pipeline = Pipeline(
|
|
db_session=db_session,
|
|
rag_engine=rag_engine,
|
|
)
|
|
|
|
documents = [
|
|
ClinicalDocument(
|
|
document_id="doc_005",
|
|
document_type="cr_medical",
|
|
content="Diagnostic: Gastrite aiguë confirmée.",
|
|
creation_date=datetime.now(),
|
|
author="Dr. Martin",
|
|
priority=2,
|
|
)
|
|
]
|
|
|
|
stay_metadata = StayMetadata(
|
|
stay_id="stay_004",
|
|
admission_date=datetime.now() - timedelta(days=1),
|
|
discharge_date=datetime.now(),
|
|
specialty="gastro-enterologie",
|
|
unit="medecine",
|
|
age=40,
|
|
sex="F",
|
|
)
|
|
|
|
result = pipeline.process_stay(documents, stay_metadata)
|
|
|
|
# Vérifier la logique de validation automatique
|
|
if result.success and result.verification_result:
|
|
if result.verification_result.decision == "accept" and not result.has_blocking_issues():
|
|
assert result.can_auto_validate() is True
|
|
else:
|
|
assert result.can_auto_validate() is False
|
|
|
|
|
|
def test_pipeline_export_audit_trail(db_session, rag_engine):
|
|
"""Test l'export de la piste d'audit."""
|
|
pipeline = Pipeline(
|
|
db_session=db_session,
|
|
rag_engine=rag_engine,
|
|
)
|
|
|
|
documents = [
|
|
ClinicalDocument(
|
|
document_id="doc_006",
|
|
document_type="cr_medical",
|
|
content="Diagnostic: Gastrite aiguë.",
|
|
creation_date=datetime.now(),
|
|
author="Dr. Dupont",
|
|
priority=2,
|
|
)
|
|
]
|
|
|
|
stay_metadata = StayMetadata(
|
|
stay_id="stay_005",
|
|
admission_date=datetime.now() - timedelta(days=1),
|
|
discharge_date=datetime.now(),
|
|
specialty="gastro-enterologie",
|
|
unit="medecine",
|
|
age=55,
|
|
sex="M",
|
|
)
|
|
|
|
# Créer le Stay dans la base de données
|
|
create_stay_in_db(db_session, stay_metadata)
|
|
|
|
# Traiter le séjour
|
|
result = pipeline.process_stay(documents, stay_metadata)
|
|
assert result.success is True
|
|
|
|
# Exporter l'audit
|
|
audit_dict = pipeline.export_audit_trail("stay_005", include_pii=False)
|
|
|
|
# Vérifications
|
|
assert audit_dict is not None
|
|
assert "stay_id" in audit_dict
|
|
assert audit_dict["stay_id"] == "stay_005"
|
|
assert "documents" in audit_dict
|
|
assert "facts" in audit_dict
|
|
assert "coding_proposal" in audit_dict
|
|
assert "verification_result" in audit_dict
|
|
assert "audit_records" in audit_dict
|
|
assert "versions" in audit_dict
|
|
|
|
|
|
def test_pipeline_error_handling(db_session, rag_engine):
|
|
"""Test la gestion des erreurs du pipeline."""
|
|
pipeline = Pipeline(
|
|
db_session=db_session,
|
|
rag_engine=rag_engine,
|
|
)
|
|
|
|
# Documents vides (devrait causer une erreur)
|
|
documents = []
|
|
|
|
stay_metadata = StayMetadata(
|
|
stay_id="stay_006",
|
|
admission_date=datetime.now() - timedelta(days=1),
|
|
discharge_date=datetime.now(),
|
|
specialty="gastro-enterologie",
|
|
unit="medecine",
|
|
age=45,
|
|
sex="M",
|
|
)
|
|
|
|
# Créer le Stay dans la base de données
|
|
create_stay_in_db(db_session, stay_metadata)
|
|
|
|
# Traiter le séjour (devrait échouer gracieusement)
|
|
result = pipeline.process_stay(documents, stay_metadata)
|
|
|
|
# Vérifications
|
|
assert result is not None
|
|
assert result.success is False
|
|
assert result.error_message is not None
|
|
# Note: avec des documents vides, le pipeline échoue avant de créer les propositions
|
|
# donc coding_proposal et verification_result peuvent être None
|
|
|
|
|
|
def test_pipeline_version_info(db_session, rag_engine):
|
|
"""Test la construction des informations de version."""
|
|
pipeline = Pipeline(
|
|
db_session=db_session,
|
|
rag_engine=rag_engine,
|
|
model_name="test-model",
|
|
model_version="2.0.0",
|
|
codeur_prompt_version="codeur-2.0.0",
|
|
verificateur_prompt_version="verificateur-2.0.0",
|
|
groupage_version="2026",
|
|
rules_version="2.0.0",
|
|
)
|
|
|
|
documents = [
|
|
ClinicalDocument(
|
|
document_id="doc_007",
|
|
document_type="cr_medical",
|
|
content="Diagnostic: Gastrite aiguë.",
|
|
creation_date=datetime.now(),
|
|
author="Dr. Martin",
|
|
priority=2,
|
|
)
|
|
]
|
|
|
|
stay_metadata = StayMetadata(
|
|
stay_id="stay_007",
|
|
admission_date=datetime.now() - timedelta(days=1),
|
|
discharge_date=datetime.now(),
|
|
specialty="gastro-enterologie",
|
|
unit="medecine",
|
|
age=50,
|
|
sex="F",
|
|
)
|
|
|
|
result = pipeline.process_stay(documents, stay_metadata)
|
|
|
|
# Vérifier les informations de version
|
|
assert result.versions is not None
|
|
assert result.versions.model_name == "test-model"
|
|
assert result.versions.model_tag == "2.0.0"
|
|
assert result.versions.model_digest is not None
|
|
assert "codeur=codeur-2.0.0" in result.versions.prompt_version
|
|
assert "verificateur=verificateur-2.0.0" in result.versions.prompt_version
|
|
assert result.versions.groupage_version == "2026"
|
|
assert result.versions.rules_version == "2.0.0"
|
|
assert result.versions.inference_params is not None
|
|
|
|
|
|
def test_pipeline_multi_document_contradictions(db_session, rag_engine):
|
|
"""
|
|
Test la détection de contradictions entre documents multiples.
|
|
|
|
Exigences: 15.4, 15.5
|
|
"""
|
|
from pipeline_mco_pmsi.database.models import StayDB
|
|
from pipeline_mco_pmsi.models.clinical import Qualifier, Span
|
|
from unittest.mock import patch
|
|
|
|
pipeline = Pipeline(
|
|
db_session=db_session,
|
|
rag_engine=rag_engine,
|
|
)
|
|
|
|
# Créer le séjour dans la base de données
|
|
stay = StayDB(
|
|
stay_id="stay_contradiction_001",
|
|
admission_date=datetime.now() - timedelta(days=1),
|
|
discharge_date=datetime.now(),
|
|
specialty="chirurgie",
|
|
unit="urgences",
|
|
age=35,
|
|
sex="M",
|
|
)
|
|
db_session.add(stay)
|
|
db_session.commit()
|
|
|
|
# Documents avec informations contradictoires
|
|
documents = [
|
|
ClinicalDocument(
|
|
document_id="doc_cro_001",
|
|
document_type="cr_operatoire",
|
|
content="""
|
|
Compte Rendu Opératoire
|
|
|
|
Diagnostic: Appendicite aiguë
|
|
|
|
Le patient présente une appendicite aiguë confirmée.
|
|
Appendicectomie réalisée.
|
|
""",
|
|
creation_date=datetime.now(),
|
|
author="Dr. Chirurgien",
|
|
priority=1, # Haute priorité (CRO)
|
|
),
|
|
ClinicalDocument(
|
|
document_id="doc_crm_001",
|
|
document_type="cr_medical",
|
|
content="""
|
|
Compte Rendu Médical
|
|
|
|
Diagnostic: Pas d'appendicite
|
|
|
|
Le patient ne présente pas d'appendicite à l'examen clinique.
|
|
Douleurs abdominales d'origine fonctionnelle.
|
|
""",
|
|
creation_date=datetime.now() - timedelta(hours=2),
|
|
author="Dr. Urgentiste",
|
|
priority=2, # Priorité moyenne (CRM)
|
|
),
|
|
]
|
|
|
|
stay_metadata = StayMetadata(
|
|
stay_id="stay_contradiction_001",
|
|
admission_date=datetime.now() - timedelta(days=1),
|
|
discharge_date=datetime.now(),
|
|
specialty="chirurgie",
|
|
unit="urgences",
|
|
age=35,
|
|
sex="M",
|
|
)
|
|
|
|
# Mock l'extraction de faits pour créer des faits contradictoires
|
|
def mock_extract_facts(structured_stay):
|
|
"""Crée des faits contradictoires pour le test."""
|
|
from pipeline_mco_pmsi.models.clinical import ClinicalFact, Evidence, Qualifier, Span
|
|
|
|
# Fait affirmé depuis le CRO
|
|
fact_affirmed = ClinicalFact(
|
|
fact_id="fact_001",
|
|
type="diagnostic",
|
|
text="appendicite aiguë",
|
|
qualifier=Qualifier(
|
|
certainty="affirmé",
|
|
markers=[],
|
|
confidence=0.95,
|
|
),
|
|
temporality="actuel",
|
|
evidence=Evidence(
|
|
document_id="doc_cro_001",
|
|
span=Span(start=50, end=67),
|
|
text="appendicite aiguë",
|
|
context="Le patient présente une appendicite aiguë confirmée",
|
|
),
|
|
confidence=0.95,
|
|
)
|
|
|
|
# Fait nié depuis le CRM - MÊME TEXTE pour être groupé
|
|
fact_negated = ClinicalFact(
|
|
fact_id="fact_002",
|
|
type="diagnostic",
|
|
text="appendicite aiguë", # Même texte que fact_affirmed
|
|
qualifier=Qualifier(
|
|
certainty="nié",
|
|
markers=["pas de"],
|
|
confidence=0.90,
|
|
),
|
|
temporality="actuel",
|
|
evidence=Evidence(
|
|
document_id="doc_crm_001",
|
|
span=Span(start=40, end=60),
|
|
text="pas d'appendicite",
|
|
context="Le patient ne présente pas d'appendicite à l'examen",
|
|
),
|
|
confidence=0.90,
|
|
)
|
|
|
|
return [fact_affirmed, fact_negated]
|
|
|
|
# Patcher l'extracteur de faits
|
|
with patch.object(pipeline.clinical_facts_extractor, 'extract_facts', side_effect=mock_extract_facts):
|
|
# Traiter le séjour
|
|
result = pipeline.process_stay(documents, stay_metadata)
|
|
|
|
# Vérifications
|
|
assert result.success is True
|
|
assert len(result.structured_stay.documents) == 2
|
|
|
|
# Vérifier qu'une contradiction a été détectée
|
|
contradiction_issues = [
|
|
issue for issue in result.validation_issues
|
|
if issue.category == "contradiction"
|
|
]
|
|
assert len(contradiction_issues) > 0
|
|
|
|
# Vérifier que la contradiction est marquée "a_revoir"
|
|
assert any(issue.severity == "a_revoir" for issue in contradiction_issues)
|
|
|
|
# Vérifier que l'action suggérée mentionne l'arbitrage TIM
|
|
assert any("arbitrage" in issue.suggested_action.lower() for issue in contradiction_issues)
|
|
|
|
|
|
def test_pipeline_multi_document_temporal_contradiction(db_session, rag_engine):
|
|
"""
|
|
Test la détection de contradictions temporelles entre documents.
|
|
|
|
Exigences: 15.4, 15.5
|
|
"""
|
|
from pipeline_mco_pmsi.database.models import StayDB
|
|
|
|
pipeline = Pipeline(
|
|
db_session=db_session,
|
|
rag_engine=rag_engine,
|
|
)
|
|
|
|
# Créer le séjour dans la base de données
|
|
stay = StayDB(
|
|
stay_id="stay_temporal_001",
|
|
admission_date=datetime.now() - timedelta(days=2),
|
|
discharge_date=datetime.now(),
|
|
specialty="endocrinologie",
|
|
unit="medecine",
|
|
age=55,
|
|
sex="M",
|
|
)
|
|
db_session.add(stay)
|
|
db_session.commit()
|
|
|
|
# Documents avec temporalité contradictoire
|
|
documents = [
|
|
ClinicalDocument(
|
|
document_id="doc_008",
|
|
document_type="cr_medical",
|
|
content="""
|
|
Compte Rendu Médical
|
|
|
|
Diagnostic: Diabète de type 2
|
|
|
|
Le patient présente un diabète de type 2 diagnostiqué lors de ce séjour.
|
|
Glycémie à jeun élevée.
|
|
""",
|
|
creation_date=datetime.now(),
|
|
author="Dr. Endocrinologue",
|
|
priority=2,
|
|
),
|
|
ClinicalDocument(
|
|
document_id="doc_009",
|
|
document_type="courrier",
|
|
content="""
|
|
Courrier de sortie
|
|
|
|
Antécédents: Diabète de type 2 connu depuis 5 ans
|
|
|
|
Le patient a un antécédent de diabète de type 2.
|
|
Traitement par metformine poursuivi.
|
|
""",
|
|
creation_date=datetime.now(),
|
|
author="Dr. Interne",
|
|
priority=5,
|
|
),
|
|
]
|
|
|
|
stay_metadata = StayMetadata(
|
|
stay_id="stay_temporal_001",
|
|
admission_date=datetime.now() - timedelta(days=2),
|
|
discharge_date=datetime.now(),
|
|
specialty="endocrinologie",
|
|
unit="medecine",
|
|
age=55,
|
|
sex="M",
|
|
)
|
|
|
|
# Traiter le séjour
|
|
result = pipeline.process_stay(documents, stay_metadata)
|
|
|
|
# Vérifications
|
|
assert result.success is True
|
|
|
|
# Vérifier qu'une contradiction temporelle a été détectée
|
|
temporal_contradictions = [
|
|
issue for issue in result.validation_issues
|
|
if issue.category == "contradiction" and "temporelle" in issue.message.lower()
|
|
]
|
|
|
|
# Note: La détection dépend de l'extraction correcte des qualificateurs temporels
|
|
# Si aucune contradiction n'est détectée, c'est que les faits n'ont pas été extraits
|
|
# avec des temporalités différentes
|
|
|
|
|
|
def test_pipeline_multi_document_source_traceability(db_session, rag_engine):
|
|
"""
|
|
Test la traçabilité des sources pour les faits multi-documents.
|
|
|
|
Exigences: 15.6
|
|
"""
|
|
from pipeline_mco_pmsi.database.models import StayDB
|
|
|
|
pipeline = Pipeline(
|
|
db_session=db_session,
|
|
rag_engine=rag_engine,
|
|
)
|
|
|
|
# Créer le séjour dans la base de données
|
|
stay = StayDB(
|
|
stay_id="stay_trace_001",
|
|
admission_date=datetime.now() - timedelta(days=1),
|
|
discharge_date=datetime.now(),
|
|
specialty="chirurgie",
|
|
unit="urgences",
|
|
age=60,
|
|
sex="F",
|
|
)
|
|
db_session.add(stay)
|
|
db_session.commit()
|
|
|
|
# Plusieurs documents
|
|
documents = [
|
|
ClinicalDocument(
|
|
document_id="doc_010",
|
|
document_type="cr_operatoire",
|
|
content="Diagnostic: Cholécystite aiguë. Cholécystectomie réalisée.",
|
|
creation_date=datetime.now(),
|
|
author="Dr. Chirurgien",
|
|
priority=1,
|
|
),
|
|
ClinicalDocument(
|
|
document_id="doc_011",
|
|
document_type="imagerie",
|
|
content="Échographie: Vésicule biliaire distendue avec calculs.",
|
|
creation_date=datetime.now() - timedelta(hours=3),
|
|
author="Dr. Radiologue",
|
|
priority=3,
|
|
),
|
|
ClinicalDocument(
|
|
document_id="doc_012",
|
|
document_type="biologie",
|
|
content="Biologie: Hyperleucocytose à 15000/mm3.",
|
|
creation_date=datetime.now() - timedelta(hours=4),
|
|
author="Laboratoire",
|
|
priority=4,
|
|
),
|
|
]
|
|
|
|
stay_metadata = StayMetadata(
|
|
stay_id="stay_trace_001",
|
|
admission_date=datetime.now() - timedelta(days=1),
|
|
discharge_date=datetime.now(),
|
|
specialty="chirurgie",
|
|
unit="urgences",
|
|
age=60,
|
|
sex="F",
|
|
)
|
|
|
|
# Traiter le séjour
|
|
result = pipeline.process_stay(documents, stay_metadata)
|
|
|
|
# Vérifications
|
|
assert result.success is True
|
|
assert len(result.structured_stay.documents) == 3
|
|
|
|
# Vérifier que tous les faits ont un document_id valide
|
|
valid_document_ids = {doc.document_id for doc in documents}
|
|
for fact in result.structured_stay.facts:
|
|
assert fact.evidence.document_id in valid_document_ids, \
|
|
f"Fait {fact.fact_id} a un document_id invalide: {fact.evidence.document_id}"
|
|
|
|
# Vérifier qu'aucun problème d'intégrité de données n'a été détecté
|
|
data_integrity_issues = [
|
|
issue for issue in result.validation_issues
|
|
if issue.category == "data_integrity"
|
|
]
|
|
assert len(data_integrity_issues) == 0
|
|
|
|
|
|
def test_pipeline_multi_document_priority_ordering(db_session, rag_engine):
|
|
"""
|
|
Test que les priorités de sources sont respectées.
|
|
|
|
Exigences: 15.2
|
|
"""
|
|
from pipeline_mco_pmsi.database.models import StayDB
|
|
|
|
pipeline = Pipeline(
|
|
db_session=db_session,
|
|
rag_engine=rag_engine,
|
|
)
|
|
|
|
# Créer le séjour dans la base de données
|
|
stay = StayDB(
|
|
stay_id="stay_priority_001",
|
|
admission_date=datetime.now() - timedelta(days=1),
|
|
discharge_date=datetime.now(),
|
|
specialty="gastro-enterologie",
|
|
unit="medecine",
|
|
age=50,
|
|
sex="M",
|
|
)
|
|
db_session.add(stay)
|
|
db_session.commit()
|
|
|
|
# Documents avec différentes priorités
|
|
documents = [
|
|
ClinicalDocument(
|
|
document_id="doc_courrier",
|
|
document_type="courrier",
|
|
content="Courrier: Gastrite.",
|
|
creation_date=datetime.now(),
|
|
author="Dr. A",
|
|
priority=5, # Basse priorité
|
|
),
|
|
ClinicalDocument(
|
|
document_id="doc_cro",
|
|
document_type="cr_operatoire",
|
|
content="CRO: Gastrite hémorragique.",
|
|
creation_date=datetime.now(),
|
|
author="Dr. B",
|
|
priority=1, # Haute priorité
|
|
),
|
|
ClinicalDocument(
|
|
document_id="doc_crm",
|
|
document_type="cr_medical",
|
|
content="CRM: Gastrite aiguë.",
|
|
creation_date=datetime.now(),
|
|
author="Dr. C",
|
|
priority=2, # Priorité moyenne
|
|
),
|
|
]
|
|
|
|
stay_metadata = StayMetadata(
|
|
stay_id="stay_priority_001",
|
|
admission_date=datetime.now() - timedelta(days=1),
|
|
discharge_date=datetime.now(),
|
|
specialty="gastro-enterologie",
|
|
unit="medecine",
|
|
age=50,
|
|
sex="M",
|
|
)
|
|
|
|
# Traiter le séjour
|
|
result = pipeline.process_stay(documents, stay_metadata)
|
|
|
|
# Vérifications
|
|
assert result.success is True
|
|
assert len(result.structured_stay.documents) == 3
|
|
|
|
# Vérifier que les documents sont triés par priorité (1 = haute, 5 = basse)
|
|
priorities = [doc.priority for doc in result.structured_stay.documents]
|
|
assert priorities == sorted(priorities), \
|
|
f"Documents non triés par priorité: {priorities}"
|
|
|
|
# Le premier document doit être le CRO (priorité 1)
|
|
assert result.structured_stay.documents[0].document_type == "cr_operatoire"
|
|
assert result.structured_stay.documents[0].priority == 1
|
|
|