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 #243 from sartography/183-remove-study-approval-code 

Remove Approvals per ticket 183
This commit is contained in:
Dan Funk 2021-02-22 18:12:02 -05:00 committed by GitHub
parent 17941ba7fc 003bf2f9b9
commit 53fa49e668
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
18 changed files with 50 additions and 1368 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

150
crc/api.yml
View File

@ -357,30 +357,6 @@ paths:
application/json:
schema:
$ref: "#/components/schemas/Study"
/study/{study_id}/approvals:
parameters:
- name: study_id
in: path
required: true
description: The id of the study for which workflows should be returned.
schema:
type: integer
format: int32
get:
operationId: crc.api.approval.get_approvals_for_study
summary: Returns approvals for a single study
tags:
- Studies
- Approvals
responses:
'200':
description: An array of approvals
content:
application/json:
schema:
type: array
items:
$ref: "#/components/schemas/Approval"
/workflow-specification:
get:
operationId: crc.api.workflow.all_specifications
@ -1135,132 +1111,6 @@ paths:
text/plain:
schema:
type: string
/approval-counts:
parameters:
- name: as_user
in: query
required: false
description: If provided, returns the approval counts for that user.
schema:
type: string
get:
operationId: crc.api.approval.get_approval_counts
summary: Provides counts for approvals by status for the given user, or all users if no user is provided
tags:
- Approvals
responses:
'200':
description: An dictionary of Approval Statuses and the counts for each
content:
application/json:
schema:
type: array
items:
$ref: "#/components/schemas/ApprovalCounts"
/all_approvals:
parameters:
- name: status
in: query
required: false
description: If set to true, returns all the approvals with any status. Defaults to false, leaving out canceled approvals.
schema:
type: boolean
get:
operationId: crc.api.approval.get_all_approvals
summary: Provides a list of all workflows approvals
tags:
- Approvals
responses:
'200':
description: An array of approvals
content:
application/json:
schema:
type: array
items:
$ref: "#/components/schemas/Approval"
/approval:
parameters:
- name: status
in: query
required: false
description: If provided, returns just approvals for the given status.
schema:
type: string
- name: as_user
in: query
required: false
description: If provided, returns the approval results as they would appear for that user.
schema:
type: string
get:
operationId: crc.api.approval.get_approvals
summary: Provides a list of workflows approvals
tags:
- Approvals
responses:
'200':
description: An array of approvals
content:
application/json:
schema:
type: array
items:
$ref: "#/components/schemas/Approval"
/approval/{approval_id}:
parameters:
- name: approval_id
in: path
required: true
description: The id of the approval in question.
schema:
type: integer
format: int32
put:
operationId: crc.api.approval.update_approval
summary: Updates an approval with the given parameters
tags:
- Approvals
requestBody:
content:
application/json:
schema:
$ref: '#/components/schemas/Approval'
responses:
'200':
description: Study updated successfully.
content:
application/json:
schema:
$ref: "#/components/schemas/Approval"
/approval/csv:
get:
operationId: crc.api.approval.get_csv
summary: Provides a list of all users for all approved studies
tags:
- Approvals
responses:
'200':
description: An array of approvals
content:
application/json:
schema:
type: object
/health_attesting:
get:
operationId: crc.api.approval.get_health_attesting_csv
summary: Returns a CSV file with health attesting records
tags:
- Approvals
responses:
'200':
description: A CSV file
content:
application/json:
schema:
type: array
items:
$ref: "#/components/schemas/Approval"
/datastore:
post:
operationId: crc.api.data_store.add_datastore

2
crc/api/admin.py
View File

@ -12,7 +12,6 @@ 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, FileDataModel
from crc.models.task_event import TaskEventModel
from crc.models.study import StudyModel
@ -83,7 +82,6 @@ class TaskEventView(AdminModelView):
admin = Admin(app)
admin.add_view(StudyView(StudyModel, db.session))
admin.add_view(ApprovalView(ApprovalModel, db.session))
admin.add_view(UserView(UserModel, db.session))
admin.add_view(WorkflowView(WorkflowModel, db.session))
admin.add_view(FileView(FileModel, db.session))

137
crc/api/approval.py
View File

@ -1,137 +0,0 @@
import csv
import io
import json
import pickle
from base64 import b64decode
from datetime import datetime
from flask import g, make_response
from crc import db, session
from crc.api.common import ApiError
from crc.models.approval import Approval, ApprovalModel, ApprovalSchema, ApprovalStatus
from crc.models.workflow import WorkflowModel
from crc.services.approval_service import ApprovalService
from crc.services.ldap_service import LdapService
# Returns counts of approvals in each status group assigned to the given user.
# The goal is to return results as quickly as possible.
def get_approval_counts(as_user=None):
uid = as_user or g.user.uid
db_user_approvals = db.session.query(ApprovalModel)\
.filter_by(approver_uid=uid)\
.filter(ApprovalModel.status != ApprovalStatus.CANCELED.name)\
.all()
study_ids = [a.study_id for a in db_user_approvals]
db_other_approvals = db.session.query(ApprovalModel)\
.filter(ApprovalModel.study_id.in_(study_ids))\
.filter(ApprovalModel.approver_uid != uid)\
.filter(ApprovalModel.status != ApprovalStatus.CANCELED.name)\
.all()
# Make a dict of the other approvals where the key is the study id and the value is the approval
# TODO: This won't work if there are more than 2 approvals with the same study_id
other_approvals = {}
for approval in db_other_approvals:
other_approvals[approval.study_id] = approval
counts = {}
for name, value in ApprovalStatus.__members__.items():
counts[name] = 0
for approval in db_user_approvals:
# Check if another approval has the same study id
if approval.study_id in other_approvals:
other_approval = other_approvals[approval.study_id]
# Other approval takes precedence over this one
if other_approval.id < approval.id:
if other_approval.status == ApprovalStatus.PENDING.name:
counts[ApprovalStatus.AWAITING.name] += 1
elif other_approval.status == ApprovalStatus.DECLINED.name:
counts[ApprovalStatus.DECLINED.name] += 1
elif other_approval.status == ApprovalStatus.CANCELED.name:
counts[ApprovalStatus.CANCELED.name] += 1
elif other_approval.status == ApprovalStatus.APPROVED.name:
counts[approval.status] += 1
else:
counts[approval.status] += 1
else:
counts[approval.status] += 1
return counts
def get_all_approvals(status=None):
approvals = ApprovalService.get_all_approvals(include_cancelled=status is True)
results = ApprovalSchema(many=True).dump(approvals)
return results
def get_approvals(status=None, as_user=None):
#status = ApprovalStatus.PENDING.value
user = g.user.uid
if as_user:
user = as_user
approvals = ApprovalService.get_approvals_per_user(user, status,
include_cancelled=False)
results = ApprovalSchema(many=True).dump(approvals)
return results
def get_approvals_for_study(study_id=None):
db_approvals = ApprovalService.get_approvals_for_study(study_id)
approvals = [Approval.from_model(approval_model) for approval_model in db_approvals]
results = ApprovalSchema(many=True).dump(approvals)
return results
def get_health_attesting_csv():
records = ApprovalService.get_health_attesting_records()
si = io.StringIO()
cw = csv.writer(si)
cw.writerows(records)
output = make_response(si.getvalue())
output.headers["Content-Disposition"] = "attachment; filename=health_attesting.csv"
output.headers["Content-type"] = "text/csv"
return output
# ----- Begin descent into madness ---- #
def get_csv():
"""A damn lie, it's a json file. A huge bit of a one-off for RRT, but 3 weeks of midnight work can convince a
man to do just about anything"""
content = ApprovalService.get_not_really_csv_content()
return content
# ----- come back to the world of the living ---- #
def update_approval(approval_id, body):
if approval_id is None:
raise ApiError('unknown_approval', 'Please provide a valid Approval ID.')
approval_model = session.query(ApprovalModel).get(approval_id)
if approval_model is None:
raise ApiError('unknown_approval', 'The approval "' + str(approval_id) + '" is not recognized.')
if approval_model.approver_uid != g.user.uid:
raise ApiError("not_your_approval", "You may not modify this approval. It belongs to another user.")
approval_model.status = body['status']
approval_model.message = body['message']
approval_model.date_approved = datetime.now()
session.add(approval_model)
session.commit()
# Called only to send emails
approver = body['approver']['uid']
ApprovalService.update_approval(approval_id, approver)
result = ApprovalSchema().dump(approval_model)
return result

