"""Tests pour la génération de contre-argumentation CPAM.""" from unittest.mock import patch, call import pytest from src.config import ( ActeCCAM, BiologieCle, ControleCPAM, Diagnostic, DossierMedical, Imagerie, RAGSource, Sejour, Traitement, ) from src.control.cpam_response import ( _build_cpam_prompt, _format_response, _search_rag_for_control, generate_cpam_response, ) def _make_dossier() -> DossierMedical: """Crée un dossier médical de test.""" return DossierMedical( source_file="test.pdf", document_type="crh", sejour=Sejour(sexe="M", age=65, duree_sejour=5), diagnostic_principal=Diagnostic( texte="Cholécystite aiguë", cim10_suggestion="K81.0", ), diagnostics_associes=[ Diagnostic(texte="Iléus réflexe", cim10_suggestion="K56.0"), ], ) def _make_dossier_complet() -> DossierMedical: """Crée un dossier médical enrichi avec traitements, imagerie, antécédents.""" return DossierMedical( source_file="test.pdf", document_type="crh", sejour=Sejour(sexe="M", age=65, duree_sejour=5, imc=31.2), diagnostic_principal=Diagnostic( texte="Cholécystite aiguë", cim10_suggestion="K81.0", ), diagnostics_associes=[ Diagnostic(texte="Iléus réflexe", cim10_suggestion="K56.0"), ], actes_ccam=[ ActeCCAM(texte="Cholécystectomie", code_ccam_suggestion="HMFC004"), ], biologie_cle=[ BiologieCle(test="CRP", valeur="180 mg/L", anomalie=True), BiologieCle(test="Créatinine", valeur="450 µmol/L", anomalie=True), ], imagerie=[ Imagerie(type="Scanner abdominal", conclusion="Lithiase cholédocienne confirmée"), ], traitements_sortie=[ Traitement(medicament="Augmentin IV", posologie="3g/j"), Traitement(medicament="Morphine SC"), ], antecedents=["HTA", "Diabète type 2"], ) def _make_controle() -> ControleCPAM: """Crée un contrôle CPAM de test.""" return ControleCPAM( numero_ogc=17, titre="Désaccord sur les DAS", arg_ucr="L'UCR confirme l'avis des médecins contrôleurs au motif que le DAS K56.0 n'est pas justifié.", decision_ucr="UCR confirme avis médecins contrôleurs", dp_ucr=None, da_ucr="K56.0", ) class TestBuildPrompt: def test_prompt_contains_dossier_info(self): dossier = _make_dossier() controle = _make_controle() prompt = _build_cpam_prompt(dossier, controle, []) assert "Cholécystite aiguë" in prompt assert "K81.0" in prompt assert "Iléus réflexe" in prompt assert "65 ans" in prompt def test_prompt_contains_cpam_argument(self): dossier = _make_dossier() controle = _make_controle() prompt = _build_cpam_prompt(dossier, controle, []) assert controle.arg_ucr in prompt assert controle.decision_ucr in prompt def test_prompt_contains_codes_contestes(self): dossier = _make_dossier() controle = _make_controle() prompt = _build_cpam_prompt(dossier, controle, []) assert "DA proposés par UCR : K56.0" in prompt def test_prompt_contains_rag_sources(self): dossier = _make_dossier() controle = _make_controle() sources = [ {"document": "guide_methodo", "page": 64, "extrait": "Texte du guide..."}, {"document": "cim10", "code": "K56.0", "extrait": "Iléus paralytique..."}, ] prompt = _build_cpam_prompt(dossier, controle, sources) assert "Guide Méthodologique MCO 2026" in prompt assert "CIM-10 FR 2026" in prompt assert "page 64" in prompt def test_prompt_contains_three_axes(self): dossier = _make_dossier() controle = _make_controle() prompt = _build_cpam_prompt(dossier, controle, []) assert "AXE MÉDICAL" in prompt assert "AXE ASYMÉTRIE D'INFORMATION" in prompt assert "AXE RÉGLEMENTAIRE" in prompt def test_prompt_contains_traitements_imagerie_when_present(self): dossier = _make_dossier_complet() controle = _make_controle() prompt = _build_cpam_prompt(dossier, controle, []) assert "Augmentin IV 3g/j" in prompt assert "Morphine SC" in prompt assert "Scanner abdominal" in prompt assert "Lithiase cholédocienne confirmée" in prompt assert "HTA" in prompt assert "Diabète type 2" in prompt assert "IMC : 31.2" in prompt def test_prompt_asymetrie_section_when_data_present(self): dossier = _make_dossier_complet() controle = _make_controle() prompt = _build_cpam_prompt(dossier, controle, []) assert "ÉLÉMENTS DU DOSSIER NON TRANSMIS À LA CPAM" in prompt assert "CRP: 180 mg/L (anormale)" in prompt assert "Cholécystectomie (HMFC004)" in prompt def test_prompt_no_asymetrie_section_when_no_data(self): dossier = DossierMedical( source_file="test.pdf", document_type="crh", sejour=Sejour(), diagnostic_principal=Diagnostic(texte="Test", cim10_suggestion="Z00"), ) controle = _make_controle() prompt = _build_cpam_prompt(dossier, controle, []) assert "ÉLÉMENTS DU DOSSIER NON TRANSMIS À LA CPAM" not in prompt def test_prompt_json_format_new_fields(self): dossier = _make_dossier() controle = _make_controle() prompt = _build_cpam_prompt(dossier, controle, []) assert "contre_arguments_medicaux" in prompt assert "contre_arguments_asymetrie" in prompt assert "contre_arguments_reglementaires" in prompt class TestFormatResponse: def test_full_response_new_format(self): parsed = { "analyse_contestation": "La CPAM conteste le DAS K56.0", "points_accord": "Aucun", "contre_arguments_medicaux": "Le guide méthodologique précise...", "contre_arguments_asymetrie": "La biologie montre une CRP à 180...", "contre_arguments_reglementaires": "L'UCR interprète restrictivement...", "references": "Guide métho p.64, CIM-10 K56.0", "conclusion": "Le DAS est justifié", } text = _format_response(parsed) assert "ANALYSE DE LA CONTESTATION" in text assert "CONTRE-ARGUMENTS MÉDICAUX" in text assert "ASYMÉTRIE D'INFORMATION" in text assert "CONTRE-ARGUMENTS RÉGLEMENTAIRES" in text assert "REFERENCES" in text assert "CONCLUSION" in text # "Aucun" ne doit pas générer la section points d'accord assert "POINTS D'ACCORD" not in text # L'ancien champ ne doit pas apparaître assert "CONTRE-ARGUMENTS\n" not in text def test_fallback_old_format(self): """L'ancien champ contre_arguments est toujours géré (réponses en cache).""" parsed = { "analyse_contestation": "Analyse...", "contre_arguments": "Arguments anciens...", "conclusion": "Conclusion...", } text = _format_response(parsed) assert "CONTRE-ARGUMENTS\nArguments anciens..." in text assert "CONCLUSION" in text def test_new_fields_override_fallback(self): """Si les nouveaux champs existent, l'ancien contre_arguments est ignoré.""" parsed = { "contre_arguments_medicaux": "Médicaux...", "contre_arguments": "Ancien fallback...", "conclusion": "Conclusion...", } text = _format_response(parsed) assert "CONTRE-ARGUMENTS MÉDICAUX" in text assert "Ancien fallback" not in text def test_partial_response(self): parsed = { "contre_arguments_medicaux": "Arguments médicaux...", "conclusion": "Conclusion...", } text = _format_response(parsed) assert "CONTRE-ARGUMENTS MÉDICAUX" in text assert "CONCLUSION" in text def test_empty_response(self): text = _format_response({}) assert text == "" class TestGenerateResponse: @patch("src.control.cpam_response.call_ollama") @patch("src.control.cpam_response._search_rag_for_control") def test_generate_success(self, mock_rag, mock_ollama): mock_rag.return_value = [ {"document": "guide_methodo", "page": 64, "extrait": "Texte guide"}, ] mock_ollama.return_value = { "analyse_contestation": "Analyse...", "contre_arguments_medicaux": "Contre-arguments médicaux...", "contre_arguments_asymetrie": "Asymétrie...", "conclusion": "Conclusion...", } dossier = _make_dossier() controle = _make_controle() text, sources = generate_cpam_response(dossier, controle) assert "Contre-arguments médicaux..." in text assert len(sources) == 1 assert sources[0].document == "guide_methodo" mock_ollama.assert_called_once() @patch("src.control.cpam_response.call_ollama") @patch("src.control.cpam_response._search_rag_for_control") def test_generate_ollama_unavailable(self, mock_rag, mock_ollama): mock_rag.return_value = [] mock_ollama.return_value = None dossier = _make_dossier() controle = _make_controle() text, sources = generate_cpam_response(dossier, controle) assert text == "" assert sources == [] class TestSearchRagForControl: """Tests pour la logique de recherche RAG multi-requêtes.""" @patch("src.medical.rag_search.search_similar_cpam") def test_multiple_queries_with_da_ucr(self, mock_search): """Avec da_ucr, on doit avoir au moins 2 requêtes (codes + argument).""" mock_search.return_value = [ {"document": "guide_methodo", "page": 10, "code": None, "score": 0.6, "extrait": "Texte guide"}, ] dossier = _make_dossier() controle = _make_controle() # da_ucr="K56.0" results = _search_rag_for_control(controle, dossier) # Au moins 2 appels : codes contestés + argument CPAM assert mock_search.call_count >= 2 @patch("src.medical.rag_search.search_similar_cpam") def test_query_codes_contains_cma(self, mock_search): """La requête codes contestés doit contenir 'CMA' pour un DA.""" mock_search.return_value = [] dossier = _make_dossier() controle = _make_controle() # da_ucr="K56.0" _search_rag_for_control(controle, dossier) # Premier appel = requête codes first_call_query = mock_search.call_args_list[0][0][0] assert "K56.0" in first_call_query assert "CMA" in first_call_query @patch("src.medical.rag_search.search_similar_cpam") def test_query_argument_contains_titre(self, mock_search): """La requête argument doit contenir le titre du contrôle.""" mock_search.return_value = [] dossier = _make_dossier() controle = _make_controle() _search_rag_for_control(controle, dossier) # Deuxième appel = requête argument second_call_query = mock_search.call_args_list[1][0][0] assert controle.titre in second_call_query @patch("src.medical.rag_search.search_similar_cpam") def test_deduplication_by_document_code_page(self, mock_search): """Les résultats dupliqués (même document/code/page) sont fusionnés.""" # Les deux requêtes retournent le même résultat shared_result = { "document": "guide_methodo", "page": 64, "code": None, "score": 0.55, "extrait": "Texte partagé", } mock_search.return_value = [shared_result.copy()] dossier = _make_dossier() controle = _make_controle() results = _search_rag_for_control(controle, dossier) # Le résultat ne doit apparaître qu'une seule fois malgré les requêtes multiples guide_results = [r for r in results if r["document"] == "guide_methodo" and r.get("page") == 64] assert len(guide_results) == 1 @patch("src.medical.rag_search.search_similar_cpam") def test_dedup_keeps_best_score(self, mock_search): """La déduplication garde le meilleur score.""" def side_effect(query, top_k=8): if "CMA" in query: return [{"document": "cim10", "code": "K56.0", "page": None, "score": 0.5, "extrait": "Iléus"}] else: return [{"document": "cim10", "code": "K56.0", "page": None, "score": 0.7, "extrait": "Iléus"}] mock_search.side_effect = side_effect dossier = _make_dossier() controle = _make_controle() results = _search_rag_for_control(controle, dossier) k56_results = [r for r in results if r.get("code") == "K56.0"] assert len(k56_results) == 1 assert k56_results[0]["score"] == 0.7 @patch("src.medical.rag_search.search_similar_cpam") def test_no_codes_contestes_only_argument_query(self, mock_search): """Sans codes contestés, seule la requête argument est lancée.""" mock_search.return_value = [] dossier = _make_dossier() controle = ControleCPAM( numero_ogc=1, titre="Désaccord sur la durée", arg_ucr="Séjour trop long selon l'UCR.", decision_ucr="Rejet", dp_ucr=None, da_ucr=None, ) _search_rag_for_control(controle, dossier) # Un seul appel : requête argument (pas de codes contestés) assert mock_search.call_count == 1 @patch("src.medical.rag_search.search_similar_cpam") def test_dp_ucr_query_contains_diagnostic_principal(self, mock_search): """Avec dp_ucr, la requête codes mentionne 'diagnostic principal'.""" mock_search.return_value = [] dossier = _make_dossier() controle = ControleCPAM( numero_ogc=2, titre="Désaccord sur le DP", arg_ucr="Le DP devrait être K80.1", decision_ucr="Rejet", dp_ucr="K81.0", da_ucr=None, ) _search_rag_for_control(controle, dossier) first_call_query = mock_search.call_args_list[0][0][0] assert "K81.0" in first_call_query assert "diagnostic principal" in first_call_query @patch("src.medical.rag_search.search_similar_cpam") def test_max_10_results(self, mock_search): """Le résultat final est limité à 10 entrées.""" mock_search.return_value = [ {"document": "guide_methodo", "page": i, "code": None, "score": 0.9 - i * 0.01, "extrait": f"Texte {i}"} for i in range(8) ] dossier = _make_dossier() controle = _make_controle() results = _search_rag_for_control(controle, dossier) assert len(results) <= 10 @patch("src.medical.rag_search.search_similar_cpam") def test_clinical_query_when_das_match(self, mock_search): """Requête clinique lancée quand da_ucr matche un DAS du dossier.""" mock_search.return_value = [] dossier = _make_dossier() # DAS K56.0 "Iléus réflexe" controle = _make_controle() # da_ucr="K56.0" _search_rag_for_control(controle, dossier) # 3 appels : codes + argument + clinique assert mock_search.call_count == 3 third_call_query = mock_search.call_args_list[2][0][0] assert "Iléus réflexe" in third_call_query assert "Cholécystite aiguë" in third_call_query