"""Tests pour le module de fusion multi-PDFs.""" import pytest from src.config import ( ActeCCAM, Diagnostic, DossierMedical, Sejour, Traitement, BiologieCle, Imagerie, ) from src.medical.fusion import ( merge_dossiers, _cim10_specificity, _prefer_most_specific_dp, _merge_sejour, _dedup_diagnostics, _dedup_actes, ) class TestCIM10Specificity: def test_none(self): assert _cim10_specificity(None) == 0 def test_short_code(self): assert _cim10_specificity("I10") == 3 def test_long_code(self): assert _cim10_specificity("K85.1") == 4 def test_longer_code(self): assert _cim10_specificity("K80.50") == 5 class TestSpecificityLongerCodeWins: def test_specificity_longer_code_wins(self): d1 = DossierMedical( diagnostic_principal=Diagnostic(texte="Calcul biliaire", cim10_suggestion="K80"), ) d2 = DossierMedical( diagnostic_principal=Diagnostic(texte="Calcul cholédoque", cim10_suggestion="K80.5"), ) dp = _prefer_most_specific_dp([d1, d2]) assert dp is not None assert dp.cim10_suggestion == "K80.5" class TestMergeSejourTrackarePriority: def test_merge_sejour_trackare_priority(self): d1 = DossierMedical( document_type="trackare", sejour=Sejour(sexe="F", age=43, date_entree="25/02/2023"), ) d2 = DossierMedical( document_type="crh", sejour=Sejour(sexe="M", age=45, date_entree="24/02/2023", mode_sortie="domicile"), ) merged = _merge_sejour([d1, d2]) assert merged.sexe == "F" # Trackare prioritaire assert merged.age == 43 assert merged.date_entree == "25/02/2023" assert merged.mode_sortie == "domicile" # Complété depuis CRH def test_merge_sejour_fills_missing(self): d1 = DossierMedical( document_type="trackare", sejour=Sejour(sexe="F"), ) d2 = DossierMedical( document_type="crh", sejour=Sejour(age=50, poids=75.0), ) merged = _merge_sejour([d1, d2]) assert merged.sexe == "F" assert merged.age == 50 assert merged.poids == 75.0 class TestDedupDiagnostics: def test_dedup_diagnostics_by_code(self): das = [ Diagnostic(texte="HTA", cim10_suggestion="I10", cim10_confidence="medium"), Diagnostic(texte="Hypertension", cim10_suggestion="I10", cim10_confidence="high"), ] result = _dedup_diagnostics(das) assert len(result) == 1 assert result[0].cim10_confidence == "high" def test_dedup_keeps_distinct_codes(self): das = [ Diagnostic(texte="HTA", cim10_suggestion="I10"), Diagnostic(texte="Diabète", cim10_suggestion="E11.9"), ] result = _dedup_diagnostics(das) assert len(result) == 2 class TestDedupActes: def test_dedup_actes_by_code(self): actes = [ ActeCCAM(texte="Cholé", code_ccam_suggestion="HMFC004"), ActeCCAM(texte="Cholécystectomie", code_ccam_suggestion="HMFC004", date="01/03"), ] result = _dedup_actes(actes) assert len(result) == 1 assert result[0].date == "01/03" # Celui avec la date est préféré class TestSingleDossierPassthrough: def test_single_dossier_passthrough(self): d = DossierMedical( source_file="test.pdf", document_type="crh", diagnostic_principal=Diagnostic(texte="HTA", cim10_suggestion="I10"), ) result = merge_dossiers([d]) assert result.diagnostic_principal.cim10_suggestion == "I10" assert result.source_files == ["test.pdf"] class TestDpNonRetainedBecomesDas: def test_dp_non_retained_becomes_das(self): d1 = DossierMedical( diagnostic_principal=Diagnostic(texte="HTA", cim10_suggestion="I10"), ) d2 = DossierMedical( diagnostic_principal=Diagnostic(texte="Calcul cholédoque", cim10_suggestion="K80.5"), ) result = merge_dossiers([d1, d2]) # K80.5 est plus spécifique, donc DP assert result.diagnostic_principal.cim10_suggestion == "K80.5" # I10 (ancien DP de d1) doit être dans les DAS das_codes = {d.cim10_suggestion for d in result.diagnostics_associes} assert "I10" in das_codes class TestFusionAlertAdded: def test_fusion_alert_added(self): d1 = DossierMedical(source_file="a.pdf", alertes_codage=["Alerte 1"]) d2 = DossierMedical(source_file="b.pdf", alertes_codage=["Alerte 2"]) result = merge_dossiers([d1, d2]) assert result.alertes_codage[0] == "FUSION: 2 documents fusionnés" assert "Alerte 1" in result.alertes_codage assert "Alerte 2" in result.alertes_codage class TestSourceFilesPopulated: def test_source_files_populated(self): d1 = DossierMedical(source_file="a.pdf") d2 = DossierMedical(source_file="b.pdf") result = merge_dossiers([d1, d2]) assert result.source_files == ["a.pdf", "b.pdf"] class TestDasEqualDpRemoved: """Vérifie que les DAS dont le code est identique au DP sont retirés après fusion.""" def test_das_same_code_as_dp_removed(self): d1 = DossierMedical( diagnostic_principal=Diagnostic(texte="HTA", cim10_suggestion="I10"), diagnostics_associes=[ Diagnostic(texte="Hypertension artérielle", cim10_suggestion="I10"), Diagnostic(texte="Diabète", cim10_suggestion="E11.9"), ], ) d2 = DossierMedical( diagnostic_principal=Diagnostic(texte="HTA essentielle", cim10_suggestion="I10"), ) result = merge_dossiers([d1, d2]) das_codes = [d.cim10_suggestion for d in result.diagnostics_associes] assert "I10" not in das_codes, "DAS=DP doit être retiré" assert "E11.9" in das_codes def test_das_different_code_kept(self): d1 = DossierMedical( diagnostic_principal=Diagnostic(texte="Cholécystite", cim10_suggestion="K81.0"), diagnostics_associes=[ Diagnostic(texte="HTA", cim10_suggestion="I10"), ], ) result = merge_dossiers([d1]) das_codes = [d.cim10_suggestion for d in result.diagnostics_associes] assert "I10" in das_codes class TestFullMergeCROTrackare: def test_full_merge_cro_trackare(self): """Cas réel : fusion Trackare + CRO.""" trackare = DossierMedical( source_file="trackare.pdf", document_type="trackare", sejour=Sejour(sexe="F", age=43, date_entree="25/02/2023", date_sortie="03/03/2023"), diagnostic_principal=Diagnostic( texte="Calcul des canaux biliaires", cim10_suggestion="K80.5", ), diagnostics_associes=[ Diagnostic(texte="HTA", cim10_suggestion="I10"), ], actes_ccam=[ ActeCCAM(texte="Cholécystectomie", code_ccam_suggestion="HMFC004", date="01/03"), ], traitements_sortie=[ Traitement(medicament="Paracétamol"), ], alertes_codage=["Alerte trackare"], ) cro = DossierMedical( source_file="cro.pdf", document_type="cro", sejour=Sejour(sexe="F"), diagnostic_principal=Diagnostic( texte="Pancréatite aiguë lithiasique", cim10_suggestion="K85.1", cim10_confidence="high", ), diagnostics_associes=[ Diagnostic(texte="Obésité", cim10_suggestion="E66.0"), Diagnostic(texte="HTA", cim10_suggestion="I10"), # doublon ], actes_ccam=[ ActeCCAM(texte="TDM", code_ccam_suggestion="ZCQK002"), ActeCCAM(texte="Cholécystectomie", code_ccam_suggestion="HMFC004"), # doublon ], traitements_sortie=[ Traitement(medicament="Paracétamol"), # doublon Traitement(medicament="Cétirizine"), ], alertes_codage=["Alerte CRO"], ) result = merge_dossiers([trackare, cro]) # DP : K85.1 est plus spécifique que K80.5 assert result.diagnostic_principal.cim10_suggestion == "K85.1" # K80.5 (ancien DP trackare) doit être dans les DAS das_codes = {d.cim10_suggestion for d in result.diagnostics_associes} assert "K80.5" in das_codes assert "I10" in das_codes assert "E66.0" in das_codes # DAS dédupliqués : I10 ne doit pas être en double i10_count = sum(1 for d in result.diagnostics_associes if d.cim10_suggestion == "I10") assert i10_count == 1 # Actes dédupliqués acte_codes = [a.code_ccam_suggestion for a in result.actes_ccam] assert acte_codes.count("HMFC004") == 1 assert "ZCQK002" in acte_codes # Traitements dédupliqués meds = [t.medicament for t in result.traitements_sortie] assert meds.count("Paracétamol") == 1 assert "Cétirizine" in meds # Source files assert result.source_files == ["trackare.pdf", "cro.pdf"] # Alertes assert result.alertes_codage[0].startswith("FUSION:") assert "Alerte trackare" in result.alertes_codage assert "Alerte CRO" in result.alertes_codage # Type prioritaire : trackare assert result.document_type == "trackare"