118
crc/models/approval.py
View File

@ -1,118 +0,0 @@
import enum
import marshmallow
from marshmallow import INCLUDE, fields
from sqlalchemy import func
from crc import db, ma, app
from crc.api.common import ApiError
from crc.models.file import FileDataModel
from crc.models.ldap import LdapSchema
from crc.models.study import StudyModel
from crc.models.workflow import WorkflowModel
from crc.services.file_service import FileService
from crc.services.ldap_service import LdapService
class ApprovalStatus(enum.Enum):
PENDING = "PENDING" # no one has done jack.
APPROVED = "APPROVED" # approved by the reviewer
DECLINED = "DECLINED" # rejected by the reviewer
CANCELED = "CANCELED" # The document was replaced with a new version and this review is no longer needed.
# Used for overall status only, never set on a task.
AWAITING = "AWAITING" # awaiting another approval
class ApprovalFile(db.Model):
file_data_id = db.Column(db.Integer, db.ForeignKey(FileDataModel.id), primary_key=True)
approval_id = db.Column(db.Integer, db.ForeignKey("approval.id"), primary_key=True)
approval = db.relationship("ApprovalModel")
file_data = db.relationship(FileDataModel)
class ApprovalModel(db.Model):
__tablename__ = 'approval'
id = db.Column(db.Integer, primary_key=True)
study_id = db.Column(db.Integer, db.ForeignKey(StudyModel.id), nullable=False)
study = db.relationship(StudyModel)
workflow_id = db.Column(db.Integer, db.ForeignKey(WorkflowModel.id), nullable=False)
workflow = db.relationship(WorkflowModel)
approver_uid = db.Column(db.String) # Not linked to user model, as they may not have logged in yet.
status = db.Column(db.String)
message = db.Column(db.String, default='')
date_created = db.Column(db.DateTime(timezone=True), default=func.now())
date_approved = db.Column(db.DateTime(timezone=True), default=None)
version = db.Column(db.Integer) # Incremented integer, so 1,2,3 as requests are made.
approval_files = db.relationship(ApprovalFile, back_populates="approval",
cascade="all, delete, delete-orphan",
order_by=ApprovalFile.file_data_id)
class Approval(object):
def __init__(self, **kwargs):
self.__dict__.update(kwargs)
@classmethod
def from_model(cls, model: ApprovalModel):
args = dict((k, v) for k, v in model.__dict__.items() if not k.startswith('_'))
instance = cls(**args)
instance.related_approvals = []
instance.title = model.study.title if model.study else ''
try:
instance.approver = LdapService.user_info(model.approver_uid)
instance.primary_investigator = LdapService.user_info(model.study.primary_investigator_id)
except ApiError as ae:
app.logger.error(f'Ldap lookup failed for approval record {model.id}', exc_info=True)
doc_dictionary = FileService.get_doc_dictionary()
instance.associated_files = []
for approval_file in model.approval_files:
try:
# fixme: This is slow because we are doing a ton of queries to find the irb code.
extra_info = doc_dictionary[approval_file.file_data.file_model.irb_doc_code]
except:
extra_info = None
associated_file = {}
associated_file['id'] = approval_file.file_data.file_model.id
if extra_info:
associated_file['name'] = '_'.join((extra_info['category1'],
approval_file.file_data.file_model.name))
associated_file['description'] = extra_info['description']
else:
associated_file['name'] = approval_file.file_data.file_model.name
associated_file['description'] = 'No description available'
associated_file['name'] = '(' + model.study.primary_investigator_id + ')' + associated_file['name']
associated_file['content_type'] = approval_file.file_data.file_model.content_type
instance.associated_files.append(associated_file)
return instance
def update_model(self, approval_model: ApprovalModel):
approval_model.status = self.status
approval_model.message = self.message
class ApprovalSchema(ma.Schema):
approver = fields.Nested(LdapSchema, dump_only=True)
primary_investigator = fields.Nested(LdapSchema, dump_only=True)
related_approvals = fields.List(fields.Nested('ApprovalSchema', allow_none=True, dump_only=True))
class Meta:
model = Approval
fields = ["id", "study_id", "workflow_id", "version", "title",
"status", "message", "approver", "primary_investigator",
"associated_files", "date_created", "date_approved",
"related_approvals"]
unknown = INCLUDE
@marshmallow.post_load
def make_approval(self, data, **kwargs):
"""Loads the basic approval data for updates to the database"""
return Approval(**data)

3
crc/models/study.py
View File

@ -222,14 +222,13 @@ class StudySchema(ma.Schema):
sponsor = fields.String(allow_none=True)
ind_number = fields.String(allow_none=True)
files = fields.List(fields.Nested(FileSchema), dump_only=True)
approvals = fields.List(fields.Nested('ApprovalSchema'), dump_only=True)
enrollment_date = fields.Date(allow_none=True)
events_history = fields.List(fields.Nested('StudyEventSchema'), dump_only=True)
class Meta:
model = Study
additional = ["id", "title", "last_updated", "primary_investigator_id", "user_uid",
"sponsor", "ind_number", "approvals", "files", "enrollment_date",
"sponsor", "ind_number", "files", "enrollment_date",
"events_history"]
unknown = INCLUDE

51
crc/scripts/request_approval.py
View File

