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 remote-tracking branch 'origin/dev' into fix/delete_workflow_endpoint

This commit is contained in:
Dan Funk 2020-08-17 15:09:11 -04:00
parent 32d47c1417 267348dae5
commit d4c5e297cc
13 changed files with 145 additions and 109 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
crc/api/study.py
View File

@ -6,7 +6,7 @@ from sqlalchemy.exc import IntegrityError
from crc import session
from crc.api.common import ApiError, ApiErrorSchema
from crc.models.protocol_builder import ProtocolBuilderStatus
from crc.models.study import Study, StudyModel, StudySchema, StudyForUpdateSchema, StudyStatus
from crc.models.study import Study, StudyEvent, StudyEventType, StudyModel, StudySchema, StudyForUpdateSchema, StudyStatus
from crc.services.study_service import StudyService
from crc.services.user_service import UserService
@ -23,8 +23,12 @@ def add_study(body):
primary_investigator_id=body['primary_investigator_id'],
last_updated=datetime.now(),
status=StudyStatus.in_progress)
session.add(study_model)
StudyService.add_study_update_event(study_model,
status=StudyStatus.in_progress,
event_type=StudyEventType.user,
user_uid=g.user.uid)
errors = StudyService._add_all_workflow_specs_to_study(study_model)
session.commit()
study = StudyService().get_study(study_model.id)
@ -34,6 +38,7 @@ def add_study(body):
def update_study(study_id, body):
"""Pretty limited, but allows manual modifications to the study status """
if study_id is None:
raise ApiError('unknown_study', 'Please provide a valid Study ID.')
@ -42,7 +47,20 @@ def update_study(study_id, body):
raise ApiError('unknown_study', 'The study "' + study_id + '" is not recognized.')
study: Study = StudyForUpdateSchema().load(body)
study.update_model(study_model)
status = StudyStatus(study.status)
study_model.last_updated = datetime.now()
if study_model.status != status:
study_model.status = status
StudyService.add_study_update_event(study_model, status, StudyEventType.user,
user_uid=UserService.current_user().uid if UserService.has_user() else None,
comment='' if not hasattr(study, 'comment') else study.comment,
)
if status == StudyStatus.open_for_enrollment:
study_model.enrollment_date = study.enrollment_date
session.add(study_model)
session.commit()
# Need to reload the full study to return it to the frontend

27
crc/models/study.py
View File

@ -61,11 +61,6 @@ class StudyModel(db.Model):
self.irb_status = IrbStatus.incomplete_in_protocol_builder
self.status = StudyStatus.in_progress
if pbs.HSRNUMBER:
self.irb_status = IrbStatus.hsr_assigned
self.status = StudyStatus.open_for_enrollment
if self.on_hold:
self.status = StudyStatus.hold
class StudyEvent(db.Model):
@ -177,28 +172,6 @@ class Study(object):
instance = cls(**args)
return instance
def update_model(self, study_model: StudyModel):
"""As the case for update was very reduced, it's mostly and specifically
updating only the study status and generating a history record
"""
status = StudyStatus(self.status)
study_model.last_updated = datetime.datetime.now()
study_model.status = status
if status == StudyStatus.open_for_enrollment:
study_model.enrollment_date = self.enrollment_date
study_event = StudyEvent(
study=study_model,
status=status,
comment='' if not hasattr(self, 'comment') else self.comment,
event_type=StudyEventType.user,
user_uid=UserService.current_user().uid if UserService.has_user() else None,
)
db.session.add(study_event)
db.session.commit()
def model_args(self):
"""Arguments that can be passed into the Study Model to update it."""
self_dict = self.__dict__.copy()

37
crc/services/study_service.py
View File

