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 #383 from sartography/feature/476_file_url 

URL for files
This commit is contained in:
Dan Funk 2021-09-30 13:47:37 -04:00 committed by GitHub
parent fbfb241c0b e002ffd363
commit 6b5a55df4c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 36 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
crc/models/file.py
View File

@ -1,6 +1,9 @@
import enum
import urllib
import connexion
import flask
from flask import url_for
from marshmallow import INCLUDE, EXCLUDE, Schema
from marshmallow.fields import Method
from marshmallow_enum import EnumField
@ -153,10 +156,14 @@ class FileSchema(Schema):
def get_url(self, obj):
token = 'not_available'
base_url = connexion.request.host_url
if obj.id is None:
return "" # We can't return a url for a file that isn't stored yet.
file_url = url_for("/v1_0.crc_api_file_get_file_data_link", file_id=obj.id)
if hasattr(flask.g, 'user'):
token = flask.g.user.encode_auth_token()
return (app.config['APPLICATION_ROOT'] + 'file/' +
str(obj.id) + '/download?auth_token=' + urllib.parse.quote_plus(token))
url = base_url + file_url + '?auth_token=' + urllib.parse.quote_plus(token)
return url
class LookupFileModel(db.Model):

42
tests/files/test_files_api.py
View File

@ -78,7 +78,8 @@ class TestFilesApi(BaseTest):
data = {'file': (io.BytesIO(b"abcdef"), 'random_fact.svg')}
rv = self.app.post('/v1.0/file?study_id=%i&workflow_id=%s&task_spec_name=%s&form_field_key=%s' %
(workflow.study_id, workflow.id, task.get_name(), correct_name), data=data, follow_redirects=True,
(workflow.study_id, workflow.id, task.get_name(), correct_name), data=data,
follow_redirects=True,
content_type='multipart/form-data', headers=self.logged_in_headers())
self.assert_success(rv)
@ -94,7 +95,8 @@ class TestFilesApi(BaseTest):
data = {'file': (io.BytesIO(b"abcdef"), 'random_fact.svg')}
rv = self.app.post('/v1.0/file?study_id=%i&workflow_id=%s&task_spec_name=%s&form_field_key=%s' %
(workflow.study_id, workflow.id, task.get_name(), correct_name), data=data, follow_redirects=True,
(workflow.study_id, workflow.id, task.get_name(), correct_name), data=data,
follow_redirects=True,
content_type='multipart/form-data', headers=self.logged_in_headers())
self.assert_success(rv)
@ -176,13 +178,25 @@ class TestFilesApi(BaseTest):
file.name = "silly_new_name.bpmn"
rv = self.app.put('/v1.0/file/%i' % file.id,
content_type="application/json",
data=json.dumps(FileModelSchema().dump(file)), headers=self.logged_in_headers())
content_type="application/json",
data=json.dumps(FileModelSchema().dump(file)), headers=self.logged_in_headers())
self.assert_success(rv)
db_file = session.query(FileModel).filter_by(id=file.id).first()
self.assertIsNotNone(db_file)
self.assertEqual(file.name, db_file.name)
def test_load_valid_url_for_files(self):
self.load_example_data()
self.create_reference_document()
file: FileModel = session.query(FileModel).filter(FileModel.is_reference == False).first()
rv = self.app.get('/v1.0/file/%i' % file.id, content_type="application/json", headers=self.logged_in_headers())
self.assert_success(rv)
file_json = json.loads(rv.get_data(as_text=True))
print(file_json)
self.assertIsNotNone(file_json['url'])
file_data_rv = self.app.get(file_json['url'])
self.assert_success(file_data_rv)
def test_update_file_data(self):
self.load_example_data()
spec = session.query(WorkflowSpecModel).first()
@ -209,7 +223,7 @@ class TestFilesApi(BaseTest):
file_data = FileService.get_file_data(file_model.id)
self.assertEqual(2, file_data.version)
rv = self.app.get('/v1.0/file/%i/data' % file_json['id'], headers=self.logged_in_headers())
rv = self.app.get('/v1.0/file/%i/data' % file_json['id'], headers=self.logged_in_headers())
self.assert_success(rv)
data = rv.get_data()
self.assertIsNotNone(data)
@ -262,7 +276,8 @@ class TestFilesApi(BaseTest):
data = {'file': (io.BytesIO(b"abcdef"), 'random_fact.svg')}
rv = self.app.post('/v1.0/file?study_id=%i&workflow_id=%s&task_spec_name=%s&form_field_key=%s' %
(workflow.study_id, workflow.id, task.get_name(), correct_name), data=data, follow_redirects=True,
(workflow.study_id, workflow.id, task.get_name(), correct_name), data=data,
follow_redirects=True,
content_type='multipart/form-data', headers=self.logged_in_headers())
self.assert_success(rv)
@ -296,7 +311,8 @@ class TestFilesApi(BaseTest):
data = {'file': (io.BytesIO(b"abcdef"), 'random_fact.svg')}
rv = self.app.post('/v1.0/file?study_id=%i&workflow_id=%s&task_spec_name=%s&form_field_key=%s' %
(workflow.study_id, workflow.id, task.get_name(), correct_name), data=data, follow_redirects=True,
(workflow.study_id, workflow.id, task.get_name(), correct_name), data=data,
follow_redirects=True,
content_type='multipart/form-data', headers=self.logged_in_headers())
self.assert_success(rv)
json_data = json.loads(rv.get_data(as_text=True))
@ -315,8 +331,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_change_primary_bpmn(self):
self.load_example_data()
spec = session.query(WorkflowSpecModel).first()
@ -332,19 +346,17 @@ class TestFilesApi(BaseTest):
file = FileModelSchema().load(json_data, session=session)
# Delete the primary BPMN file for the workflow.
orig_model = session.query(FileModel).\
filter(FileModel.primary == True).\
orig_model = session.query(FileModel). \
filter(FileModel.primary == True). \
filter(FileModel.workflow_spec_id == spec.id).first()
rv = self.app.delete('/v1.0/file?file_id=%s' % orig_model.id, headers=self.logged_in_headers())
# Set that new file to be the primary BPMN, assure it has a primary_process_id
file.primary = True
rv = self.app.put('/v1.0/file/%i' % file.id,
content_type="application/json",
data=json.dumps(FileModelSchema().dump(file)), headers=self.logged_in_headers())
content_type="application/json",
data=json.dumps(FileModelSchema().dump(file)), headers=self.logged_in_headers())
self.assert_success(rv)
json_data = json.loads(rv.get_data(as_text=True))
self.assertTrue(json_data['primary'])
self.assertIsNotNone(json_data['primary_process_id'])