#!/usr/bin/env python3 """ Test Final du Backend Visual Workflow Builder Auteur : Dom, Alice, Kiro - 08 janvier 2026 Ce script valide que toutes les corrections ont été appliquées et que le backend fonctionne correctement. """ import os import sys import time import json import subprocess import signal from pathlib import Path from typing import Optional def print_section(title: str): """Affiche une section avec formatage.""" print(f"\n{'='*60}") print(f" {title}") print(f"{'='*60}") def print_subsection(title: str): """Affiche une sous-section.""" print(f"\n{'-'*40}") print(f" {title}") print(f"{'-'*40}") def check_files_exist(): """Vérifie que tous les fichiers nécessaires existent.""" print_subsection("Vérification des Fichiers") required_files = [ "visual_workflow_builder/backend/app_lightweight.py", "visual_workflow_builder/backend/start_fast.sh", "visual_workflow_builder/backend/test_backend.py", "visual_workflow_builder/backend/models.py", "visual_workflow_builder/backend/services/serialization.py", "visual_workflow_builder/backend/api/workflows.py", "visual_workflow_builder/backend/api/errors.py" ] all_exist = True for file_path in required_files: if Path(file_path).exists(): print(f"✅ {file_path}") else: print(f"❌ {file_path} - MANQUANT") all_exist = False return all_exist def test_import_models(): """Test l'import des modèles.""" print_subsection("Test d'Import des Modèles") try: # Changer vers le répertoire backend original_cwd = os.getcwd() backend_path = Path("visual_workflow_builder/backend") os.chdir(backend_path) sys.path.insert(0, str(backend_path.absolute())) # Test d'import from models import VisualWorkflow, VisualNode, VisualEdge, Variable print("✅ Import models réussi") # Test de création d'un workflow workflow = VisualWorkflow( id="test_wf_001", name="Test Workflow", description="Workflow de test" ) # Test de sérialisation workflow_dict = workflow.to_dict() print("✅ Sérialisation workflow réussie") # Test de désérialisation workflow_restored = VisualWorkflow.from_dict(workflow_dict) print("✅ Désérialisation workflow réussie") # Restaurer le répertoire os.chdir(original_cwd) sys.path.remove(str(backend_path.absolute())) return True except Exception as e: print(f"❌ Erreur import models: {e}") # Restaurer le répertoire os.chdir(original_cwd) if str(backend_path.absolute()) in sys.path: sys.path.remove(str(backend_path.absolute())) return False def start_backend_server() -> Optional[subprocess.Popen]: """Démarre le serveur backend.""" print_subsection("Démarrage du Serveur Backend") try: # Trouver un port libre import socket sock = socket.socket() sock.bind(('', 0)) port = sock.getsockname()[1] sock.close() print(f"🚀 Démarrage sur le port {port}") # Démarrer le serveur env = os.environ.copy() env['PORT'] = str(port) process = subprocess.Popen( [sys.executable, 'app_lightweight.py'], cwd='visual_workflow_builder/backend', env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) # Attendre que le serveur soit prêt print("⏳ Attente du démarrage...") time.sleep(3) # Vérifier que le processus est toujours en vie if process.poll() is None: print(f"✅ Serveur démarré (PID: {process.pid})") return process, port else: stdout, stderr = process.communicate() print(f"❌ Échec du démarrage") print(f"STDOUT: {stdout}") print(f"STDERR: {stderr}") return None, None except Exception as e: print(f"❌ Erreur démarrage serveur: {e}") return None, None def test_api_endpoints(port: int): """Test les endpoints de l'API.""" print_subsection("Test des Endpoints API") base_url = f"http://localhost:{port}" try: import urllib.request import urllib.parse # Test 1: Health check print("🧪 Test health check...") try: with urllib.request.urlopen(f"{base_url}/health", timeout=5) as response: if response.status == 200: data = json.loads(response.read().decode()) print(f"✅ Health check OK - {data.get('status')}") else: print(f"❌ Health check échoué: {response.status}") return False except Exception as e: print(f"❌ Erreur health check: {e}") return False # Test 2: Liste des workflows print("🧪 Test liste workflows...") try: with urllib.request.urlopen(f"{base_url}/api/workflows", timeout=5) as response: if response.status == 200: workflows = json.loads(response.read().decode()) print(f"✅ Liste workflows OK ({len(workflows)} workflows)") else: print(f"❌ Liste workflows échoué: {response.status}") return False except Exception as e: print(f"❌ Erreur liste workflows: {e}") return False # Test 3: Création d'un workflow print("🧪 Test création workflow...") try: test_workflow = { "name": "Test Workflow Final", "description": "Workflow de test automatique final", "created_by": "test_script_final" } data = json.dumps(test_workflow).encode() req = urllib.request.Request( f"{base_url}/api/workflows", data=data, headers={'Content-Type': 'application/json'}, method='POST' ) with urllib.request.urlopen(req, timeout=5) as response: if response.status == 201: created_workflow = json.loads(response.read().decode()) workflow_id = created_workflow['id'] print(f"✅ Création workflow OK (ID: {workflow_id})") # Test 4: Récupération du workflow créé print("🧪 Test récupération workflow...") with urllib.request.urlopen(f"{base_url}/api/workflows/{workflow_id}", timeout=5) as get_response: if get_response.status == 200: retrieved_workflow = json.loads(get_response.read().decode()) if retrieved_workflow['name'] == test_workflow['name']: print("✅ Récupération workflow OK") return True else: print("❌ Données workflow incorrectes") return False else: print(f"❌ Récupération workflow échoué: {get_response.status}") return False else: print(f"❌ Création workflow échoué: {response.status}") error_data = response.read().decode() print(f" Erreur: {error_data}") return False except Exception as e: print(f"❌ Erreur création workflow: {e}") return False except ImportError: print("❌ urllib non disponible pour les tests HTTP") return False def stop_server(process: subprocess.Popen): """Arrête le serveur.""" print_subsection("Arrêt du Serveur") try: # Envoyer SIGTERM process.terminate() # Attendre un peu try: process.wait(timeout=5) print("✅ Serveur arrêté proprement") except subprocess.TimeoutExpired: # Forcer l'arrêt process.kill() process.wait() print("⚠️ Serveur forcé à s'arrêter") except Exception as e: print(f"❌ Erreur arrêt serveur: {e}") def test_performance(): """Test les performances de démarrage.""" print_subsection("Test de Performance") try: start_time = time.time() # Démarrer et arrêter rapidement le serveur env = os.environ.copy() env['PORT'] = '0' # Port automatique process = subprocess.Popen( [sys.executable, 'app_lightweight.py'], cwd='visual_workflow_builder/backend', env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) # Attendre un peu puis arrêter time.sleep(1) process.terminate() process.wait(timeout=5) startup_time = time.time() - start_time if startup_time < 5.0: print(f"✅ Performance OK - Démarrage en {startup_time:.2f}s") return True else: print(f"⚠️ Performance lente - Démarrage en {startup_time:.2f}s") return False except Exception as e: print(f"❌ Erreur test performance: {e}") return False def main(): """Fonction principale de test.""" print_section("TEST FINAL BACKEND VISUAL WORKFLOW BUILDER") print("Auteur : Dom, Alice, Kiro - 08 janvier 2026") # Tests tests_passed = 0 total_tests = 5 # Test 1: Fichiers if check_files_exist(): tests_passed += 1 print("✅ Test 1/5 : Fichiers - PASSÉ") else: print("❌ Test 1/5 : Fichiers - ÉCHOUÉ") # Test 2: Import des modèles if test_import_models(): tests_passed += 1 print("✅ Test 2/5 : Import modèles - PASSÉ") else: print("❌ Test 2/5 : Import modèles - ÉCHOUÉ") # Test 3: Performance if test_performance(): tests_passed += 1 print("✅ Test 3/5 : Performance - PASSÉ") else: print("❌ Test 3/5 : Performance - ÉCHOUÉ") # Test 4 & 5: Serveur et API server_result = start_backend_server() if server_result[0] is not None: process, port = server_result tests_passed += 1 print("✅ Test 4/5 : Démarrage serveur - PASSÉ") # Test API if test_api_endpoints(port): tests_passed += 1 print("✅ Test 5/5 : API endpoints - PASSÉ") else: print("❌ Test 5/5 : API endpoints - ÉCHOUÉ") # Arrêter le serveur stop_server(process) else: print("❌ Test 4/5 : Démarrage serveur - ÉCHOUÉ") print("❌ Test 5/5 : API endpoints - IGNORÉ") # Résumé print_section("RÉSUMÉ DES TESTS") success_rate = (tests_passed / total_tests) * 100 print(f"Tests passés : {tests_passed}/{total_tests}") print(f"Taux de réussite : {success_rate:.1f}%") if tests_passed == total_tests: print("🎉 TOUS LES TESTS SONT PASSÉS!") print("✅ Le backend Visual Workflow Builder est 100% opérationnel") print("") print("Pour utiliser le backend:") print("1. cd visual_workflow_builder/backend") print("2. ./start_fast.sh") return True else: print("⚠️ CERTAINS TESTS ONT ÉCHOUÉ") print("❌ Des corrections supplémentaires peuvent être nécessaires") return False if __name__ == "__main__": success = main() sys.exit(0 if success else 1)