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

Adding unit tests to WorkflowPlayer.py

Browse Source
This commit is contained in:
Kodjo Sossouvi
2025-07-13 12:23:25 +02:00
parent fdf05edec3
commit f3deeaefd1
9 changed files with 252 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -11,7 +11,7 @@ from components.workflows.commands import WorkflowDesignerCommandManager
from components.workflows.components.WorkflowPlayer import WorkflowPlayer
from components.workflows.constants import WORKFLOW_DESIGNER_INSTANCE_ID, ProcessorTypes
from components.workflows.db_management import WorkflowsDesignerSettings, WorkflowComponent, \
Connection, WorkflowsDesignerDbManager, WorkflowsPlayerSettings, WorkflowComponentRuntimeState, ComponentState
Connection, WorkflowsDesignerDbManager, ComponentState
from components_helpers import apply_boundaries, mk_tooltip, mk_dialog_buttons, mk_icon
from core.instance_manager import InstanceManager
from core.utils import get_unique_id, make_safe_id
@@ -71,9 +71,7 @@ class WorkflowDesigner(BaseComponent):
WorkflowPlayer,
settings_manager=self._settings_manager,
tabs_manager=self.tabs_manager,
player_settings=WorkflowsPlayerSettings(workflow_name,
list(self._state.components.values()),
self._state.connections),
designer=self,
boundaries=boundaries)
self._error_message = None
@@ -202,10 +200,10 @@ class WorkflowDesigner(BaseComponent):
# change the tab and display the results
self._player.set_boundaries(boundaries)
self.tabs_manager.add_tab(f"Workflow {self._designer_settings.workflow_name}", self._player, self._player.key)
return self.tabs_manager.refresh()
def stop_workflow(self, boundaries):
def stop_workflow(self):
self._error_message = None
self._player.run()
return self.tabs_manager.refresh()
@@ -220,6 +218,15 @@ class WorkflowDesigner(BaseComponent):
return self.refresh_properties()
def get_workflow_name(self):
return self._designer_settings.workflow_name
def get_workflow_components(self):
return self._state.components.values()
def get_workflow_connections(self):
return self._state.connections
def __ft__(self):
return Div(
H1(f"{self._designer_settings.workflow_name}", cls="text-xl font-bold"),
@@ -227,7 +234,8 @@ class WorkflowDesigner(BaseComponent):
Div(
self._mk_media(),
self._mk_error_message(),
cls="flex mb-2"
cls="flex mb-2",
id=f"t_{self._id}"
),
self._mk_designer(),
Div(cls="wkf-splitter", id=f"s_{self._id}"),
@@ -268,21 +276,23 @@ class WorkflowDesigner(BaseComponent):
</svg>
"""
def _mk_component(self, component: WorkflowComponent, runtime_state: WorkflowComponentRuntimeState):
def _mk_component(self, component: WorkflowComponent):
runtime_state = self._player.get_component_runtime_state(component.id)
info = COMPONENT_TYPES[component.type]
is_selected = self._state.selected_component_id == component.id
tooltip_content = None
tooltip_class = ""
if runtime_state.state == ComponentState.FAILURE:
state_class = 'error' # To be styled with a red highlight
tooltip_content = runtime_state.error_message
tooltip_class = "mmt-tooltip"
elif runtime_state.state == ComponentState.NOT_RUN:
state_class = 'not-run' # To be styled as greyed-out
else:
state_class = ''
if runtime_state.state == ComponentState.FAILURE:
tooltip_content = runtime_state.error_message
tooltip_class = "mmt-tooltip"
else:
tooltip_content = None
tooltip_class = ""
return Div(
# Input connection point
@@ -315,8 +325,7 @@ class WorkflowDesigner(BaseComponent):
*[NotStr(self._mk_connection_svg(conn)) for conn in self._state.connections],
# Render components
*[self._mk_component(comp, state) for comp, state in zip(self._state.components.values(),
self._player.runtime_states.values())],
*[self._mk_component(comp) for comp in self._state.components.values()],
)
def _mk_canvas(self, oob=False):