mirror of
https://github.com/status-im/spiff-arena.git
synced 2025-01-17 21:51:43 +00:00
1bed0fb3ee
cd4da465 Merge pull request #264 from sartography/bugfix/dmn-equality-with-boolean 414a59eb disambiguate DMN expressions eea53c91 Merge pull request #263 from sartography/feature/cleanup-task-completion d248d5b1 execute postscript before other complete hook tasks c09f1a90 streamline predict & remove some duplicated calls to it 64c21791 remove duplicate calls to update 4ca1076d move task update to _on_complete to ensure data is copied consistently after task related activities are done d037a7eb small changes for readability 025bc30f Quick patch -- is_executable needs to be accurate immediately. 14d3d8c3 Merge pull request #262 from sartography/feature/parser_info_features 849c223e We are jumping through a lot of complex xml parsing in SpiffWorkflow-Backend because we need to know some basic information about a BPMN process at the moment it is saved. Rather than do that work in the backend, it seems better to have SpiffWorkflow handle parsing the xml and providing a bit of metadata, including: git-subtree-dir: SpiffWorkflow git-subtree-split: cd4da465e125ca1ae1b57d227bfa324d9d4c507c
99 lines
3.8 KiB
Python
99 lines
3.8 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import unittest
|
|
import random
|
|
|
|
from SpiffWorkflow.bpmn.workflow import BpmnWorkflow
|
|
from tests.SpiffWorkflow.camunda.BaseTestCase import BaseTestCase
|
|
|
|
__author__ = 'matth'
|
|
|
|
debug = True
|
|
|
|
class MultiInstanceParallelArrayTest(BaseTestCase):
|
|
"""The example bpmn diagram tests both a set cardinality from user input
|
|
as well as looping over an existing array."""
|
|
|
|
def setUp(self):
|
|
spec, subprocesses = self.load_workflow_spec('multi_instance_array_parallel.bpmn', 'MultiInstanceArray')
|
|
self.workflow = BpmnWorkflow(spec, subprocesses)
|
|
|
|
def testRunThroughHappy(self):
|
|
self.actual_test(False)
|
|
|
|
def testRunThroughSaveRestore(self):
|
|
self.actual_test(True)
|
|
|
|
def actual_test(self, save_restore=False):
|
|
|
|
first_task = self.workflow.task_tree
|
|
|
|
# A previous task (in this case the root task) will set the data
|
|
# so it must be found later.
|
|
first_task.update_data({"FamilySize": 3})
|
|
self.workflow.do_engine_steps()
|
|
if save_restore: self.reload_save_restore()
|
|
# Set initial array size to 3 in the first user form.
|
|
task = self.workflow.get_ready_user_tasks()[0]
|
|
self.assertEqual("Activity_FamSize", task.task_spec.name)
|
|
task.update_data({"FamilySize": 3})
|
|
self.workflow.complete_task_from_id(task.id)
|
|
if save_restore: self.reload_save_restore()
|
|
self.workflow.do_engine_steps()
|
|
|
|
# Set the names of the 3 family members.
|
|
for i in range(3):
|
|
|
|
tasks = self.workflow.get_ready_user_tasks()
|
|
self.assertEqual(len(tasks),1) # still with sequential MI
|
|
task = tasks[0]
|
|
if i > 0:
|
|
self.assertEqual("FamilyMemberTask"+"_%d"%(i-1), task.task_spec.name)
|
|
else:
|
|
self.assertEqual("FamilyMemberTask", task.task_spec.name)
|
|
task.update_data({"FamilyMember": {"FirstName": "The Funk #%i" % i}})
|
|
self.workflow.complete_task_from_id(task.id)
|
|
self.workflow.do_engine_steps()
|
|
if save_restore:
|
|
self.reload_save_restore()
|
|
tasks = self.workflow.get_ready_user_tasks()
|
|
|
|
self.assertEqual(3,len(tasks))
|
|
# Set the birthdays of the 3 family members.
|
|
for i in range(3): # emulate random Access
|
|
task = random.choice(tasks)
|
|
x = task.internal_data['runtimes'] -1
|
|
self.assertEqual("FamilyMemberBday", task.task_spec.name[:16])
|
|
self.assertEqual({"FirstName": "The Funk #%i" % x},
|
|
task.data["CurrentFamilyMember"])
|
|
task.update_data(
|
|
{"CurrentFamilyMember": {"Birthdate": "10/05/1985" + str(x)}})
|
|
self.workflow.do_engine_steps()
|
|
self.workflow.complete_task_from_id(task.id)
|
|
# We used to check that the current data variable was available in the task,
|
|
# but there's no reason to preserve it after the task completes. We removed it
|
|
# in some cases and left it in others, which just adds to the confusion.
|
|
self.workflow.do_engine_steps()
|
|
if save_restore:
|
|
self.reload_save_restore()
|
|
self.workflow.do_engine_steps()
|
|
|
|
tasks = self.workflow.get_ready_user_tasks()
|
|
|
|
self.workflow.do_engine_steps()
|
|
if save_restore:
|
|
self.reload_save_restore()
|
|
|
|
names = task.data['FamilyMembers']
|
|
bdays = task.data['FamilyMemberBirthday']
|
|
for x in list(names.keys()):
|
|
self.assertEqual(str(names[x]['FirstName'][-1]),str(bdays[x]['Birthdate'][-1]))
|
|
self.assertTrue(self.workflow.is_completed())
|
|
|
|
|
|
def suite():
|
|
return unittest.TestLoader().loadTestsFromTestCase(MultiInstanceParallelArrayTest)
|
|
|
|
if __name__ == '__main__':
|
|
unittest.TextTestRunner(verbosity=2).run(suite())
|