cleveractors-core/features/steps/langgraph_isolated_steps.py

"""Isolated step definitions for LangGraph tests to avoid conflicts."""

import json
from unittest.mock import MagicMock, patch

from behave import given, then, when
from rx.subject import Subject

from cleveractors.core.application import ReactiveCleverAgentsApp
from cleveractors.langgraph.nodes import NodeType


def create_isolated_mock_langgraph(
    name="test_graph", nodes=None, edges=None, config_dict=None
):
    """Create a mock LangGraph for isolated testing."""
    mock_graph = MagicMock()
    mock_graph.name = name
    mock_graph.config = MagicMock()
    mock_graph.config.entry_point = "start"

    # Set defaults
    mock_graph.config.checkpointing = False
    mock_graph.config.enable_time_travel = False
    mock_graph.config.parallel_execution = False

    # Override with configuration if provided
    if config_dict:
        mock_graph.config.checkpointing = config_dict.get("checkpointing", False)
        mock_graph.config.enable_time_travel = config_dict.get(
            "enable_time_travel", False
        )
        mock_graph.config.parallel_execution = config_dict.get(
            "parallel_execution", False
        )
    mock_graph.config.nodes = nodes or {}
    # Create proper edge objects with conditions
    mock_edges = []
    if edges:
        for edge in edges:
            mock_edge = MagicMock()
            mock_edge.source = edge.get("source")
            mock_edge.target = edge.get("target")
            mock_edge.condition = edge.get("condition")
            mock_edges.append(mock_edge)
    mock_graph.config.edges = mock_edges
    # Ensure nodes include start, end, and any custom nodes
    graph_nodes = {}
    if nodes:
        for node_name, node_config in nodes.items():
            mock_node = MagicMock()
            mock_node.config = MagicMock()
            mock_node.config.type = NodeType.FUNCTION  # default
            if node_config.get("type") == "agent":
                mock_node.config.type = NodeType.AGENT
                mock_node.config.agent = node_config.get("agent")
            elif node_config.get("type") == "conditional":
                mock_node.config.type = NodeType.CONDITIONAL
            graph_nodes[node_name] = mock_node
    # Always add start and end nodes
    graph_nodes.update({"start": MagicMock(), "end": MagicMock()})
    mock_graph.nodes = graph_nodes

    # Mock state manager
    mock_graph.state_manager = MagicMock()
    mock_graph.state_manager.get_state = MagicMock()
    mock_graph.state_manager.update_state = MagicMock()
    mock_graph.state_manager.get_state_observable = MagicMock(return_value=Subject())

    # Mock execute method
    async def mock_execute(input_data):
        state = MagicMock()
        state.messages = [{"role": "assistant", "content": "Mock response"}]
        state.to_dict = MagicMock(return_value={"messages": state.messages})
        return state

    mock_graph.execute = mock_execute
    mock_graph.get_execution_history = MagicMock(return_value=[])
    mock_graph.visualize = MagicMock(return_value="Mock visualization")

    return mock_graph


@given(
    "the CleverAgents reactive system is available with LangGraph support for isolated testing"
)
def step_impl(context):
    """Initialize the reactive system with LangGraph support for isolated testing."""
    # Clear any existing state to prevent pollution from other tests
    if hasattr(context, "isolated_app"):
        context.isolated_app = None
    if hasattr(context, "isolated_graphs"):
        context.isolated_graphs.clear()
    if hasattr(context, "isolated_config_dict"):
        context.isolated_config_dict.clear()

    # Create fresh instances with isolated names
    context.isolated_app = ReactiveCleverAgentsApp(verbose=True)
    context.isolated_graphs = {}
    context.isolated_config_dict = {}


@given("I have a LangGraph configuration for isolated testing")
def step_impl(context):
    """Parse LangGraph configuration from text for isolated testing."""
    context.isolated_graph_config = json.loads(context.text)


@when("I create the graph for isolated testing")
def step_impl(context):
    """Create a LangGraph from configuration for isolated testing."""
    try:
        # Ensure we have valid graph config
        if (
            not hasattr(context, "isolated_graph_config")
            or not context.isolated_graph_config
        ):
            raise ValueError("No graph configuration provided")

        # Ensure we have a valid app instance
        if not hasattr(context, "isolated_app") or not context.isolated_app:
            raise ValueError("No ReactiveCleverAgentsApp instance available")

        graph_name = context.isolated_graph_config.get("name", "test_graph")

        with patch("cleveractors.langgraph.bridge.LangGraph") as mock_langgraph_class:
            # Create mock with configuration details
            mock_graph = create_isolated_mock_langgraph(
                name=graph_name,
                nodes=context.isolated_graph_config.get("nodes", {}),
                edges=context.isolated_graph_config.get("edges", []),
                config_dict=context.isolated_graph_config,
            )
            mock_langgraph_class.return_value = mock_graph

            # Create the graph through the bridge
            graph = context.isolated_app.langgraph_bridge.create_graph_from_config(
                context.isolated_graph_config
            )
            context.isolated_graphs[graph.name] = graph
            context.isolated_current_graph = graph
            context.isolated_graph_created = True
    except Exception as e:
        context.isolated_graph_created = False
        context.isolated_error = str(e)


@then("the graph should be created successfully for isolated testing")
def step_impl(context):
    """Verify graph creation for isolated testing."""
    assert context.isolated_graph_created, (
        f"Graph creation failed: {getattr(context, 'isolated_error', 'Unknown error')}"
    )
    assert context.isolated_current_graph is not None


@then(
    "the graph should have {count:d} custom node plus start and end nodes for isolated testing"
)
def step_impl(context, count):
    """Verify node count for isolated testing."""
    graph = context.isolated_current_graph
    # Total nodes = custom nodes + start + end
    assert len(graph.nodes) == count + 2


@then("the graph should support conditional routing for isolated testing")
def step_impl(context):
    """Verify conditional routing support for isolated testing."""
    graph = context.isolated_current_graph

    # Check for conditional node
    check_node = graph.nodes.get("check")
    assert check_node is not None
    assert check_node.config.type == NodeType.CONDITIONAL

    # Check for conditional edges
    conditional_edges = [e for e in graph.config.edges if e.condition is not None]
    assert len(conditional_edges) > 0