@ -13,7 +13,8 @@ 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
from crc.models.study import StudyModel, Study, StudyStatus, Category, WorkflowMetadata, StudyEvent
from crc.models.study import StudyModel, Study, StudyStatus, Category, WorkflowMetadata, StudyEventType, StudyEvent, \
IrbStatus
from crc.models.task_event import TaskEventModel, TaskEvent
from crc.models.workflow import WorkflowSpecCategoryModel, WorkflowModel, WorkflowSpecModel, WorkflowState, \
WorkflowStatus, WorkflowSpecDependencyFile
@ -79,7 +80,6 @@ class StudyService(object):
@staticmethod
def delete_study(study_id):
session.query(TaskEventModel).filter_by(study_id=study_id).delete()
# session.query(StudyEvent).filter_by(study_id=study_id).delete()
for workflow in session.query(WorkflowModel).filter_by(study_id=study_id):
StudyService.delete_workflow(workflow.id)
study = session.query(StudyModel).filter_by(id=study_id).first()
@ -262,25 +262,54 @@ class StudyService(object):
db_studies = session.query(StudyModel).filter_by(user_uid=user.uid).all()
# Update all studies from the protocol builder, create new studies as needed.
# Futher assures that every active study (that does exist in the protocol builder)
# Further assures that every active study (that does exist in the protocol builder)
# has a reference to every available workflow (though some may not have started yet)
for pb_study in pb_studies:
new_status = None
db_study = next((s for s in db_studies if s.id == pb_study.STUDYID), None)
if not db_study:
db_study = StudyModel(id=pb_study.STUDYID)
db_study.status = None # Force a new sa
new_status = StudyStatus.in_progress
session.add(db_study)
db_studies.append(db_study)
if pb_study.HSRNUMBER:
db_study.irb_status = IrbStatus.hsr_assigned
if db_study.status != StudyStatus.open_for_enrollment:
new_status = StudyStatus.open_for_enrollment
db_study.update_from_protocol_builder(pb_study)
StudyService._add_all_workflow_specs_to_study(db_study)
# If there is a new automatic status change and there isn't a manual change in place, record it.
if new_status and db_study.status != StudyStatus.hold:
db_study.status = new_status
StudyService.add_study_update_event(db_study,
status=new_status,
event_type=StudyEventType.automatic)
# Mark studies as inactive that are no longer in Protocol Builder
for study in db_studies:
pb_study = next((pbs for pbs in pb_studies if pbs.STUDYID == study.id), None)
if not pb_study:
if not pb_study and study.status != StudyStatus.abandoned:
study.status = StudyStatus.abandoned
StudyService.add_study_update_event(study,
status=StudyStatus.abandoned,
event_type=StudyEventType.automatic)
db.session.commit()
@staticmethod
def add_study_update_event(study, status, event_type, user_uid=None, comment=''):
study_event = StudyEvent(study=study,
status=status,
event_type=event_type,
user_uid=user_uid,
comment=comment)
db.session.add(study_event)
db.session.commit()
@staticmethod
def __update_status_of_workflow_meta(workflow_metas, status):
# Update the status on each workflow

0
tests/approval/__init__.py Normal file
View File

0
tests/emails/__init__.py Normal file
View File

0
tests/files/__init__.py Normal file
View File

0
tests/ldap/__init__.py Normal file
View File

0
tests/study/__init__.py Normal file
View File

26
tests/study/test_study_api.py
View File

