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
cr-connect-workflow/tests/test_workflow_spec_api.py

105 lines
5.2 KiB
Python
Raw Normal View History

Protocol builder connections
2020-02-20 18:30:04 +00:00
import json
from crc import session
from crc.models.file import FileModel
Adds test for updating workflow spec
2020-03-16 16:32:39 +00:00
from crc.models.workflow import WorkflowSpecModel, WorkflowSpecModelSchema, WorkflowModel, WorkflowSpecCategoryModel
Protocol builder connections
2020-02-20 18:30:04 +00:00
from tests.base_test import BaseTest
class TestWorkflowSpec(BaseTest):
def test_list_workflow_specifications(self):
self.load_example_data()
spec = session.query(WorkflowSpecModel).first()
rv = self.app.get('/v1.0/workflow-specification',
follow_redirects=True,
It became impossible to use the Swagger ui when we started adding authentication to all of the calls. I discovered Connexion and Swagger have a default way of handing JTW authentication and this cleans up our code quite a bit, moves the securing of endpoints into the API Definition, which is quite nice, and removes a whole library dependency (I never get to do that!) I added a SWAGGER_AUTH_KEY that can be used in non-production environments to allow users to quickly authenticate from the Swagger ui. I also removed all api calls to simple little happy api services, because that is just mean and pointless.
2020-03-24 18:15:21 +00:00
content_type="application/json",headers=self.logged_in_headers())
Protocol builder connections
2020-02-20 18:30:04 +00:00
self.assert_success(rv)
json_data = json.loads(rv.get_data(as_text=True))
specs = WorkflowSpecModelSchema(many=True).load(json_data, session=session)
spec2 = specs[0]
self.assertEqual(spec.id, spec2.id)
self.assertEqual(spec.display_name, spec2.display_name)
self.assertEqual(spec.description, spec2.description)
def test_add_new_workflow_specification(self):
self.load_example_data();
num_before = session.query(WorkflowSpecModel).count()
Adds workflow spec category to API
2020-03-16 16:10:32 +00:00
spec = WorkflowSpecModel(id='make_cookies', name='make_cookies', display_name='Cooooookies',
Protocol builder connections
2020-02-20 18:30:04 +00:00
description='Om nom nom delicious cookies')
Requires login
2020-03-11 19:27:22 +00:00
rv = self.app.post('/v1.0/workflow-specification',
headers=self.logged_in_headers(),
content_type="application/json",
Protocol builder connections
2020-02-20 18:30:04 +00:00
data=json.dumps(WorkflowSpecModelSchema().dump(spec)))
self.assert_success(rv)
db_spec = session.query(WorkflowSpecModel).filter_by(id='make_cookies').first()
self.assertEqual(spec.display_name, db_spec.display_name)
num_after = session.query(WorkflowSpecModel).count()
self.assertEqual(num_after, num_before + 1)
def test_get_workflow_specification(self):
self.load_example_data()
db_spec = session.query(WorkflowSpecModel).first()
Requires login
2020-03-11 19:27:22 +00:00
rv = self.app.get('/v1.0/workflow-specification/%s' % db_spec.id, headers=self.logged_in_headers())
Protocol builder connections
2020-02-20 18:30:04 +00:00
self.assert_success(rv)
json_data = json.loads(rv.get_data(as_text=True))
api_spec = WorkflowSpecModelSchema().load(json_data, session=session)
self.assertEqual(db_spec, api_spec)
Adds test for updating workflow spec
2020-03-16 16:32:39 +00:00
def test_update_workflow_specification(self):
self.load_example_data()
category = WorkflowSpecCategoryModel(id=0, name='trap', display_name="It's a trap!")
session.add(category)
session.commit()
db_spec_before: WorkflowSpecModel = session.query(WorkflowSpecModel).first()
spec_id = db_spec_before.id
self.assertIsNone(db_spec_before.workflow_spec_category_id)
db_spec_before.workflow_spec_category_id = 0
rv = self.app.put('/v1.0/workflow-specification/%s' % spec_id,
content_type="application/json",
headers=self.logged_in_headers(),
data=json.dumps(WorkflowSpecModelSchema().dump(db_spec_before)))
self.assert_success(rv)
json_data = json.loads(rv.get_data(as_text=True))
api_spec = WorkflowSpecModelSchema().load(json_data, session=session)
self.assertEqual(db_spec_before, api_spec)
db_spec_after: WorkflowSpecModel = session.query(WorkflowSpecModel).filter_by(id=spec_id).first()
self.assertIsNotNone(db_spec_after.workflow_spec_category_id)
self.assertIsNotNone(db_spec_after.workflow_spec_category)
self.assertEqual(db_spec_after.workflow_spec_category.display_name, category.display_name)
Protocol builder connections
2020-02-20 18:30:04 +00:00
def test_delete_workflow_specification(self):
self.load_example_data()
spec_id = 'random_fact'
Standardizing the script tasks that can be executed on the server, adding tons of error messages for when things go wrong. All scripts must exist in side of the crc/scripts directory. Adding a new script that script tasks can use to add in data about the study. Moving all the test workflow specifications out of the main load. fixing a pile of tests so they can find workflow specs that are now moved into the test directory.
2020-03-03 18:50:22 +00:00
self.load_test_spec(spec_id)
Protocol builder connections
2020-02-20 18:30:04 +00:00
num_specs_before = session.query(WorkflowSpecModel).filter_by(id=spec_id).count()
self.assertEqual(num_specs_before, 1)
num_files_before = session.query(FileModel).filter_by(workflow_spec_id=spec_id).count()
num_workflows_before = session.query(WorkflowModel).filter_by(workflow_spec_id=spec_id).count()
self.assertGreater(num_files_before + num_workflows_before, 0)
Requires login
2020-03-11 19:27:22 +00:00
rv = self.app.delete('/v1.0/workflow-specification/' + spec_id, headers=self.logged_in_headers())
Protocol builder connections
2020-02-20 18:30:04 +00:00
self.assert_success(rv)
num_specs_after = session.query(WorkflowSpecModel).filter_by(id=spec_id).count()
self.assertEqual(0, num_specs_after)
# Make sure that all items in the database with the workflow spec ID are deleted as well.
num_files_after = session.query(FileModel).filter_by(workflow_spec_id=spec_id).count()
num_workflows_after = session.query(WorkflowModel).filter_by(workflow_spec_id=spec_id).count()
self.assertEqual(num_files_after + num_workflows_after, 0)
Removed all self-referential calls in the study_api. One api endpoint should never call another api endpoint. Moved the logic for updating a study to the study Model, rather than checking and setting dictionary values which will become very hard to maintain. The protocol builder service now returns real models, not dictionaries, forcing proper validation and fail-fast behavior. Changed the name of the "status" spec, to "top_level_workflow" and removing any connection a workflow or study has with this specification. It is only unused to determine status in real time, and is not reused or tracked. Modified the required documents script to return a dictionary and not an array, making it easier to speak to specific values in the BPMN and DMN. Working on new ways to test the top_level_workflow in the context of updates, this is still a work in progress. Making use of several modifications to the Spiff library that enables more complex expressions in DMN models. This is evident in the new DMN models for the top_level_workflow
2020-03-26 16:51:53 +00:00
# def test_validate_workflow_specification(self):
# self.load_example_data()
# db_spec = session.query(WorkflowSpecModel).first()
# rv = self.app.get('/v1.0/workflow-specification/%s/validate' % db_spec.id, headers=self.logged_in_headers())
# self.assert_success(rv)
# json_data = json.loads(rv.get_data(as_text=True))
# api_spec = WorkflowSpecModelSchema().load(json_data, session=session)
# self.assertEqual(db_spec, api_spec)