sartography/cr-connect-workflow
1
0
Fork 0
mirror of https://github.com/sartography/cr-connect-workflow.git synced 2025-02-20 19:58:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #160 from sartography/feature/ldap_lookup_script

Browse Source 
Ldap lookup script
This commit is contained in:
Dan Funk 2020-07-30 11:27:38 -04:00 committed by GitHub
commit 37d1ba5d5a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 318 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

78
crc/scripts/ldap_lookup.py Normal file
View File

@ -0,0 +1,78 @@
import copy
from crc import app
from crc.api.common import ApiError
from crc.scripts.script import Script
from crc.services.ldap_service import LdapService
USER_DETAILS = {
"PIComputingID": {
"value": "",
"data": {
},
"label": "invalid uid"
}
}
class LdapLookup(Script):
"""This Script allows to be introduced as part of a workflow and called from there, taking
a UID as input and looking it up through LDAP to return the person's details """
def get_description(self):
return """
Attempts to create a dictionary with person details, using the
provided argument (a UID) and look it up through LDAP.
Example:
LdapLookup PIComputingID
"""
def do_task_validate_only(self, task, *args, **kwargs):
self.get_user_info(task, args)
def do_task(self, task, *args, **kwargs):
args = [arg for arg in args if type(arg) == str]
user_info = self.get_user_info(task, args)
user_details = copy.deepcopy(USER_DETAILS)
user_details['PIComputingID']['value'] = user_info['uid']
if len(user_info.keys()) > 1:
user_details['PIComputingID']['label'] = user_info.pop('label')
else:
user_info.pop('uid')
user_details['PIComputingID']['data'] = user_info
return user_details
def get_user_info(self, task, args):
if len(args) < 1:
raise ApiError(code="missing_argument",
message="Ldap lookup script requires one argument. The "
"UID for the person we want to look up")
arg = args.pop() # Extracting only one value for now
uid = task.workflow.script_engine.evaluate_expression(task, arg)
if not isinstance(uid, str):
raise ApiError(code="invalid_argument",
message="Ldap lookup script requires one 1 UID argument, of type string.")
user_info_dict = {}
try:
user_info = LdapService.user_info(uid)
user_info_dict = {
"display_name": user_info.display_name,
"given_name": user_info.given_name,
"email_address": user_info.email_address,
"telephone_number": user_info.telephone_number,
"title": user_info.title,
"department": user_info.department,
"affiliation": user_info.affiliation,
"sponsor_type": user_info.sponsor_type,
"uid": user_info.uid,
"label": user_info.proper_name()
}
except:
user_info_dict['uid'] = uid
app.logger.error(f'Ldap lookup failed for UID {uid}')
return user_info_dict

60
crc/scripts/ldap_replace.py Normal file
View File

@ -0,0 +1,60 @@
import copy
from crc import app
from crc.api.common import ApiError
from crc.scripts.script import Script
from crc.services.ldap_service import LdapService
class LdapReplace(Script):
"""This Script allows to be introduced as part of a workflow and called from there, taking
a UID (or several) as input and looking it up through LDAP to return the person's details """
def get_description(self):
return """
Attempts to create a dictionary with person details, using the
provided argument (a UID) and look it up through LDAP.
Examples:
#! LdapReplace supervisor
#! LdapReplace supervisor collaborator
#! LdapReplace supervisor cosupervisor collaborator
"""
def do_task_validate_only(self, task, *args, **kwargs):
self.set_users_info_in_task(task, args)
def do_task(self, task, *args, **kwargs):
args = [arg for arg in args if type(arg) == str]
self.set_users_info_in_task(task, args)
def set_users_info_in_task(self, task, args):
if len(args) < 1:
raise ApiError(code="missing_argument",
message="Ldap replace script requires at least one argument. The "
"UID for the person(s) we want to look up")
users_info = {}
for arg in args:
uid = task.workflow.script_engine.evaluate_expression(task, arg)
if not isinstance(uid, str):
raise ApiError(code="invalid_argument",
message="Ldap replace script found an invalid argument, type string is required")
user_info_dict = {}
try:
user_info = LdapService.user_info(uid)
user_info_dict = {
"display_name": user_info.display_name,
"given_name": user_info.given_name,
"email_address": user_info.email_address,
"telephone_number": user_info.telephone_number,
"title": user_info.title,
"department": user_info.department,
"affiliation": user_info.affiliation,
"sponsor_type": user_info.sponsor_type,
"uid": user_info.uid,
"proper_name": user_info.proper_name()
}
except:
app.logger.error(f'Ldap replace failed for UID {uid}')
task.data[arg] = user_info_dict

