Spaces:
Running
Running
| """ | |
| Test fonctionnel pour l'API FastAPI de prédiction ML. | |
| Ce test démarre réellement le serveur et fait une vraie requête HTTP sur /health. | |
| """ | |
| import subprocess | |
| import time | |
| import sys | |
| import httpx | |
| import signal | |
| import os | |
| def wait_for_server(url: str, timeout: int = 30) -> bool: | |
| """Attend que le serveur soit prêt.""" | |
| start_time = time.time() | |
| while time.time() - start_time < timeout: | |
| try: | |
| response = httpx.get(url, timeout=2) | |
| if response.status_code == 200: | |
| return True | |
| except httpx.RequestError: | |
| pass | |
| time.sleep(0.5) | |
| return False | |
| def run_functional_tests(): | |
| """Exécute le test fonctionnel sur l'endpoint /health.""" | |
| base_url = "http://127.0.0.1:7860" | |
| server_process = None | |
| try: | |
| # Démarrer le serveur uvicorn | |
| print("🚀 Démarrage du serveur uvicorn...") | |
| server_process = subprocess.Popen( | |
| ["uvicorn", "main:app", "--host", "127.0.0.1", "--port", "7860"], | |
| stdout=subprocess.PIPE, | |
| stderr=subprocess.PIPE, | |
| preexec_fn=os.setsid | |
| ) | |
| # Attendre que le serveur soit prêt | |
| print("⏳ Attente du serveur...") | |
| if not wait_for_server(f"{base_url}/health"): | |
| print("❌ Le serveur n'a pas démarré dans le délai imparti") | |
| return False | |
| print("✅ Serveur démarré avec succès") | |
| # ============================================ | |
| # TEST: Endpoint /health | |
| # ============================================ | |
| print("\n📋 Test fonctionnel: Endpoint /health") | |
| response = httpx.get(f"{base_url}/health") | |
| assert response.status_code == 200, f"Expected 200, got {response.status_code}" | |
| data = response.json() | |
| assert data["status"] == "ok", f"Expected status 'ok', got {data['status']}" | |
| assert data["model_loaded"] is True, "Model should be loaded" | |
| assert "available_items" in data, "Clé 'available_items' absente" | |
| assert "available_areas" in data, "Clé 'available_areas' absente" | |
| assert data["available_items"] > 0, "Should have items" | |
| assert data["available_areas"] > 0, "Should have areas" | |
| print(f" ✅ Status: {data['status']}") | |
| print(f" ✅ Modèle chargé: {data['model_loaded']}") | |
| print(f" ✅ Cultures disponibles: {data['available_items']}") | |
| print(f" ✅ Pays disponibles: {data['available_areas']}") | |
| # ============================================ | |
| # Résultat final | |
| # ============================================ | |
| print("\n" + "=" * 50) | |
| print("🎉 TEST FONCTIONNEL /health RÉUSSI !") | |
| print("=" * 50) | |
| return True | |
| except AssertionError as e: | |
| print(f"\n❌ ÉCHEC DU TEST: {e}") | |
| return False | |
| except Exception as e: | |
| print(f"\n❌ ERREUR: {e}") | |
| return False | |
| finally: | |
| # Arrêter le serveur | |
| if server_process: | |
| print("\n🛑 Arrêt du serveur...") | |
| os.killpg(os.getpgid(server_process.pid), signal.SIGTERM) | |
| server_process.wait() | |
| if __name__ == "__main__": | |
| success = run_functional_tests() | |
| sys.exit(0 if success else 1) | |