test_api.py

#!/usr/bin/env python3
"""
Script de test local pour l'application Hugging Face
Teste l'API et l'interface avant déploiement
"""

import asyncio
import httpx
import json
from pathlib import Path

# Configuration
API_BASE_URL = "http://localhost:7860"
TEST_DATA_FILE = Path("data/documents.json")

# Couleurs pour le terminal
class Colors:
    GREEN = '\033[92m'
    RED = '\033[91m'
    YELLOW = '\033[93m'
    BLUE = '\033[94m'
    END = '\033[0m'

def print_success(msg):
    print(f"{Colors.GREEN}✅ {msg}{Colors.END}")

def print_error(msg):
    print(f"{Colors.RED}❌ {msg}{Colors.END}")

def print_info(msg):
    print(f"{Colors.BLUE}ℹ️  {msg}{Colors.END}")

def print_warning(msg):
    print(f"{Colors.YELLOW}⚠️  {msg}{Colors.END}")

async def test_health():
    """Test du health check"""
    print_info("Test du health check...")
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(f"{API_BASE_URL}/api/health", timeout=5.0)
            if response.status_code == 200:
                data = response.json()
                print_success(f"Health check OK - {data.get('documents_loaded', 0)} documents chargés")
                return True
            else:
                print_error(f"Health check échoué - Status: {response.status_code}")
                return False
    except Exception as e:
        print_error(f"Erreur health check: {e}")
        return False

async def test_search_post():
    """Test de la recherche (POST)"""
    print_info("Test de la recherche (POST)...")
    try:
        async with httpx.AsyncClient() as client:
            payload = {
                "query": "togo",
                "limit": 5,
                "fuzzy": True
            }
            response = await client.post(
                f"{API_BASE_URL}/api/search",
                json=payload,
                timeout=10.0
            )
            if response.status_code == 200:
                data = response.json()
                print_success(f"Recherche OK - {data['total']} résultats en {data['execution_time_ms']}ms")
                if data['results']:
                    print_info(f"Premier résultat: {data['results'][0].get('titre', 'N/A')[:50]}...")
                return True
            else:
                print_error(f"Recherche échouée - Status: {response.status_code}")
                return False
    except Exception as e:
        print_error(f"Erreur recherche POST: {e}")
        return False

async def test_search_get():
    """Test de la recherche (GET)"""
    print_info("Test de la recherche (GET)...")
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{API_BASE_URL}/api/search?q=économie&pays=Togo&limit=5",
                timeout=10.0
            )
            if response.status_code == 200:
                data = response.json()
                print_success(f"Recherche GET OK - {data['total']} résultats")
                return True
            else:
                print_error(f"Recherche GET échouée - Status: {response.status_code}")
                return False
    except Exception as e:
        print_error(f"Erreur recherche GET: {e}")
        return False

async def test_fuzzy_search():
    """Test de la recherche fuzzy (avec fautes)"""
    print_info("Test de la recherche fuzzy (tolérance aux fautes)...")
    try:
        async with httpx.AsyncClient() as client:
            # Recherche avec faute de frappe volontaire
            payload = {
                "query": "economie",  # Sans accent
                "limit": 5,
                "fuzzy": True
            }
            response = await client.post(
                f"{API_BASE_URL}/api/search",
                json=payload,
                timeout=10.0
            )
            if response.status_code == 200:
                data = response.json()
                if data['total'] > 0:
                    print_success(f"Fuzzy search OK - {data['total']} résultats malgré la faute")
                else:
                    print_warning("Fuzzy search OK mais aucun résultat (base vide?)")
                return True
            else:
                print_error(f"Fuzzy search échouée - Status: {response.status_code}")
                return False
    except Exception as e:
        print_error(f"Erreur fuzzy search: {e}")
        return False

async def test_stats():
    """Test des statistiques"""
    print_info("Test des statistiques...")
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(f"{API_BASE_URL}/api/stats", timeout=5.0)
            if response.status_code == 200:
                data = response.json()
                print_success(f"Stats OK - {data['total_documents']} documents")
                if data['pays']:
                    print_info(f"Pays: {', '.join(data['pays'].keys())}")
                return True
            else:
                print_error(f"Stats échouées - Status: {response.status_code}")
                return False
    except Exception as e:
        print_error(f"Erreur stats: {e}")
        return False

async def test_documents():
    """Test de la liste des documents"""
    print_info("Test de la liste des documents...")
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{API_BASE_URL}/api/documents?skip=0&limit=5",
                timeout=5.0
            )
            if response.status_code == 200:
                data = response.json()
                print_success(f"Liste documents OK - {data['total']} total, {len(data['documents'])} retournés")
                return True
            else:
                print_error(f"Liste documents échouée - Status: {response.status_code}")
                return False
    except Exception as e:
        print_error(f"Erreur liste documents: {e}")
        return False

def check_data_file():
    """Vérifie si le fichier de données existe"""
    print_info("Vérification du fichier de données...")
    if TEST_DATA_FILE.exists():
        try:
            with open(TEST_DATA_FILE, 'r', encoding='utf-8') as f:
                data = json.load(f)
                print_success(f"Fichier de données trouvé - {len(data)} documents")
                return True
        except Exception as e:
            print_error(f"Erreur lecture fichier: {e}")
            return False
    else:
        print_warning("Fichier de données non trouvé - La base sera vide")
        print_info("Lancez 'python scrape_togo_massive.py' pour scraper des données")
        return False

async def run_all_tests():
    """Lance tous les tests"""
    print("\n" + "="*60)
    print("🧪 TESTS DE L'APPLICATION SCRAP-DJI")
    print("="*60 + "\n")
    
    # Vérification préliminaire
    check_data_file()
    print()
    
    # Tests API
    tests = [
        ("Health Check", test_health),
        ("Recherche POST", test_search_post),
        ("Recherche GET", test_search_get),
        ("Recherche Fuzzy", test_fuzzy_search),
        ("Statistiques", test_stats),
        ("Liste Documents", test_documents),
    ]
    
    results = []
    for name, test_func in tests:
        result = await test_func()
        results.append((name, result))
        print()
    
    # Résumé
    print("="*60)
    print("📊 RÉSUMÉ DES TESTS")
    print("="*60)
    
    passed = sum(1 for _, result in results if result)
    total = len(results)
    
    for name, result in results:
        status = "✅ PASS" if result else "❌ FAIL"
        print(f"{status} - {name}")
    
    print()
    print(f"Résultat: {passed}/{total} tests réussis")
    
    if passed == total:
        print_success("Tous les tests sont passés ! 🎉")
        print_info("L'application est prête pour le déploiement sur Hugging Face")
    else:
        print_warning(f"{total - passed} test(s) échoué(s)")
        print_info("Vérifiez que l'application est bien lancée avec 'python app.py'")
    
    print()

if __name__ == "__main__":
    print_info("Assurez-vous que l'application est lancée avec 'python app.py'")
    print_info("Attendez quelques secondes puis appuyez sur Entrée pour continuer...")
    input()
    
    asyncio.run(run_all_tests())