2
crc/scripts/request_approval.py
View File

@ -11,7 +11,7 @@ class RequestApproval(Script):
return """
Creates an approval request on this workflow, by the given approver_uid(s),"
Takes multiple arguments, which should point to data located in current task
or be quoted strings. The order is important. Approvals will be processed
or be quoted strings. The order is important. Approvals will be processed
in this order.
Example:

69
tests/data/ldap_replace/ldap_replace.bpmn Normal file
View File

@ -0,0 +1,69 @@
<?xml version="1.0" encoding="UTF-8"?>
<bpmn:definitions xmlns:bpmn="http://www.omg.org/spec/BPMN/20100524/MODEL" xmlns:bpmndi="http://www.omg.org/spec/BPMN/20100524/DI" xmlns:dc="http://www.omg.org/spec/DD/20100524/DC" xmlns:camunda="http://camunda.org/schema/1.0/bpmn" xmlns:di="http://www.omg.org/spec/DD/20100524/DI" id="Definitions_0y2dq4f" targetNamespace="http://bpmn.io/schema/bpmn" exporter="Camunda Modeler" exporterVersion="3.7.3">
<bpmn:process id="Process_0tad5ma" name="Set Recipients" isExecutable="true">
<bpmn:startEvent id="StartEvent_1">
<bpmn:outgoing>Flow_1synsig</bpmn:outgoing>
</bpmn:startEvent>
<bpmn:endEvent id="Event_0izrcj4">
<bpmn:incoming>Flow_11e7jgz</bpmn:incoming>
</bpmn:endEvent>
<bpmn:scriptTask id="Activity_0s5v97n" name="Ldap Replace">
<bpmn:incoming>Flow_08n2npe</bpmn:incoming>
<bpmn:outgoing>Flow_1xlrgne</bpmn:outgoing>
<bpmn:script>#! LdapReplace Supervisor Investigator</bpmn:script>
</bpmn:scriptTask>
<bpmn:sequenceFlow id="Flow_1synsig" sourceRef="StartEvent_1" targetRef="Activity_1l9vih3" />
<bpmn:sequenceFlow id="Flow_1xlrgne" sourceRef="Activity_0s5v97n" targetRef="Activity_0f78ek5" />
<bpmn:sequenceFlow id="Flow_08n2npe" sourceRef="Activity_1l9vih3" targetRef="Activity_0s5v97n" />
<bpmn:userTask id="Activity_1l9vih3" name="Set UIDs">
<bpmn:extensionElements>
<camunda:formData>
<camunda:formField id="Supervisor" label="Approver" type="string" />
<camunda:formField id="Investigator" label="Primary Investigator" type="string" />
</camunda:formData>
</bpmn:extensionElements>
<bpmn:incoming>Flow_1synsig</bpmn:incoming>
<bpmn:outgoing>Flow_08n2npe</bpmn:outgoing>
</bpmn:userTask>
<bpmn:sequenceFlow id="Flow_11e7jgz" sourceRef="Activity_0f78ek5" targetRef="Event_0izrcj4" />
<bpmn:userTask id="Activity_0f78ek5" name="Read UIDs">
<bpmn:incoming>Flow_1xlrgne</bpmn:incoming>
<bpmn:outgoing>Flow_11e7jgz</bpmn:outgoing>
</bpmn:userTask>
</bpmn:process>
<bpmndi:BPMNDiagram id="BPMNDiagram_1">
<bpmndi:BPMNPlane id="BPMNPlane_1" bpmnElement="Process_0tad5ma">
<bpmndi:BPMNEdge id="Flow_08n2npe_di" bpmnElement="Flow_08n2npe">
<di:waypoint x="370" y="117" />
<di:waypoint x="450" y="117" />
</bpmndi:BPMNEdge>
<bpmndi:BPMNEdge id="Flow_1xlrgne_di" bpmnElement="Flow_1xlrgne">
<di:waypoint x="550" y="117" />
<di:waypoint x="620" y="117" />
</bpmndi:BPMNEdge>
<bpmndi:BPMNEdge id="Flow_1synsig_di" bpmnElement="Flow_1synsig">
<di:waypoint x="215" y="117" />
<di:waypoint x="270" y="117" />
</bpmndi:BPMNEdge>
<bpmndi:BPMNEdge id="Flow_11e7jgz_di" bpmnElement="Flow_11e7jgz">
<di:waypoint x="720" y="117" />
<di:waypoint x="802" y="117" />
</bpmndi:BPMNEdge>
<bpmndi:BPMNShape id="_BPMNShape_StartEvent_2" bpmnElement="StartEvent_1">
<dc:Bounds x="179" y="99" width="36" height="36" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Activity_04imfm6_di" bpmnElement="Activity_0s5v97n">
<dc:Bounds x="450" y="77" width="100" height="80" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Activity_0xugr62_di" bpmnElement="Activity_1l9vih3">
<dc:Bounds x="270" y="77" width="100" height="80" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Event_0izrcj4_di" bpmnElement="Event_0izrcj4">
<dc:Bounds x="802" y="99" width="36" height="36" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Activity_17h05g6_di" bpmnElement="Activity_0f78ek5">
<dc:Bounds x="620" y="77" width="100" height="80" />
</bpmndi:BPMNShape>
</bpmndi:BPMNPlane>
</bpmndi:BPMNDiagram>
</bpmn:definitions>

110
tests/ldap/test_ldap_lookup_script.py Normal file
View File

@ -0,0 +1,110 @@
from tests.base_test import BaseTest
from crc.services.workflow_processor import WorkflowProcessor
from crc.scripts.ldap_replace import LdapReplace
from crc import db, mail
class TestLdapLookupScript(BaseTest):
def test_get_existing_user_details(self):
self.load_example_data()
self.create_reference_document()
workflow = self.create_workflow('empty_workflow')
processor = WorkflowProcessor(workflow)
task = processor.next_task()
task.data = {
'PIComputingID': 'dhf8r'
}
script = LdapReplace()
user_details = script.do_task(task, workflow.study_id, workflow.id, "PIComputingID")
self.assertEqual(task.data['PIComputingID']['display_name'], 'Dan Funk')
self.assertEqual(task.data['PIComputingID']['given_name'], 'Dan')
self.assertEqual(task.data['PIComputingID']['email_address'], 'dhf8r@virginia.edu')
self.assertEqual(task.data['PIComputingID']['telephone_number'], '+1 (434) 924-1723')
self.assertEqual(task.data['PIComputingID']['title'], 'E42:He\'s a hoopy frood')
self.assertEqual(task.data['PIComputingID']['department'], 'E0:EN-Eng Study of Parallel Universes')
self.assertEqual(task.data['PIComputingID']['affiliation'], 'faculty')
self.assertEqual(task.data['PIComputingID']['sponsor_type'], 'Staff')
self.assertEqual(task.data['PIComputingID']['uid'], 'dhf8r')
self.assertEqual(task.data['PIComputingID']['proper_name'], 'Dan Funk - (dhf8r)')
def test_get_existing_users_details(self):
self.load_example_data()
self.create_reference_document()
workflow = self.create_workflow('empty_workflow')
processor = WorkflowProcessor(workflow)
task = processor.next_task()
task.data = {
'supervisor': 'dhf8r',
'investigator': 'lb3dp'
}
script = LdapReplace()
user_details = script.do_task(task, workflow.study_id, workflow.id, "supervisor", "investigator")
self.assertEqual(task.data['supervisor']['display_name'], 'Dan Funk')
self.assertEqual(task.data['supervisor']['given_name'], 'Dan')
self.assertEqual(task.data['supervisor']['email_address'], 'dhf8r@virginia.edu')
self.assertEqual(task.data['supervisor']['telephone_number'], '+1 (434) 924-1723')
self.assertEqual(task.data['supervisor']['title'], 'E42:He\'s a hoopy frood')
self.assertEqual(task.data['supervisor']['department'], 'E0:EN-Eng Study of Parallel Universes')
self.assertEqual(task.data['supervisor']['affiliation'], 'faculty')
self.assertEqual(task.data['supervisor']['sponsor_type'], 'Staff')
self.assertEqual(task.data['supervisor']['uid'], 'dhf8r')
self.assertEqual(task.data['supervisor']['proper_name'], 'Dan Funk - (dhf8r)')
self.assertEqual(task.data['investigator']['display_name'], 'Laura Barnes')
self.assertEqual(task.data['investigator']['given_name'], 'Laura')
self.assertEqual(task.data['investigator']['email_address'], 'lb3dp@virginia.edu')
self.assertEqual(task.data['investigator']['telephone_number'], '+1 (434) 924-1723')
self.assertEqual(task.data['investigator']['title'], 'E0:Associate Professor of Systems and Information Engineering')
self.assertEqual(task.data['investigator']['department'], 'E0:EN-Eng Sys and Environment')
self.assertEqual(task.data['investigator']['affiliation'], 'faculty')
self.assertEqual(task.data['investigator']['sponsor_type'], 'Staff')
self.assertEqual(task.data['investigator']['uid'], 'lb3dp')
self.assertEqual(task.data['investigator']['proper_name'], 'Laura Barnes - (lb3dp)')
def test_get_invalid_user_details(self):
self.load_example_data()
self.create_reference_document()
workflow = self.create_workflow('empty_workflow')
processor = WorkflowProcessor(workflow)
task = processor.next_task()
task.data = {
'PIComputingID': 'rec3z'
}
script = LdapReplace()
user_details = script.do_task(task, workflow.study_id, workflow.id, "PIComputingID")
self.assertEqual(task.data['PIComputingID'], {})
def test_bpmn_task_receives_user_details(self):
workflow = self.create_workflow('ldap_replace')
task_data = {
'Supervisor': 'dhf8r',
'Investigator': 'lb3dp'
}
task = self.get_workflow_api(workflow).next_task
self.complete_form(workflow, task, task_data)
task = self.get_workflow_api(workflow).next_task
self.assertEqual(task.data['Supervisor']['display_name'], 'Dan Funk')
self.assertEqual(task.data['Supervisor']['given_name'], 'Dan')
self.assertEqual(task.data['Supervisor']['email_address'], 'dhf8r@virginia.edu')
self.assertEqual(task.data['Supervisor']['telephone_number'], '+1 (434) 924-1723')
self.assertEqual(task.data['Supervisor']['title'], 'E42:He\'s a hoopy frood')
self.assertEqual(task.data['Supervisor']['department'], 'E0:EN-Eng Study of Parallel Universes')
self.assertEqual(task.data['Supervisor']['affiliation'], 'faculty')
self.assertEqual(task.data['Supervisor']['sponsor_type'], 'Staff')
self.assertEqual(task.data['Supervisor']['uid'], 'dhf8r')
self.assertEqual(task.data['Supervisor']['proper_name'], 'Dan Funk - (dhf8r)')