Spaces:
Running
Running
File size: 3,274 Bytes
b008a5a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 | """
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)
|