sartography
/
cr-connect-workflow
mirror of https://github.com/sartography/cr-connect-workflow.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #487 from sartography/chore/update-get-workflow-status-664 

Chore/update get workflow status #664
This commit is contained in:
Dan Funk 2022-03-10 14:30:32 -05:00 committed by GitHub
parent e1625d45f0 d90bf1c977
commit 2d5517bd15
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 32 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
crc/scripts/get_workflow_status.py
View File

@ -1,6 +1,6 @@
from crc import session
from crc.api.common import ApiError
from crc.models.workflow import WorkflowModel
from crc.models.workflow import WorkflowModel, WorkflowStatus
from crc.scripts.script import Script
@ -19,21 +19,29 @@ Examples:
"""
def do_task_validate_only(self, task, study_id, workflow_id, *args, **kwargs):
return self.do_task(task, study_id, workflow_id, *args, **kwargs)
if 'workflow_spec_id' in kwargs.keys() or len(args) > 0:
return WorkflowStatus.not_started.value
else:
raise ApiError.from_task(code='missing_argument',
message='You must include a workflow_spec_id when calling the `get_workflow_status` script.',
task=task)
def do_task(self, task, study_id, workflow_id, *args, **kwargs):
if 'search_workflow_id' in kwargs.keys() or len(args) > 0:
if 'search_workflow_id' in kwargs.keys():
search_workflow_id = kwargs['search_workflow_id']
if 'workflow_spec_id' in kwargs.keys() or len(args) > 0:
if 'workflow_spec_id' in kwargs.keys():
workflow_spec_id = kwargs['workflow_spec_id']
else:
search_workflow_id = args[0]
workflow_model = session.query(WorkflowModel).filter(WorkflowModel.id == search_workflow_id).first()
workflow_spec_id = args[0]
workflow_model = session.query(WorkflowModel). \
filter(WorkflowModel.workflow_spec_id == workflow_spec_id). \
filter(WorkflowModel.study_id == study_id).\
first()
if workflow_model:
return workflow_model.status.value
else:
return f'No model found for workflow {search_workflow_id}.'
return WorkflowStatus.not_started.value
else:
raise ApiError.from_task(code='missing_argument',
message='You must include a workflow_id when calling the `get_workflow_status` script.',
message='You must include a workflow_spec_id when calling the `get_workflow_status` script.',
task=task)

26
tests/data/get_workflow_status/get_workflow_status.bpmn
View File

@ -8,7 +8,7 @@
<bpmn:scriptTask id="Activity_0y5dzit" name="Get Workflow Status Arg">
<bpmn:incoming>Flow_0wp4z9u</bpmn:incoming>
<bpmn:outgoing>Flow_0fl7rsj</bpmn:outgoing>
<bpmn:script>status_arg = get_workflow_status(search_workflow_id)</bpmn:script>
<bpmn:script>status_arg = get_workflow_status(workflow_spec_id)</bpmn:script>
</bpmn:scriptTask>
<bpmn:sequenceFlow id="Flow_0fl7rsj" sourceRef="Activity_0y5dzit" targetRef="Activity_StatusArg" />
<bpmn:manualTask id="Activity_StatusArg" name="Display Workflow Status Arg">
@ -21,7 +21,7 @@
<bpmn:scriptTask id="Activity_09p0m5x" name="Get Workflow Status Kwarg">
<bpmn:incoming>Flow_00x8h5p</bpmn:incoming>
<bpmn:outgoing>Flow_0lt7dwr</bpmn:outgoing>
<bpmn:script>status_kwarg = get_workflow_status(search_workflow_id=search_workflow_id)</bpmn:script>
<bpmn:script>status_kwarg = get_workflow_status(workflow_spec_id=workflow_spec_id)</bpmn:script>
</bpmn:scriptTask>
<bpmn:manualTask id="Activity_StatusKwarg" name="Display Workflow Status Kwarg">
<bpmn:documentation># Status Kwarg
@ -38,7 +38,7 @@
<bpmn:userTask id="Activity_WorkflowID" name="Get Workflow ID" camunda:formKey="WorkflowIDForm">
<bpmn:extensionElements>
<camunda:formData>
<camunda:formField id="search_workflow_id" label="'Workflow ID'" type="long" defaultValue="1">
<camunda:formField id="workflow_spec_id" label="&#39;Workflow Spec ID&#39;" type="string">
<camunda:validation>
<camunda:constraint name="required" config="True" />
</camunda:validation>
@ -51,6 +51,10 @@
</bpmn:process>
<bpmndi:BPMNDiagram id="BPMNDiagram_1">
<bpmndi:BPMNPlane id="BPMNPlane_1" bpmnElement="Process_0k8r3a2">
<bpmndi:BPMNEdge id="Flow_0wp4z9u_di" bpmnElement="Flow_0wp4z9u">
<di:waypoint x="340" y="117" />
<di:waypoint x="410" y="117" />
</bpmndi:BPMNEdge>
<bpmndi:BPMNEdge id="Flow_0eg806h_di" bpmnElement="Flow_0eg806h">
<di:waypoint x="990" y="117" />
<di:waypoint x="1052" y="117" />
@ -71,22 +75,15 @@
<di:waypoint x="188" y="117" />
<di:waypoint x="240" y="117" />
</bpmndi:BPMNEdge>
<bpmndi:BPMNEdge id="Flow_0wp4z9u_di" bpmnElement="Flow_0wp4z9u">
<di:waypoint x="340" y="117" />
<di:waypoint x="410" y="117" />
</bpmndi:BPMNEdge>
<bpmndi:BPMNShape id="_BPMNShape_StartEvent_2" bpmnElement="StartEvent_1">
<dc:Bounds x="152" y="99" width="36" height="36" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Activity_0bsr1f1_di" bpmnElement="Activity_0y5dzit">
<dc:Bounds x="410" y="77" width="100" height="80" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Activity_18ywo1x_di" bpmnElement="Activity_StatusArg">
<dc:Bounds x="570" y="77" width="100" height="80" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="_BPMNShape_StartEvent_2" bpmnElement="StartEvent_1">
<dc:Bounds x="152" y="99" width="36" height="36" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Activity_1s8egpj_di" bpmnElement="Activity_WorkflowID">
<dc:Bounds x="240" y="77" width="100" height="80" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Activity_0lhah9s_di" bpmnElement="Activity_09p0m5x">
<dc:Bounds x="730" y="77" width="100" height="80" />
</bpmndi:BPMNShape>
@ -96,6 +93,9 @@
<bpmndi:BPMNShape id="Event_1nvmkni_di" bpmnElement="Event_1nvmkni">
<dc:Bounds x="1052" y="99" width="36" height="36" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Activity_1s8egpj_di" bpmnElement="Activity_WorkflowID">
<dc:Bounds x="240" y="77" width="100" height="80" />
</bpmndi:BPMNShape>
</bpmndi:BPMNPlane>
</bpmndi:BPMNDiagram>
</bpmn:definitions>

4
tests/scripts/test_get_workflow_status.py
View File

@ -16,13 +16,13 @@ class TestGetWorkflowStatus(BaseTest):
def test_get_workflow_status(self):
self.create_workflow('random_fact')
workflow_model_1 = session.query(WorkflowModel).filter(WorkflowModel.id == 1).first()
search_workflow_id = workflow_model_1.id
workflow_spec_id = workflow_model_1.workflow_spec_id
workflow = self.create_workflow('get_workflow_status')
workflow_api = self.get_workflow_api(workflow)
task = workflow_api.next_task
# calls get_workflow_status(search_workflow_id)
workflow_api = self.complete_form(workflow, task, {'search_workflow_id': search_workflow_id})
workflow_api = self.complete_form(workflow, task, {'workflow_spec_id': workflow_spec_id})
task = workflow_api.next_task
self.assertEqual('Activity_StatusArg', task.name)
self.assertEqual(task.data['status_arg'], workflow_model_1.status.value)