342 lines
9.8 KiB
Python
342 lines
9.8 KiB
Python
"""
|
|
Configuration pytest et Hypothesis pour les tests du Pipeline MCO PMSI.
|
|
|
|
Ce fichier configure :
|
|
- Les profils Hypothesis pour différents environnements
|
|
- Les fixtures communes à tous les tests
|
|
- Les hooks pytest personnalisés
|
|
"""
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Generator
|
|
|
|
import pytest
|
|
from hypothesis import HealthCheck, Phase, Verbosity, settings
|
|
|
|
# ============================================================================
|
|
# Configuration Hypothesis
|
|
# ============================================================================
|
|
|
|
# Profil par défaut : équilibré entre vitesse et couverture
|
|
settings.register_profile(
|
|
"default",
|
|
max_examples=100,
|
|
deadline=5000, # 5 secondes par test case
|
|
derandomize=False,
|
|
print_blob=True,
|
|
verbosity=Verbosity.normal,
|
|
phases=[Phase.explicit, Phase.reuse, Phase.generate, Phase.target, Phase.shrink],
|
|
suppress_health_check=[HealthCheck.too_slow],
|
|
)
|
|
|
|
# Profil CI : plus d'exemples pour meilleure couverture
|
|
settings.register_profile(
|
|
"ci",
|
|
max_examples=200,
|
|
deadline=10000, # 10 secondes par test case
|
|
derandomize=True, # Reproductible en CI
|
|
print_blob=True,
|
|
verbosity=Verbosity.verbose,
|
|
phases=[Phase.explicit, Phase.reuse, Phase.generate, Phase.target, Phase.shrink],
|
|
)
|
|
|
|
# Profil dev : rapide pour développement
|
|
settings.register_profile(
|
|
"dev",
|
|
max_examples=20,
|
|
deadline=2000, # 2 secondes par test case
|
|
derandomize=False,
|
|
print_blob=False,
|
|
verbosity=Verbosity.normal,
|
|
phases=[Phase.explicit, Phase.reuse, Phase.generate, Phase.shrink],
|
|
suppress_health_check=[HealthCheck.too_slow, HealthCheck.data_too_large],
|
|
)
|
|
|
|
# Profil debug : minimal pour debugging
|
|
settings.register_profile(
|
|
"debug",
|
|
max_examples=10,
|
|
deadline=None, # Pas de deadline pour debugging
|
|
derandomize=True,
|
|
print_blob=True,
|
|
verbosity=Verbosity.debug,
|
|
phases=[Phase.explicit, Phase.reuse, Phase.generate],
|
|
)
|
|
|
|
# Profil exhaustif : pour validation finale
|
|
settings.register_profile(
|
|
"exhaustive",
|
|
max_examples=1000,
|
|
deadline=30000, # 30 secondes par test case
|
|
derandomize=True,
|
|
print_blob=True,
|
|
verbosity=Verbosity.verbose,
|
|
phases=[Phase.explicit, Phase.reuse, Phase.generate, Phase.target, Phase.shrink],
|
|
)
|
|
|
|
# Charger le profil depuis la variable d'environnement ou utiliser 'default'
|
|
profile = os.getenv("HYPOTHESIS_PROFILE", "default")
|
|
settings.load_profile(profile)
|
|
|
|
|
|
# ============================================================================
|
|
# Fixtures communes
|
|
# ============================================================================
|
|
|
|
@pytest.fixture(scope="session")
|
|
def project_root() -> Path:
|
|
"""Retourne le chemin racine du projet."""
|
|
return Path(__file__).parent.parent
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def data_dir(project_root: Path) -> Path:
|
|
"""Retourne le répertoire data/."""
|
|
return project_root / "data"
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def referentiels_dir(data_dir: Path) -> Path:
|
|
"""Retourne le répertoire des référentiels."""
|
|
return data_dir / "referentiels"
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def config_dir(project_root: Path) -> Path:
|
|
"""Retourne le répertoire config/."""
|
|
return project_root / "config"
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def logs_dir(project_root: Path) -> Path:
|
|
"""Retourne le répertoire logs/."""
|
|
logs = project_root / "logs"
|
|
logs.mkdir(exist_ok=True)
|
|
return logs
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_db_path(tmp_path: Path) -> Path:
|
|
"""Crée un chemin pour une base de données temporaire."""
|
|
return tmp_path / "test_db.sqlite"
|
|
|
|
|
|
# ============================================================================
|
|
# Hooks pytest
|
|
# ============================================================================
|
|
|
|
def pytest_configure(config: pytest.Config) -> None:
|
|
"""Configuration pytest au démarrage."""
|
|
# Créer le répertoire logs s'il n'existe pas
|
|
logs_dir = Path("logs")
|
|
logs_dir.mkdir(exist_ok=True)
|
|
|
|
# Afficher le profil Hypothesis utilisé
|
|
print(f"\n🔬 Hypothesis profile: {profile}")
|
|
print(f" Max examples: {settings.default.max_examples}")
|
|
print(f" Deadline: {settings.default.deadline}ms\n")
|
|
|
|
|
|
def pytest_collection_modifyitems(config: pytest.Config, items: list) -> None:
|
|
"""Modifier les items de test collectés."""
|
|
# Ajouter automatiquement le marker 'property' aux tests utilisant Hypothesis
|
|
for item in items:
|
|
if "hypothesis" in item.keywords:
|
|
item.add_marker(pytest.mark.property)
|
|
item.add_marker(pytest.mark.pbt)
|
|
|
|
|
|
def pytest_addoption(parser: pytest.Parser) -> None:
|
|
"""Ajouter des options de ligne de commande personnalisées."""
|
|
parser.addoption(
|
|
"--run-slow",
|
|
action="store_true",
|
|
default=False,
|
|
help="Exécuter les tests marqués comme 'slow'",
|
|
)
|
|
parser.addoption(
|
|
"--run-gpu",
|
|
action="store_true",
|
|
default=False,
|
|
help="Exécuter les tests nécessitant un GPU",
|
|
)
|
|
|
|
|
|
def pytest_runtest_setup(item: pytest.Item) -> None:
|
|
"""Hook exécuté avant chaque test."""
|
|
# Skip les tests 'slow' sauf si --run-slow est spécifié
|
|
if "slow" in item.keywords and not item.config.getoption("--run-slow"):
|
|
pytest.skip("need --run-slow option to run")
|
|
|
|
# Skip les tests 'gpu' sauf si --run-gpu est spécifié
|
|
if "gpu" in item.keywords and not item.config.getoption("--run-gpu"):
|
|
pytest.skip("need --run-gpu option to run")
|
|
|
|
|
|
# ============================================================================
|
|
# Fixtures pour les tests de propriétés
|
|
# ============================================================================
|
|
|
|
@pytest.fixture
|
|
def hypothesis_seed() -> int:
|
|
"""Seed fixe pour reproductibilité des tests de propriétés."""
|
|
return 42
|
|
|
|
|
|
# ============================================================================
|
|
# Utilitaires de test
|
|
# ============================================================================
|
|
|
|
@pytest.fixture
|
|
def sample_clinical_text() -> str:
|
|
"""Texte clinique d'exemple pour les tests."""
|
|
return """
|
|
COMPTE RENDU MÉDICAL
|
|
|
|
Patient admis pour douleurs abdominales aiguës.
|
|
|
|
ANAMNÈSE:
|
|
Douleurs abdominales depuis 48h, localisées au niveau de la fosse iliaque droite.
|
|
Pas de fièvre. Pas de vomissements.
|
|
|
|
EXAMEN CLINIQUE:
|
|
Défense abdominale à la palpation de la FID.
|
|
Signe de Blumberg positif.
|
|
|
|
DIAGNOSTIC:
|
|
Appendicite aiguë.
|
|
|
|
TRAITEMENT:
|
|
Appendicectomie réalisée le 15/01/2024.
|
|
Suites opératoires simples.
|
|
"""
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_clinical_text_with_negation() -> str:
|
|
"""Texte clinique avec négations pour tester les qualificateurs."""
|
|
return """
|
|
COMPTE RENDU MÉDICAL
|
|
|
|
Patient admis pour bilan.
|
|
|
|
ANAMNÈSE:
|
|
Pas de douleurs thoraciques.
|
|
Absence de dyspnée.
|
|
Pas d'antécédent de diabète.
|
|
|
|
EXAMEN CLINIQUE:
|
|
Auscultation cardiaque normale.
|
|
Pas de souffle cardiaque.
|
|
|
|
DIAGNOSTIC:
|
|
Bilan de santé normal.
|
|
"""
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_clinical_text_with_suspicion() -> str:
|
|
"""Texte clinique avec suspicions pour tester les qualificateurs."""
|
|
return """
|
|
COMPTE RENDU MÉDICAL
|
|
|
|
Patient admis pour exploration.
|
|
|
|
ANAMNÈSE:
|
|
Possible infection urinaire.
|
|
Suspicion de pyélonéphrite.
|
|
|
|
EXAMENS:
|
|
ECBU en attente.
|
|
|
|
DIAGNOSTIC:
|
|
À confirmer après résultats ECBU.
|
|
"""
|
|
|
|
|
|
# ============================================================================
|
|
# Fixtures pour le pipeline
|
|
# ============================================================================
|
|
|
|
@pytest.fixture
|
|
def db_session(temp_db_path: Path) -> Generator:
|
|
"""Crée une session de base de données temporaire pour les tests."""
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from pipeline_mco_pmsi.database.base import Base
|
|
|
|
# Créer le moteur de base de données
|
|
engine = create_engine(f"sqlite:///{temp_db_path}")
|
|
|
|
# Créer toutes les tables
|
|
Base.metadata.create_all(engine)
|
|
|
|
# Créer une session
|
|
Session = sessionmaker(bind=engine)
|
|
session = Session()
|
|
|
|
try:
|
|
yield session
|
|
finally:
|
|
session.close()
|
|
engine.dispose()
|
|
|
|
|
|
@pytest.fixture
|
|
def rag_engine(tmp_path):
|
|
"""Crée un moteur RAG mock pour les tests."""
|
|
from pipeline_mco_pmsi.rag.rag_engine import RAGEngine
|
|
from pipeline_mco_pmsi.rag.referentiels_manager import ReferentielsManager
|
|
|
|
# Créer un répertoire temporaire pour les référentiels
|
|
ref_dir = tmp_path / "referentiels"
|
|
ref_dir.mkdir(exist_ok=True)
|
|
|
|
# Créer un ReferentielsManager mock
|
|
referentiels_manager = ReferentielsManager(
|
|
data_dir=ref_dir,
|
|
embedding_model="mock",
|
|
)
|
|
|
|
# Créer le RAG Engine
|
|
rag_engine = RAGEngine(
|
|
referentiels_manager=referentiels_manager,
|
|
data_dir=ref_dir,
|
|
)
|
|
|
|
return rag_engine
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_stay(db_session):
|
|
"""Crée un séjour d'exemple pour les tests."""
|
|
from datetime import datetime, timedelta
|
|
from pipeline_mco_pmsi.models.metadata import StayMetadata
|
|
from pipeline_mco_pmsi.database.models import StayDB
|
|
|
|
stay_metadata = StayMetadata(
|
|
stay_id="test_stay_001",
|
|
admission_date=datetime.now() - timedelta(days=3),
|
|
discharge_date=datetime.now(),
|
|
specialty="gastro-enterologie",
|
|
unit="medecine",
|
|
age=45,
|
|
sex="M",
|
|
)
|
|
|
|
# Créer le 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()
|
|
|
|
return stay_metadata
|