sartography/cr-connect-workflow
1
0
Fork 0
mirror of https://github.com/sartography/cr-connect-workflow.git synced 2025-02-22 12:48:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

I had to give up and live with the idea that we can only render documentation on the current task, not on the previous or next tasks. I think this is ok. If you want to view a task, you need to make it the active task to assure all the parts and pieces are in place.

Browse Source
This commit is contained in:
Dan Funk 2020-05-06 13:01:38 -04:00
parent 8ded625c7d
commit 1571986c0e
4 changed files with 30 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
crc/api/workflow.py
View File

@ -103,7 +103,7 @@ def __get_workflow_api_model(processor: WorkflowProcessor):
) )
next_task = processor.next_task() next_task = processor.next_task()
if next_task: if next_task:
workflow_api.next_task = WorkflowService.spiff_task_to_api_task(next_task) workflow_api.next_task = WorkflowService.spiff_task_to_api_task(next_task, add_docs_and_forms=True)
return workflow_api return workflow_api

8
crc/services/workflow_service.py
View File

@ -62,7 +62,7 @@ class WorkflowService(object):
we.sender) we.sender)
@staticmethod @staticmethod
def spiff_task_to_api_task(spiff_task): def spiff_task_to_api_task(spiff_task, add_docs_and_forms=False):
task_type = spiff_task.task_spec.__class__.__name__ task_type = spiff_task.task_spec.__class__.__name__
if isinstance(spiff_task.task_spec, UserTask): if isinstance(spiff_task.task_spec, UserTask):
@ -109,8 +109,10 @@ class WorkflowService(object):
process_name=spiff_task.task_spec._wf_spec.description, process_name=spiff_task.task_spec._wf_spec.description,
properties=props) properties=props)
# Only process the form and documentation if this is something that is ready or completed. # Only process the form and documentation if requested.
if not (spiff_task._is_predicted()): # The task should be in a completed or a ready state, and should
# not be a previously completed MI Task.
if add_docs_and_forms:
if hasattr(spiff_task.task_spec, "form"): if hasattr(spiff_task.task_spec, "form"):
task.form = spiff_task.task_spec.form task.form = spiff_task.task_spec.form
for field in task.form.fields: for field in task.form.fields:

12
tests/data/multi_instance_parallel/multi_instance_parallel.bpmn
View File

@ -49,18 +49,18 @@
<dc:Bounds x="129" y="142" width="64" height="14" /> <dc:Bounds x="129" y="142" width="64" height="14" />
</bpmndi:BPMNLabel> </bpmndi:BPMNLabel>
</bpmndi:BPMNShape> </bpmndi:BPMNShape>
<bpmndi:BPMNShape id="ScriptTask_0cbbirp_di" bpmnElement="Task_1v0e2zu">
<dc:Bounds x="250" y="77" width="100" height="80" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Activity_1iyilui_di" bpmnElement="MutiInstanceTask">
<dc:Bounds x="410" y="77" width="100" height="80" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Event_1g0pmib_di" bpmnElement="Event_End"> <bpmndi:BPMNShape id="Event_1g0pmib_di" bpmnElement="Event_End">
<dc:Bounds x="582" y="99" width="36" height="36" /> <dc:Bounds x="582" y="99" width="36" height="36" />
<bpmndi:BPMNLabel> <bpmndi:BPMNLabel>
<dc:Bounds x="575" y="142" width="54" height="14" /> <dc:Bounds x="575" y="142" width="54" height="14" />
</bpmndi:BPMNLabel> </bpmndi:BPMNLabel>
</bpmndi:BPMNShape> </bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Activity_1iyilui_di" bpmnElement="MutiInstanceTask">
<dc:Bounds x="410" y="77" width="100" height="80" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="ScriptTask_0cbbirp_di" bpmnElement="Task_1v0e2zu">
<dc:Bounds x="250" y="77" width="100" height="80" />
</bpmndi:BPMNShape>
</bpmndi:BPMNPlane> </bpmndi:BPMNPlane>
</bpmndi:BPMNDiagram> </bpmndi:BPMNDiagram>
</bpmn:definitions> </bpmn:definitions>

35
tests/test_tasks_api.py
View File