@ -1,51 +0,0 @@
from crc.api.common import ApiError
from crc.scripts.script import Script
from crc.services.approval_service import ApprovalService
class RequestApproval(Script):
"""This still needs to be fully wired up as a Script task callable from the workflow
But the basic logic is here just to get the tests passing and logic sound. """
def get_description(self):
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
in this order.
Example:
RequestApproval approver1 "dhf8r"
"""
def do_task_validate_only(self, task, study_id, workflow_id, *args, **kwargs):
self.get_uids(task, args)
def do_task(self, task, study_id, workflow_id, *args, **kwargs):
uids = self.get_uids(task, args)
if isinstance(uids, str):
ApprovalService.add_approval(study_id, workflow_id, args)
elif isinstance(uids, list):
for id in uids:
if id: ## Assure it's not empty or null
ApprovalService.add_approval(study_id, workflow_id, id)
def get_uids(self, task, args):
if len(args) < 1:
raise ApiError(code="missing_argument",
message="The RequestApproval script requires at least one argument. The "
"the name of the variable in the task data that contains user"
"id to process. Multiple arguments are accepted.")
uids = []
for arg in args:
id = task.workflow.script_engine.evaluate_expression(task, arg)
uids.append(id)
if not isinstance(id, str):
raise ApiError(code="invalid_argument",
message="The RequestApproval script requires 1 argument. The "
"the name of the variable in the task data that contains user"
"ids to process. This must point to an array or a string, but "
"it currently points to a %s " % uids.__class__.__name__)
return uids

385
crc/services/approval_service.py
View File

@ -1,385 +0,0 @@
import json
import pickle
import sys
from base64 import b64decode
from datetime import datetime, timedelta
from sqlalchemy import desc, func
from crc import app, db, session
from crc.api.common import ApiError
from crc.models.approval import ApprovalModel, ApprovalStatus, ApprovalFile, Approval
from crc.models.study import StudyModel
from crc.models.workflow import WorkflowModel
from crc.services.file_service import FileService
from crc.services.ldap_service import LdapService
from crc.services.mails import (
send_ramp_up_submission_email,
send_ramp_up_approval_request_email,
send_ramp_up_approval_request_first_review_email,
send_ramp_up_approved_email,
send_ramp_up_denied_email,
send_ramp_up_denied_email_to_approver
)
class ApprovalService(object):
"""Provides common tools for working with an Approval"""
@staticmethod
def __one_approval_from_study(study, approver_uid = None, status=None,
include_cancelled=True):
"""Returns one approval, with all additional approvals as 'related_approvals',
the main approval can be pinned to an approver with an optional argument.
Will return null if no approvals exist on the study."""
main_approval = None
related_approvals = []
query = db.session.query(ApprovalModel).filter(ApprovalModel.study_id == study.id)
if not include_cancelled:
query=query.filter(ApprovalModel.status != ApprovalStatus.CANCELED.value)
approvals = query.all() # All non-cancelled approvals.
for approval_model in approvals:
if approval_model.approver_uid == approver_uid:
main_approval = approval_model
else:
related_approvals.append(approval_model)
# IF WE ARE JUST RETURNING ALL OF THE APPROVALS PER STUDY
if not main_approval and len(related_approvals) > 0:
main_approval = related_approvals[0]
related_approvals = related_approvals[1:]
if main_approval is not None: # May be null if the study has no approvals.
final_status = ApprovalService.__calculate_overall_approval_status(main_approval, related_approvals)
if status and final_status != status: return # Now that we are certain of the status, filter on it.
main_approval = Approval.from_model(main_approval)
main_approval.status = final_status
for ra in related_approvals:
main_approval.related_approvals.append(Approval.from_model(ra))
return main_approval
@staticmethod
def __calculate_overall_approval_status(approval, related):
# In the case of pending approvals, check to see if there is a related approval
# that proceeds this approval - and if it is declined, or still pending, then change
# the state of the approval to be Declined, or Waiting respectively.
if approval.status == ApprovalStatus.PENDING.value:
for ra in related:
if ra.id < approval.id:
if ra.status == ApprovalStatus.DECLINED.value or ra.status == ApprovalStatus.CANCELED.value:
return ra.status # If any prior approval id declined or cancelled so is this approval.
elif ra.status == ApprovalStatus.PENDING.value:
return ApprovalStatus.AWAITING.value # if any prior approval is pending, then this is waiting.
return approval.status
else:
return approval.status
@staticmethod
def get_approvals_per_user(approver_uid, status=None, include_cancelled=False):
"""Returns a list of approval objects (not db models) for the given
approver. """
studies = db.session.query(StudyModel).join(ApprovalModel).\
filter(ApprovalModel.approver_uid == approver_uid).all()
approvals = []
for study in studies:
approval = ApprovalService.__one_approval_from_study(study, approver_uid,
status, include_cancelled)
if approval:
approvals.append(approval)
return approvals
@staticmethod
def get_all_approvals(include_cancelled=True):
"""Returns a list of all approval objects (not db models), one record
per study, with any associated approvals grouped under the first approval."""
studies = db.session.query(StudyModel).all()
approvals = []
for study in studies:
approval = ApprovalService.__one_approval_from_study(study, include_cancelled=include_cancelled)
if approval:
approvals.append(approval)
return approvals
@staticmethod
def get_approvals_for_study(study_id, include_cancelled=True):
"""Returns an array of Approval objects for the study, it does not
compute the related approvals."""
query = session.query(ApprovalModel).filter_by(study_id=study_id)
if not include_cancelled:
query = query.filter(ApprovalModel.status != ApprovalStatus.CANCELED.value)
db_approvals = query.all()
return [Approval.from_model(approval_model) for approval_model in db_approvals]
@staticmethod
def get_approval_details(approval):
"""Returns a list of packed approval details, obtained from
the task data sent during the workflow """
def extract_value(task, key):
if key in task['data']:
return pickle.loads(b64decode(task['data'][key]['__bytes__']))
else:
return ""
def find_task(uuid, task):
if task['id']['__uuid__'] == uuid:
return task
for child in task['children']:
task = find_task(uuid, child)
if task:
return task
if approval.status != ApprovalStatus.APPROVED.value:
return {}
for related_approval in approval.related_approvals:
if related_approval.status != ApprovalStatus.APPROVED.value:
continue
workflow = db.session.query(WorkflowModel).filter(WorkflowModel.id == approval.workflow_id).first()
data = json.loads(workflow.bpmn_workflow_json)
last_task = find_task(data['last_task']['__uuid__'], data['task_tree'])
personnel = extract_value(last_task, 'personnel')
training_val = extract_value(last_task, 'RequiredTraining')
pi_supervisor = extract_value(last_task, 'PISupervisor')['value']
review_complete = 'AllRequiredTraining' in training_val
pi_uid = workflow.study.primary_investigator_id
pi_details = LdapService.user_info(pi_uid)
details = {
'Supervisor': pi_supervisor,
'PI_Details': pi_details,
'Review': review_complete
}
details['person_details'] = []
details['person_details'].append(pi_details)
for person in personnel:
uid = person['PersonnelComputingID']['value']
details['person_details'].append(LdapService.user_info(uid))
return details
@staticmethod
def get_health_attesting_records():
"""Return a list with prepared information related to all approvals """
approvals = ApprovalService.get_all_approvals(include_cancelled=False)
health_attesting_rows = [
['university_computing_id',
'last_name',
'first_name',
'department',
'job_title',
'supervisor_university_computing_id']
]
for approval in approvals:
try:
details = ApprovalService.get_approval_details(approval)
if not details:
continue
for person in details['person_details']:
first_name = person.given_name
last_name = person.display_name.replace(first_name, '').strip()
record = [
person.uid,
last_name,
first_name,
'',
'Academic Researcher',
details['Supervisor'] if person.uid == details['person_details'][0].uid else 'askresearch'
]
if record not in health_attesting_rows:
health_attesting_rows.append(record)
except Exception as e:
app.logger.error(f'Error pulling data for workflow {approval.workflow_id}', exc_info=True)
return health_attesting_rows
@staticmethod
def get_not_really_csv_content():
approvals = ApprovalService.get_all_approvals(include_cancelled=False)
output = []
errors = []
for approval in approvals:
try:
details = ApprovalService.get_approval_details(approval)
for person in details['person_details']:
record = {
"study_id": approval.study_id,
"pi_uid": details['PI_Details'].uid,
"pi": details['PI_Details'].display_name,
"name": person.display_name,
"uid": person.uid,
"email": person.email_address,
"supervisor": details['Supervisor'] if person.uid == details['person_details'][0].uid else "",
"review_complete": details['Review'],
}
output.append(record)
except Exception as e:
errors.append(
f'Error pulling data for workflow #{approval.workflow_id} '
f'(Approval status: {approval.status} - '
f'More details in Sentry): {str(e)}'
)
# Detailed information sent to Sentry
app.logger.error(f'Error pulling data for workflow {approval.workflow_id}', exc_info=True)
return {"results": output, "errors": errors }
@staticmethod
def update_approval(approval_id, approver_uid):
"""Update a specific approval
NOTE: Actual update happens in the API layer, this
funtion is currently in charge of only sending
corresponding emails
"""
db_approval = session.query(ApprovalModel).get(approval_id)
status = db_approval.status
if db_approval:
if status == ApprovalStatus.APPROVED.value:
# second_approval = ApprovalModel().query.filter_by(
# study_id=db_approval.study_id, workflow_id=db_approval.workflow_id,
# status=ApprovalStatus.PENDING.value, version=db_approval.version).first()
# if second_approval:
# send rrp approval request for second approver
ldap_service = LdapService()
pi_user_info = ldap_service.user_info(db_approval.study.primary_investigator_id)
approver_info = ldap_service.user_info(approver_uid)
# send rrp submission
mail_result = send_ramp_up_approved_email(
'askresearch@virginia.edu',
[pi_user_info.email_address],
f'{approver_info.display_name} - ({approver_info.uid})'
)
if mail_result:
app.logger.error(mail_result, exc_info=True)
elif status == ApprovalStatus.DECLINED.value:
ldap_service = LdapService()
pi_user_info = ldap_service.user_info(db_approval.study.primary_investigator_id)
approver_info = ldap_service.user_info(approver_uid)
# send rrp submission
mail_result = send_ramp_up_denied_email(
'askresearch@virginia.edu',
[pi_user_info.email_address],
f'{approver_info.display_name} - ({approver_info.uid})'
)
if mail_result:
app.logger.error(mail_result, exc_info=True)
first_approval = ApprovalModel().query.filter_by(
study_id=db_approval.study_id, workflow_id=db_approval.workflow_id,
status=ApprovalStatus.APPROVED.value, version=db_approval.version).first()
if first_approval:
# Second approver denies
first_approver_info = ldap_service.user_info(first_approval.approver_uid)
approver_email = [first_approver_info.email_address] if first_approver_info.email_address else app.config['FALLBACK_EMAILS']
# send rrp denied by second approver email to first approver
mail_result = send_ramp_up_denied_email_to_approver(
'askresearch@virginia.edu',
approver_email,
f'{pi_user_info.display_name} - ({pi_user_info.uid})',
f'{approver_info.display_name} - ({approver_info.uid})'
)
if mail_result:
app.logger.error(mail_result, exc_info=True)
return db_approval
@staticmethod
def add_approval(study_id, workflow_id, approver_uid):
"""we might have multiple approvals for a workflow, so I would expect this
method to get called multiple times for the same workflow. This will
only add a new approval if no approval already exists for the approver_uid,
unless the workflow has changed, at which point, it will CANCEL any
pending approvals and create a new approval for the latest version
of the workflow."""
# Find any existing approvals for this workflow.
latest_approval_requests = db.session.query(ApprovalModel). \
filter(ApprovalModel.workflow_id == workflow_id). \
order_by(desc(ApprovalModel.version))
latest_approver_request = latest_approval_requests.filter(ApprovalModel.approver_uid == approver_uid).first()
# Construct as hash of the latest files to see if things have changed since
# the last approval.
workflow = db.session.query(WorkflowModel).filter(WorkflowModel.id == workflow_id).first()
workflow_data_files = FileService.get_workflow_data_files(workflow_id)
current_data_file_ids = list(data_file.id for data_file in workflow_data_files)
if len(current_data_file_ids) == 0:
raise ApiError("invalid_workflow_approval", "You can't create an approval for a workflow that has"
"no files to approve in it.")
# If an existing approval request exists and no changes were made, do nothing.
# If there is an existing approval request for a previous version of the workflow
# then add a new request, and cancel any waiting/pending requests.
if latest_approver_request:
request_file_ids = list(file.file_data_id for file in latest_approver_request.approval_files)
current_data_file_ids.sort()
request_file_ids.sort()
other_approver = latest_approval_requests.filter(ApprovalModel.approver_uid != approver_uid).first()
if current_data_file_ids == request_file_ids:
return # This approval already exists or we're updating other approver.
else:
for approval_request in latest_approval_requests:
if (approval_request.version == latest_approver_request.version and
approval_request.status != ApprovalStatus.CANCELED.value):
approval_request.status = ApprovalStatus.CANCELED.value
db.session.add(approval_request)
version = latest_approver_request.version + 1
else:
version = 1
model = ApprovalModel(study_id=study_id, workflow_id=workflow_id,
approver_uid=approver_uid, status=ApprovalStatus.PENDING.value,
message="", date_created=datetime.now(),
version=version)
approval_files = ApprovalService._create_approval_files(workflow_data_files, model)
# Check approvals count
approvals_count = ApprovalModel().query.filter_by(study_id=study_id, workflow_id=workflow_id,
version=version).count()
db.session.add(model)
db.session.add_all(approval_files)
db.session.commit()
# Send first email
if approvals_count == 0:
ldap_service = LdapService()
pi_user_info = ldap_service.user_info(model.study.primary_investigator_id)
approver_info = ldap_service.user_info(approver_uid)
# send rrp submission
mail_result = send_ramp_up_submission_email(
'askresearch@virginia.edu',
[pi_user_info.email_address],
f'{approver_info.display_name} - ({approver_info.uid})'
)
if mail_result:
app.logger.error(mail_result, exc_info=True)
# send rrp approval request for first approver
# enhance the second part in case it bombs
approver_email = [approver_info.email_address] if approver_info.email_address else app.config['FALLBACK_EMAILS']
mail_result = send_ramp_up_approval_request_first_review_email(
'askresearch@virginia.edu',
approver_email,
f'{pi_user_info.display_name} - ({pi_user_info.uid})'
)
if mail_result:
app.logger.error(mail_result, exc_info=True)
@staticmethod
def _create_approval_files(workflow_data_files, approval):
"""Currently based exclusively on the status of files associated with a workflow."""
file_approval_models = []
for file_data in workflow_data_files:
file_approval_models.append(ApprovalFile(file_data_id=file_data.id,
approval=approval))
return file_approval_models

33
crc/services/study_service.py
View File

@ -9,7 +9,6 @@ from ldap3.core.exceptions import LDAPSocketOpenError
from crc import db, session, app
from crc.api.common import ApiError
from crc.models.approval import ApprovalFile, ApprovalModel
from crc.models.file import FileDataModel, FileModel, FileModelSchema, File, LookupFileModel, LookupDataModel
from crc.models.ldap import LdapSchema
from crc.models.protocol_builder import ProtocolBuilderStudy, ProtocolBuilderStatus
@ -18,7 +17,6 @@ from crc.models.study import StudyModel, Study, StudyStatus, Category, WorkflowM
from crc.models.task_event import TaskEventModel, TaskEvent
from crc.models.workflow import WorkflowSpecCategoryModel, WorkflowModel, WorkflowSpecModel, WorkflowState, \
WorkflowStatus, WorkflowSpecDependencyFile
from crc.services.approval_service import ApprovalService
from crc.services.file_service import FileService
from crc.services.ldap_service import LdapService
from crc.services.protocol_builder import ProtocolBuilderService
@ -59,7 +57,6 @@ class StudyService(object):
study = Study.from_model(study_model)
study.categories = StudyService.get_categories()
workflow_metas = StudyService.__get_workflow_metas(study_id)
study.approvals = ApprovalService.get_approvals_for_study(study.id)
files = FileService.get_files_for_study(study.id)
files = (File.from_models(model, FileService.get_file_data(model.id),
FileService.get_doc_dictionary()) for model in files)
@ -96,10 +93,6 @@ class StudyService(object):
session.query(WorkflowSpecDependencyFile).filter_by(workflow_id=workflow_id).delete(synchronize_session='fetch')
session.query(FileModel).filter_by(workflow_id=workflow_id).update({'archived': True, 'workflow_id': None})
# Todo: Remove approvals completely.
session.query(ApprovalFile).filter(ApprovalModel.workflow_id == workflow_id).delete(synchronize_session='fetch')
session.query(ApprovalModel).filter_by(workflow_id=workflow.id).delete()
session.delete(workflow)
session.commit()
@ -113,32 +106,6 @@ class StudyService(object):
categories.append(Category(cat_model))
return categories
@staticmethod
def get_approvals(study_id):
"""Returns a list of non-hidden approval workflows."""
study = StudyService.get_study(study_id)
cat = next(c for c in study.categories if c.name == 'approvals')
approvals = []
for wf in cat.workflows:
if wf.state is WorkflowState.hidden:
continue
workflow = db.session.query(WorkflowModel).filter_by(id=wf.id).first()
approvals.append({
'study_id': study_id,
'workflow_id': wf.id,
'display_name': wf.display_name,
'display_order': wf.display_order or 0,
'name': wf.name,
'state': wf.state.value,
'status': wf.status.value,
'workflow_spec_id': workflow.workflow_spec_id,
})
approvals.sort(key=lambda k: k['display_order'])
return approvals
@staticmethod
def get_documents_status(study_id):
"""Returns a list of documents related to the study, and any file information

