Files
aivanov_CIM/tests/test_tim_api.py
2026-03-05 01:20:14 +01:00

726 lines
23 KiB
Python

"""
Tests pour l'API TIM.
Exigences : 10.1, 10.6, 10.7, 10.8, 10.9
"""
import json
from datetime import datetime
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from pipeline_mco_pmsi.api.tim_api import app, get_db
from pipeline_mco_pmsi.database.base import Base
# Import ALL models at module level to ensure they are registered with Base.metadata
from pipeline_mco_pmsi.database.models import (
StayDB,
ClinicalDocumentDB,
ClinicalFactDB,
CodeDB,
EvidenceDB,
ValidationIssueDB,
QuestionDB,
AuditRecordDB,
TIMCorrectionDB,
VerificationResultDB,
GroupageResultDB,
ReferentielVersionDB,
)
# Fixture pour la base de données de test
@pytest.fixture
def test_db(tmp_path):
"""Crée une base de données de test en mémoire."""
# Use a temporary file instead of :memory: to ensure all connections see the same database
db_path = tmp_path / "test.db"
# Configure SQLite for testing with check_same_thread=False
engine = create_engine(
f"sqlite:///{db_path}",
connect_args={"check_same_thread": False},
)
# Create all tables - models are already imported at module level
Base.metadata.create_all(engine)
TestingSessionLocal = sessionmaker(bind=engine)
db = TestingSessionLocal()
yield db
db.close()
engine.dispose()
@pytest.fixture
def client(test_db):
"""Crée un client de test FastAPI."""
# Get the engine from the test_db session
engine = test_db.get_bind()
# Create a session factory for this engine
TestingSessionLocal = sessionmaker(bind=engine)
def override_get_db():
db = TestingSessionLocal()
try:
yield db
finally:
db.close()
app.dependency_overrides[get_db] = override_get_db
with TestClient(app) as test_client:
yield test_client
app.dependency_overrides.clear()
@pytest.fixture
def sample_stay(test_db):
"""Crée un séjour de test avec codes et preuves."""
# Créer le séjour
stay = StayDB(
stay_id="STAY_TEST_001",
admission_date=datetime(2024, 1, 15),
discharge_date=datetime(2024, 1, 20),
specialty="chirurgie",
unit="Chirurgie digestive",
age=45,
sex="M",
)
test_db.add(stay)
test_db.flush()
# Créer un document
document = ClinicalDocumentDB(
stay_id=stay.id,
document_id="DOC001",
document_type="cr_operatoire",
content="Patient opéré pour appendicite aiguë. Intervention réalisée sous anesthésie générale.",
creation_date=datetime(2024, 1, 15, 10, 0),
author="Dr. Martin",
priority=1,
)
test_db.add(document)
test_db.flush()
# Créer un code DP
code_dp = CodeDB(
stay_id=stay.id,
code="K35.8",
label="Appendicite aiguë, autres et sans précision",
type="dp",
confidence=0.92,
reasoning="Diagnostic principal basé sur le compte-rendu opératoire",
referentiel_version="CIM-10 2024",
status="proposed",
model_name="test-model",
model_digest="a" * 64, # Hash SHA-256 valide (64 caractères hexadécimaux)
prompt_version="v1.0",
)
test_db.add(code_dp)
test_db.flush()
# Créer une preuve pour le code DP
evidence = EvidenceDB(
code_id=code_dp.id,
document_id=document.id,
span_start=19,
span_end=36,
text="appendicite aiguë",
context="Patient opéré pour appendicite aiguë. Intervention réalisée",
)
test_db.add(evidence)
# Créer un code CCAM
code_ccam = CodeDB(
stay_id=stay.id,
code="HHFA015",
label="Appendicectomie",
type="ccam",
confidence=0.95,
reasoning="Acte chirurgical réalisé",
referentiel_version="CCAM V81",
status="proposed",
model_name="test-model",
model_digest="b" * 64, # Hash SHA-256 valide (64 caractères hexadécimaux)
prompt_version="v1.0",
)
test_db.add(code_ccam)
test_db.flush()
# Créer une preuve pour le code CCAM
evidence_ccam = EvidenceDB(
code_id=code_ccam.id,
document_id=document.id,
span_start=0,
span_end=15,
text="Patient opéré",
context="Patient opéré pour appendicite aiguë",
)
test_db.add(evidence_ccam)
test_db.commit()
return stay
class TestGetCodingProposal:
"""Tests pour l'endpoint GET /stays/{stay_id}/coding-proposal."""
def test_get_coding_proposal_success(self, client, sample_stay):
"""Test récupération réussie d'une proposition de codage."""
response = client.get(f"/stays/{sample_stay.stay_id}/coding-proposal")
assert response.status_code == 200
data = response.json()
assert data["stay_id"] == sample_stay.stay_id
assert data["dp"] is not None
assert data["dp"]["code"] == "K35.8"
assert data["dp"]["confidence"] == 0.92
assert len(data["dp"]["evidence"]) == 1
assert len(data["ccam"]) == 1
assert data["ccam"][0]["code"] == "HHFA015"
assert "confidence_scores" in data
assert data["confidence_scores"]["dp"] == 0.92
def test_get_coding_proposal_not_found(self, client):
"""Test avec un séjour inexistant."""
response = client.get("/stays/NONEXISTENT/coding-proposal")
assert response.status_code == 404
assert "not found" in response.json()["detail"].lower()
def test_get_coding_proposal_no_codes(self, client, test_db):
"""Test avec un séjour sans codes proposés."""
# Créer un séjour sans codes
stay = StayDB(
stay_id="STAY_NO_CODES",
admission_date=datetime(2024, 1, 15),
discharge_date=datetime(2024, 1, 20),
specialty="medecine",
)
test_db.add(stay)
test_db.commit()
response = client.get(f"/stays/{stay.stay_id}/coding-proposal")
assert response.status_code == 404
assert "no coding proposal" in response.json()["detail"].lower()
class TestCorrectCode:
"""Tests pour l'endpoint POST /stays/{stay_id}/correct-code."""
def test_correct_code_success(self, client, sample_stay):
"""Test correction réussie d'un code."""
correction_data = {
"stay_id": sample_stay.stay_id,
"original_code": "K35.8",
"corrected_code": "K35.2",
"corrected_label": "Appendicite aiguë avec péritonite généralisée",
"comment": "Précision du diagnostic après relecture",
}
response = client.post(
f"/stays/{sample_stay.stay_id}/correct-code",
json=correction_data,
)
assert response.status_code == 200
data = response.json()
assert data["stay_id"] == sample_stay.stay_id
assert data["original_code"] == "K35.8"
assert data["corrected_code"] == "K35.2"
assert "timestamp" in data
def test_correct_code_stay_not_found(self, client):
"""Test correction avec séjour inexistant."""
correction_data = {
"stay_id": "NONEXISTENT",
"original_code": "K35.8",
"corrected_code": "K35.2",
}
response = client.post(
"/stays/NONEXISTENT/correct-code",
json=correction_data,
)
assert response.status_code == 404
def test_correct_code_mismatch_stay_id(self, client, sample_stay):
"""Test correction avec stay_id incohérent."""
correction_data = {
"stay_id": "DIFFERENT_STAY",
"original_code": "K35.8",
"corrected_code": "K35.2",
}
response = client.post(
f"/stays/{sample_stay.stay_id}/correct-code",
json=correction_data,
)
assert response.status_code == 400
assert "mismatch" in response.json()["detail"].lower()
def test_correct_code_not_found(self, client, sample_stay):
"""Test correction d'un code inexistant."""
correction_data = {
"stay_id": sample_stay.stay_id,
"original_code": "Z99.9", # Code inexistant
"corrected_code": "K35.2",
}
response = client.post(
f"/stays/{sample_stay.stay_id}/correct-code",
json=correction_data,
)
assert response.status_code == 404
class TestValidateStay:
"""Tests pour l'endpoint POST /stays/{stay_id}/validate."""
def test_validate_stay_accepted(self, client, sample_stay):
"""Test validation acceptée d'un séjour."""
validation_data = {
"stay_id": sample_stay.stay_id,
"validation_status": "accepted",
"comment": "Codage conforme",
}
response = client.post(
f"/stays/{sample_stay.stay_id}/validate",
json=validation_data,
)
assert response.status_code == 200
data = response.json()
assert data["stay_id"] == sample_stay.stay_id
assert data["validation_status"] == "accepted"
assert "timestamp" in data
def test_validate_stay_rejected(self, client, sample_stay):
"""Test validation rejetée d'un séjour."""
validation_data = {
"stay_id": sample_stay.stay_id,
"validation_status": "rejected",
"comment": "Codes incorrects",
}
response = client.post(
f"/stays/{sample_stay.stay_id}/validate",
json=validation_data,
)
assert response.status_code == 200
data = response.json()
assert data["validation_status"] == "rejected"
def test_validate_stay_needs_review(self, client, sample_stay):
"""Test validation à revoir d'un séjour."""
validation_data = {
"stay_id": sample_stay.stay_id,
"validation_status": "needs_review",
}
response = client.post(
f"/stays/{sample_stay.stay_id}/validate",
json=validation_data,
)
assert response.status_code == 200
data = response.json()
assert data["validation_status"] == "needs_review"
def test_validate_stay_not_found(self, client):
"""Test validation avec séjour inexistant."""
validation_data = {
"stay_id": "NONEXISTENT",
"validation_status": "accepted",
}
response = client.post(
"/stays/NONEXISTENT/validate",
json=validation_data,
)
assert response.status_code == 404
class TestAddComment:
"""Tests pour l'endpoint POST /stays/{stay_id}/comment."""
def test_add_comment_success(self, client, sample_stay):
"""Test ajout réussi d'un commentaire."""
comment_data = {
"stay_id": sample_stay.stay_id,
"code": "K35.8",
"comment": "Vérifier la présence de péritonite",
}
response = client.post(
f"/stays/{sample_stay.stay_id}/comment",
json=comment_data,
)
assert response.status_code == 200
data = response.json()
assert data["stay_id"] == sample_stay.stay_id
assert data["code"] == "K35.8"
assert data["comment"] == comment_data["comment"]
assert "timestamp" in data
def test_add_comment_stay_not_found(self, client):
"""Test ajout de commentaire avec séjour inexistant."""
comment_data = {
"stay_id": "NONEXISTENT",
"code": "K35.8",
"comment": "Test",
}
response = client.post(
"/stays/NONEXISTENT/comment",
json=comment_data,
)
assert response.status_code == 404
def test_add_comment_code_not_found(self, client, sample_stay):
"""Test ajout de commentaire pour un code inexistant."""
comment_data = {
"stay_id": sample_stay.stay_id,
"code": "Z99.9", # Code inexistant
"comment": "Test",
}
response = client.post(
f"/stays/{sample_stay.stay_id}/comment",
json=comment_data,
)
assert response.status_code == 404
class TestExportAudit:
"""Tests pour l'endpoint POST /stays/{stay_id}/audit/export."""
def test_export_audit_plain(self, client, sample_stay):
"""Test export d'audit non chiffré."""
export_data = {
"stay_id": sample_stay.stay_id,
"include_pii": False,
"encrypt": False,
}
response = client.post(
f"/stays/{sample_stay.stay_id}/audit/export",
json=export_data,
)
assert response.status_code == 200
data = response.json()
assert data["stay_id"] == sample_stay.stay_id
assert data["encrypted"] is False
assert "data" in data
assert "export_date" in data
def test_export_audit_encrypted(self, client, sample_stay):
"""Test export d'audit chiffré."""
export_data = {
"stay_id": sample_stay.stay_id,
"include_pii": False,
"encrypt": True,
}
response = client.post(
f"/stays/{sample_stay.stay_id}/audit/export",
json=export_data,
)
assert response.status_code == 200
data = response.json()
assert data["stay_id"] == sample_stay.stay_id
assert data["encrypted"] is True
assert "data" in data
assert "encryption_key" in data
assert "export_date" in data
# Vérifier que les données sont en base64
import base64
try:
base64.b64decode(data["data"])
except Exception:
pytest.fail("Les données chiffrées ne sont pas en base64 valide")
def test_export_audit_stay_not_found(self, client):
"""Test export d'audit avec séjour inexistant."""
export_data = {
"stay_id": "NONEXISTENT",
"include_pii": False,
"encrypt": False,
}
response = client.post(
"/stays/NONEXISTENT/audit/export",
json=export_data,
)
assert response.status_code == 404
class TestGetEvidence:
"""Tests pour l'endpoint GET /stays/{stay_id}/evidence/{code}."""
def test_get_evidence_success(self, client, sample_stay):
"""Test récupération réussie des preuves."""
response = client.get(f"/stays/{sample_stay.stay_id}/evidence/K35.8")
assert response.status_code == 200
data = response.json()
assert data["stay_id"] == sample_stay.stay_id
assert data["code"] == "K35.8"
assert len(data["evidences"]) == 1
evidence = data["evidences"][0]
assert evidence["document_id"] == "DOC001"
assert evidence["text"] == "appendicite aiguë"
assert "document_link" in evidence
def test_get_evidence_stay_not_found(self, client):
"""Test récupération de preuves avec séjour inexistant."""
response = client.get("/stays/NONEXISTENT/evidence/K35.8")
assert response.status_code == 404
def test_get_evidence_code_not_found(self, client, sample_stay):
"""Test récupération de preuves pour un code inexistant."""
response = client.get(f"/stays/{sample_stay.stay_id}/evidence/Z99.9")
assert response.status_code == 404
class TestGetDocument:
"""Tests pour l'endpoint GET /documents/{document_id}."""
def test_get_document_success(self, client, sample_stay):
"""Test récupération réussie d'un document."""
response = client.get("/documents/DOC001")
assert response.status_code == 200
data = response.json()
assert data["document_id"] == "DOC001"
assert data["document_type"] == "cr_operatoire"
assert "content" in data
assert "Patient opéré" in data["content"]
assert "creation_date" in data
assert data["author"] == "Dr. Martin"
def test_get_document_not_found(self, client):
"""Test récupération d'un document inexistant."""
response = client.get("/documents/NONEXISTENT")
assert response.status_code == 404
class TestEvidenceNavigation:
"""Tests pour la navigation preuves → texte source (Exigences 10.2, 10.3)."""
def test_evidence_to_document_navigation(self, client, sample_stay):
"""Test navigation complète depuis une preuve vers le document source."""
# 1. Récupérer les preuves pour un code
evidence_response = client.get(f"/stays/{sample_stay.stay_id}/evidence/K35.8")
assert evidence_response.status_code == 200
evidence_data = evidence_response.json()
# Vérifier que les preuves contiennent les informations nécessaires
assert len(evidence_data["evidences"]) > 0
evidence = evidence_data["evidences"][0]
assert "document_id" in evidence
assert "span" in evidence
assert "start" in evidence["span"]
assert "end" in evidence["span"]
assert "document_link" in evidence
# 2. Utiliser le lien pour récupérer le document
document_id = evidence["document_id"]
document_response = client.get(f"/documents/{document_id}")
assert document_response.status_code == 200
document_data = document_response.json()
# 3. Vérifier que le texte de la preuve correspond au span dans le document
span_start = evidence["span"]["start"]
span_end = evidence["span"]["end"]
document_content = document_data["content"]
extracted_text = document_content[span_start:span_end]
assert extracted_text == evidence["text"]
def test_evidence_contains_context(self, client, sample_stay):
"""Test que les preuves contiennent le contexte pour faciliter la navigation."""
response = client.get(f"/stays/{sample_stay.stay_id}/evidence/K35.8")
assert response.status_code == 200
data = response.json()
evidence = data["evidences"][0]
assert "context" in evidence
assert evidence["context"] is not None
# Le contexte doit contenir le texte de la preuve
assert evidence["text"] in evidence["context"]
def test_multiple_evidences_navigation(self, client, test_db):
"""Test navigation avec plusieurs preuves pour un même code."""
# Créer un séjour avec plusieurs documents et preuves
stay = StayDB(
stay_id="STAY_MULTI_EVIDENCE",
admission_date=datetime(2024, 1, 15),
discharge_date=datetime(2024, 1, 20),
specialty="medecine",
)
test_db.add(stay)
test_db.flush()
# Document 1
doc1 = ClinicalDocumentDB(
stay_id=stay.id,
document_id="DOC_MULTI_1",
document_type="cr_medical",
content="Patient présente une hypertension artérielle essentielle.",
creation_date=datetime(2024, 1, 15),
priority=2,
)
test_db.add(doc1)
test_db.flush()
# Document 2
doc2 = ClinicalDocumentDB(
stay_id=stay.id,
document_id="DOC_MULTI_2",
document_type="biologie",
content="Tension artérielle élevée confirmée.",
creation_date=datetime(2024, 1, 16),
priority=4,
)
test_db.add(doc2)
test_db.flush()
# Code avec plusieurs preuves
code = CodeDB(
stay_id=stay.id,
code="I10",
label="Hypertension essentielle",
type="dp",
confidence=0.88,
reasoning="Diagnostic confirmé par plusieurs documents",
referentiel_version="CIM-10 2024",
status="proposed",
model_name="test-model",
model_digest="test-digest",
prompt_version="v1.0",
)
test_db.add(code)
test_db.flush()
# Preuve 1
ev1 = EvidenceDB(
code_id=code.id,
document_id=doc1.id,
span_start=21,
span_end=56,
text="hypertension artérielle essentielle",
context="Patient présente une hypertension artérielle essentielle.",
)
test_db.add(ev1)
# Preuve 2
ev2 = EvidenceDB(
code_id=code.id,
document_id=doc2.id,
span_start=0,
span_end=25,
text="Tension artérielle élevée",
context="Tension artérielle élevée confirmée.",
)
test_db.add(ev2)
test_db.commit()
# Récupérer les preuves
response = client.get(f"/stays/{stay.stay_id}/evidence/I10")
assert response.status_code == 200
data = response.json()
# Vérifier qu'on a bien 2 preuves
assert len(data["evidences"]) == 2
# Vérifier que chaque preuve pointe vers le bon document
doc_ids = [ev["document_id"] for ev in data["evidences"]]
assert "DOC_MULTI_1" in doc_ids
assert "DOC_MULTI_2" in doc_ids
# Vérifier la navigation vers chaque document
for evidence in data["evidences"]:
doc_response = client.get(f"/documents/{evidence['document_id']}")
assert doc_response.status_code == 200
doc_data = doc_response.json()
# Vérifier que le span correspond
span_start = evidence["span"]["start"]
span_end = evidence["span"]["end"]
extracted = doc_data["content"][span_start:span_end]
assert extracted == evidence["text"]
def test_evidence_document_metadata(self, client, sample_stay):
"""Test que les métadonnées du document sont disponibles pour la navigation."""
response = client.get(f"/stays/{sample_stay.stay_id}/evidence/K35.8")
assert response.status_code == 200
data = response.json()
evidence = data["evidences"][0]
# Vérifier les métadonnées
assert "document_type" in evidence
assert evidence["document_type"] == "cr_operatoire"
# Récupérer le document complet
doc_response = client.get(f"/documents/{evidence['document_id']}")
assert doc_response.status_code == 200
doc_data = doc_response.json()
# Vérifier que les métadonnées sont cohérentes
assert doc_data["document_type"] == evidence["document_type"]
assert "priority" in doc_data
assert "creation_date" in doc_data
class TestRootEndpoint:
"""Tests pour l'endpoint racine."""
def test_root_returns_html(self, client):
"""Test que l'endpoint racine retourne l'interface HTML."""
response = client.get("/")
# Devrait retourner un fichier HTML ou une redirection
assert response.status_code in [200, 307] # 200 OK ou 307 Temporary Redirect