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

Merge branch 'dev' into feature/admin_impersonations

Browse Source
This commit is contained in:
Aaron Louie 2020-07-30 13:01:46 -04:00
commit f8944da31f
12 changed files with 519 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
Pipfile
View File

@ -45,6 +45,7 @@ webtest = "*"
werkzeug = "*"
xlrd = "*"
xlsxwriter = "*"
pygithub = "*"
[requires]
python_version = "3.7"

41
Pipfile.lock generated
View File

@ -1,7 +1,7 @@
{
"_meta": {
"hash": {
"sha256": "2057a84011229daa6b8a9491d729a0bae5225e6ce11c7ca45136d3c1fad85ec0"
"sha256": "381d29428eb328ad6167774b510b9d818bd1505b95f50454a19f1564782326cc"
},
"pipfile-spec": 6,
"requires": {
@ -242,6 +242,13 @@
"index": "pypi",
"version": "==5.2.1"
},
"deprecated": {
"hashes": [
"sha256:525ba66fb5f90b07169fdd48b6373c18f1ee12728ca277ca44567a367d9d7f74",
"sha256:a766c1dccb30c5f6eb2b203f87edd1d8588847709c78589e1521d769addc8218"
],
"version": "==1.2.10"
},
"docutils": {
"hashes": [
"sha256:0c5b78adfbf7762415433f5515cd5c9e762339e23369dbe8000d84a4bf4ab3af",
@ -693,6 +700,14 @@
"markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'",
"version": "==2.20"
},
"pygithub": {
"hashes": [
"sha256:8375a058ec651cc0774244a3bc7395cf93617298735934cdd59e5bcd9a1df96e",
"sha256:d2d17d1e3f4474e070353f201164685a95b5a92f5ee0897442504e399c7bc249"
],
"index": "pypi",
"version": "==1.51"
},
"pygments": {
"hashes": [
"sha256:647344a061c249a3b74e230c739f434d7ea4d8b1d5f3721bc0f3558049b38f44",
@ -1033,12 +1048,18 @@
"index": "pypi",
"version": "==1.0.1"
},
"wrapt": {
"hashes": [
"sha256:b62ffa81fb85f4332a4f609cab4ac40709470da05643a082ec1eb88e6d9b97d7"
],
"version": "==1.12.1"
},
"wtforms": {
"hashes": [
"sha256:6ff8635f4caeed9f38641d48cfe019d0d3896f41910ab04494143fc027866e1b",
"sha256:861a13b3ae521d6700dac3b2771970bd354a63ba7043ecc3a82b5288596a1972"
"sha256:43f19879b2a9b8dfd81d2e4e427ce44d3e5c09dbe08f2af8f4be9586b7dfc33d",
"sha256:715ebd303f47384bf6468fd9dfff52c6acc400e71204df8acfa6ef7bf40e1c27"
],
"version": "==2.3.1"
"version": "==2.3.2"
},
"xlrd": {
"hashes": [
@ -1050,11 +1071,11 @@
},
"xlsxwriter": {
"hashes": [
"sha256:828b3285fc95105f5b1946a6a015b31cf388bd5378fdc6604e4d1b7839df2e77",
"sha256:82a3b0e73e3913483da23791d1a25e4d2dbb3837d1be4129473526b9a270a5cc"
"sha256:3015f707cf237d277cf1b2d7805f409f0387e32bc52f3c76db9f85098980e828",
"sha256:ee3fc2f32890246aba44dd14d777d6b3135e3454f865d8cc669618e20152296b"
],
"index": "pypi",
"version": "==1.2.9"
"version": "==1.3.0"
},
"zipp": {
"hashes": [
@ -1178,11 +1199,11 @@
},
"pytest": {
"hashes": [
"sha256:869ec27f9b89964ccfe4fbdd5ccb8d3f285aaa3e9aa16a8491b9c8829148c230",
"sha256:a64d8fb4c15cdc70dae047352e980a197d855747cc885eb332cb73ddcc769168"
"sha256:85228d75db9f45e06e57ef9bf4429267f81ac7c0d742cc9ed63d09886a9fe6f4",
"sha256:8b6007800c53fdacd5a5c192203f4e531eb2a1540ad9c752e052ec0f7143dbad"
],
"index": "pypi",
"version": "==6.0.0"
"version": "==6.0.1"
},
"six": {
"hashes": [

3
config/default.py
View File

@ -46,6 +46,9 @@ PB_STUDY_DETAILS_URL = environ.get('PB_STUDY_DETAILS_URL', default=PB_BASE_URL +
LDAP_URL = environ.get('LDAP_URL', default="ldap.virginia.edu").strip('/') # No trailing slash or http://
LDAP_TIMEOUT_SEC = int(environ.get('LDAP_TIMEOUT_SEC', default=1))
# Github token
GITHUB_TOKEN = environ.get('GITHUB_TOKEN', None)
# Email configuration
DEFAULT_SENDER = 'askresearch@virginia.edu'
FALLBACK_EMAILS = ['askresearch@virginia.edu', 'sartographysupport@googlegroups.com']

24
crc/api/admin.py
View File

@ -3,19 +3,22 @@ import json
from flask import url_for
from flask_admin import Admin
from flask_admin.actions import action
from flask_admin.contrib import sqla
from flask_admin.contrib.sqla import ModelView
from sqlalchemy import desc
from werkzeug.utils import redirect
from jinja2 import Markup
from crc import db, app
from crc.api.user import verify_token, verify_token_admin
from crc.models.approval import ApprovalModel
from crc.models.file import FileModel
from crc.models.file import FileModel, FileDataModel
from crc.models.task_event import TaskEventModel
from crc.models.study import StudyModel
from crc.models.user import UserModel
from crc.models.workflow import WorkflowModel
from crc.services.file_service import FileService
class AdminModelView(sqla.ModelView):
@ -34,26 +37,40 @@ class AdminModelView(sqla.ModelView):
# redirect to login page if user doesn't have access
return redirect(url_for('home'))
class UserView(AdminModelView):
column_filters = ['uid']
class StudyView(AdminModelView):
column_filters = ['id', 'primary_investigator_id']
column_searchable_list = ['title']
class ApprovalView(AdminModelView):
column_filters = ['study_id', 'approver_uid']
class WorkflowView(AdminModelView):
column_filters = ['study_id', 'id']
class FileView(AdminModelView):
column_filters = ['workflow_id']
column_filters = ['workflow_id', 'type']
@action('publish', 'Publish', 'Are you sure you want to publish this file(s)?')
def action_publish(self, ids):
FileService.publish_to_github(ids)
@action('update', 'Update', 'Are you sure you want to update this file(s)?')
def action_update(self, ids):
FileService.update_from_github(ids)
def json_formatter(view, context, model, name):
value = getattr(model, name)
json_value = json.dumps(value, ensure_ascii=False, indent=2)
return Markup('<pre>{}</pre>'.format(json_value))
return Markup(f'<pre>{json_value}</pre>')
class TaskEventView(AdminModelView):
column_filters = ['workflow_id', 'action']
@ -62,6 +79,7 @@ class TaskEventView(AdminModelView):
'form_data': json_formatter,
}
admin = Admin(app)
admin.add_view(StudyView(StudyModel, db.session))

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:

49
crc/services/file_service.py
View File

@ -2,6 +2,7 @@ import hashlib
import json
import os
from datetime import datetime
from github import Github, UnknownObjectException
from uuid import UUID
from lxml import etree
@ -333,3 +334,51 @@ class FileService(object):
file_model.archived = True
session.commit()
app.logger.info("Failed to delete file, so archiving it instead. %i, due to %s" % (file_id, str(ie)))
@staticmethod
def update_from_github(file_ids):
gh_token = app.config['GITHUB_TOKEN']
_github = Github(gh_token)
repo = _github.get_user().get_repo('crispy-fiesta')
for file_id in file_ids:
file_data_model = FileDataModel.query.filter_by(
file_model_id=file_id
).order_by(
desc(FileDataModel.version)
).first()
try:
repo_file = repo.get_contents(file_data_model.file_model.name)
except UnknownObjectException:
# TODO: Add message indicating file is not in the repo
pass
else:
file_data_model.data = repo_file.decoded_content
session.add(file_data_model)
session.commit()
@staticmethod
def publish_to_github(file_ids):
gh_token = app.config['GITHUB_TOKEN']
_github = Github(gh_token)
repo = _github.get_user().get_repo('crispy-fiesta')
for file_id in file_ids:
file_data_model = FileDataModel.query.filter_by(file_model_id=file_id).first()
try:
repo_file = repo.get_contents(file_data_model.file_model.name)
except UnknownObjectException:
repo.create_file(
path=file_data_model.file_model.name,
message=f'Creating {file_data_model.file_model.name}',
content=file_data_model.data
)
return {'created': True}
else:
updated = repo.update_file(
path=repo_file.path,
message=f'Updating {file_data_model.file_model.name}',
content=file_data_model.data,
sha=repo_file.sha
)
return {'updated': True}

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>

95
tests/files/test_file_service.py
View File

@ -1,9 +1,45 @@
from github import UnknownObjectException
from sqlalchemy import desc
from tests.base_test import BaseTest
from unittest.mock import patch, Mock
from crc import db
from crc.models.file import FileDataModel
from crc.services.file_service import FileService
from crc.services.workflow_processor import WorkflowProcessor
class FakeGithubCreates(Mock):
def get_user(var):
class FakeUser(Mock):
def get_repo(var, name):
class FakeRepo(Mock):
def get_contents(var, filename):
raise UnknownObjectException(status='Failure', data='Failed data')
def update_file(var, path, message, content, sha):
pass
return FakeRepo()
return FakeUser()
class FakeGithub(Mock):
def get_user(var):
class FakeUser(Mock):
def get_repo(var, name):
class FakeRepo(Mock):
def get_contents(var, filename):
fake_file = Mock()
fake_file.decoded_content = b'Some bytes'
fake_file.path = '/el/path/'
fake_file.data = 'Serious data'
fake_file.sha = 'Sha'
return fake_file
def update_file(var, path, message, content, sha):
pass
return FakeRepo()
return FakeUser()
class TestFileService(BaseTest):
"""Largely tested via the test_file_api, and time is tight, but adding new tests here."""
@ -103,3 +139,62 @@ class TestFileService(BaseTest):
binary_data=b'5678')
file_models = FileService.get_workflow_files(workflow_id=workflow.id)
self.assertEqual(2, len(file_models))
@patch('crc.services.file_service.Github')
def test_update_from_github(self, mock_github):
mock_github.return_value = FakeGithub()
self.load_example_data()
self.create_reference_document()
workflow = self.create_workflow('file_upload_form')
processor = WorkflowProcessor(workflow)
task = processor.next_task()
irb_code = "UVACompl_PRCAppr" # The first file referenced in pb required docs.
file_model = FileService.add_workflow_file(workflow_id=workflow.id,
irb_doc_code=irb_code,
name="anything.png", content_type="text",
binary_data=b'1234')
FileService.update_from_github([file_model.id])
file_model_data = FileDataModel.query.filter_by(
file_model_id=file_model.id
).order_by(
desc(FileDataModel.version)
).first()
self.assertEqual(file_model_data.data, b'Some bytes')
@patch('crc.services.file_service.Github')
def test_publish_to_github_creates(self, mock_github):
mock_github.return_value = FakeGithubCreates()
self.load_example_data()
self.create_reference_document()
workflow = self.create_workflow('file_upload_form')
processor = WorkflowProcessor(workflow)
task = processor.next_task()
irb_code = "UVACompl_PRCAppr" # The first file referenced in pb required docs.
file_model = FileService.add_workflow_file(workflow_id=workflow.id,
irb_doc_code=irb_code,
name="anything.png", content_type="text",
binary_data=b'1234')
result = FileService.publish_to_github([file_model.id])
self.assertEqual(result['created'], True)
@patch('crc.services.file_service.Github')
def test_publish_to_github_updates(self, mock_github):
mock_github.return_value = FakeGithub()
self.load_example_data()
self.create_reference_document()
workflow = self.create_workflow('file_upload_form')
processor = WorkflowProcessor(workflow)
task = processor.next_task()
irb_code = "UVACompl_PRCAppr" # The first file referenced in pb required docs.
file_model = FileService.add_workflow_file(workflow_id=workflow.id,
irb_doc_code=irb_code,
name="anything.png", content_type="text",
binary_data=b'1234')
result = FileService.publish_to_github([file_model.id])
self.assertEqual(result['updated'], True)

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)')

2
tests/test_ldap_service.py
View File

@ -30,4 +30,4 @@ class TestLdapService(BaseTest):
user_info = LdapService.user_info("nosuch")
self.assertFalse(True, "An API error should be raised.")
except ApiError as ae:
self.assertEqual("missing_ldap_record", ae.code)
self.assertEqual("missing_ldap_record", ae.code)