48
migrations/versions/ff29528a9909_.py Normal file
View File

@ -0,0 +1,48 @@
"""empty message
Revision ID: ff29528a9909
Revises: c6261ac7a7bc
Create Date: 2021-02-16 09:00:38.674050
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
# revision identifiers, used by Alembic.
revision = 'ff29528a9909'
down_revision = 'c6261ac7a7bc'
branch_labels = None
depends_on = None
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table('approval_file')
op.drop_table('approval')
# ### end Alembic commands ###
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('approval_file',
sa.Column('approval_id', sa.INTEGER(), autoincrement=False, nullable=False),
sa.Column('file_data_id', sa.INTEGER(), autoincrement=False, nullable=False),
sa.ForeignKeyConstraint(['approval_id'], ['approval.id'], name='approval_file_approval_id_fkey'),
sa.ForeignKeyConstraint(['file_data_id'], ['file_data.id'], name='approval_file_file_data_id_fkey')
)
op.create_table('approval',
sa.Column('id', sa.INTEGER(), autoincrement=True, nullable=False),
sa.Column('study_id', sa.INTEGER(), autoincrement=False, nullable=False),
sa.Column('workflow_id', sa.INTEGER(), autoincrement=False, nullable=False),
sa.Column('approver_uid', sa.VARCHAR(), autoincrement=False, nullable=True),
sa.Column('status', sa.VARCHAR(), autoincrement=False, nullable=True),
sa.Column('message', sa.VARCHAR(), autoincrement=False, nullable=True),
sa.Column('date_created', postgresql.TIMESTAMP(timezone=True), autoincrement=False, nullable=True),
sa.Column('version', sa.INTEGER(), autoincrement=False, nullable=True),
sa.Column('date_approved', postgresql.TIMESTAMP(timezone=True), autoincrement=False, nullable=True),
sa.ForeignKeyConstraint(['study_id'], ['study.id'], name='approval_study_id_fkey'),
sa.ForeignKeyConstraint(['workflow_id'], ['workflow.id'], name='approval_workflow_id_fkey'),
sa.PrimaryKeyConstraint('id', name='approval_pkey')
)
# ### end Alembic commands ###

0
tests/approval/__init__.py
View File

239
tests/approval/test_approvals_api.py
View File

@ -1,239 +0,0 @@
import json
import random
import string
from flask import g
from tests.base_test import BaseTest
from crc import session, db
from crc.models.approval import ApprovalModel, ApprovalStatus
from crc.models.study import StudyModel
from crc.models.workflow import WorkflowModel
class TestApprovals(BaseTest):
def setUp(self):
"""Initial setup shared by all TestApprovals tests"""
self.load_example_data()
# Add a study with 2 approvers
study_workflow_approvals_1 = self._create_study_workflow_approvals(
user_uid="dhf8r", title="first study", primary_investigator_id="lb3dp",
approver_uids=["lb3dp", "dhf8r"], statuses=[ApprovalStatus.PENDING.value, ApprovalStatus.PENDING.value]
)
self.study = study_workflow_approvals_1['study']
self.workflow = study_workflow_approvals_1['workflow']
self.approval = study_workflow_approvals_1['approvals'][0]
self.approval_2 = study_workflow_approvals_1['approvals'][1]
# Add a study with 1 approver
study_workflow_approvals_2 = self._create_study_workflow_approvals(
user_uid="dhf8r", title="second study", primary_investigator_id="dhf8r",
approver_uids=["lb3dp"], statuses=[ApprovalStatus.PENDING.value]
)
self.unrelated_study = study_workflow_approvals_2['study']
self.unrelated_workflow = study_workflow_approvals_2['workflow']
self.approval_3 = study_workflow_approvals_2['approvals'][0]
def test_list_approvals_per_approver(self):
"""Only approvals associated with approver should be returned"""
approver_uid = self.approval_2.approver_uid
rv = self.app.get(f'/v1.0/approval', headers=self.logged_in_headers())
self.assert_success(rv)
response = json.loads(rv.get_data(as_text=True))
# Stored approvals are 3
approvals_count = ApprovalModel.query.count()
self.assertEqual(approvals_count, 3)
# but Dan's approvals should be only 1
self.assertEqual(len(response), 1)
# Confirm approver UID matches returned payload
approval = response[0]
self.assertEqual(approval['approver']['uid'], approver_uid)
def test_list_approvals_as_user(self):
"""All approvals as different user"""
rv = self.app.get('/v1.0/approval?as_user=lb3dp', headers=self.logged_in_headers())
self.assert_success(rv)
response = json.loads(rv.get_data(as_text=True))
# Returned approvals should match what's in the db for user ld3dp, we should get one
# approval back per study (2 studies), and that approval should have one related approval.
response_count = len(response)
self.assertEqual(2, response_count)
rv = self.app.get('/v1.0/approval', headers=self.logged_in_headers())
self.assert_success(rv)
response = json.loads(rv.get_data(as_text=True))
response_count = len(response)
self.assertEqual(1, response_count)
self.assertEqual(1, len(response[0]['related_approvals'])) # this approval has a related approval.
def test_update_approval_fails_if_not_the_approver(self):
approval = session.query(ApprovalModel).filter_by(approver_uid='lb3dp').first()
data = {'id': approval.id,
"approver_uid": "dhf8r",
'message': "Approved. I like the cut of your jib.",
'status': ApprovalStatus.APPROVED.value}
self.assertEqual(approval.status, ApprovalStatus.PENDING.value)
rv = self.app.put(f'/v1.0/approval/{approval.id}',
content_type="application/json",
headers=self.logged_in_headers(), # As dhf8r
data=json.dumps(data))
self.assert_failure(rv)
def test_accept_approval(self):
approval = session.query(ApprovalModel).filter_by(approver_uid='dhf8r').first()
data = {'id': approval.id,
"approver": {"uid": "dhf8r"},
'message': "Approved. I like the cut of your jib.",
'status': ApprovalStatus.APPROVED.value}
self.assertEqual(approval.status, ApprovalStatus.PENDING.value)
rv = self.app.put(f'/v1.0/approval/{approval.id}',
content_type="application/json",
headers=self.logged_in_headers(), # As dhf8r
data=json.dumps(data))
self.assert_success(rv)
session.refresh(approval)
# Updated record should now have the data sent to the endpoint
self.assertEqual(approval.message, data['message'])
self.assertEqual(approval.status, ApprovalStatus.APPROVED.value)
def test_decline_approval(self):
approval = session.query(ApprovalModel).filter_by(approver_uid='dhf8r').first()
data = {'id': approval.id,
"approver": {"uid": "dhf8r"},
'message': "Approved. I find the cut of your jib lacking.",
'status': ApprovalStatus.DECLINED.value}
self.assertEqual(approval.status, ApprovalStatus.PENDING.value)
rv = self.app.put(f'/v1.0/approval/{approval.id}',
content_type="application/json",
headers=self.logged_in_headers(), # As dhf8r
data=json.dumps(data))
self.assert_success(rv)
session.refresh(approval)
# Updated record should now have the data sent to the endpoint
self.assertEqual(approval.message, data['message'])
self.assertEqual(approval.status, ApprovalStatus.DECLINED.value)
def test_csv_export(self):
self.load_test_spec('two_forms')
self._add_lots_of_random_approvals(n=50, workflow_spec_name='two_forms')
# Get all workflows
workflows = db.session.query(WorkflowModel).filter_by(workflow_spec_id='two_forms').all()
# For each workflow, complete all tasks
for workflow in workflows:
workflow_api = self.get_workflow_api(workflow, user_uid=workflow.study.user_uid)
self.assertEqual('two_forms', workflow_api.workflow_spec_id)
# Log current user out.
g.user = None
self.assertIsNone(g.user)
# Complete the form for Step one and post it.
self.complete_form(workflow, workflow_api.next_task, {"color": "blue"}, error_code=None, user_uid=workflow.study.user_uid)
# Get the next Task
workflow_api = self.get_workflow_api(workflow, user_uid=workflow.study.user_uid)
self.assertEqual("StepTwo", workflow_api.next_task.name)
# Get all user Tasks and check that the data have been saved
task = workflow_api.next_task
self.assertIsNotNone(task.data)
for val in task.data.values():
self.assertIsNotNone(val)
rv = self.app.get(f'/v1.0/approval/csv', headers=self.logged_in_headers())
self.assert_success(rv)
def test_all_approvals(self):
self._add_lots_of_random_approvals()
not_canceled = session.query(ApprovalModel).filter(ApprovalModel.status != 'CANCELED').all()
not_canceled_study_ids = []
for a in not_canceled:
if a.study_id not in not_canceled_study_ids:
not_canceled_study_ids.append(a.study_id)
rv_all = self.app.get(f'/v1.0/all_approvals?status=false', headers=self.logged_in_headers())
self.assert_success(rv_all)
all_data = json.loads(rv_all.get_data(as_text=True))
self.assertEqual(len(all_data), len(not_canceled_study_ids), 'Should return all non-canceled approvals, grouped by study')
all_approvals = session.query(ApprovalModel).all()
all_approvals_study_ids = []
for a in all_approvals:
if a.study_id not in all_approvals_study_ids:
all_approvals_study_ids.append(a.study_id)
rv_all = self.app.get(f'/v1.0/all_approvals?status=true', headers=self.logged_in_headers())
self.assert_success(rv_all)
all_data = json.loads(rv_all.get_data(as_text=True))
self.assertEqual(len(all_data), len(all_approvals_study_ids), 'Should return all approvals, grouped by study')
def test_approvals_counts(self):
statuses = [name for name, value in ApprovalStatus.__members__.items()]
self._add_lots_of_random_approvals()
# Get the counts
rv_counts = self.app.get(f'/v1.0/approval-counts', headers=self.logged_in_headers())
self.assert_success(rv_counts)
counts = json.loads(rv_counts.get_data(as_text=True))
# Get the actual approvals
rv_approvals = self.app.get(f'/v1.0/approval', headers=self.logged_in_headers())
self.assert_success(rv_approvals)
approvals = json.loads(rv_approvals.get_data(as_text=True))
# Tally up the number of approvals in each status category
manual_counts = {}
for status in statuses:
manual_counts[status] = 0
for approval in approvals:
manual_counts[approval['status']] += 1
# Numbers in each category should match
for status in statuses:
self.assertEqual(counts[status], manual_counts[status], 'Approval counts for status %s should match' % status)
# Total number of approvals should match
total_counts = sum(counts[status] for status in statuses)
self.assertEqual(total_counts, len(approvals), 'Total approval counts for user should match number of approvals for user')
def _add_lots_of_random_approvals(self, n=100, workflow_spec_name="random_fact"):
num_studies_before = db.session.query(StudyModel).count()
statuses = [name for name, value in ApprovalStatus.__members__.items()]
# Add a whole bunch of approvals with random statuses
for i in range(n):
approver_uids = random.choices(["lb3dp", "dhf8r"])
self._create_study_workflow_approvals(
user_uid=random.choice(["lb3dp", "dhf8r"]),
title="".join(random.choices(string.ascii_lowercase, k=64)),
primary_investigator_id=random.choice(["lb3dp", "dhf8r"]),
approver_uids=approver_uids,
statuses=random.choices(statuses, k=len(approver_uids)),
workflow_spec_name=workflow_spec_name
)
session.flush()
num_studies_after = db.session.query(StudyModel).count()
self.assertEqual(num_studies_after, num_studies_before + n)

125
tests/approval/test_approvals_service.py
View File

@ -1,125 +0,0 @@
from tests.base_test import BaseTest
from crc import db
from crc.models.approval import ApprovalModel
from crc.services.approval_service import ApprovalService, ApprovalStatus
from crc.services.file_service import FileService
from crc.services.workflow_processor import WorkflowProcessor
class TestApprovalsService(BaseTest):
def test_create_approval_record(self):
self.create_reference_document()
workflow = self.create_workflow("empty_workflow")
FileService.add_workflow_file(workflow_id=workflow.id,
name="anything.png", content_type="text",
binary_data=b'5678', irb_doc_code="UVACompl_PRCAppr" )
ApprovalService.add_approval(study_id=workflow.study_id, workflow_id=workflow.id, approver_uid="dhf8r")
self.assertEqual(1, db.session.query(ApprovalModel).count())
model = db.session.query(ApprovalModel).first()
self.assertEqual(workflow.study_id, model.study_id)
self.assertEqual(workflow.id, model.workflow_id)
self.assertEqual("dhf8r", model.approver_uid)
self.assertEqual(1, model.version)
def test_new_requests_dont_add_if_approval_exists_for_current_workflow(self):
self.create_reference_document()
workflow = self.create_workflow("empty_workflow")
FileService.add_workflow_file(workflow_id=workflow.id,
name="anything.png", content_type="text",
binary_data=b'5678', irb_doc_code="UVACompl_PRCAppr" )
ApprovalService.add_approval(study_id=workflow.study_id, workflow_id=workflow.id, approver_uid="dhf8r")
ApprovalService.add_approval(study_id=workflow.study_id, workflow_id=workflow.id, approver_uid="dhf8r")
self.assertEqual(1, db.session.query(ApprovalModel).count())
model = db.session.query(ApprovalModel).first()
self.assertEqual(1, model.version)
def test_new_approval_requests_after_file_modification_create_new_requests(self):
self.load_example_data()
self.create_reference_document()
workflow = self.create_workflow('empty_workflow')
FileService.add_workflow_file(workflow_id=workflow.id,
name="anything.png", content_type="text",
binary_data=b'5678', irb_doc_code="AD_CoCAppr")
ApprovalService.add_approval(study_id=workflow.study_id, workflow_id=workflow.id, approver_uid="dhf8r")
FileService.add_workflow_file(workflow_id=workflow.id,
name="anything.png", content_type="text",
binary_data=b'5678', irb_doc_code="UVACompl_PRCAppr")
ApprovalService.add_approval(study_id=workflow.study_id, workflow_id=workflow.id, approver_uid="dhf8r")
self.assertEqual(2, db.session.query(ApprovalModel).count())
models = db.session.query(ApprovalModel).order_by(ApprovalModel.version).all()
self.assertEqual(1, models[0].version)
self.assertEqual(2, models[1].version)
def test_get_health_attesting_records(self):
self.load_example_data()
self.create_reference_document()
workflow = self.create_workflow('empty_workflow')
FileService.add_workflow_file(workflow_id=workflow.id,
name="anything.png", content_type="text",
binary_data=b'5678', irb_doc_code="AD_CoCAppr")
ApprovalService.add_approval(study_id=workflow.study_id, workflow_id=workflow.id, approver_uid="dhf8r")
records = ApprovalService.get_health_attesting_records()
self.assertEqual(len(records), 1)
def test_get_not_really_csv_content(self):
self.load_example_data()
self.create_reference_document()
workflow = self.create_workflow('empty_workflow')
FileService.add_workflow_file(workflow_id=workflow.id,
name="anything.png", content_type="text",
binary_data=b'5678', irb_doc_code="AD_CoCAppr")
ApprovalService.add_approval(study_id=workflow.study_id, workflow_id=workflow.id, approver_uid="dhf8r")
records = ApprovalService.get_not_really_csv_content()
self.assertEqual(len(records), 2)
def test_new_approval_cancels_all_previous_approvals(self):
self.create_reference_document()
workflow = self.create_workflow("empty_workflow")
FileService.add_workflow_file(workflow_id=workflow.id,
name="anything.png", content_type="text",
binary_data=b'5678', irb_doc_code="UVACompl_PRCAppr" )
ApprovalService.add_approval(study_id=workflow.study_id, workflow_id=workflow.id, approver_uid="dhf8r")
ApprovalService.add_approval(study_id=workflow.study_id, workflow_id=workflow.id, approver_uid="lb3dp")
current_count = ApprovalModel.query.count()
self.assertTrue(current_count, 2)
FileService.add_workflow_file(workflow_id=workflow.id,
name="borderline.png", content_type="text",
binary_data=b'906090', irb_doc_code="AD_CoCAppr" )
ApprovalService.add_approval(study_id=workflow.study_id, workflow_id=workflow.id, approver_uid="dhf8r")
current_count = ApprovalModel.query.count()
canceled_count = ApprovalModel.query.filter(ApprovalModel.status == ApprovalStatus.CANCELED.value)
self.assertTrue(current_count, 2)
self.assertTrue(current_count, 3)
ApprovalService.add_approval(study_id=workflow.study_id, workflow_id=workflow.id, approver_uid="lb3dp")
current_count = ApprovalModel.query.count()
self.assertTrue(current_count, 4)
def test_new_approval_sends_proper_emails(self):
self.assertEqual(1, 1)
def test_new_approval_failed_ldap_lookup(self):
# failed lookup should send email to sartographysupport@googlegroups.com + Cheryl
self.assertEqual(1, 1)
def test_approve_approval_sends_proper_emails(self):
self.assertEqual(1, 1)
def test_deny_approval_sends_proper_emails(self):
self.assertEqual(1, 1)

68
tests/approval/test_request_approval_script.py
View File

@ -1,68 +0,0 @@
from tests.base_test import BaseTest
from crc.services.file_service import FileService
from crc.scripts.request_approval import RequestApproval
from crc.services.workflow_processor import WorkflowProcessor
from crc.api.common import ApiError
from crc import db
from crc.models.approval import ApprovalModel
class TestRequestApprovalScript(BaseTest):
def test_do_task(self):
self.load_example_data()
self.create_reference_document()
workflow = self.create_workflow('empty_workflow')
processor = WorkflowProcessor(workflow)
task = processor.next_task()
task.data = {"study": {"approval1": "dhf8r", 'approval2':'lb3dp'}}
FileService.add_workflow_file(workflow_id=workflow.id,
irb_doc_code="UVACompl_PRCAppr",
name="anything.png", content_type="text",
binary_data=b'1234')
script = RequestApproval()
script.do_task(task, workflow.study_id, workflow.id, "study.approval1", "study.approval2")
self.assertEqual(2, db.session.query(ApprovalModel).count())
def test_do_task_with_blank_second_approver(self):
self.load_example_data()
self.create_reference_document()
workflow = self.create_workflow('empty_workflow')
processor = WorkflowProcessor(workflow)
task = processor.next_task()
task.data = {"study": {"approval1": "dhf8r", 'approval2':''}}
FileService.add_workflow_file(workflow_id=workflow.id,
irb_doc_code="UVACompl_PRCAppr",
name="anything.png", content_type="text",
binary_data=b'1234')
script = RequestApproval()
script.do_task(task, workflow.study_id, workflow.id, "study.approval1", "study.approval2")
self.assertEqual(1, db.session.query(ApprovalModel).count())
def test_do_task_with_incorrect_argument(self):
"""This script should raise an error if it can't figure out the approvers."""
self.load_example_data()
self.create_reference_document()
workflow = self.create_workflow('empty_workflow')
processor = WorkflowProcessor(workflow)
task = processor.next_task()
task.data = {"approvals": {'dhf8r':["invalid"], 'lb3dp':"invalid"}}
script = RequestApproval()
with self.assertRaises(ApiError):
script.do_task(task, workflow.study_id, workflow.id, "approvals")
def test_do_task_validate_only(self):
self.load_example_data()
self.create_reference_document()
workflow = self.create_workflow('empty_workflow')
processor = WorkflowProcessor(workflow)
task = processor.next_task()
task.data = {"study": {"approval1": "dhf8r", 'approval2':'lb3dp'}}
script = RequestApproval()
script.do_task_validate_only(task, workflow.study_id, workflow.id, "study.approval1")
self.assertEqual(0, db.session.query(ApprovalModel).count())

