#!/usr/bin/env python3 """ Test de sérialisation - Visual Workflow Builder Test manuel du système de sérialisation complet. """ import sys import json from pathlib import Path sys.path.insert(0, '.') from services.serialization import ( WorkflowSerializer, WorkflowDatabase, SerializationError, ValidationError, create_empty_workflow ) from models.visual_workflow import ( VisualNode, VisualEdge, Variable, Position, Size, Port ) def test_id_generation(): """Test de génération d'ID unique""" print("🧪 Test 1: Génération d'ID unique") ids = set() for _ in range(100): wf_id = WorkflowSerializer.generate_workflow_id() assert wf_id not in ids, "ID dupliqué détecté!" ids.add(wf_id) assert wf_id.startswith('wf_'), f"Format d'ID invalide: {wf_id}" print("✅ 100 IDs uniques générés avec succès") def test_empty_workflow_creation(): """Test de création de workflow vide""" print("\n🧪 Test 2: Création de workflow vide") workflow = create_empty_workflow( name="Test Workflow", description="Un workflow de test", created_by="test_user" ) assert workflow.id.startswith('wf_'), "ID invalide" assert workflow.name == "Test Workflow" assert workflow.description == "Un workflow de test" assert workflow.created_by == "test_user" assert len(workflow.nodes) == 0 assert len(workflow.edges) == 0 assert len(workflow.variables) == 0 print(f"✅ Workflow créé: {workflow.id}") def test_serialization_json(): """Test de sérialisation JSON""" print("\n🧪 Test 3: Sérialisation JSON") workflow = create_empty_workflow("Test JSON") # Ajouter un node workflow.nodes.append(VisualNode( id="click_1", type="click", position=Position(100, 100), size=Size(200, 80), parameters={'target': 'button'}, input_ports=[Port('in', 'Input', 'input')], output_ports=[Port('out', 'Output', 'output')] )) # Sérialiser json_str = WorkflowSerializer.serialize(workflow, format='json') assert isinstance(json_str, str) assert len(json_str) > 0 # Vérifier que c'est du JSON valide data = json.loads(json_str) assert data['id'] == workflow.id assert data['name'] == "Test JSON" assert len(data['nodes']) == 1 assert '_serialization' in data print(f"✅ Sérialisation JSON réussie ({len(json_str)} caractères)") def test_deserialization_json(): """Test de désérialisation JSON""" print("\n🧪 Test 4: Désérialisation JSON") # Créer et sérialiser original = create_empty_workflow("Test Deserialize") original.nodes.append(VisualNode( id="type_1", type="type", position=Position(200, 200), size=Size(200, 80), parameters={'target': 'input', 'text': 'Hello'}, input_ports=[Port('in', 'Input', 'input')], output_ports=[Port('out', 'Output', 'output')] )) json_str = WorkflowSerializer.serialize(original, format='json') # Désérialiser restored = WorkflowSerializer.deserialize(json_str, format='json') assert restored.id == original.id assert restored.name == original.name assert len(restored.nodes) == len(original.nodes) assert restored.nodes[0].id == "type_1" assert restored.nodes[0].type == "type" assert restored.nodes[0].parameters['text'] == 'Hello' print("✅ Désérialisation JSON réussie") def test_round_trip(): """Test de round-trip (sérialisation + désérialisation)""" print("\n🧪 Test 5: Round-trip complet") # Créer un workflow complexe workflow = create_empty_workflow("Complex Workflow") # Ajouter des nodes workflow.nodes.extend([ VisualNode( id="click_1", type="click", position=Position(100, 100), size=Size(200, 80), parameters={'target': 'button'}, input_ports=[Port('in', 'Input', 'input')], output_ports=[Port('out', 'Output', 'output')] ), VisualNode( id="type_1", type="type", position=Position(400, 100), size=Size(200, 80), parameters={'target': 'input', 'text': 'Test'}, input_ports=[Port('in', 'Input', 'input')], output_ports=[Port('out', 'Output', 'output')] ) ]) # Ajouter un edge workflow.edges.append(VisualEdge( id="edge_1", source="click_1", target="type_1", source_port="out", target_port="in" )) # Ajouter une variable workflow.variables.append(Variable( name="username", type="string", value="test_user" )) # Round-trip json_str = WorkflowSerializer.serialize(workflow, format='json') restored = WorkflowSerializer.deserialize(json_str, format='json') # Vérifications assert restored.id == workflow.id assert len(restored.nodes) == 2 assert len(restored.edges) == 1 assert len(restored.variables) == 1 assert restored.variables[0].name == "username" print("✅ Round-trip réussi avec 2 nodes, 1 edge, 1 variable") def test_validation_errors(): """Test de détection d'erreurs de validation""" print("\n🧪 Test 6: Validation d'erreurs") workflow = create_empty_workflow("Invalid Workflow") # Ajouter un edge avec des nodes inexistants workflow.edges.append(VisualEdge( id="edge_1", source="nonexistent_1", target="nonexistent_2", source_port="out", target_port="in" )) # La validation devrait échouer errors = workflow.validate() assert len(errors) > 0, "Des erreurs auraient dû être détectées" print(f"✅ {len(errors)} erreurs détectées comme prévu:") for error in errors: print(f" - {error}") def test_database_operations(): """Test des opérations de base de données""" print("\n🧪 Test 7: Opérations de base de données") # Créer une DB temporaire db = WorkflowDatabase("test_data/workflows") # Créer et sauvegarder un workflow workflow = create_empty_workflow("DB Test") workflow.nodes.append(VisualNode( id="wait_1", type="wait", position=Position(300, 300), size=Size(200, 80), parameters={'duration': 1000}, input_ports=[Port('in', 'Input', 'input')], output_ports=[Port('out', 'Output', 'output')] )) db.save(workflow) print(f"✅ Workflow sauvegardé: {workflow.id}") # Charger le workflow loaded = db.load(workflow.id) assert loaded is not None assert loaded.id == workflow.id assert len(loaded.nodes) == 1 print(f"✅ Workflow chargé: {loaded.id}") # Lister tous les workflows all_workflows = db.list_all() assert len(all_workflows) >= 1 print(f"✅ {len(all_workflows)} workflow(s) dans la DB") # Supprimer le workflow deleted = db.delete(workflow.id) assert deleted == True print(f"✅ Workflow supprimé: {workflow.id}") # Vérifier qu'il n'existe plus assert db.load(workflow.id) is None print("✅ Vérification de suppression OK") def test_file_persistence(): """Test de persistance dans des fichiers""" print("\n🧪 Test 8: Persistance fichier") workflow = create_empty_workflow("File Test") filepath = Path("test_data/test_workflow.json") # Sauvegarder WorkflowSerializer.save_to_file(workflow, filepath, format='json') assert filepath.exists() print(f"✅ Fichier créé: {filepath}") # Charger loaded = WorkflowSerializer.load_from_file(filepath, format='json') assert loaded.id == workflow.id print(f"✅ Fichier chargé: {loaded.id}") # Nettoyer filepath.unlink() print("✅ Fichier nettoyé") def main(): """Exécute tous les tests""" print("=" * 60) print("🚀 Tests de Sérialisation - Visual Workflow Builder") print("=" * 60) try: test_id_generation() test_empty_workflow_creation() test_serialization_json() test_deserialization_json() test_round_trip() test_validation_errors() test_database_operations() test_file_persistence() print("\n" + "=" * 60) print("✅ TOUS LES TESTS RÉUSSIS!") print("=" * 60) return 0 except AssertionError as e: print(f"\n❌ ÉCHEC: {e}") return 1 except Exception as e: print(f"\n❌ ERREUR: {e}") import traceback traceback.print_exc() return 1 if __name__ == '__main__': sys.exit(main())