""" Tests complets pour l'API Notariale 4NK """ import pytest import asyncio import json from fastapi.testclient import TestClient from unittest.mock import Mock, patch, AsyncMock import tempfile import os # Import de l'application import sys sys.path.append('services/host_api') from app import app client = TestClient(app) class TestNotaryAPI: """Tests pour l'API Notariale""" def test_health_check(self): """Test du health check""" response = client.get("/api/health") assert response.status_code == 200 data = response.json() assert "status" in data assert data["status"] == "healthy" def test_upload_document_success(self): """Test d'upload de document réussi""" # Création d'un fichier PDF de test with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp_file: tmp_file.write(b"%PDF-1.4\n1 0 obj\n<<\n/Type /Catalog\n/Pages 2 0 R\n>>\nendobj\n") tmp_file.flush() with open(tmp_file.name, "rb") as f: response = client.post( "/api/notary/upload", files={"file": ("test.pdf", f, "application/pdf")}, data={ "id_dossier": "TEST-001", "etude_id": "E-001", "utilisateur_id": "U-123", "source": "upload" } ) os.unlink(tmp_file.name) assert response.status_code == 200 data = response.json() assert "document_id" in data assert data["status"] == "queued" assert "message" in data def test_upload_document_invalid_type(self): """Test d'upload avec type de fichier invalide""" with tempfile.NamedTemporaryFile(suffix=".txt", delete=False) as tmp_file: tmp_file.write(b"Ceci est un fichier texte") tmp_file.flush() with open(tmp_file.name, "rb") as f: response = client.post( "/api/notary/upload", files={"file": ("test.txt", f, "text/plain")}, data={ "id_dossier": "TEST-001", "etude_id": "E-001", "utilisateur_id": "U-123" } ) os.unlink(tmp_file.name) assert response.status_code == 415 data = response.json() assert "Type de fichier non supporté" in data["detail"] def test_upload_document_missing_fields(self): """Test d'upload avec champs manquants""" with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp_file: tmp_file.write(b"%PDF-1.4") tmp_file.flush() with open(tmp_file.name, "rb") as f: response = client.post( "/api/notary/upload", files={"file": ("test.pdf", f, "application/pdf")}, data={ "id_dossier": "TEST-001" # etude_id et utilisateur_id manquants } ) os.unlink(tmp_file.name) assert response.status_code == 422 # Validation error def test_get_document_status(self): """Test de récupération du statut d'un document""" # Mock d'un document existant with patch('services.host_api.routes.notary_documents.get_document_status') as mock_status: mock_status.return_value = { "document_id": "test-123", "status": "processing", "progress": 50, "current_step": "extraction_entites" } response = client.get("/api/notary/document/test-123/status") assert response.status_code == 200 data = response.json() assert data["status"] == "processing" assert data["progress"] == 50 def test_get_document_analysis(self): """Test de récupération de l'analyse d'un document""" # Mock d'une analyse complète with patch('services.host_api.routes.notary_documents.get_document_analysis') as mock_analysis: mock_analysis.return_value = { "document_id": "test-123", "type_detecte": "acte_vente", "confiance_classification": 0.95, "texte_extrait": "Texte de test", "entites_extraites": { "identites": [ {"nom": "DUPONT", "prenom": "Jean", "type": "vendeur"} ] }, "verifications_externes": { "cadastre": {"status": "verified", "confidence": 0.9} }, "score_vraisemblance": 0.92, "avis_synthese": "Document cohérent", "recommandations": ["Vérifier l'identité"], "timestamp_analyse": "2025-01-09 10:30:00" } response = client.get("/api/notary/document/test-123/analysis") assert response.status_code == 200 data = response.json() assert data["type_detecte"] == "acte_vente" assert data["score_vraisemblance"] == 0.92 def test_list_documents(self): """Test de la liste des documents""" with patch('services.host_api.routes.notary_documents.list_documents') as mock_list: mock_list.return_value = { "documents": [ { "document_id": "test-123", "filename": "test.pdf", "status": "completed", "created_at": "2025-01-09T10:00:00" } ], "total": 1, "limit": 50, "offset": 0 } response = client.get("/api/notary/documents") assert response.status_code == 200 data = response.json() assert len(data["documents"]) == 1 assert data["total"] == 1 def test_get_processing_stats(self): """Test des statistiques de traitement""" with patch('services.host_api.routes.notary_documents.get_processing_stats') as mock_stats: mock_stats.return_value = { "documents_traites": 100, "documents_en_cours": 5, "taux_reussite": 0.98, "temps_moyen_traitement": 90, "types_documents": { "acte_vente": 50, "acte_donation": 20, "cni": 30 } } response = client.get("/api/notary/stats") assert response.status_code == 200 data = response.json() assert data["documents_traites"] == 100 assert data["taux_reussite"] == 0.98 class TestOCRProcessor: """Tests pour le processeur OCR""" @pytest.mark.asyncio async def test_ocr_processing(self): """Test du traitement OCR""" from services.host_api.utils.ocr_processor import OCRProcessor processor = OCRProcessor() # Mock d'une image de test with patch('cv2.imread') as mock_imread: mock_imread.return_value = None # Simuler une image with patch('pytesseract.image_to_string') as mock_tesseract: mock_tesseract.return_value = "Texte extrait par OCR" with patch('pytesseract.image_to_data') as mock_data: mock_data.return_value = { 'text': ['Texte', 'extrait'], 'conf': [90, 85] } # Test avec un fichier inexistant (sera mocké) result = await processor.process_document("test_image.jpg") assert "text" in result assert result["confidence"] > 0 class TestDocumentClassifier: """Tests pour le classificateur de documents""" @pytest.mark.asyncio async def test_classification_by_rules(self): """Test de classification par règles""" from services.host_api.utils.document_classifier import DocumentClassifier classifier = DocumentClassifier() # Texte d'un acte de vente text = """ ACTE DE VENTE Entre les soussignés : VENDEUR : M. DUPONT Jean ACHETEUR : Mme MARTIN Marie Prix de vente : 250 000 euros """ result = classifier._classify_by_rules(text) assert result["type"] == "acte_vente" assert result["confidence"] > 0 @pytest.mark.asyncio async def test_classification_by_llm(self): """Test de classification par LLM""" from services.host_api.utils.document_classifier import DocumentClassifier classifier = DocumentClassifier() # Mock de la réponse LLM with patch.object(classifier.llm_client, 'generate_response') as mock_llm: mock_llm.return_value = ''' { "type": "acte_vente", "confidence": 0.95, "reasoning": "Document contient vendeur, acheteur et prix", "key_indicators": ["vendeur", "acheteur", "prix"] } ''' result = await classifier._classify_by_llm("Test document", None) assert result["type"] == "acte_vente" assert result["confidence"] == 0.95 class TestEntityExtractor: """Tests pour l'extracteur d'entités""" @pytest.mark.asyncio async def test_entity_extraction(self): """Test d'extraction d'entités""" from services.host_api.utils.entity_extractor import EntityExtractor extractor = EntityExtractor() text = """ VENDEUR : M. DUPONT Jean, né le 15/03/1980 ACHETEUR : Mme MARTIN Marie Adresse : 123 rue de la Paix, 75001 Paris Prix : 250 000 euros """ result = await extractor.extract_entities(text, "acte_vente") assert "identites" in result assert "adresses" in result assert "montants" in result assert len(result["identites"]) > 0 class TestExternalAPIs: """Tests pour les APIs externes""" @pytest.mark.asyncio async def test_cadastre_verification(self): """Test de vérification cadastre""" from services.host_api.utils.external_apis import ExternalAPIManager api_manager = ExternalAPIManager() # Mock de la réponse API with patch('aiohttp.ClientSession.get') as mock_get: mock_response = AsyncMock() mock_response.status = 200 mock_response.json.return_value = { "features": [ { "properties": { "id": "1234", "section": "A", "numero": "1" } } ] } mock_get.return_value.__aenter__.return_value = mock_response result = await api_manager.verify_cadastre("123 rue de la Paix, Paris") assert result.status == "verified" assert result.confidence > 0 class TestVerificationEngine: """Tests pour le moteur de vérification""" def test_credibility_score_calculation(self): """Test du calcul du score de vraisemblance""" from services.host_api.utils.verification_engine import VerificationEngine engine = VerificationEngine() # Données de test ocr_result = {"confidence": 85, "word_count": 100} classification_result = {"confidence": 0.9, "type": "acte_vente"} entities = { "identites": [{"confidence": 0.8}], "adresses": [{"confidence": 0.9}] } verifications = { "cadastre": {"status": "verified", "confidence": 0.9} } # Test synchrone (le calcul est synchrone) score = asyncio.run(engine.calculate_credibility_score( ocr_result, classification_result, entities, verifications )) assert 0 <= score <= 1 assert score > 0.5 # Score raisonnable pour des données de test class TestLLMClient: """Tests pour le client LLM""" @pytest.mark.asyncio async def test_llm_generation(self): """Test de génération LLM""" from services.host_api.utils.llm_client import LLMClient client = LLMClient() # Mock de la réponse Ollama with patch('aiohttp.ClientSession.post') as mock_post: mock_response = AsyncMock() mock_response.status = 200 mock_response.json.return_value = { "response": "Réponse de test du LLM" } mock_post.return_value.__aenter__.return_value = mock_response result = await client.generate_response("Test prompt") assert "Réponse de test du LLM" in result # Tests d'intégration class TestIntegration: """Tests d'intégration""" def test_full_pipeline_simulation(self): """Test de simulation du pipeline complet""" # Ce test simule le pipeline complet sans les vraies APIs externes # 1. Upload with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp_file: tmp_file.write(b"%PDF-1.4") tmp_file.flush() with open(tmp_file.name, "rb") as f: upload_response = client.post( "/api/notary/upload", files={"file": ("test.pdf", f, "application/pdf")}, data={ "id_dossier": "INTEGRATION-001", "etude_id": "E-001", "utilisateur_id": "U-123" } ) os.unlink(tmp_file.name) assert upload_response.status_code == 200 document_id = upload_response.json()["document_id"] # 2. Statut (simulé) with patch('services.host_api.routes.notary_documents.get_document_status') as mock_status: mock_status.return_value = { "document_id": document_id, "status": "completed", "progress": 100 } status_response = client.get(f"/api/notary/document/{document_id}/status") assert status_response.status_code == 200 # 3. Analyse (simulée) with patch('services.host_api.routes.notary_documents.get_document_analysis') as mock_analysis: mock_analysis.return_value = { "document_id": document_id, "type_detecte": "acte_vente", "score_vraisemblance": 0.85, "avis_synthese": "Document analysé avec succès" } analysis_response = client.get(f"/api/notary/document/{document_id}/analysis") assert analysis_response.status_code == 200 if __name__ == "__main__": pytest.main([__file__, "-v"])