1
tests/base_test.py
View File

@ -13,7 +13,6 @@ from sqlalchemy import Sequence
from crc import app, db, session
from crc.models.api_models import WorkflowApiSchema, MultiInstanceType
from crc.models.approval import ApprovalModel, ApprovalStatus
from crc.models.file import FileModel, FileDataModel, CONTENT_TYPES
from crc.models.task_event import TaskEventModel
from crc.models.study import StudyModel, StudyStatus

1
tests/emails/test_email_service.py
View File

@ -1,7 +1,6 @@
from tests.base_test import BaseTest
from crc import session
from crc.models.approval import ApprovalModel, ApprovalStatus
from crc.models.email import EmailModel
from crc.services.email_service import EmailService

35
tests/files/test_files_api.py
View File

@ -9,8 +9,6 @@ from crc.models.workflow import WorkflowSpecModel
from crc.services.file_service import FileService
from crc.services.workflow_processor import WorkflowProcessor
from example_data import ExampleDataLoader
from crc.services.approval_service import ApprovalService
from crc.models.approval import ApprovalModel, ApprovalStatus
class TestFilesApi(BaseTest):
@ -251,39 +249,6 @@ class TestFilesApi(BaseTest):
rv = self.app.get('/v1.0/file/%i' % file_id, headers=self.logged_in_headers())
self.assertEqual(404, rv.status_code)
def test_delete_file_after_approval(self):
self.create_reference_document()
workflow = self.create_workflow("empty_workflow")
FileService.add_workflow_file(workflow_id=workflow.id,
name="anything.png", content_type="text",
binary_data=b'5678', irb_doc_code="UVACompl_PRCAppr")
FileService.add_workflow_file(workflow_id=workflow.id,
name="anotother_anything.png", content_type="text",
binary_data=b'1234', irb_doc_code="Study_App_Doc")
ApprovalService.add_approval(study_id=workflow.study_id, workflow_id=workflow.id, approver_uid="dhf8r")
file = session.query(FileModel).\
filter(FileModel.workflow_id == workflow.id).\
filter(FileModel.name == "anything.png").first()
self.assertFalse(file.archived)
rv = self.app.get('/v1.0/file/%i' % file.id, headers=self.logged_in_headers())
self.assert_success(rv)
rv = self.app.delete('/v1.0/file/%i' % file.id, headers=self.logged_in_headers())
self.assert_success(rv)
session.refresh(file)
self.assertTrue(file.archived)
ApprovalService.add_approval(study_id=workflow.study_id, workflow_id=workflow.id, approver_uid="dhf8r")
approvals = session.query(ApprovalModel)\
.filter(ApprovalModel.status == ApprovalStatus.PENDING.value)\
.filter(ApprovalModel.study_id == workflow.study_id).all()
self.assertEqual(1, len(approvals))
self.assertEqual(1, len(approvals[0].approval_files))
def test_change_primary_bpmn(self):

