import pytest from fastcore.basics import NotStr from fasthtml.components import * from fasthtml.xtend import Script 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.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 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""" """ 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('