rpa_vision_v3/tests/integration/test_graph_to_visual.py

"""
Tests du GraphToVisualConverter — conversion core Workflow → VWB VisualWorkflow.

Vérifie que le pont inverse (GraphBuilder → VWB) fonctionne correctement :
- Chaque WorkflowNode produit un VisualNode avec position, type, ports
- Chaque WorkflowEdge produit un VisualEdge avec source/target
- L'ordre topologique est respecté (entry → end)
- Les métadonnées visuelles (couleurs, labels) sont cohérentes
"""

import sys
from pathlib import Path

import pytest

_ROOT = str(Path(__file__).resolve().parents[2])
if _ROOT not in sys.path:
    sys.path.insert(0, _ROOT)


# =========================================================================
# Fixtures
# =========================================================================

def _make_core_workflow(num_nodes=3):
    """Crée un core Workflow minimal pour les tests."""
    from core.models.workflow_graph import (
        Workflow,
        WorkflowNode,
        WorkflowEdge,
        Action,
        TargetSpec,
        ScreenTemplate,
        WindowConstraint,
        TextConstraint,
        UIConstraint,
        EmbeddingPrototype,
        EdgeConstraints,
        PostConditions,
        EdgeStats,
        SafetyRules,
        WorkflowStats,
        LearningConfig,
    )

    nodes = []
    for i in range(num_nodes):
        node = WorkflowNode(
            node_id=f"node_{i}",
            name=f"Étape {i}",
            description=f"Description nœud {i}",
            template=ScreenTemplate(
                window=WindowConstraint(title_pattern=f"App{i}"),
                text=TextConstraint(),
                ui=UIConstraint(),
                embedding=EmbeddingPrototype(
                    provider="test",
                    vector_id=f"vec_{i}",
                    min_cosine_similarity=0.8,
                    sample_count=1,
                ),
            ),
            is_entry=(i == 0),
            is_end=(i == num_nodes - 1),
            metadata={
                "visual_type": "click" if i > 0 and i < num_nodes - 1 else ("start" if i == 0 else "end"),
                "parameters": {"target": f"button_{i}"},
            },
        )
        nodes.append(node)

    edges = []
    for i in range(num_nodes - 1):
        edge = WorkflowEdge(
            edge_id=f"edge_{i}_to_{i+1}",
            from_node=f"node_{i}",
            to_node=f"node_{i+1}",
            action=Action(
                type="mouse_click",
                target=TargetSpec(by_text=f"button_{i}"),
            ),
            constraints=EdgeConstraints(),
            post_conditions=PostConditions(expected_node=f"node_{i+1}"),
            stats=EdgeStats(),
        )
        edges.append(edge)

    from datetime import datetime
    now = datetime.now()

    return Workflow(
        workflow_id="test_wf_001",
        name="Test Workflow",
        description="Workflow de test pour conversion",
        version=1,
        learning_state="OBSERVATION",
        created_at=now,
        updated_at=now,
        entry_nodes=["node_0"],
        end_nodes=[f"node_{num_nodes - 1}"],
        nodes=nodes,
        edges=edges,
        safety_rules=SafetyRules(),
        stats=WorkflowStats(),
        learning=LearningConfig(),
        metadata={"tags": ["test"], "source": "test"},
    )


# =========================================================================
# Tests
# =========================================================================


