mirror of
https://github.com/sartography/spiff-arena.git
synced 2025-01-27 01:40:48 +00:00
c237e218b2
0e61be85 Merge pull request #289 from sartography/improvement/execution-and-serialization-cleanup 527684da fix some typos in the class & method docs 0dff44a4 Merge branch 'main' into improvement/execution-and-serialization-cleanup 64737498 Allow for other PythonScriptEngine environments besides task data (#288) dd63e916 remove some unused tests & diagrams 24aae519 clean up various small stuff 3b2dc35d use context when opening files for parsing 69eec3eb update class/method docs 24528dfb move all spec conversion classes to top level 5af33b11 remove some unused methods related to old serializer 931b90fb reorganize serializer 4e81ed29 consolidate pointless serializer classes d62acf02 change task_spec._update_hook to return a boolean indicating whether the task is ready git-subtree-dir: SpiffWorkflow git-subtree-split: 0e61be85c47474a33037e6f398e64c96e02f13ad
87 lines
3.7 KiB
Python
87 lines
3.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import datetime
|
|
import unittest
|
|
|
|
from SpiffWorkflow.bpmn.PythonScriptEngine import PythonScriptEngine
|
|
from SpiffWorkflow.bpmn.PythonScriptEngineEnvironment import TaskDataEnvironment
|
|
from SpiffWorkflow.bpmn.workflow import BpmnWorkflow
|
|
from tests.SpiffWorkflow.bpmn.BpmnWorkflowTestCase import BpmnWorkflowTestCase
|
|
|
|
__author__ = 'sartography'
|
|
|
|
class CustomScriptEngine(PythonScriptEngine):
|
|
"""This is a custom script processor that can be easily injected into Spiff Workflow.
|
|
It will execute python code read in from the bpmn. It will also make any scripts in the
|
|
scripts directory available for execution. """
|
|
def __init__(self):
|
|
environment = TaskDataEnvironment({
|
|
'timedelta': datetime.timedelta,
|
|
})
|
|
super().__init__(environment=environment)
|
|
|
|
class TooManyLoopsTest(BpmnWorkflowTestCase):
|
|
|
|
"""Looping back around many times would cause the tree of tasks to grow
|
|
for each loop, doing this a 100 or 1000 times would cause the system to
|
|
run fail in various ways. This assures that is no longer the case."""
|
|
|
|
def testRunThroughHappy(self):
|
|
self.actual_test(save_restore=False)
|
|
|
|
def testThroughSaveRestore(self):
|
|
self.actual_test(save_restore=True)
|
|
|
|
def actual_test(self,save_restore = False):
|
|
spec, subprocesses = self.load_workflow_spec('too_many_loops*.bpmn', 'loops')
|
|
self.workflow = BpmnWorkflow(spec, subprocesses, script_engine=CustomScriptEngine())
|
|
counter = 0
|
|
data = {}
|
|
while not self.workflow.is_completed():
|
|
self.workflow.do_engine_steps()
|
|
self.workflow.refresh_waiting_tasks()
|
|
if (self.workflow.last_task.data != data):
|
|
data = self.workflow.last_task.data
|
|
counter += 1 # There is a 10 millisecond wait task.
|
|
if save_restore:
|
|
self.save_restore()
|
|
self.workflow.script_engine = CustomScriptEngine()
|
|
self.assertEqual(20, self.workflow.last_task.data['counter'])
|
|
|
|
def test_with_sub_process(self):
|
|
# Found an issue where looping back would fail when it happens
|
|
# right after a sub-process. So assuring this is fixed.
|
|
counter = 0
|
|
spec, subprocesses = self.load_workflow_spec('too_many_loops_sub_process.bpmn', 'loops_sub')
|
|
self.workflow = BpmnWorkflow(spec, subprocesses, script_engine=CustomScriptEngine())
|
|
data = {}
|
|
while not self.workflow.is_completed():
|
|
self.workflow.do_engine_steps()
|
|
self.workflow.refresh_waiting_tasks()
|
|
if (self.workflow.last_task.data != data):
|
|
data = self.workflow.last_task.data
|
|
counter += 1 # There is a 10 millisecond wait task.
|
|
# self.save_restore()
|
|
self.assertEqual(20, self.workflow.last_task.data['counter'])
|
|
# One less, because we don't go back through once the first counter
|
|
# hits 20.
|
|
self.assertEqual(19, self.workflow.last_task.data['counter2'])
|
|
|
|
def test_with_two_call_activities(self):
|
|
spec, subprocess = self.load_workflow_spec('sub_in_loop*.bpmn', 'main')
|
|
self.workflow = BpmnWorkflow(spec, subprocess, script_engine=CustomScriptEngine())
|
|
self.workflow.do_engine_steps()
|
|
for loop in range(3):
|
|
ready = self.workflow.get_ready_user_tasks()
|
|
ready[0].data = { 'done': True if loop == 3 else False }
|
|
ready[0].complete()
|
|
self.workflow.refresh_waiting_tasks()
|
|
self.workflow.do_engine_steps()
|
|
self.save_restore()
|
|
self.workflow.script_engine = CustomScriptEngine()
|
|
|
|
def suite():
|
|
return unittest.TestLoader().loadTestsFromTestCase(TooManyLoopsTest)
|
|
if __name__ == '__main__':
|
|
unittest.TextTestRunner(verbosity=2).run(suite())
|