kodjo/MyManagingTools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Started unit test for Workflows.py and WorkflowDesigner.py

Browse Source
This commit is contained in:
Kodjo Sossouvi
2025-07-06 11:02:57 +02:00
parent 9df32e3b5f
commit 60872a0aec
9 changed files with 348 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/assets/icons.py
View File

@@ -12,7 +12,7 @@ icon_dismiss_regular = NotStr(
)
# Fluent Add16Regular
icon_add_regular = NotStr("""<svg name="addd" xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" viewBox="0 0 16 16">
icon_add_regular = NotStr("""<svg name="add" xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" viewBox="0 0 16 16">
<g fill="none">
<path d="M8 2.5a.5.5 0 0 0-1 0V7H2.5a.5.5 0 0 0 0 1H7v4.5a.5.5 0 0 0 1 0V8h4.5a.5.5 0 0 0 0-1H8V2.5z" fill="currentColor">
</path>

11
src/components/workflows/assets/Workflows.css
View File

@@ -89,6 +89,17 @@
transition: none;
}
.wkf-component-content {
padding: 0.75rem; /* p-3 in Tailwind */
border-radius: 0.5rem; /* rounded-lg in Tailwind */
border-width: 2px; /* border-2 in Tailwind */
background-color: white; /* bg-white in Tailwind */
box-shadow: 0 10px 15px -3px rgba(0, 0, 0, 0.1), 0 4px 6px -2px rgba(0, 0, 0, 0.05); /* shadow-lg in Tailwind */
display: flex; /* flex in Tailwind */
align-items: center; /* items-center in Tailwind */
}
.wkf-connection-line {
position: absolute;
pointer-events: none;

26
src/components/workflows/components/WorkflowDesigner.py
View File

@@ -6,7 +6,7 @@ from fasthtml.xtend import Script
from components.BaseComponent import BaseComponent
from components.workflows.commands import WorkflowDesignerCommandManager
from components.workflows.constants import WORKFLOW_DESIGNER_INSTANCE_ID
from components.workflows.constants import WORKFLOW_DESIGNER_INSTANCE_ID, ProcessorTypes
from components.workflows.db_management import WorkflowsDesignerSettings, WorkflowComponent, \
Connection, WorkflowsDesignerDbManager
from components_helpers import apply_boundaries, mk_tooltip, mk_dialog_buttons
@@ -17,19 +17,19 @@ logger = logging.getLogger("WorkflowDesigner")
# Component templates
COMPONENT_TYPES = {
"producer": {
ProcessorTypes.Producer: {
"title": "Data Producer",
"description": "Generates or loads data",
"icon": "📊",
"color": "bg-green-100 border-green-300 text-neutral"
},
"filter": {
ProcessorTypes.Filter: {
"title": "Data Filter",
"description": "Filters and transforms data",
"icon": "🔍",
"color": "bg-blue-100 border-blue-300 text-neutral"
},
"presenter": {
ProcessorTypes.Presenter: {
"title": "Data Presenter",
"description": "Displays or exports data",
"icon": "📋",
@@ -38,9 +38,9 @@ COMPONENT_TYPES = {
}
PROCESSOR_TYPES = {
"producer": ["Repository", "Jira"],
"filter": ["Default"],
"presenter": ["Default"]}
ProcessorTypes.Producer: ["Repository", "Jira"],
ProcessorTypes.Filter: ["Default"],
ProcessorTypes.Presenter: ["Default"]}
class WorkflowDesigner(BaseComponent):
@@ -229,7 +229,7 @@ class WorkflowDesigner(BaseComponent):
# Render components
*[self._mk_workflow_component(comp) for comp in self._state.components.values()],
),
)
def _mk_canvas(self, oob=False):
return Div(
@@ -264,9 +264,9 @@ class WorkflowDesigner(BaseComponent):
return self._mk_jira_processor_details(component)
elif processor_name == "Repository":
return self._mk_repository_processor_details(component)
elif component.type == "filter" and processor_name == "Default":
elif component.type == ProcessorTypes.Filter and processor_name == "Default":
return self._mk_filter_processor_details(component)
elif component.type == "presenter" and processor_name == "Default":
elif component.type == ProcessorTypes.Presenter and processor_name == "Default":
return self._mk_presenter_processor_details(component)
return Div('Not defined yet !')
@@ -403,9 +403,9 @@ class WorkflowDesigner(BaseComponent):
def _mk_presenter_processor_details(component):
return Div(
Fieldset(
Legend("Filter", cls="fieldset-legend"),
Legend("Presenter", cls="fieldset-legend"),
Input(type="text",
name="filter",
name="presenter",
value=component.properties.get("filter", ""),
placeholder="Enter filter expression",
cls="input w-full"),
@@ -454,7 +454,7 @@ class WorkflowDesigner(BaseComponent):
Div(
Span(info["icon"], cls="text-xl mb-1"),
H4(component.title, cls="font-semibold text-xs"),
cls=f"p-3 rounded-lg border-2 {info['color']} bg-white shadow-lg flex items-center"
cls=f"wkf-component-content {info['color']}"
),
# Output connection point

2
src/components/workflows/components/Workflows.py
View File

@@ -75,7 +75,7 @@ class Workflows(BaseComponentSingleton):
self.tabs_manager.select_tab_by_key(tab_key)
self.db.select_workflow(workflow_name)
return self.tabs_manager.refresh(), self.refresh()
return self.refresh(), self.tabs_manager.refresh()
def refresh(self):
return self._mk_workflows(True)

5
src/components/workflows/constants.py
View File

@@ -5,6 +5,11 @@ WORKFLOW_DESIGNER_DB_ENTRY = "WorkflowDesigner"
WORKFLOW_DESIGNER_DB_SETTINGS_ENTRY = "Settings"
WORKFLOW_DESIGNER_DB_STATE_ENTRY = "State"
class ProcessorTypes:
Producer = "producer"
Filter = "filter"
Presenter = "presenter"
ROUTE_ROOT = "/workflows"

13
tests/helpers.py
View File

@@ -420,7 +420,8 @@ def matches(actual, expected, path=""):
assert matches(actual_child, expected_child)
elif isinstance(expected, NotStr):
assert actual.s.lstrip('\n').startswith(expected.s), \
to_compare = actual.s.lstrip('\n').lstrip()
assert to_compare.startswith(expected.s), \
f"{print_path(path)}NotStr are different: '{actual.s.lstrip('\n')}' != '{expected.s}'."
elif hasattr(actual, "tag"):
@@ -741,10 +742,20 @@ def _get_element_value(element):
def icon(name: str):
"""
Test if an element is an icon
:param name:
:return:
"""
return NotStr(f'<svg name="{name}"')
def div_icon(name: str):
"""
Test if an element is an icon wrapped in a div
:param name:
:return:
"""
return Div(NotStr(f'<svg name="{name}"'))

2
tests/test_repositories.py
View File

@@ -25,7 +25,7 @@ def tabs_manager():
self._called_methods: list[tuple] = []
def add_tab(self, *args, **kwargs):
self._called_methods.append(("set_tab_content", args, kwargs))
self._called_methods.append(("add_tab", args, kwargs))
table_name, content, key = args
self.tabs.append({"table_name": table_name, "content": content, "key": key})

151
tests/test_workflow_designer.py Normal file
View File

@@ -0,0 +1,151 @@
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
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
actual = designer._mk_workflow_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)

153
tests/test_workflows.py Normal file
View File

@@ -0,0 +1,153 @@
from unittest.mock import MagicMock
import pytest
from fasthtml.components import *
from components.form.components.MyForm import FormField, MyForm
from components.tabs.components.MyTabs import MyTabs
from components.workflows.components.Workflows import Workflows
from core.settings_management import SettingsManager, MemoryDbEngine
from helpers import matches, div_icon, search_elements_by_name, Contains
TEST_WORKFLOWS_ID = "testing_repositories_id"
@pytest.fixture
def tabs_manager():
class MockTabsManager(MagicMock):
def __init__(self, *args, **kwargs):
super().__init__(*args, spec=MyTabs, **kwargs)
self.request_new_tab_id = MagicMock(side_effect =["new_tab_id", "new_tab_2", "new_tab_3", StopIteration])
self.tabs = {}
self.tabs_by_key = {}
def add_tab(self, title, content, key: str | tuple = None, tab_id: str = None, icon=None):
self.tabs[tab_id] = (title, content)
self.tabs_by_key[key] = (title, content)
def set_tab_content(self, tab_id, content, title=None, key: str | tuple = None, active=None):
self.tabs[tab_id] = (title, content)
self.tabs_by_key[key] = (title, content)
def refresh(self):
return Div(
Div(
[Div(title) for title in self.tabs.keys()]
),
list(self.tabs.values())[-1]
)
return MockTabsManager()
@pytest.fixture
def workflows(session, tabs_manager):
return Workflows(session=session, _id=TEST_WORKFLOWS_ID,
settings_manager=SettingsManager(engine=MemoryDbEngine()),
tabs_manager=tabs_manager)
def test_render_no_workflow(workflows):
actual = workflows.__ft__()
expected = Div(
Div(cls="divider"),
Div(
Div("Workflows"),
div_icon("add"), # icon to add a new workflow
cls="flex"
),
Div(id=f"w_{workflows.get_id()}", ), # list of workflow
id=workflows.get_id(),
)
assert matches(actual, expected)
def test_render_with_workflows_defined(workflows):
workflows.db.add_workflow("workflow 1")
workflows.db.add_workflow("workflow 2")
actual = workflows.__ft__()
expected = Div(
Div(cls="divider"),
Div(), # title + icon 'Add'
Div(
Div("workflow 1"),
Div("workflow 2"),
id=f"w_{workflows.get_id()}"
), # list of workflows
id=workflows.get_id(),
)
assert matches(actual, expected)
def test_i_can_see_selected_workflow(workflows):
workflows.db.add_workflow("workflow 1")
workflows.db.add_workflow("workflow 2")
workflows.db.select_workflow("workflow 2")
actual = workflows.__ft__()
to_compare = search_elements_by_name(actual, "div", attrs={"id": f"w_{workflows.get_id()}"})[0]
expected = Div(
Div("workflow 1"),
Div(Div("workflow 2"), cls=Contains("mmt-selected")),
id=f"w_{workflows.get_id()}"
)
assert matches(to_compare, expected)
def test_i_can_request_for_a_new_workflow(workflows, tabs_manager):
res = workflows.request_new_workflow()
tabs_manager.request_new_tab_id.assert_called_once()
assert "new_tab_id" in res.tabs
tab_def = res.tabs["new_tab_id"]
assert tab_def[0] == "Add Workflow"
content = tab_def[1]
assert isinstance(content, MyForm)
assert content.title == "Add Workflow"
assert content.fields == [FormField("name", 'Workflow Name', 'input')]
def test_i_can_add_a_new_workflow(workflows, tabs_manager):
res = workflows.request_new_workflow()
tab_id = list(res.tabs.keys())[0]
actual = workflows.add_new_workflow(tab_id, "Not relevant here", "New Workflow", {})
expected = (
Div(
Div("New Workflow"),
id=f"w_{workflows.get_id()}"
), # list of workflows
Div(), # Workflow Designer embedded in the tab
)
assert matches(actual, expected)
# check that the workflow was added
assert workflows.db.exists_workflow("New Workflow")
def test_i_can_select_a_workflow(workflows):
workflows.add_new_workflow("tab_id_1", "Not relevant", "workflow 1", {})
workflows.add_new_workflow("tab_id_2", "Not relevant", "workflow 2", {})
workflows.add_new_workflow("tab_id_3", "Not relevant", "workflow 3", {})
actual = workflows.show_workflow("workflow 2", {})
expected = (
Div(
Div("workflow 1"),
Div(Div("workflow 2"), cls=Contains("mmt-selected")),
Div("workflow 3"),
id=f"w_{workflows.get_id()}"
), # list of workflows
Div(), # Workflow Designer embedded in the tab
)
assert matches(actual, expected)