class TestGraphToVisualConverter:
    """Tests de conversion core Workflow → VisualWorkflow."""

    def test_basic_conversion(self):
        """Un workflow 3 nodes se convertit sans erreur."""
        sys.path.insert(0, str(Path(_ROOT) / "visual_workflow_builder" / "backend"))
        from services.graph_to_visual_converter import GraphToVisualConverter

        wf = _make_core_workflow(3)
        converter = GraphToVisualConverter()
        visual = converter.convert(wf)

        assert visual.id == "test_wf_001"
        assert visual.name == "Test Workflow"
        assert len(visual.nodes) == 3
        assert len(visual.edges) == 2

    def test_node_ids_preserved(self):
        """Les IDs des nodes sont préservés."""
        sys.path.insert(0, str(Path(_ROOT) / "visual_workflow_builder" / "backend"))
        from services.graph_to_visual_converter import GraphToVisualConverter

        wf = _make_core_workflow(4)
        visual = GraphToVisualConverter().convert(wf)

        visual_ids = {n.id for n in visual.nodes}
        assert visual_ids == {"node_0", "node_1", "node_2", "node_3"}

    def test_edge_source_target_preserved(self):
        """Les edges connectent les bons nodes."""
        sys.path.insert(0, str(Path(_ROOT) / "visual_workflow_builder" / "backend"))
        from services.graph_to_visual_converter import GraphToVisualConverter

        wf = _make_core_workflow(3)
        visual = GraphToVisualConverter().convert(wf)

        edge_pairs = [(e.source, e.target) for e in visual.edges]
        assert ("node_0", "node_1") in edge_pairs
        assert ("node_1", "node_2") in edge_pairs

    def test_visual_types_inferred(self):
        """Les types visuels sont correctement inférés depuis les métadonnées."""
        sys.path.insert(0, str(Path(_ROOT) / "visual_workflow_builder" / "backend"))
        from services.graph_to_visual_converter import GraphToVisualConverter

        wf = _make_core_workflow(3)
        visual = GraphToVisualConverter().convert(wf)

        types = {n.id: n.type for n in visual.nodes}
        assert types["node_0"] == "start"
        assert types["node_1"] == "click"
        assert types["node_2"] == "end"

    def test_positions_ordered_vertically(self):
        """Les nodes sont positionnés de haut en bas."""
        sys.path.insert(0, str(Path(_ROOT) / "visual_workflow_builder" / "backend"))
        from services.graph_to_visual_converter import GraphToVisualConverter

        wf = _make_core_workflow(5)
        visual = GraphToVisualConverter().convert(wf)

        y_positions = [n.position.y for n in visual.nodes]
        assert y_positions == sorted(y_positions), "Les nodes doivent être ordonnés verticalement"

    def test_start_node_has_no_input_port(self):
        """Le node 'start' n'a pas de port d'entrée."""
        sys.path.insert(0, str(Path(_ROOT) / "visual_workflow_builder" / "backend"))
        from services.graph_to_visual_converter import GraphToVisualConverter

        wf = _make_core_workflow(3)
        visual = GraphToVisualConverter().convert(wf)

        start_node = [n for n in visual.nodes if n.type == "start"][0]
        assert len(start_node.input_ports) == 0
        assert len(start_node.output_ports) == 1

    def test_end_node_has_no_output_port(self):
        """Le node 'end' n'a pas de port de sortie."""
        sys.path.insert(0, str(Path(_ROOT) / "visual_workflow_builder" / "backend"))
        from services.graph_to_visual_converter import GraphToVisualConverter

        wf = _make_core_workflow(3)
        visual = GraphToVisualConverter().convert(wf)

        end_node = [n for n in visual.nodes if n.type == "end"][0]
        assert len(end_node.input_ports) == 1
        assert len(end_node.output_ports) == 0

    def test_to_dict_roundtrip(self):
        """Le VisualWorkflow produit un dict valide et reconstructible."""
        sys.path.insert(0, str(Path(_ROOT) / "visual_workflow_builder" / "backend"))
        from services.graph_to_visual_converter import GraphToVisualConverter

        wf = _make_core_workflow(3)
        visual = GraphToVisualConverter().convert(wf)

        d = visual.to_dict()
        assert d["id"] == "test_wf_001"
        assert len(d["nodes"]) == 3
        assert len(d["edges"]) == 2

        # Vérifier que les nodes dict ont les bons champs
        node0 = d["nodes"][0]
        assert "id" in node0
        assert "type" in node0
        assert "position" in node0

    def test_large_workflow(self):
        """Un workflow de 20 nodes se convertit correctement."""
        sys.path.insert(0, str(Path(_ROOT) / "visual_workflow_builder" / "backend"))
        from services.graph_to_visual_converter import GraphToVisualConverter

        wf = _make_core_workflow(20)
        visual = GraphToVisualConverter().convert(wf)

        assert len(visual.nodes) == 20
        assert len(visual.edges) == 19

    def test_colors_assigned(self):
        """Chaque type de node a une couleur."""
        sys.path.insert(0, str(Path(_ROOT) / "visual_workflow_builder" / "backend"))
        from services.graph_to_visual_converter import GraphToVisualConverter

        wf = _make_core_workflow(3)
        visual = GraphToVisualConverter().convert(wf)

        for node in visual.nodes:
            assert node.color is not None
            assert node.color.startswith("#")

    def test_utility_function(self):
        """La fonction utilitaire convert_graph_to_visual fonctionne."""
        sys.path.insert(0, str(Path(_ROOT) / "visual_workflow_builder" / "backend"))
        from services.graph_to_visual_converter import convert_graph_to_visual

        wf = _make_core_workflow(3)
        visual = convert_graph_to_visual(wf)

        assert visual.name == "Test Workflow"
        assert len(visual.nodes) == 3