@ -79,10 +79,11 @@ class TestTasksApi(BaseTest):
def test_get_current_user_tasks(self): def test_get_current_user_tasks(self):
self.load_example_data() self.load_example_data()
workflow = self.create_workflow('random_fact') workflow = self.create_workflow('random_fact')
tasks = self.get_workflow_api(workflow).user_tasks workflow = self.get_workflow_api(workflow)
self.assertEqual("Task_User_Select_Type", tasks[0].name) task = workflow.next_task
self.assertEqual(3, len(tasks[0].form["fields"][0]["options"])) self.assertEqual("Task_User_Select_Type", task['name'])
self.assertIsNotNone(tasks[0].documentation) self.assertEqual(3, len(task['form']["fields"][0]["options"]))
self.assertIsNotNone(task['documentation'])
expected_docs = """# h1 Heading 8-) expected_docs = """# h1 Heading 8-)
## h2 Heading ## h2 Heading
### h3 Heading ### h3 Heading
@ -90,7 +91,7 @@ class TestTasksApi(BaseTest):
##### h5 Heading ##### h5 Heading
###### h6 Heading ###### h6 Heading
""" """
self.assertTrue(str.startswith(tasks[0].documentation, expected_docs)) self.assertTrue(str.startswith(task['documentation'], expected_docs))
def test_two_forms_task(self): def test_two_forms_task(self):
# Set up a new workflow # Set up a new workflow
@ -100,7 +101,7 @@ class TestTasksApi(BaseTest):
workflow_api = self.get_workflow_api(workflow) workflow_api = self.get_workflow_api(workflow)
self.assertEqual('two_forms', workflow_api.workflow_spec_id) self.assertEqual('two_forms', workflow_api.workflow_spec_id)
self.assertEqual(2, len(workflow_api.user_tasks)) self.assertEqual(2, len(workflow_api.user_tasks))
self.assertIsNotNone(workflow_api.user_tasks[0].form) self.assertIsNotNone(workflow_api.next_task['form'])
self.assertEqual("UserTask", workflow_api.next_task['type']) self.assertEqual("UserTask", workflow_api.next_task['type'])
self.assertEqual("StepOne", workflow_api.next_task['name']) self.assertEqual("StepOne", workflow_api.next_task['name'])
self.assertEqual(1, len(workflow_api.next_task['form']['fields'])) self.assertEqual(1, len(workflow_api.next_task['form']['fields']))
@ -194,10 +195,10 @@ class TestTasksApi(BaseTest):
self.load_example_data() self.load_example_data()
workflow = self.create_workflow('random_fact') workflow = self.create_workflow('random_fact')
workflow_api = self.get_workflow_api(workflow) workflow_api = self.get_workflow_api(workflow)
tasks = workflow_api.user_tasks task = workflow_api.next_task
self.assertEqual("Task_User_Select_Type", tasks[0].name) self.assertEqual("Task_User_Select_Type", task['name'])
self.assertEqual(3, len(tasks[0].form["fields"][0]["options"])) self.assertEqual(3, len(task['form']["fields"][0]["options"]))
self.assertIsNotNone(tasks[0].documentation) self.assertIsNotNone(task['documentation'])
self.complete_form(workflow, workflow_api.user_tasks[0], {"type": "norris"}) self.complete_form(workflow, workflow_api.user_tasks[0], {"type": "norris"})
workflow_api = self.get_workflow_api(workflow) workflow_api = self.get_workflow_api(workflow)
self.assertEqual("EndEvent_0u1cgrf", workflow_api.next_task['name']) self.assertEqual("EndEvent_0u1cgrf", workflow_api.next_task['name'])
@ -303,11 +304,11 @@ class TestTasksApi(BaseTest):
self.load_example_data() self.load_example_data()
workflow = self.create_workflow('enum_options_with_search') workflow = self.create_workflow('enum_options_with_search')
# get the first form in the two form workflow. # get the first form in the two form workflow.
tasks = self.get_workflow_api(workflow).user_tasks workflow = self.get_workflow_api(workflow)
task = tasks[0] task = workflow.next_task
field_id = task.form['fields'][0]['id'] field_id = task['form']['fields'][0]['id']
rv = self.app.get('/v1.0/workflow/%i/task/%s/lookup/%s?query=%s&limit=5' % rv = self.app.get('/v1.0/workflow/%i/task/%s/lookup/%s?query=%s&limit=5' %
(workflow.id, task.id, field_id, 'c'), # All records with a word that starts with 'c' (workflow.id, task['id'], field_id, 'c'), # All records with a word that starts with 'c'
headers=self.logged_in_headers(), headers=self.logged_in_headers(),
content_type="application/json") content_type="application/json")
self.assert_success(rv) self.assert_success(rv)
@ -395,13 +396,13 @@ class TestTasksApi(BaseTest):
self.assertEquals("MutiInstanceTask", tasks[0].name) self.assertEquals("MutiInstanceTask", tasks[0].name)
self.assertEquals("Gather more information", tasks[0].title) self.assertEquals("Gather more information", tasks[0].title)
self.complete_form(workflow, tasks[0], {"email": "dhf8r@virginia.edu"}) self.complete_form(workflow, tasks[0], {"investigator":{"email": "dhf8r@virginia.edu"}})
tasks = self.get_workflow_api(workflow).user_tasks tasks = self.get_workflow_api(workflow).user_tasks
self.complete_form(workflow, tasks[2], {"email": "abc@virginia.edu"}) self.complete_form(workflow, tasks[2], {"investigator":{"email": "abc@virginia.edu"}})
tasks = self.get_workflow_api(workflow).user_tasks tasks = self.get_workflow_api(workflow).user_tasks
self.complete_form(workflow, tasks[1], {"email": "def@virginia.edu"}) self.complete_form(workflow, tasks[1], {"investigator":{"email": "def@virginia.edu"}})
tasks = self.get_workflow_api(workflow).user_tasks tasks = self.get_workflow_api(workflow).user_tasks
workflow = self.get_workflow_api(workflow) workflow = self.get_workflow_api(workflow)