919 lines
33 KiB
Python
919 lines
33 KiB
Python
"""
|
|
Tests unitaires pour le ReferentielsManager.
|
|
|
|
Ces tests vérifient l'import, le hashing et le chunking des référentiels ATIH.
|
|
"""
|
|
|
|
import hashlib
|
|
import tempfile
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from pipeline_mco_pmsi.rag import ReferentielsManager
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_data_dir():
|
|
"""Crée un répertoire temporaire pour les tests."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
yield Path(tmpdir)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_pdf():
|
|
"""Crée un PDF de test simple."""
|
|
from pypdf import PdfWriter
|
|
|
|
with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp:
|
|
writer = PdfWriter()
|
|
writer.add_blank_page(width=200, height=200)
|
|
writer.write(tmp)
|
|
tmp_path = Path(tmp.name)
|
|
|
|
yield tmp_path
|
|
|
|
# Cleanup
|
|
if tmp_path.exists():
|
|
tmp_path.unlink()
|
|
|
|
|
|
class TestReferentielsManagerInit:
|
|
"""Tests d'initialisation du ReferentielsManager."""
|
|
|
|
def test_init_creates_data_dir(self, temp_data_dir):
|
|
"""Test que l'initialisation crée le répertoire de données."""
|
|
data_dir = temp_data_dir / "referentiels"
|
|
manager = ReferentielsManager(data_dir=data_dir)
|
|
|
|
assert data_dir.exists()
|
|
assert manager.data_dir == data_dir
|
|
assert manager.embedding_model_name == "camembert-bio"
|
|
|
|
def test_init_with_custom_embedding_model(self, temp_data_dir):
|
|
"""Test l'initialisation avec un modèle d'embeddings personnalisé."""
|
|
manager = ReferentielsManager(
|
|
data_dir=temp_data_dir,
|
|
embedding_model="drbert"
|
|
)
|
|
|
|
assert manager.embedding_model_name == "drbert"
|
|
|
|
|
|
class TestImportReferentiel:
|
|
"""Tests d'import de référentiels."""
|
|
|
|
def test_import_referentiel_invalid_type(self, temp_data_dir, sample_pdf):
|
|
"""Test que l'import rejette un type de référentiel invalide."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
with pytest.raises(ValueError, match="Type de référentiel invalide"):
|
|
manager.import_referentiel(
|
|
file_path=str(sample_pdf),
|
|
referentiel_type="invalid_type",
|
|
version="2026"
|
|
)
|
|
|
|
def test_import_referentiel_file_not_found(self, temp_data_dir):
|
|
"""Test que l'import échoue si le fichier n'existe pas."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
with pytest.raises(FileNotFoundError):
|
|
manager.import_referentiel(
|
|
file_path="/nonexistent/file.pdf",
|
|
referentiel_type="cim10",
|
|
version="2026"
|
|
)
|
|
|
|
def test_import_referentiel_generates_hash(self, temp_data_dir, sample_pdf):
|
|
"""Test que l'import génère un hash SHA-256."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
# Calculer le hash attendu
|
|
with open(sample_pdf, "rb") as f:
|
|
expected_hash = hashlib.sha256(f.read()).hexdigest()
|
|
|
|
result = manager.import_referentiel(
|
|
file_path=str(sample_pdf),
|
|
referentiel_type="cim10",
|
|
version="2026"
|
|
)
|
|
|
|
assert result.file_hash == expected_hash
|
|
assert len(result.file_hash) == 64 # SHA-256 = 64 caractères hex
|
|
|
|
def test_import_referentiel_creates_version(self, temp_data_dir, sample_pdf):
|
|
"""Test que l'import crée une ReferentielVersion correcte."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
result = manager.import_referentiel(
|
|
file_path=str(sample_pdf),
|
|
referentiel_type="guide_mco",
|
|
version="2026"
|
|
)
|
|
|
|
assert result.type == "guide_mco"
|
|
assert result.version == "2026"
|
|
assert isinstance(result.import_date, datetime)
|
|
assert result.file_hash is not None
|
|
assert len(result.file_hash) == 64
|
|
|
|
def test_import_referentiel_saves_text(self, temp_data_dir, sample_pdf):
|
|
"""Test que l'import sauvegarde le texte extrait."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
manager.import_referentiel(
|
|
file_path=str(sample_pdf),
|
|
referentiel_type="ccam",
|
|
version="2025"
|
|
)
|
|
|
|
text_file = temp_data_dir / "ccam_2025_text.txt"
|
|
assert text_file.exists()
|
|
|
|
def test_import_referentiel_caches_version(self, temp_data_dir, sample_pdf):
|
|
"""Test que l'import met en cache la version."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
result = manager.import_referentiel(
|
|
file_path=str(sample_pdf),
|
|
referentiel_type="cim10",
|
|
version="2026"
|
|
)
|
|
|
|
# Vérifier que la version est dans le cache
|
|
cached_version = manager.get_version_info("cim10")
|
|
assert cached_version is not None
|
|
assert cached_version.type == "cim10"
|
|
assert cached_version.version == "2026"
|
|
assert cached_version.file_hash == result.file_hash
|
|
|
|
|
|
class TestGetVersionInfo:
|
|
"""Tests de récupération des informations de version."""
|
|
|
|
def test_get_version_info_not_found(self, temp_data_dir):
|
|
"""Test que get_version_info retourne None si non trouvé."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
result = manager.get_version_info("cim10")
|
|
assert result is None
|
|
|
|
def test_get_version_info_returns_cached(self, temp_data_dir, sample_pdf):
|
|
"""Test que get_version_info retourne la version en cache."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
imported = manager.import_referentiel(
|
|
file_path=str(sample_pdf),
|
|
referentiel_type="guide_mco",
|
|
version="2026"
|
|
)
|
|
|
|
result = manager.get_version_info("guide_mco")
|
|
assert result is not None
|
|
assert result.type == imported.type
|
|
assert result.version == imported.version
|
|
assert result.file_hash == imported.file_hash
|
|
|
|
|
|
class TestChunkReferentiel:
|
|
"""Tests de chunking des référentiels."""
|
|
|
|
def test_chunk_referentiel_guide_mco(self, temp_data_dir, sample_pdf):
|
|
"""Test le chunking du Guide MCO."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
# Import d'abord
|
|
ref_version = manager.import_referentiel(
|
|
file_path=str(sample_pdf),
|
|
referentiel_type="guide_mco",
|
|
version="2026"
|
|
)
|
|
|
|
# Chunking
|
|
chunks = manager.chunk_referentiel(ref_version)
|
|
|
|
assert len(chunks) > 0
|
|
for chunk in chunks:
|
|
assert chunk.referentiel_type == "guide_mco"
|
|
assert chunk.referentiel_version == "2026"
|
|
assert chunk.content is not None
|
|
assert chunk.chunk_id.startswith("guide_mco_2026_")
|
|
|
|
def test_chunk_referentiel_cim10(self, temp_data_dir, sample_pdf):
|
|
"""Test le chunking de la CIM-10."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
ref_version = manager.import_referentiel(
|
|
file_path=str(sample_pdf),
|
|
referentiel_type="cim10",
|
|
version="2026"
|
|
)
|
|
|
|
chunks = manager.chunk_referentiel(ref_version)
|
|
|
|
assert len(chunks) > 0
|
|
for chunk in chunks:
|
|
assert chunk.referentiel_type == "cim10"
|
|
assert chunk.chunk_id.startswith("cim10_2026_")
|
|
|
|
def test_chunk_referentiel_ccam(self, temp_data_dir, sample_pdf):
|
|
"""Test le chunking de la CCAM."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
ref_version = manager.import_referentiel(
|
|
file_path=str(sample_pdf),
|
|
referentiel_type="ccam",
|
|
version="2025"
|
|
)
|
|
|
|
chunks = manager.chunk_referentiel(ref_version)
|
|
|
|
assert len(chunks) > 0
|
|
for chunk in chunks:
|
|
assert chunk.referentiel_type == "ccam"
|
|
assert chunk.chunk_id.startswith("ccam_2025_")
|
|
|
|
def test_chunk_referentiel_file_not_found(self, temp_data_dir):
|
|
"""Test que le chunking échoue si le fichier texte n'existe pas."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
from pipeline_mco_pmsi.models.metadata import ReferentielVersion
|
|
|
|
# Créer une version sans avoir importé le fichier
|
|
fake_version = ReferentielVersion(
|
|
type="cim10",
|
|
version="2026",
|
|
import_date=datetime.now(),
|
|
file_hash="a" * 64,
|
|
chunk_count=0,
|
|
index_hash="0" * 64 # Placeholder hash
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="Fichier texte du référentiel introuvable"):
|
|
manager.chunk_referentiel(fake_version)
|
|
|
|
|
|
class TestChunkGuideMCO:
|
|
"""Tests spécifiques pour le chunking du Guide MCO."""
|
|
|
|
def test_chunk_guide_mco_preserves_rules(self, temp_data_dir):
|
|
"""Test que le chunking préserve les règles complètes."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
# Créer un texte de test avec des règles
|
|
test_text = """CHAPITRE 1 - Le recueil d'information
|
|
|
|
1.1 Règles générales
|
|
|
|
Règle 1: Le diagnostic principal (DP) est la pathologie ayant motivé l'admission.
|
|
|
|
Critère d'éligibilité:
|
|
- Le DP doit être documenté dans le dossier médical
|
|
- Le DP doit être codé selon la CIM-10
|
|
|
|
Exclusion: Les antécédents ne peuvent pas être DP.
|
|
|
|
1.2 Règles spécifiques
|
|
|
|
Règle 2: Les diagnostics associés significatifs (DAS) sont les comorbidités.
|
|
"""
|
|
|
|
# Sauvegarder le texte
|
|
text_file = temp_data_dir / "guide_mco_2026_text.txt"
|
|
with open(text_file, "w", encoding="utf-8") as f:
|
|
f.write(test_text)
|
|
|
|
chunks = manager.chunk_guide_mco(test_text, "2026")
|
|
|
|
# Vérifier qu'on a des chunks
|
|
assert len(chunks) > 0
|
|
|
|
# Vérifier que les métadonnées contiennent la section
|
|
for chunk in chunks:
|
|
assert "section" in chunk.metadata
|
|
assert chunk.metadata["chunk_type"] == "section"
|
|
|
|
# Vérifier qu'aucune règle n'est coupée au milieu
|
|
# (une règle commence par "Règle" et se termine avant la prochaine règle ou section)
|
|
full_content = "\n".join([c.content for c in chunks])
|
|
assert "Règle 1:" in full_content
|
|
assert "Règle 2:" in full_content
|
|
assert "Exclusion:" in full_content
|
|
|
|
def test_chunk_guide_mco_respects_size_limits(self, temp_data_dir):
|
|
"""Test que les chunks respectent les limites de taille."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
# Créer un texte long
|
|
test_text = "CHAPITRE 1\n\n" + ("Paragraphe de test. " * 500)
|
|
|
|
chunks = manager.chunk_guide_mco(test_text, "2026")
|
|
|
|
# Vérifier que les chunks ne dépassent pas la taille max
|
|
for chunk in chunks:
|
|
assert len(chunk.content) <= 4096 # max_chunk_size
|
|
|
|
def test_chunk_guide_mco_creates_overlap(self, temp_data_dir):
|
|
"""Test que le chunking crée un overlap entre chunks."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
# Créer un texte avec plusieurs sections
|
|
test_text = """CHAPITRE 1
|
|
|
|
Section 1.1
|
|
""" + ("Contenu de la section 1.1. " * 200) + """
|
|
|
|
Section 1.2
|
|
""" + ("Contenu de la section 1.2. " * 200)
|
|
|
|
chunks = manager.chunk_guide_mco(test_text, "2026")
|
|
|
|
# Si on a plusieurs chunks, vérifier qu'il y a un overlap
|
|
if len(chunks) > 1:
|
|
# Le dernier contenu du chunk N devrait apparaître au début du chunk N+1
|
|
for i in range(len(chunks) - 1):
|
|
chunk_n_end = chunks[i].content[-200:] # Derniers 200 caractères
|
|
chunk_n1_start = chunks[i + 1].content[:400] # Premiers 400 caractères
|
|
|
|
# Vérifier qu'il y a un overlap (au moins quelques mots en commun)
|
|
# On cherche des mots de plus de 5 caractères
|
|
words_n = [w for w in chunk_n_end.split() if len(w) > 5]
|
|
if words_n:
|
|
# Au moins un mot devrait être dans le chunk suivant
|
|
assert any(word in chunk_n1_start for word in words_n[:5])
|
|
|
|
|
|
class TestChunkCIM10:
|
|
"""Tests spécifiques pour le chunking de la CIM-10."""
|
|
|
|
def test_chunk_cim10_preserves_notes(self, temp_data_dir):
|
|
"""Test que le chunking préserve les notes d'inclusion/exclusion."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
# Créer un texte de test avec des codes et notes
|
|
test_text = """CHAPITRE I - Maladies infectieuses
|
|
|
|
A00-A09 Maladies intestinales infectieuses
|
|
|
|
A00 Choléra
|
|
A00.0 Choléra dû à Vibrio cholerae 01, biovar cholerae
|
|
A00.1 Choléra dû à Vibrio cholerae 01, biovar El Tor
|
|
A00.9 Choléra, sans précision
|
|
|
|
Inclus: infection à Vibrio cholerae
|
|
Exclut: intoxication alimentaire (A05.-)
|
|
|
|
A01 Fièvres typhoïde et paratyphoïde
|
|
A01.0 Fièvre typhoïde
|
|
|
|
Note: La fièvre typhoïde est causée par Salmonella typhi.
|
|
Comprend: infection à Salmonella typhi
|
|
"""
|
|
|
|
# Sauvegarder le texte
|
|
text_file = temp_data_dir / "cim10_2026_text.txt"
|
|
with open(text_file, "w", encoding="utf-8") as f:
|
|
f.write(test_text)
|
|
|
|
chunks = manager.chunk_cim10(test_text, "2026")
|
|
|
|
# Vérifier qu'on a des chunks
|
|
assert len(chunks) > 0
|
|
|
|
# Vérifier que les métadonnées contiennent le chapitre
|
|
for chunk in chunks:
|
|
assert "chapter" in chunk.metadata
|
|
assert chunk.metadata["chunk_type"] == "code_block"
|
|
|
|
# Vérifier que les notes ne sont pas coupées
|
|
full_content = "\n".join([c.content for c in chunks])
|
|
assert "Inclus:" in full_content
|
|
assert "Exclut:" in full_content
|
|
assert "Note:" in full_content
|
|
|
|
def test_chunk_cim10_respects_size_limits(self, temp_data_dir):
|
|
"""Test que les chunks respectent les limites de taille."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
# Créer un texte long
|
|
test_text = "CHAPITRE I\n\n" + ("A00 Code de test\n" * 300)
|
|
|
|
chunks = manager.chunk_cim10(test_text, "2026")
|
|
|
|
# Vérifier que les chunks ne dépassent pas la taille max
|
|
for chunk in chunks:
|
|
assert len(chunk.content) <= 4096 # max_chunk_size
|
|
|
|
def test_chunk_cim10_does_not_cut_note_blocks(self, temp_data_dir):
|
|
"""Test que les blocs de notes ne sont jamais coupés."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
# Créer un texte avec un long bloc de notes
|
|
test_text = """A00 Choléra
|
|
|
|
Inclus:
|
|
- infection à Vibrio cholerae
|
|
- choléra classique
|
|
- choléra El Tor
|
|
- choléra asiatique
|
|
- choléra épidémique
|
|
|
|
Exclut:
|
|
- intoxication alimentaire (A05.-)
|
|
- gastro-entérite non infectieuse (K52.-)
|
|
|
|
A01 Fièvre typhoïde
|
|
"""
|
|
|
|
chunks = manager.chunk_cim10(test_text, "2026")
|
|
|
|
# Vérifier que chaque chunk contient soit le bloc complet, soit rien du bloc
|
|
for chunk in chunks:
|
|
if "Inclus:" in chunk.content:
|
|
# Si le chunk contient "Inclus:", il doit contenir toute la liste
|
|
assert "- infection à Vibrio cholerae" in chunk.content
|
|
assert "- choléra épidémique" in chunk.content
|
|
|
|
|
|
class TestChunkCCAM:
|
|
"""Tests spécifiques pour le chunking de la CCAM."""
|
|
|
|
def test_chunk_ccam_preserves_extensions(self, temp_data_dir):
|
|
"""Test que le chunking préserve les extensions ATIH."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
# Créer un texte de test avec des codes CCAM et extensions
|
|
test_text = """CHAPITRE 1 - Actes diagnostiques
|
|
|
|
SECTION 1.1 - Imagerie
|
|
|
|
YYYY001 Radiographie du thorax
|
|
Description: Radiographie standard du thorax de face et de profil
|
|
|
|
Extensions ATIH:
|
|
+ABC Extension pour urgence
|
|
+DEF Extension pour patient hospitalisé
|
|
+GHI Extension pour acte itératif
|
|
|
|
Note technique: Cet acte nécessite une prescription médicale.
|
|
Condition d'application: Patient en position debout ou allongé.
|
|
|
|
YYYY002 Scanner thoracique
|
|
Description: Tomodensitométrie du thorax avec injection
|
|
"""
|
|
|
|
# Sauvegarder le texte
|
|
text_file = temp_data_dir / "ccam_2025_text.txt"
|
|
with open(text_file, "w", encoding="utf-8") as f:
|
|
f.write(test_text)
|
|
|
|
chunks = manager.chunk_ccam(test_text, "2025")
|
|
|
|
# Vérifier qu'on a des chunks
|
|
assert len(chunks) > 0
|
|
|
|
# Vérifier que les métadonnées contiennent la section
|
|
for chunk in chunks:
|
|
assert "section" in chunk.metadata
|
|
assert chunk.metadata["chunk_type"] == "acte"
|
|
|
|
# Vérifier que les extensions ne sont pas coupées
|
|
full_content = "\n".join([c.content for c in chunks])
|
|
assert "Extensions ATIH:" in full_content
|
|
assert "+ABC" in full_content
|
|
assert "+DEF" in full_content
|
|
assert "+GHI" in full_content
|
|
|
|
def test_chunk_ccam_respects_size_limits(self, temp_data_dir):
|
|
"""Test que les chunks respectent les limites de taille."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
# Créer un texte long
|
|
test_text = "CHAPITRE 1\n\n" + ("YYYY001 Acte de test\n" * 300)
|
|
|
|
chunks = manager.chunk_ccam(test_text, "2025")
|
|
|
|
# Vérifier que les chunks ne dépassent pas la taille max
|
|
for chunk in chunks:
|
|
assert len(chunk.content) <= 4096 # max_chunk_size
|
|
|
|
def test_chunk_ccam_does_not_cut_technical_notes(self, temp_data_dir):
|
|
"""Test que les notes techniques ne sont jamais coupées."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
# Créer un texte avec une longue note technique
|
|
test_text = """YYYY001 Acte chirurgical
|
|
|
|
Note technique:
|
|
Cet acte nécessite:
|
|
- Une anesthésie générale
|
|
- Un plateau technique complet
|
|
- Une équipe chirurgicale de 3 personnes minimum
|
|
- Un contrôle radiologique per-opératoire
|
|
- Une surveillance post-opératoire de 24h
|
|
|
|
Condition d'application: Patient à jeun depuis 6h.
|
|
|
|
YYYY002 Autre acte
|
|
"""
|
|
|
|
chunks = manager.chunk_ccam(test_text, "2025")
|
|
|
|
# Vérifier que chaque chunk contient soit la note complète, soit rien de la note
|
|
for chunk in chunks:
|
|
if "Note technique:" in chunk.content:
|
|
# Si le chunk contient "Note technique:", il doit contenir toute la note
|
|
assert "- Une anesthésie générale" in chunk.content
|
|
assert "- Une surveillance post-opératoire de 24h" in chunk.content
|
|
|
|
|
|
class TestBuildIndex:
|
|
"""Tests de construction d'index vectoriel."""
|
|
|
|
def test_build_index_empty_chunks(self, temp_data_dir):
|
|
"""Test que build_index rejette une liste vide de chunks."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
with pytest.raises(ValueError, match="La liste de chunks ne peut pas être vide"):
|
|
manager.build_index([])
|
|
|
|
def test_build_index_creates_vector_index(self, temp_data_dir, sample_pdf):
|
|
"""Test que build_index crée un index vectoriel."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
# Import et chunking
|
|
ref_version = manager.import_referentiel(
|
|
file_path=str(sample_pdf),
|
|
referentiel_type="cim10",
|
|
version="2026"
|
|
)
|
|
chunks = manager.chunk_referentiel(ref_version)
|
|
|
|
# Construction de l'index
|
|
vector_index = manager.build_index(chunks)
|
|
|
|
# Vérifications
|
|
assert vector_index.index_hash is not None
|
|
assert len(vector_index.index_hash) == 64 # SHA-256
|
|
assert vector_index.dimension > 0
|
|
assert vector_index.num_vectors == len(chunks)
|
|
assert vector_index.index_type == "HNSW"
|
|
assert isinstance(vector_index.created_at, datetime)
|
|
|
|
def test_build_index_saves_to_disk(self, temp_data_dir, sample_pdf):
|
|
"""Test que build_index sauvegarde l'index sur disque."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
# Import et chunking
|
|
ref_version = manager.import_referentiel(
|
|
file_path=str(sample_pdf),
|
|
referentiel_type="guide_mco",
|
|
version="2026"
|
|
)
|
|
chunks = manager.chunk_referentiel(ref_version)
|
|
|
|
# Construction de l'index
|
|
manager.build_index(chunks)
|
|
|
|
# Vérifier que le fichier d'index existe
|
|
index_file = temp_data_dir / "guide_mco_2026_index.faiss"
|
|
assert index_file.exists()
|
|
|
|
# Vérifier que le fichier de chunks existe
|
|
chunks_file = temp_data_dir / "guide_mco_2026_chunks.json"
|
|
assert chunks_file.exists()
|
|
|
|
|
|
class TestRebuildIndex:
|
|
"""Tests de reconstruction d'index après mise à jour."""
|
|
|
|
def test_rebuild_index_with_code_mapper(self, temp_data_dir, sample_pdf):
|
|
"""Test la reconstruction d'index avec code mapper."""
|
|
from pipeline_mco_pmsi.referentiels import CodeMapper
|
|
from pipeline_mco_pmsi.referentiels.code_mapper import CodeMapping
|
|
from pipeline_mco_pmsi.models.metadata import ReferentielVersion
|
|
from datetime import datetime
|
|
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
code_mapper = CodeMapper(mappings_dir=temp_data_dir / "mappings")
|
|
|
|
# Ajouter un mapping de test
|
|
mapping = CodeMapping(
|
|
obsolete_code="A00.0",
|
|
current_code="A00.1",
|
|
obsolete_label="Ancien code",
|
|
current_label="Nouveau code",
|
|
effective_date=datetime.now(),
|
|
reason="obsolete"
|
|
)
|
|
code_mapper.add_mapping(mapping, "cim10")
|
|
|
|
# Import initial
|
|
ref_version = manager.import_referentiel(
|
|
file_path=str(sample_pdf),
|
|
referentiel_type="cim10",
|
|
version="2026"
|
|
)
|
|
|
|
# Chunking et indexation initiale
|
|
chunks = manager.chunk_referentiel(ref_version)
|
|
initial_index = manager.build_index(chunks)
|
|
|
|
# Mise à jour du référentiel version dans le cache
|
|
# Créer une nouvelle instance car ReferentielVersion est frozen
|
|
updated_ref_version = ReferentielVersion(
|
|
type=ref_version.type,
|
|
version=ref_version.version,
|
|
import_date=ref_version.import_date,
|
|
file_hash=ref_version.file_hash,
|
|
chunk_count=len(chunks),
|
|
index_hash=initial_index.index_hash
|
|
)
|
|
manager._versions_cache[f"cim10_2026"] = updated_ref_version
|
|
|
|
# Reconstruction avec code mapper
|
|
rebuilt_index = manager.rebuild_index("cim10", "2026", code_mapper=code_mapper)
|
|
|
|
# Vérifications
|
|
assert rebuilt_index.index_hash is not None
|
|
assert rebuilt_index.num_vectors > 0
|
|
|
|
# Le hash devrait être différent car le contenu a changé
|
|
# (même si dans ce test le PDF est vide, la logique est testée)
|
|
assert rebuilt_index.index_hash is not None
|
|
|
|
|
|
|
|
def test_rebuild_index_with_code_mapper(self, temp_data_dir, sample_pdf):
|
|
"""Test la reconstruction d'index avec code mapper."""
|
|
from pipeline_mco_pmsi.referentiels import CodeMapper
|
|
from datetime import datetime
|
|
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
code_mapper = CodeMapper(mappings_dir=temp_data_dir / "mappings")
|
|
|
|
# Ajouter un mapping de test
|
|
from pipeline_mco_pmsi.referentiels.code_mapper import CodeMapping
|
|
mapping = CodeMapping(
|
|
obsolete_code="A00.0",
|
|
current_code="A00.1",
|
|
obsolete_label="Ancien code",
|
|
current_label="Nouveau code",
|
|
effective_date=datetime.now(),
|
|
reason="obsolete"
|
|
)
|
|
code_mapper.add_mapping(mapping, "cim10")
|
|
|
|
# Import initial
|
|
ref_version = manager.import_referentiel(
|
|
file_path=str(sample_pdf),
|
|
referentiel_type="cim10",
|
|
version="2026"
|
|
)
|
|
|
|
# Chunking et indexation initiale
|
|
chunks = manager.chunk_referentiel(ref_version)
|
|
initial_index = manager.build_index(chunks)
|
|
|
|
# Mise à jour du référentiel version dans le cache
|
|
# Créer une nouvelle instance car ReferentielVersion est frozen
|
|
from pipeline_mco_pmsi.models.metadata import ReferentielVersion
|
|
updated_ref_version = ReferentielVersion(
|
|
type=ref_version.type,
|
|
version=ref_version.version,
|
|
import_date=ref_version.import_date,
|
|
file_hash=ref_version.file_hash,
|
|
chunk_count=len(chunks),
|
|
index_hash=initial_index.index_hash
|
|
)
|
|
manager._versions_cache[f"cim10_2026"] = updated_ref_version
|
|
|
|
# Reconstruction avec code mapper
|
|
rebuilt_index = manager.rebuild_index("cim10", "2026", code_mapper=code_mapper)
|
|
|
|
# Vérifications
|
|
assert rebuilt_index.index_hash is not None
|
|
assert rebuilt_index.num_vectors > 0
|
|
|
|
# Le hash devrait être différent car le contenu a changé
|
|
# (même si dans ce test le PDF est vide, la logique est testée)
|
|
assert rebuilt_index.index_hash is not None
|
|
|
|
def test_rebuild_index_referentiel_not_found(self, temp_data_dir):
|
|
"""Test que rebuild_index échoue si le référentiel n'est pas trouvé."""
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
with pytest.raises(RuntimeError, match="Version du référentiel .* non trouvée"):
|
|
manager.rebuild_index("cim10", "2026")
|
|
|
|
def test_rebuild_index_text_file_not_found(self, temp_data_dir):
|
|
"""Test que rebuild_index échoue si le fichier texte n'existe pas."""
|
|
from pipeline_mco_pmsi.models.metadata import ReferentielVersion
|
|
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
# Créer une version sans fichier texte
|
|
fake_version = ReferentielVersion(
|
|
type="cim10",
|
|
version="2026",
|
|
import_date=datetime.now(),
|
|
file_hash="a" * 64,
|
|
chunk_count=0,
|
|
index_hash="0" * 64
|
|
)
|
|
manager._versions_cache["cim10_2026"] = fake_version
|
|
|
|
with pytest.raises(RuntimeError, match="Fichier texte du référentiel introuvable"):
|
|
manager.rebuild_index("cim10", "2026")
|
|
|
|
def test_rebuild_index_updates_metadata(self, temp_data_dir, sample_pdf):
|
|
"""Test que rebuild_index met à jour les métadonnées de version."""
|
|
from pipeline_mco_pmsi.models.metadata import ReferentielVersion
|
|
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
|
|
# Import initial
|
|
ref_version = manager.import_referentiel(
|
|
file_path=str(sample_pdf),
|
|
referentiel_type="ccam",
|
|
version="2025"
|
|
)
|
|
|
|
# Chunking et indexation initiale
|
|
chunks = manager.chunk_referentiel(ref_version)
|
|
initial_index = manager.build_index(chunks)
|
|
|
|
# Mise à jour du référentiel version dans le cache
|
|
# Créer une nouvelle instance car ReferentielVersion est frozen
|
|
updated_ref_version = ReferentielVersion(
|
|
type=ref_version.type,
|
|
version=ref_version.version,
|
|
import_date=ref_version.import_date,
|
|
file_hash=ref_version.file_hash,
|
|
chunk_count=len(chunks),
|
|
index_hash=initial_index.index_hash
|
|
)
|
|
manager._versions_cache[f"ccam_2025"] = updated_ref_version
|
|
|
|
# Sauvegarder les valeurs initiales
|
|
initial_hash = updated_ref_version.index_hash
|
|
initial_count = updated_ref_version.chunk_count
|
|
|
|
# Reconstruction
|
|
rebuilt_index = manager.rebuild_index("ccam", "2025")
|
|
|
|
# Vérifier que les métadonnées ont été mises à jour
|
|
updated_version = manager.get_version_info("ccam")
|
|
assert updated_version.index_hash == rebuilt_index.index_hash
|
|
assert updated_version.chunk_count == rebuilt_index.num_vectors
|
|
|
|
# Les valeurs devraient être cohérentes
|
|
assert updated_version.index_hash is not None
|
|
assert updated_version.chunk_count > 0
|
|
|
|
|
|
|
|
class TestApplyCodeMappings:
|
|
"""Tests d'application des mappings de codes."""
|
|
|
|
def test_apply_code_mappings_cim10(self, temp_data_dir):
|
|
"""Test l'application des mappings pour CIM-10."""
|
|
from pipeline_mco_pmsi.referentiels import CodeMapper
|
|
from pipeline_mco_pmsi.referentiels.code_mapper import CodeMapping
|
|
from datetime import datetime
|
|
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
code_mapper = CodeMapper()
|
|
|
|
# Ajouter un mapping
|
|
mapping = CodeMapping(
|
|
obsolete_code="A00.0",
|
|
current_code="A00.9",
|
|
obsolete_label="Ancien",
|
|
current_label="Nouveau",
|
|
effective_date=datetime.now(),
|
|
reason="obsolete"
|
|
)
|
|
code_mapper.add_mapping(mapping, "cim10")
|
|
|
|
# Texte avec code obsolète
|
|
text = """CHAPITRE I
|
|
|
|
A00.0 Choléra ancien
|
|
A00.1 Choléra actuel
|
|
A00.9 Choléra sans précision
|
|
"""
|
|
|
|
# Appliquer les mappings
|
|
updated_text = manager._apply_code_mappings(text, "cim10", code_mapper)
|
|
|
|
# Vérifier que le code obsolète a été remplacé
|
|
assert "A00.9 Choléra ancien" in updated_text
|
|
assert "A00.1 Choléra actuel" in updated_text
|
|
# Le code A00.0 ne devrait plus apparaître seul (remplacé par A00.9)
|
|
import re
|
|
# Compter les occurrences de A00.0 comme code (pas dans A00.01 par exemple)
|
|
a00_0_count = len(re.findall(r'\bA00\.0\b', updated_text))
|
|
assert a00_0_count == 0
|
|
|
|
def test_apply_code_mappings_ccam(self, temp_data_dir):
|
|
"""Test l'application des mappings pour CCAM."""
|
|
from pipeline_mco_pmsi.referentiels import CodeMapper
|
|
from pipeline_mco_pmsi.referentiels.code_mapper import CodeMapping
|
|
from datetime import datetime
|
|
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
code_mapper = CodeMapper()
|
|
|
|
# Ajouter un mapping
|
|
mapping = CodeMapping(
|
|
obsolete_code="YYYY001",
|
|
current_code="YYYY002",
|
|
obsolete_label="Ancien acte",
|
|
current_label="Nouvel acte",
|
|
effective_date=datetime.now(),
|
|
reason="merged"
|
|
)
|
|
code_mapper.add_mapping(mapping, "ccam")
|
|
|
|
# Texte avec code obsolète
|
|
text = """CHAPITRE 1
|
|
|
|
YYYY001 Acte ancien
|
|
YYYY002 Acte actuel
|
|
YYYY003 Autre acte
|
|
"""
|
|
|
|
# Appliquer les mappings
|
|
updated_text = manager._apply_code_mappings(text, "ccam", code_mapper)
|
|
|
|
# Vérifier que le code obsolète a été remplacé
|
|
assert "YYYY002 Acte ancien" in updated_text
|
|
assert "YYYY002 Acte actuel" in updated_text
|
|
assert "YYYY003 Autre acte" in updated_text
|
|
# Le code YYYY001 ne devrait plus apparaître
|
|
assert "YYYY001" not in updated_text
|
|
|
|
def test_apply_code_mappings_with_aliases(self, temp_data_dir):
|
|
"""Test l'application des mappings avec aliases."""
|
|
from pipeline_mco_pmsi.referentiels import CodeMapper
|
|
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
code_mapper = CodeMapper()
|
|
|
|
# Ajouter un alias
|
|
code_mapper.add_alias("A00.X", "A00.9", "cim10")
|
|
|
|
# Texte avec alias
|
|
text = """A00.X Choléra (alias)
|
|
A00.9 Choléra sans précision
|
|
"""
|
|
|
|
# Appliquer les mappings
|
|
updated_text = manager._apply_code_mappings(text, "cim10", code_mapper)
|
|
|
|
# Vérifier que l'alias a été résolu
|
|
assert "A00.9 Choléra (alias)" in updated_text
|
|
assert "A00.X" not in updated_text
|
|
|
|
def test_apply_code_mappings_guide_mco_unchanged(self, temp_data_dir):
|
|
"""Test que les mappings ne modifient pas le guide MCO."""
|
|
from pipeline_mco_pmsi.referentiels import CodeMapper
|
|
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
code_mapper = CodeMapper()
|
|
|
|
# Texte du guide
|
|
text = """CHAPITRE 1 - Le recueil
|
|
|
|
Règle 1: Le DP doit être codé selon la CIM-10.
|
|
"""
|
|
|
|
# Appliquer les mappings (ne devrait rien changer)
|
|
updated_text = manager._apply_code_mappings(text, "guide_mco", code_mapper)
|
|
|
|
# Le texte devrait être inchangé
|
|
assert updated_text == text
|
|
|
|
def test_apply_code_mappings_preserves_unknown_codes(self, temp_data_dir):
|
|
"""Test que les codes inconnus sont préservés."""
|
|
from pipeline_mco_pmsi.referentiels import CodeMapper
|
|
|
|
manager = ReferentielsManager(data_dir=temp_data_dir)
|
|
code_mapper = CodeMapper() # Aucun mapping
|
|
|
|
# Texte avec codes
|
|
text = """A00.0 Code 1
|
|
A00.1 Code 2
|
|
A00.2 Code 3
|
|
"""
|
|
|
|
# Appliquer les mappings (ne devrait rien changer)
|
|
updated_text = manager._apply_code_mappings(text, "cim10", code_mapper)
|
|
|
|
# Le texte devrait être inchangé
|
|
assert updated_text == text
|
|
|