154 lines
5.0 KiB
Python
154 lines
5.0 KiB
Python
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, \
|
|
WorkflowComponentRuntimeState
|
|
from core.settings_management import SettingsManager, MemoryDbEngine
|
|
from helpers import matches, Contains
|
|
|
|
TEST_WORKFLOW_DESIGNER_ID = "workflow_designer_id"
|
|
|
|
|
|
@pytest.fixture
|
|
def designer(session):
|
|
return WorkflowDesigner(session=session, _id=TEST_WORKFLOW_DESIGNER_ID,
|
|
settings_manager=SettingsManager(engine=MemoryDbEngine()),
|
|
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"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
|
|
component_state = WorkflowComponentRuntimeState(component.id)
|
|
actual = designer._mk_component(component, component_state)
|
|
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)
|