rpa_vision_v3/visual_workflow_builder/backend/test_converter.py

#!/usr/bin/env python3
"""
Test du Convertisseur Visual → WorkflowGraph

Test manuel de la conversion de workflows visuels en WorkflowGraph exécutables.
"""

import sys
from pathlib import Path

sys.path.insert(0, '.')

from services.converter import VisualToGraphConverter, ConversionError, convert_visual_to_graph
from services.serialization import create_empty_workflow
from models.visual_workflow import (
    VisualNode,
    VisualEdge,
    Position,
    Size,
    Port
)


def test_empty_workflow_conversion():
    """Test de conversion d'un workflow vide"""
    print("🧪 Test 1: Conversion d'un workflow vide")
    
    workflow = create_empty_workflow("Empty Workflow")
    
    converter = VisualToGraphConverter()
    
    try:
        result = converter.convert(workflow)
        print("❌ Devrait échouer pour un workflow vide")
        return False
    except ConversionError as e:
        print(f"✅ Erreur attendue: {e}")
        return True


def test_simple_workflow_conversion():
    """Test de conversion d'un workflow simple"""
    print("\n🧪 Test 2: Conversion d'un workflow simple (2 nodes, 1 edge)")
    
    workflow = create_empty_workflow("Simple Workflow")
    
    # Ajouter 2 nodes
    workflow.nodes.append(VisualNode(
        id="click_1",
        type="click",
        position=Position(100, 100),
        size=Size(200, 80),
        parameters={'target': {'text': 'Login Button', 'role': 'button'}},
        input_ports=[],
        output_ports=[Port('out', 'Output', 'output')]
    ))
    
    workflow.nodes.append(VisualNode(
        id="type_1",
        type="type",
        position=Position(400, 100),
        size=Size(200, 80),
        parameters={'target': {'role': 'textfield'}, 'text': 'username'},
        input_ports=[Port('in', 'Input', 'input')],
        output_ports=[]
    ))
    
    # Ajouter 1 edge
    workflow.edges.append(VisualEdge(
        id="edge_1",
        source="click_1",
        target="type_1",
        source_port="out",
        target_port="in"
    ))
    
    # Convertir
    converter = VisualToGraphConverter()
    
    try:
        result = converter.convert(workflow)
        
        # Vérifications
        assert result.workflow_id == workflow.id
        assert result.name == "Simple Workflow"
        assert len(result.nodes) == 2
        assert len(result.edges) == 1
        assert len(result.entry_nodes) == 1
        assert len(result.end_nodes) == 1
        assert result.entry_nodes[0] == "click_1"
        assert result.end_nodes[0] == "type_1"
        
        # Vérifier les nodes
        node1 = result.get_node("click_1")
        assert node1 is not None
        assert node1.is_entry == True
        assert node1.metadata['visual_type'] == 'click'
        
        node2 = result.get_node("type_1")
        assert node2 is not None
        assert node2.is_end == True
        assert node2.metadata['visual_type'] == 'type'
        
        # Vérifier l'edge
        edge1 = result.get_edge("edge_1")
        assert edge1 is not None
        assert edge1.from_node == "click_1"
        assert edge1.to_node == "type_1"
        assert edge1.action.type == "mouse_click"
        
        print(f"✅ Conversion réussie:")
        print(f"   - {len(result.nodes)} nodes")
        print(f"   - {len(result.edges)} edges")
        print(f"   - Entry: {result.entry_nodes}")
        print(f"   - End: {result.end_nodes}")
        
        return True
        
    except Exception as e:
        print(f"❌ Erreur: {e}")
        import traceback
        traceback.print_exc()
        return False


def test_complex_workflow_conversion():
    """Test de conversion d'un workflow complexe"""
    print("\n🧪 Test 3: Conversion d'un workflow complexe (4 nodes, 3 edges)")
    
    workflow = create_empty_workflow("Complex Workflow")
    
    # Ajouter 4 nodes
    workflow.nodes.extend([
        VisualNode(
            id="navigate_1",
            type="navigate",
            position=Position(100, 100),
            size=Size(200, 80),
            parameters={'url': 'https://example.com'},
            input_ports=[],
            output_ports=[Port('out', 'Output', 'output')]
        ),
        VisualNode(
            id="click_1",
            type="click",
            position=Position(400, 100),
            size=Size(200, 80),
            parameters={'target': {'text': 'Login'}},
            input_ports=[Port('in', 'Input', 'input')],
            output_ports=[Port('out', 'Output', 'output')]
        ),
        VisualNode(
            id="type_1",
            type="type",
            position=Position(700, 100),
            size=Size(200, 80),
            parameters={'target': {'role': 'textfield'}, 'text': '${username}'},
            input_ports=[Port('in', 'Input', 'input')],
            output_ports=[Port('out', 'Output', 'output')]
        ),
        VisualNode(
            id="wait_1",
            type="wait",
            position=Position(1000, 100),
            size=Size(200, 80),
            parameters={'duration': 2000},
            input_ports=[Port('in', 'Input', 'input')],
            output_ports=[]
        )
    ])
    
    # Ajouter 3 edges
    workflow.edges.extend([
        VisualEdge(
            id="edge_1",
            source="navigate_1",
            target="click_1",
            source_port="out",
            target_port="in"
        ),
        VisualEdge(
            id="edge_2",
            source="click_1",
            target="type_1",
            source_port="out",
            target_port="in"
        ),
        VisualEdge(
            id="edge_3",
            source="type_1",
            target="wait_1",
            source_port="out",
            target_port="in"
        )
    ])
    
    # Convertir
    converter = VisualToGraphConverter()
    
    try:
        result = converter.convert(workflow)
        
        # Vérifications
        assert len(result.nodes) == 4
        assert len(result.edges) == 3
        assert result.entry_nodes[0] == "navigate_1"
        assert result.end_nodes[0] == "wait_1"
        
        # Vérifier les types d'actions
        edge1 = result.get_edge("edge_1")
        assert edge1.action.type == "navigate"
        assert edge1.action.parameters['url'] == 'https://example.com'
        
        edge2 = result.get_edge("edge_2")
        assert edge2.action.type == "mouse_click"
        
        edge3 = result.get_edge("edge_3")
        assert edge3.action.type == "text_input"
        assert edge3.action.parameters['text'] == '${username}'
        
        print(f"✅ Conversion réussie:")
        print(f"   - {len(result.nodes)} nodes")
        print(f"   - {len(result.edges)} edges")
        print(f"   - Types d'actions: navigate, mouse_click, text_input, wait")
        
        return True
        
    except Exception as e:
        print(f"❌ Erreur: {e}")
        import traceback
        traceback.print_exc()
        return False


