MyManagingTools/tests/test_workflow_designer.py

from unittest.mock import MagicMock

import pytest
from fastcore.basics import NotStr
from fasthtml.components import *
from fasthtml.xtend import Script

from components.undo_redo.components.UndoRedo import UndoRedo
from components.workflows.components.WorkflowDesigner import WorkflowDesigner, COMPONENT_TYPES
from components.workflows.constants import ProcessorTypes
from components.workflows.db_management import WorkflowsDesignerSettings, WorkflowComponent, Connection
from core.instance_manager import InstanceManager
from core.settings_management import SettingsManager, MemoryDbEngine
from helpers import matches, Contains
from my_mocks import tabs_manager

TEST_WORKFLOW_DESIGNER_ID = "workflow_designer_id"


@pytest.fixture(autouse=True)
def mock_undo_redo(session):
  # Create a mock UndoRedo instance
  undo_redo = MagicMock(spec=UndoRedo)
  
  # Store original get method
  original_get = InstanceManager.get
  
  def mock_get(sess, instance_id, *args, **kwargs):
    if instance_id == UndoRedo.create_component_id(sess):
      return undo_redo
    return original_get(sess, instance_id, *args, **kwargs)
  
  # Replace get method with our mock
  InstanceManager.get = mock_get
  
  yield undo_redo
  
  # Restore original get method after test
  InstanceManager.get = original_get

@pytest.fixture
def designer(session, tabs_manager):
  return WorkflowDesigner(session=session, _id=TEST_WORKFLOW_DESIGNER_ID,
                          settings_manager=SettingsManager(engine=MemoryDbEngine()),
                          tabs_manager=tabs_manager,
                          key=TEST_WORKFLOW_DESIGNER_ID,
                          designer_settings=WorkflowsDesignerSettings("Workflow Name"),
                          boundaries={"height": 500, "width": 800}
                          )


@pytest.fixture
def producer_component():
  return WorkflowComponent(
    "comp_producer",
    ProcessorTypes.Producer,
    10,
    100,
    COMPONENT_TYPES[ProcessorTypes.Producer]["title"],
    COMPONENT_TYPES[ProcessorTypes.Producer]["description"],
    {"processor_name": ProcessorTypes.Producer[0]}
  )


@pytest.fixture
def filter_component():
  return WorkflowComponent(
    "comp_filter",
    ProcessorTypes.Filter,
    40,
    100,
    COMPONENT_TYPES[ProcessorTypes.Filter]["title"],
    COMPONENT_TYPES[ProcessorTypes.Filter]["description"],
    {"processor_name": ProcessorTypes.Filter[0]}
  )


@pytest.fixture
def presenter_component():
  return WorkflowComponent(
    "comp_presenter",
    ProcessorTypes.Presenter,
    70,
    100,
    COMPONENT_TYPES[ProcessorTypes.Presenter]["title"],
    COMPONENT_TYPES[ProcessorTypes.Presenter]["description"],
    {"processor_name": ProcessorTypes.Presenter[0]}
  )


@pytest.fixture
def components(producer_component, filter_component, presenter_component):
  return [producer_component, filter_component, presenter_component]


def test_i_can_render_no_component(designer):
  actual = designer.__ft__()
  expected = Div(
    H1("Workflow Name"),
    P("Drag components from the toolbox to the canvas to create your workflow."),
    Div(id=f"t_{designer.get_id()}"),  # media + error message
    Div(id=f"d_{designer.get_id()}"),  # designer container
    Div(cls="wkf-splitter"),
    Div(id=f"p_{designer.get_id()}"),  # properties panel
    Script(f"bindWorkflowDesigner('{designer.get_id()}');"),
    id=designer.get_id(),
  )
  
  assert matches(actual, expected)


def test_i_can_render_a_producer(designer, producer_component):
  component = producer_component
  actual = designer._mk_component(component)
  expected = Div(
    # input connection point
    Div(cls="wkf-connection-point wkf-input-point",
        data_component_id=component.id,
        data_point_type="input"
        ),
    
    # Component content
    Div(
      Span(COMPONENT_TYPES[component.type]["icon"]),
      H4(component.title),
      cls=Contains("wkf-component-content")
    ),
    
    # Output connection point
    Div(cls="wkf-connection-point wkf-output-point",
        data_component_id=component.id,
        data_point_type="output"
        ),
    
    cls=Contains("wkf-workflow-component"),
    style=f"left: {component.x}px; top: {component.y}px;",
    data_component_id=component.id,
    draggable="true"
  )
  
  assert matches(actual, expected)


def test_i_can_render_a_connection(designer, components):
  designer._state.components = {c.id: c for c in components}
  connection = Connection("conn_1", "comp_producer", "comp_presenter")
  
  actual = designer._mk_connection_svg(connection)
  path = "M 138 132 C 104.0 132, 104.0 132, 70 132"
  expected = f"""
      <svg class="wkf-connection-line" style="left: 0; top: 0; width: 100%; height: 100%;"
           data-from-id="{connection.from_id}" data-to-id="{connection.to_id}">
          <path d="{path}" class="wkf-connection-path-thick"/>
          <path d="{path}" class="wkf-connection-path" marker-end="url(#arrowhead)"/>

          <defs>
              <marker id="arrowhead" markerWidth="10" markerHeight="7" refX="9" refY="3.5" orient="auto">
                  <polygon points="0 0, 10 3.5, 0 7" class="wkf-connection-path-arrowhead"/>
              </marker>
          </defs>
      </svg>
      """
  assert actual == expected


def test_i_can_render_elements_with_connections(designer, components):
  designer._state.components = {c.id: c for c in components}
  designer._state.connections = [Connection("conn_1", components[0].id, components[1].id),
                                 Connection("conn_2", components[1].id, components[2].id)]
  actual = designer._mk_elements()
  expected = Div(
    NotStr('<svg class="wkf-connection-line"'),  # connection 1
    NotStr('<svg class="wkf-connection-line"'),  # connection 2
    Div(cls=Contains("wkf-workflow-component")),
    Div(cls=Contains("wkf-workflow-component")),
    Div(cls=Contains("wkf-workflow-component")),
  )
  
  assert matches(actual, expected)