19
tests/study/test_study_api.py
View File

@ -9,14 +9,13 @@ from unittest.mock import patch
from crc import session, app
from crc.models.protocol_builder import ProtocolBuilderStatus, \
ProtocolBuilderStudySchema
from crc.models.approval import ApprovalStatus
from crc.models.file import FileModel
from crc.models.task_event import TaskEventModel
from crc.models.study import StudyEvent, StudyModel, StudySchema, StudyStatus, StudyEventType
from crc.models.workflow import WorkflowSpecModel, WorkflowModel
from crc.services.file_service import FileService
from crc.services.workflow_processor import WorkflowProcessor
from crc.services.workflow_service import WorkflowService
class TestStudyApi(BaseTest):
@ -100,22 +99,6 @@ class TestStudyApi(BaseTest):
# TODO: WRITE A TEST FOR STUDY FILES
def test_get_study_has_details_about_approvals(self):
self.load_example_data()
full_study = self._create_study_workflow_approvals(
user_uid="dhf8r", title="first study", primary_investigator_id="lb3dp",
approver_uids=["lb3dp", "dhf8r"], statuses=[ApprovalStatus.PENDING.value, ApprovalStatus.PENDING.value]
)
api_response = self.app.get('/v1.0/study/%i' % full_study['study'].id,
headers=self.logged_in_headers(), content_type="application/json")
self.assert_success(api_response)
study = StudySchema().loads(api_response.get_data(as_text=True))
self.assertEqual(len(study.approvals), 2)
for approval in study.approvals:
self.assertEqual(full_study['study'].title, approval['title'])
def test_add_study(self):
self.load_example_data()

3
tests/study/test_study_service.py
View File

@ -100,9 +100,6 @@ class TestStudyService(BaseTest):
self.assertEqual(1, workflow.completed_tasks)
# Get approvals
approvals = StudyService.get_approvals(studies[0].id)
self.assertGreater(len(approvals), 0)
self.assertIsNotNone(approvals[0]['display_order'])
@patch('crc.services.protocol_builder.ProtocolBuilderService.get_required_docs') # mock_docs
def test_get_required_docs(self, mock_docs):