def test_workflow_with_variables():
    """Test de conversion avec variables"""
    print("\n🧪 Test 4: Conversion avec variables")
    
    workflow = create_empty_workflow("Workflow with Variables")
    
    # Ajouter des variables
    from models.visual_workflow import Variable
    workflow.variables.extend([
        Variable(name="username", type="string", value="test_user"),
        Variable(name="password", type="string", value="secret123")
    ])
    
    # Ajouter des nodes utilisant les variables
    workflow.nodes.extend([
        VisualNode(
            id="type_user",
            type="type",
            position=Position(100, 100),
            size=Size(200, 80),
            parameters={'target': {'role': 'textfield'}, 'text': '${username}'},
            input_ports=[],
            output_ports=[Port('out', 'Output', 'output')]
        ),
        VisualNode(
            id="type_pass",
            type="type",
            position=Position(400, 100),
            size=Size(200, 80),
            parameters={'target': {'role': 'textfield'}, 'text': '${password}'},
            input_ports=[Port('in', 'Input', 'input')],
            output_ports=[]
        )
    ])
    
    workflow.edges.append(VisualEdge(
        id="edge_1",
        source="type_user",
        target="type_pass",
        source_port="out",
        target_port="in"
    ))
    
    # Convertir
    converter = VisualToGraphConverter()
    
    try:
        result = converter.convert(workflow)
        
        # Vérifier que les variables sont préservées dans les métadonnées
        assert len(result.nodes) == 2
        
        # Vérifier que les références de variables sont préservées
        edge1 = result.get_edge("edge_1")
        assert '${username}' in edge1.action.parameters['text']
        
        print(f"✅ Conversion avec variables réussie")
        print(f"   - Variables préservées dans les paramètres")
        
        return True
        
    except Exception as e:
        print(f"❌ Erreur: {e}")
        import traceback
        traceback.print_exc()
        return False


def test_conversion_utility_function():
    """Test de la fonction utilitaire convert_visual_to_graph"""
    print("\n🧪 Test 5: Fonction utilitaire convert_visual_to_graph")
    
    workflow = create_empty_workflow("Utility Test")
    
    workflow.nodes.append(VisualNode(
        id="click_1",
        type="click",
        position=Position(100, 100),
        size=Size(200, 80),
        parameters={'target': 'button'},
        input_ports=[],
        output_ports=[]
    ))
    
    try:
        result = convert_visual_to_graph(workflow)
        
        assert result.workflow_id == workflow.id
        assert len(result.nodes) == 1
        
        print("✅ Fonction utilitaire fonctionne")
        
        return True
        
    except Exception as e:
        print(f"❌ Erreur: {e}")
        return False


def main():
    """Exécute tous les tests"""
    print("=" * 60)
    print("🚀 Tests du Convertisseur Visual → WorkflowGraph")
    print("=" * 60)
    
    tests = [
        test_empty_workflow_conversion,
        test_simple_workflow_conversion,
        test_complex_workflow_conversion,
        test_workflow_with_variables,
        test_conversion_utility_function
    ]
    
    results = []
    for test in tests:
        try:
            result = test()
            results.append(result)
        except Exception as e:
            print(f"\n❌ Test échoué avec exception: {e}")
            import traceback
            traceback.print_exc()
            results.append(False)
    
    print("\n" + "=" * 60)
    passed = sum(results)
    total = len(results)
    
    if passed == total:
        print(f"✅ TOUS LES TESTS RÉUSSIS! ({passed}/{total})")
        print("=" * 60)
        return 0
    else:
        print(f"❌ {total - passed} test(s) échoué(s) sur {total}")
        print("=" * 60)
        return 1


if __name__ == '__main__':
    sys.exit(main())