status-im
/
spiff-arena
mirror of https://github.com/status-im/spiff-arena.git
2
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fixing feedback from @jasquat

This commit is contained in:
danfunk 2023-05-12 09:19:46 -04:00
parent 1d37eed860
commit ca62c49de8
2 changed files with 10 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
spiffworkflow-backend/src/spiffworkflow_backend/services/process_instance_processor.py
View File

@ -40,7 +40,7 @@ from SpiffWorkflow.bpmn.serializer.task_spec import ( # type: ignore
EventBasedGatewayConverter,
)
from SpiffWorkflow.bpmn.serializer.workflow import BpmnWorkflowSerializer # type: ignore
from SpiffWorkflow.bpmn.specs.BpmnProcessSpec import BpmnProcessSpec # type: ignore
from SpiffWorkflow.bpmn.specs.bpmn_process_spec import BpmnProcessSpec # type: ignore
from SpiffWorkflow.bpmn.workflow import BpmnWorkflow # type: ignore
from SpiffWorkflow.exceptions import WorkflowException # type: ignore
from SpiffWorkflow.serializer.exceptions import MissingSpecError # type: ignore
@ -1111,7 +1111,6 @@ class ProcessInstanceProcessor:
def manual_complete_task(self, task_id: str, execute: bool) -> None:
"""Mark the task complete optionally executing it."""
time.time()
spiff_task = self.bpmn_process_instance.get_task_from_id(UUID(task_id))
event_type = ProcessInstanceEventType.task_skipped.value
start_time = time.time()

22
spiffworkflow-backend/tests/spiffworkflow_backend/unit/test_process_instance_processor.py
View File

@ -508,24 +508,20 @@ class TestProcessInstanceProcessor(BaseTest):
processor.manual_complete_task(str(human_task_one.task_id), execute=True)
processor.save()
processor = ProcessInstanceProcessor(process_instance)
assert (
processor.get_task_by_bpmn_identifier("step_1", processor.bpmn_process_instance).state # type: ignore
== TaskState.COMPLETED
)
assert (
processor.get_task_by_bpmn_identifier("Gateway_Open", processor.bpmn_process_instance).state # type: ignore
== TaskState.READY
)
step1_task = processor.get_task_by_bpmn_identifier("step_1", processor.bpmn_process_instance)
assert (step1_task is not None)
assert (step1_task.state == TaskState.COMPLETED)
gateway_task = processor.get_task_by_bpmn_identifier("Gateway_Open", processor.bpmn_process_instance)
assert (gateway_task is not None)
assert (gateway_task.state == TaskState.READY)
gateway_task = processor.bpmn_process_instance.get_tasks(TaskState.READY)[0]
processor.manual_complete_task(str(gateway_task.id), execute=True)
processor.save()
processor = ProcessInstanceProcessor(process_instance)
assert (
processor.get_task_by_bpmn_identifier("Gateway_Open", processor.bpmn_process_instance).state # type: ignore
== TaskState.COMPLETED
)
print(processor)
gateway_task = processor.get_task_by_bpmn_identifier("Gateway_Open", processor.bpmn_process_instance)
assert (gateway_task is not None)
assert (gateway_task.state == TaskState.COMPLETED)
def test_properly_saves_tasks_when_running(
self,