@ -133,13 +133,20 @@ class TestStudyApi(BaseTest):
error_count = len(study["errors"])
self.assertEqual(workflow_spec_count, workflow_count + error_count)
study_event = session.query(StudyEvent).first()
self.assertIsNotNone(study_event)
self.assertEqual(study_event.status, StudyStatus.in_progress)
self.assertEqual(study_event.event_type, StudyEventType.user)
self.assertFalse(study_event.comment)
self.assertEqual(study_event.user_uid, self.test_uid)
def test_update_study(self):
self.load_example_data()
update_comment = 'Updating the study'
study: StudyModel = session.query(StudyModel).first()
study.title = "Pilot Study of Fjord Placement for Single Fraction Outcomes to Cortisol Susceptibility"
study_schema = StudySchema().dump(study)
study_schema['status'] = StudyStatus.in_progress.value
study_schema['status'] = StudyStatus.hold.value
study_schema['comment'] = update_comment
rv = self.app.put('/v1.0/study/%i' % study.id,
content_type="application/json",
@ -153,7 +160,8 @@ class TestStudyApi(BaseTest):
# Making sure events history is being properly recorded
study_event = session.query(StudyEvent).first()
self.assertIsNotNone(study_event)
self.assertEqual(study_event.status, StudyStatus.in_progress)
self.assertEqual(study_event.status, StudyStatus.hold)
self.assertEqual(study_event.event_type, StudyEventType.user)
self.assertEqual(study_event.comment, update_comment)
self.assertEqual(study_event.user_uid, self.test_uid)
@ -191,7 +199,6 @@ class TestStudyApi(BaseTest):
self.assert_success(api_response)
json_data = json.loads(api_response.get_data(as_text=True))
num_incomplete = 0
num_abandoned = 0
num_in_progress = 0
num_open = 0
@ -210,9 +217,18 @@ class TestStudyApi(BaseTest):
self.assertEqual(num_abandoned, 1)
self.assertEqual(num_open, 1)
self.assertEqual(num_in_progress, 2)
self.assertEqual(num_incomplete, 0)
self.assertEqual(len(json_data), num_db_studies_after)
self.assertEqual(num_open + num_in_progress + num_incomplete + num_abandoned, num_db_studies_after)
self.assertEqual(num_open + num_in_progress + num_abandoned, num_db_studies_after)
# Automatic events check
in_progress_events = session.query(StudyEvent).filter_by(status=StudyStatus.in_progress)
self.assertEqual(in_progress_events.count(), 1) # 1 study is in progress
abandoned_events = session.query(StudyEvent).filter_by(status=StudyStatus.abandoned)
self.assertEqual(abandoned_events.count(), 1) # 1 study has been abandoned
open_for_enrollment_events = session.query(StudyEvent).filter_by(status=StudyStatus.open_for_enrollment)
self.assertEqual(open_for_enrollment_events.count(), 1) # 1 study was moved to open for enrollment
@patch('crc.services.protocol_builder.ProtocolBuilderService.get_investigators') # mock_studies
@patch('crc.services.protocol_builder.ProtocolBuilderService.get_required_docs') # mock_docs

126
tests/test_user_roles.py
View File

@ -33,7 +33,7 @@ class TestTasksApi(BaseTest):
except ApiError as ae:
error = ae
self.assertIsNotNone(error, "An error should be raised.")
self.assertEquals("invalid_role", error.code)
self.assertEqual("invalid_role", error.code)
def test_raise_error_if_user_does_not_have_the_correct_role(self):
submitter = self.create_user(uid='lje5u')
@ -62,8 +62,8 @@ class TestTasksApi(BaseTest):
workflow_api = self.get_workflow_api(workflow, user_uid=submitter.uid)
nav = workflow_api.navigation
self.assertEquals(5, len(nav))
self.assertEquals("supervisor", nav[1]['lane'])
self.assertEqual(5, len(nav))
self.assertEqual("supervisor", nav[1]['lane'])
def test_get_outstanding_tasks_awaiting_current_user(self):
submitter = self.create_user(uid='lje5u')
@ -81,7 +81,7 @@ class TestTasksApi(BaseTest):
task_logs = db.session.query(TaskEventModel). \
filter(TaskEventModel.user_uid == supervisor.uid). \
filter(TaskEventModel.action == WorkflowService.TASK_ACTION_ASSIGNMENT).all()
self.assertEquals(1, len(task_logs))
self.assertEqual(1, len(task_logs))
# A call to the /task endpoint as the supervisor user should return a list of
# tasks that need their attention.
@ -91,10 +91,10 @@ class TestTasksApi(BaseTest):
self.assert_success(rv)
json_data = json.loads(rv.get_data(as_text=True))
tasks = TaskEventSchema(many=True).load(json_data)
self.assertEquals(1, len(tasks))
self.assertEquals(workflow.id, tasks[0]['workflow']['id'])
self.assertEquals(workflow.study.id, tasks[0]['study']['id'])
self.assertEquals("Test Workflows", tasks[0]['workflow']['category_display_name'])
self.assertEqual(1, len(tasks))
self.assertEqual(workflow.id, tasks[0]['workflow']['id'])
self.assertEqual(workflow.study.id, tasks[0]['study']['id'])
self.assertEqual("Test Workflows", tasks[0]['workflow']['category_display_name'])
# Assure we can say something sensible like:
# You have a task called "Approval" to be completed in the "Supervisor Approval" workflow
@ -103,9 +103,9 @@ class TestTasksApi(BaseTest):
# Display name isn't set in the tests, so just checking name, but the full workflow details are included.
# I didn't delve into the full user details to keep things decoupled from ldap, so you just get the
# uid back, but could query to get the full entry.
self.assertEquals("roles", tasks[0]['workflow']['name'])
self.assertEquals("Beer consumption in the bipedal software engineer", tasks[0]['study']['title'])
self.assertEquals("lje5u", tasks[0]['study']['user_uid'])
self.assertEqual("roles", tasks[0]['workflow']['name'])
self.assertEqual("Beer consumption in the bipedal software engineer", tasks[0]['study']['title'])
self.assertEqual("lje5u", tasks[0]['study']['user_uid'])
# Completing the next step of the workflow will close the task.
data['approval'] = True
@ -121,39 +121,39 @@ class TestTasksApi(BaseTest):
# Navigation as Submitter with ready task.
workflow_api = self.get_workflow_api(workflow, user_uid=submitter.uid)
nav = workflow_api.navigation
self.assertEquals(5, len(nav))
self.assertEquals('READY', nav[0]['state']) # First item is ready, no progress yet.
self.assertEquals('LOCKED', nav[1]['state']) # Second item is locked, it is the review and doesn't belong to this user.
self.assertEquals('LOCKED', nav[2]['state']) # third item is a gateway, and belongs to no one, and is locked.
self.assertEquals('NOOP', nav[3]['state']) # Approved Path, has no operation
self.assertEquals('NOOP', nav[4]['state']) # Rejected Path, has no operation.
self.assertEquals('READY', workflow_api.next_task.state)
self.assertEqual(5, len(nav))
self.assertEqual('READY', nav[0]['state']) # First item is ready, no progress yet.
self.assertEqual('LOCKED', nav[1]['state']) # Second item is locked, it is the review and doesn't belong to this user.
self.assertEqual('LOCKED', nav[2]['state']) # third item is a gateway, and belongs to no one, and is locked.
self.assertEqual('NOOP', nav[3]['state']) # Approved Path, has no operation
self.assertEqual('NOOP', nav[4]['state']) # Rejected Path, has no operation.
self.assertEqual('READY', workflow_api.next_task.state)
# Navigation as Submitter after handoff to supervisor
data = workflow_api.next_task.data
data['supervisor'] = supervisor.uid
workflow_api = self.complete_form(workflow, workflow_api.next_task, data, user_uid=submitter.uid)
nav = workflow_api.navigation
self.assertEquals('COMPLETED', nav[0]['state']) # First item is ready, no progress yet.
self.assertEquals('LOCKED', nav[1]['state']) # Second item is locked, it is the review and doesn't belong to this user.
self.assertEquals('LOCKED', nav[2]['state']) # third item is a gateway, and belongs to no one, and is locked.
self.assertEquals('LOCKED', workflow_api.next_task.state)
self.assertEqual('COMPLETED', nav[0]['state']) # First item is ready, no progress yet.
self.assertEqual('LOCKED', nav[1]['state']) # Second item is locked, it is the review and doesn't belong to this user.
self.assertEqual('LOCKED', nav[2]['state']) # third item is a gateway, and belongs to no one, and is locked.
self.assertEqual('LOCKED', workflow_api.next_task.state)
# In the event the next task is locked, we should say something sensible here.
# It is possible to look at the role of the task, and say The next task "TASK TITLE" will
# be handled by 'dhf8r', who is full-filling the role of supervisor. the Task Data
# is guaranteed to have a supervisor attribute in it that will contain the users uid, which
# could be looked up through an ldap service.
self.assertEquals('supervisor', workflow_api.next_task.lane)
self.assertEqual('supervisor', workflow_api.next_task.lane)
# Navigation as Supervisor
workflow_api = self.get_workflow_api(workflow, user_uid=supervisor.uid)
nav = workflow_api.navigation
self.assertEquals(5, len(nav))
self.assertEquals('LOCKED', nav[0]['state']) # First item belongs to the submitter, and is locked.
self.assertEquals('READY', nav[1]['state']) # Second item is locked, it is the review and doesn't belong to this user.
self.assertEquals('LOCKED', nav[2]['state']) # third item is a gateway, and belongs to no one, and is locked.
self.assertEquals('READY', workflow_api.next_task.state)
self.assertEqual(5, len(nav))
self.assertEqual('LOCKED', nav[0]['state']) # First item belongs to the submitter, and is locked.
self.assertEqual('READY', nav[1]['state']) # Second item is locked, it is the review and doesn't belong to this user.
self.assertEqual('LOCKED', nav[2]['state']) # third item is a gateway, and belongs to no one, and is locked.
self.assertEqual('READY', workflow_api.next_task.state)
data = workflow_api.next_task.data
data["approval"] = False
@ -161,47 +161,47 @@ class TestTasksApi(BaseTest):
# Navigation as Supervisor, after completing task.
nav = workflow_api.navigation
self.assertEquals(5, len(nav))
self.assertEquals('LOCKED', nav[0]['state']) # First item belongs to the submitter, and is locked.
self.assertEquals('COMPLETED', nav[1]['state']) # Second item is locked, it is the review and doesn't belong to this user.
self.assertEquals('COMPLETED', nav[2]['state']) # third item is a gateway, and is now complete.
self.assertEquals('LOCKED', workflow_api.next_task.state)
self.assertEqual(5, len(nav))
self.assertEqual('LOCKED', nav[0]['state']) # First item belongs to the submitter, and is locked.
self.assertEqual('COMPLETED', nav[1]['state']) # Second item is locked, it is the review and doesn't belong to this user.
self.assertEqual('COMPLETED', nav[2]['state']) # third item is a gateway, and is now complete.
self.assertEqual('LOCKED', workflow_api.next_task.state)
# Navigation as Submitter, coming back in to a rejected workflow to view the rejection message.
workflow_api = self.get_workflow_api(workflow, user_uid=submitter.uid)
nav = workflow_api.navigation
self.assertEquals(5, len(nav))
self.assertEquals('COMPLETED', nav[0]['state']) # First item belongs to the submitter, and is locked.
self.assertEquals('LOCKED', nav[1]['state']) # Second item is locked, it is the review and doesn't belong to this user.
self.assertEquals('LOCKED', nav[2]['state']) # third item is a gateway belonging to the supervisor, and is locked.
self.assertEquals('READY', workflow_api.next_task.state)
self.assertEqual(5, len(nav))
self.assertEqual('COMPLETED', nav[0]['state']) # First item belongs to the submitter, and is locked.
self.assertEqual('LOCKED', nav[1]['state']) # Second item is locked, it is the review and doesn't belong to this user.
self.assertEqual('LOCKED', nav[2]['state']) # third item is a gateway belonging to the supervisor, and is locked.
self.assertEqual('READY', workflow_api.next_task.state)
# Navigation as Submitter, re-completing the original request a second time, and sending it for review.
workflow_api = self.complete_form(workflow, workflow_api.next_task, data, user_uid=submitter.uid)
nav = workflow_api.navigation
self.assertEquals(5, len(nav))
self.assertEquals('READY', nav[0]['state'])
self.assertEquals('LOCKED', nav[1]['state']) # Second item is locked, it is the review and doesn't belong to this user.
self.assertEquals('LOCKED', nav[2]['state']) # third item is a gateway belonging to the supervisor, and is locked.
self.assertEquals('READY', workflow_api.next_task.state)
self.assertEqual(5, len(nav))
self.assertEqual('READY', nav[0]['state']) # When you loop back the task is again in the ready state.
self.assertEqual('LOCKED', nav[1]['state']) # Second item is locked, it is the review and doesn't belong to this user.
self.assertEqual('LOCKED', nav[2]['state']) # third item is a gateway belonging to the supervisor, and is locked.
self.assertEqual('READY', workflow_api.next_task.state)
data["favorite_color"] = "blue"
data["quest"] = "to seek the holy grail"
workflow_api = self.complete_form(workflow, workflow_api.next_task, data, user_uid=submitter.uid)
self.assertEquals('LOCKED', workflow_api.next_task.state)
self.assertEqual('LOCKED', workflow_api.next_task.state)
workflow_api = self.get_workflow_api(workflow, user_uid=supervisor.uid)
self.assertEquals('READY', workflow_api.next_task.state)
self.assertEqual('READY', workflow_api.next_task.state)
data = workflow_api.next_task.data
data["approval"] = True
workflow_api = self.complete_form(workflow, workflow_api.next_task, data, user_uid=supervisor.uid)
self.assertEquals('LOCKED', workflow_api.next_task.state)
self.assertEqual('LOCKED', workflow_api.next_task.state)
workflow_api = self.get_workflow_api(workflow, user_uid=submitter.uid)
self.assertEquals('COMPLETED', workflow_api.next_task.state)
self.assertEquals('EndEvent', workflow_api.next_task.type) # Are are at the end.
self.assertEquals(WorkflowStatus.complete, workflow_api.status)
self.assertEqual('COMPLETED', workflow_api.next_task.state)
self.assertEqual('EndEvent', workflow_api.next_task.type) # Are are at the end.
self.assertEqual(WorkflowStatus.complete, workflow_api.status)
def get_assignment_task_events(self, uid):
return db.session.query(TaskEventModel). \
@ -222,31 +222,31 @@ class TestTasksApi(BaseTest):
# At this point there should be a task_log with an action of ASSIGNMENT on it for
# the supervisor.
self.assertEquals(1, len(self.get_assignment_task_events(supervisor.uid)))
self.assertEqual(1, len(self.get_assignment_task_events(supervisor.uid)))
# Resetting the workflow at this point should clear the event log.
workflow_api = self.get_workflow_api(workflow, hard_reset=True, user_uid=submitter.uid)
self.assertEquals(0, len(self.get_assignment_task_events(supervisor.uid)))
self.assertEqual(0, len(self.get_assignment_task_events(supervisor.uid)))
# Re-complete first task, and awaiting tasks should shift to 0 for for submitter, and 1 for supervisor
workflow_api = self.complete_form(workflow, workflow_api.next_task, data, user_uid=submitter.uid)
self.assertEquals(0, len(self.get_assignment_task_events(submitter.uid)))
self.assertEquals(1, len(self.get_assignment_task_events(supervisor.uid)))
self.assertEqual(0, len(self.get_assignment_task_events(submitter.uid)))
self.assertEqual(1, len(self.get_assignment_task_events(supervisor.uid)))
# Complete the supervisor task with rejected approval, and the assignments should switch.
workflow_api = self.get_workflow_api(workflow, user_uid=supervisor.uid)
data = workflow_api.next_task.data
data["approval"] = False
workflow_api = self.complete_form(workflow, workflow_api.next_task, data, user_uid=supervisor.uid)
self.assertEquals(1, len(self.get_assignment_task_events(submitter.uid)))
self.assertEquals(0, len(self.get_assignment_task_events(supervisor.uid)))
self.assertEqual(1, len(self.get_assignment_task_events(submitter.uid)))
self.assertEqual(0, len(self.get_assignment_task_events(supervisor.uid)))
# Mark the return form review page as complete, and then recomplete the form, and assignments switch yet again.
workflow_api = self.get_workflow_api(workflow, user_uid=submitter.uid)
workflow_api = self.complete_form(workflow, workflow_api.next_task, data, user_uid=submitter.uid)
workflow_api = self.complete_form(workflow, workflow_api.next_task, data, user_uid=submitter.uid)
self.assertEquals(0, len(self.get_assignment_task_events(submitter.uid)))
self.assertEquals(1, len(self.get_assignment_task_events(supervisor.uid)))
self.assertEqual(0, len(self.get_assignment_task_events(submitter.uid)))
self.assertEqual(1, len(self.get_assignment_task_events(supervisor.uid)))
# Complete the supervisor task, accepting the approval, and the workflow is completed.
# When it is all done, there should be no outstanding assignments.
@ -254,14 +254,14 @@ class TestTasksApi(BaseTest):
data = workflow_api.next_task.data
data["approval"] = True
workflow_api = self.complete_form(workflow, workflow_api.next_task, data, user_uid=supervisor.uid)
self.assertEquals(WorkflowStatus.complete, workflow_api.status)
self.assertEquals('EndEvent', workflow_api.next_task.type) # Are are at the end.
self.assertEquals(0, len(self.get_assignment_task_events(submitter.uid)))
self.assertEquals(0, len(self.get_assignment_task_events(supervisor.uid)))
self.assertEqual(WorkflowStatus.complete, workflow_api.status)
self.assertEqual('EndEvent', workflow_api.next_task.type) # Are are at the end.
self.assertEqual(0, len(self.get_assignment_task_events(submitter.uid)))
self.assertEqual(0, len(self.get_assignment_task_events(supervisor.uid)))
# Sending any subsequent complete forms does not result in a new task event
with self.assertRaises(AssertionError) as _api_error:
workflow_api = self.complete_form(workflow, workflow_api.next_task, data, user_uid=submitter.uid)
self.assertEquals(0, len(self.get_assignment_task_events(submitter.uid)))
self.assertEquals(0, len(self.get_assignment_task_events(supervisor.uid)))
self.assertEqual(0, len(self.get_assignment_task_events(submitter.uid)))
self.assertEqual(0, len(self.get_assignment_task_events(supervisor.uid)))

0
tests/workflow/__init__.py Normal file
View File

2
tests/workflow/test_workflow_processor.py
View File

@ -388,5 +388,5 @@ class TestWorkflowProcessor(BaseTest):
self._populate_form_with_random_data(task)
processor.complete_task(task)
supervisor_task = processor.next_user_tasks()[0]
self.assertEquals("supervisor", supervisor_task.task_spec.lane)
self.assertEqual("supervisor", supervisor_task.task_spec.lane)

12
tests/workflow/test_workflow_processor_multi_instance.py
View File

@ -158,7 +158,7 @@ class TestWorkflowProcessorMultiInstance(BaseTest):
self.assertEqual(3, len(next_user_tasks))
# There should be six tasks in the navigation: start event, the script task, end event, and three tasks
# for the three executions of hte multi-instance.
self.assertEquals(6, len(processor.bpmn_workflow.get_nav_list()))
self.assertEqual(6, len(processor.bpmn_workflow.get_nav_list()))
# We can complete the tasks out of order.
task = next_user_tasks[2]
@ -171,12 +171,12 @@ class TestWorkflowProcessorMultiInstance(BaseTest):
# Assure navigation picks up the label of the current element variable.
nav = WorkflowService.processor_to_workflow_api(processor, task).navigation
self.assertEquals("Primary Investigator", nav[2].title)
self.assertEqual("Primary Investigator", nav[2].title)
task.update_data({"investigator": {"email": "dhf8r@virginia.edu"}})
processor.complete_task(task)
processor.do_engine_steps()
self.assertEquals(6, len(processor.bpmn_workflow.get_nav_list()))
self.assertEqual(6, len(processor.bpmn_workflow.get_nav_list()))
task = next_user_tasks[0]
api_task = WorkflowService.spiff_task_to_api_task(task)
@ -184,7 +184,7 @@ class TestWorkflowProcessorMultiInstance(BaseTest):
task.update_data({"investigator":{"email":"asd3v@virginia.edu"}})
processor.complete_task(task)
processor.do_engine_steps()
self.assertEquals(6, len(processor.bpmn_workflow.get_nav_list()))
self.assertEqual(6, len(processor.bpmn_workflow.get_nav_list()))
task = next_user_tasks[1]
api_task = WorkflowService.spiff_task_to_api_task(task)
@ -192,7 +192,7 @@ class TestWorkflowProcessorMultiInstance(BaseTest):
task.update_data({"investigator":{"email":"asdf32@virginia.edu"}})
processor.complete_task(task)
processor.do_engine_steps()
self.assertEquals(6, len(processor.bpmn_workflow.get_nav_list()))
self.assertEqual(6, len(processor.bpmn_workflow.get_nav_list()))
# Completing the tasks out of order, still provides the correct information.
expected = self.mock_investigator_response
@ -203,4 +203,4 @@ class TestWorkflowProcessorMultiInstance(BaseTest):
task.data['StudyInfo']['investigators'])
self.assertEqual(WorkflowStatus.complete, processor.get_status())
self.assertEquals(6, len(processor.bpmn_workflow.get_nav_list()))
self.assertEqual(6, len(processor.bpmn_workflow.get_nav_list()))