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-12 18:40:36 +02:00
parent bdd954b243
commit fdf05edec3
5 changed files with 239 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
requirements.txt
View File

@@ -31,6 +31,7 @@ pydantic-settings==2.9.1
pydantic_core==2.33.2
Pygments==2.19.1
pytest==8.3.3
pytest-mock==3.14.1
python-dateutil==2.9.0.post0
python-dotenv==1.0.1
python-fasthtml==0.12.21

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

@@ -58,7 +58,7 @@ class WorkflowPlayer(BaseComponent):
self._datagrid.set_boundaries(boundaries)
def run(self):
self._reset_state(state=ComponentState.NOT_RUN)
self._reset_state()
components_by_id = {c.id: c for c in self._player_settings.components}
@@ -123,6 +123,9 @@ class WorkflowPlayer(BaseComponent):
def stop(self):
self._reset_state()
def get_dataframe(self):
return self._datagrid.get_dataframe()
def __ft__(self):
return Div(

34
tests/my_mocks.py Normal file
View File

@@ -0,0 +1,34 @@
from unittest.mock import MagicMock
import pytest
from fasthtml.components import *
from components.tabs.components.MyTabs import MyTabs
@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()

192
tests/test_workflow_player.py Normal file
View File

@@ -0,0 +1,192 @@
from unittest.mock import MagicMock
import pandas as pd
import pytest
from pandas.testing import assert_frame_equal
from components.workflows.components.WorkflowDesigner import COMPONENT_TYPES
from components.workflows.components.WorkflowPlayer import WorkflowPlayer, WorkflowsPlayerError
from components.workflows.constants import ProcessorTypes
from components.workflows.db_management import WorkflowsPlayerSettings, WorkflowComponent, Connection, ComponentState
from core.settings_management import SettingsManager, MemoryDbEngine
from my_mocks import tabs_manager
TEST_WORKFLOW_PLAYER_ID = "workflow_player_id"
@pytest.fixture
def player(session, tabs_manager):
"""
Sets up a standard WorkflowPlayer instance with a 3-component linear workflow.
A helper method 'get_dataframe' is attached for easier testing.
"""
components = [
WorkflowComponent(
"comp_producer",
ProcessorTypes.Producer,
10, 100,
COMPONENT_TYPES[ProcessorTypes.Producer]["title"],
COMPONENT_TYPES[ProcessorTypes.Producer]["description"],
{"processor_name": "Repository"}
),
WorkflowComponent(
"comp_filter",
ProcessorTypes.Filter,
40, 100,
COMPONENT_TYPES[ProcessorTypes.Filter]["title"],
COMPONENT_TYPES[ProcessorTypes.Filter]["description"],
{"processor_name": "Default"}
),
WorkflowComponent(
"comp_presenter",
ProcessorTypes.Presenter,
70, 100,
COMPONENT_TYPES[ProcessorTypes.Presenter]["title"],
COMPONENT_TYPES[ProcessorTypes.Presenter]["description"],
{"processor_name": "Default"}
)
]
connections = [
Connection("conn_1", "comp_producer", "comp_filter"),
Connection("conn_2", "comp_filter", "comp_presenter"),
]
return WorkflowPlayer(session=session,
_id=TEST_WORKFLOW_PLAYER_ID,
settings_manager=SettingsManager(engine=MemoryDbEngine()),
tabs_manager=tabs_manager,
player_settings=WorkflowsPlayerSettings("Workflow Name", components, connections),
boundaries={"height": 500, "width": 800}
)
def test_run_successful_workflow(player, mocker):
"""
Tests the "happy path" where the workflow runs successfully from start to finish.
"""
# 1. Arrange: Mock a successful engine run
mock_engine = MagicMock()
mock_engine.has_error = False
mock_result_data = [
MagicMock(as_dict=lambda: {'col_a': 1, 'col_b': 'x'}),
MagicMock(as_dict=lambda: {'col_a': 2, 'col_b': 'y'})
]
mock_engine.run_to_list.return_value = mock_result_data
mocker.patch.object(player, '_get_engine', return_value=mock_engine)
# 2. Act
player.run()
# 3. Assert: Check for success state and correct data
assert not player.has_error
assert player.global_error is None
for component_id, state in player.runtime_states.items():
assert state.state == ComponentState.SUCCESS
player._get_engine.assert_called_once()
mock_engine.run_to_list.assert_called_once()
expected_df = pd.DataFrame([row.as_dict() for row in mock_result_data])
assert_frame_equal(player.get_dataframe(), expected_df)
def test_run_with_cyclical_dependency(player, mocker):
"""
Tests that a workflow with a cycle is detected and handled before execution.
"""
# 1. Arrange: Introduce a cycle and spy on engine creation
player._player_settings.connections.append(Connection("conn_3", "comp_presenter", "comp_producer"))
spy_get_engine = mocker.spy(player, '_get_engine')
# 2. Act
player.run()
# 3. Assert: Check for the specific cycle error
assert player.has_error
assert "Workflow configuration error: A cycle was detected" in player.global_error
assert player.get_dataframe().empty
spy_get_engine.assert_not_called()
def test_run_with_component_initialization_failure(player, mocker):
"""
Tests that an error during a component's initialization is handled correctly.
"""
# 1. Arrange: Make the engine creation fail for a specific component
failing_component_id = "comp_filter"
error = ValueError("Missing a required property")
mocker.patch.object(player, '_get_engine', side_effect=WorkflowsPlayerError(failing_component_id, error))
# 2. Act
player.run()
# 3. Assert: Check that the specific component is marked as failed
assert player.has_error
assert f"Failed to init component '{failing_component_id}'" in player.global_error
assert player.runtime_states[failing_component_id].state == ComponentState.FAILURE
assert str(error) in player.runtime_states[failing_component_id].error_message
assert player.runtime_states["comp_producer"].state == ComponentState.NOT_RUN
def test_run_with_failure_in_middle_component(player, mocker):
"""
Tests failure in a middle component updates all component states correctly.
"""
# 1. Arrange: Mock an engine that fails at the filter component
mock_engine = MagicMock()
mock_engine.has_error = True
failing_component_id = "comp_filter"
error = RuntimeError("Data processing failed unexpectedly")
mock_engine.errors = {failing_component_id: error}
mock_engine.run_to_list.return_value = []
mocker.patch.object(player, '_get_engine', return_value=mock_engine)
# 2. Act
player.run()
# 3. Assert: Check the state of each component in the chain
assert player.has_error
assert f"Error in component 'Default': {error}" in player.global_error
assert player.runtime_states["comp_producer"].state == ComponentState.SUCCESS
assert player.runtime_states[failing_component_id].state == ComponentState.FAILURE
assert str(error) in player.runtime_states[failing_component_id].error_message
assert player.runtime_states["comp_presenter"].state == ComponentState.NOT_RUN
def test_run_with_empty_workflow(player, mocker):
"""
Tests that running a workflow with no components completes without errors.
"""
# 1. Arrange: Clear components and connections
player._player_settings.components = []
player._player_settings.connections = []
player.runtime_states = {}
spy_get_engine = mocker.spy(player, '_get_engine')
# 2. Act
player.run()
# 3. Assert: Ensure it finishes cleanly with no data
assert not player.has_error
assert player.global_error is None
assert player.get_dataframe().empty
spy_get_engine.assert_called_once()
def test_run_with_global_engine_error(player, mocker):
"""
Tests a scenario where the engine reports a global error not tied to a specific component.
"""
# 1. Arrange: Mock a global engine failure
mock_engine = MagicMock()
mock_engine.has_error = True
mock_engine.errors = {} # No specific component error
mock_engine.global_error = "A simulated global engine failure"
mock_engine.run_to_list.return_value = []
mocker.patch.object(player, '_get_engine', return_value=mock_engine)
# 2. Act
player.run()
# 3. Assert: The player should report the global error from the engine
assert player.has_error
assert player.global_error == mock_engine.global_error

45
tests/test_workflows.py
View File

@@ -1,44 +1,15 @@
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
from my_mocks import tabs_manager
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()
boundaries = {"height": 500, "width": 800}
@pytest.fixture
def workflows(session, tabs_manager):
@@ -117,7 +88,7 @@ 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", {})
actual = workflows.add_new_workflow(tab_id, "Not relevant here", "New Workflow", boundaries)
expected = (
Div(
@@ -134,11 +105,11 @@ def test_i_can_add_a_new_workflow(workflows, tabs_manager):
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", {})
workflows.add_new_workflow("tab_id_1", "Not relevant", "workflow 1", boundaries)
workflows.add_new_workflow("tab_id_2", "Not relevant", "workflow 2", boundaries)
workflows.add_new_workflow("tab_id_3", "Not relevant", "workflow 3", boundaries)
actual = workflows.show_workflow("workflow 2", {})
actual = workflows.show_workflow("workflow 2", boundaries)
expected = (
Div(
@@ -150,4 +121,4 @@ def test_i_can_select_a_workflow(workflows):
Div(), # Workflow Designer embedded in the tab
)
assert matches(actual, expected)
assert matches(actual, expected)