sartography/spiff-arena
1
0
Fork 0
mirror of https://github.com/sartography/spiff-arena.git synced 2025-01-28 02:05:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

do not pass mocked data on unit tests to multi instance task types w/ burnettk

Browse Source
This commit is contained in:
jasquat 2023-09-25 15:48:06 -04:00
parent 7f666357ce
commit 4be95df48c
1 changed files with 16 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
spiffworkflow-backend/src/spiffworkflow_backend/services/process_model_test_runner_service.py
View File

@ -345,7 +345,15 @@ class ProcessModelTestRunner:
self, spiff_task: SpiffTask, test_case_task_key: str | None, test_case_task_properties: dict | None self, spiff_task: SpiffTask, test_case_task_key: str | None, test_case_task_properties: dict | None
) -> None: ) -> None:
task_data_for_submit = None task_data_for_submit = None
if test_case_task_key and test_case_task_properties and "data" in test_case_task_properties:
# do not bother mocking data on multi-instance task types since the mocked data should go to the children
# which will be the same type as the actual task like "UserTask" or "ManualTask".
if (
test_case_task_key
and test_case_task_properties
and "data" in test_case_task_properties
and not self._is_multi_instance_task(spiff_task)
):
if test_case_task_key not in self.task_data_index: if test_case_task_key not in self.task_data_index:
self.task_data_index[test_case_task_key] = 0 self.task_data_index[test_case_task_key] = 0
task_data_length = len(test_case_task_properties["data"]) task_data_length = len(test_case_task_properties["data"])
@ -359,6 +367,13 @@ class ProcessModelTestRunner:
self.task_data_index[test_case_task_key] += 1 self.task_data_index[test_case_task_key] += 1
self.process_model_test_runner_delegate.execute_task(spiff_task, task_data_for_submit) self.process_model_test_runner_delegate.execute_task(spiff_task, task_data_for_submit)
def _is_multi_instance_task(self, spiff_task: SpiffTask) -> bool:
return spiff_task.task_spec.__class__.__name__ in [
"ParallelMultiInstanceTask",
"SequentialMultiINstanceTask",
"StandardLoopTask",
]
def _get_next_task(self, bpmn_process_instance: BpmnWorkflow) -> SpiffTask | None: def _get_next_task(self, bpmn_process_instance: BpmnWorkflow) -> SpiffTask | None:
return self.process_model_test_runner_delegate.get_next_task(bpmn_process_instance) return self.process_model_test_runner_delegate.get_next_task(bpmn_process_instance)