Merge pull request #371 from sartography/zip-documents-379

Zip documents #379

crc/scripts/get_zipped_files.py

@@ -0,0 +1,58 @@
+from crc import session
+from crc.api.common import ApiError
+from crc.models.file import FileModel, FileDataModel
+from crc.scripts.script import Script
+from crc.services.file_service import FileService
+from crc.services.study_service import StudyService
+
+import os
+import zipfile
+
+
+class GetZippedFiles(Script):
+
+    """This script creates a zip document from a list of file ids"""
+
+    def get_description(self):
+        """Creates a zip file from a list of file_ids.
+           This is meant to use as an attachment to an email message"""
+
+    def do_task_validate_only(self, task, study_id, workflow_id, *args, **kwargs):
+        if 'file_ids' in kwargs.keys():
+            return True
+        return False
+
+    def do_task(self, task, study_id, workflow_id, *args, **kwargs):
+
+        if 'file_ids' in kwargs.keys():
+
+            doc_info = StudyService().get_documents_status(study_id)
+            if 'filename' in kwargs.keys():
+                zip_filename = kwargs['filename']
+            else:
+                zip_filename = 'attachments.zip'
+
+            file_ids = kwargs['file_ids']
+            files = session.query(FileModel).filter(FileModel.id.in_(file_ids)).all()
+            if files:
+                # Create a temporary zipfile with the requested files
+                with zipfile.ZipFile(zip_filename, mode='x', compression=zipfile.ZIP_DEFLATED) as zfw:
+                    for file in files:
+                        zip_key_words = doc_info[file.irb_doc_code]['zip_key_words']
+                        file_name = f'{study_id} {zip_key_words} {file.name}'
+                        file_data = session.query(FileDataModel).filter(FileDataModel.file_model_id == file.id).first()
+                        zfw.writestr(file_name, file_data.data)
+
+                # Add the zipfile to the DB, and grab the file_model
+                with open(zip_filename, mode='rb') as handle:
+                    file_model = FileService().add_workflow_file(workflow_id, None, task.name, zip_filename, 'application/zip', handle.read())
+
+                # Delete the temporary zipfile
+                os.remove(zip_filename)
+
+                # Return the file_model
+                return file_model
+
+        else:
+            raise ApiError(code='missing_file_ids',
+                           message='You must include a list of file_ids.')

tests/files/test_files_api.py

@@ -301,7 +301,7 @@ class TestFilesApi(BaseTest):
         self.assert_success(rv)
         json_data = json.loads(rv.get_data(as_text=True))
         self.assertEqual('Ancillary Document', json_data['document']['category1'])
-        self.assertEqual('CRC', json_data['document']['Who Uploads?'])
+        self.assertEqual('Study Team', json_data['document']['who_uploads?'])
 
     def test_delete_file(self):
         self.load_example_data()

tests/scripts/test_get_zipped_file_script.py

@@ -0,0 +1,55 @@
+from tests.base_test import BaseTest
+
+from crc import session
+from crc.models.file import FileModel, FileDataModel
+from crc.models.workflow import WorkflowModel
+from crc.models.study import StudyModel
+from crc.scripts.get_zipped_files import GetZippedFiles
+from crc.services.file_service import FileService
+
+import io
+import os
+import zipfile
+
+
+class TestGetZippedFiles(BaseTest):
+
+    def test_get_zipped_files(self):
+        self.load_example_data()
+        study = session.query(StudyModel).order_by(StudyModel.id.desc()).first()
+        workflow = session.query(WorkflowModel).first()
+        workflow_api = self.get_workflow_api(workflow)
+        task = workflow_api.next_task
+
+        # Add files to use in the test
+        FileService.add_workflow_file(workflow_id=workflow.id,
+                                      name="document_1.png", content_type="text",
+                                      task_spec_name=task.name,
+                                      binary_data=b'1234', irb_doc_code='Study_Protocol_Document')
+        FileService.add_workflow_file(workflow_id=workflow.id,
+                                      name="document_2.txt", content_type="text",
+                                      task_spec_name=task.name,
+                                      binary_data=b'1234', irb_doc_code='Study_App_Doc')
+        FileService.add_workflow_file(workflow_id=workflow.id,
+                                      name="document_3.pdf", content_type="text",
+                                      task_spec_name=task.name,
+                                      binary_data=b'1234', irb_doc_code='Admin_AdvChklst')
+
+        # Gather the file_ids
+        file_ids = []
+        files = session.query(FileModel).filter(FileModel.irb_doc_code != '').all()
+        for file in files:
+            file_ids.append(file.id)
+
+        # Send files to the zipper, receive file_model in return
+        file_model = GetZippedFiles().do_task(task, study.id, workflow.id, file_ids=file_ids, filename='another_attachment.zip')
+        file_data = session.query(FileDataModel).filter(FileDataModel.file_model_id == file_model.id).first()
+
+        # Test what we get back in the file_model
+        with zipfile.ZipFile(io.BytesIO(file_data.data), 'r') as zf:
+            self.assertIsInstance(zf, zipfile.ZipFile)
+            for name in zf.namelist():
+                info = zf.getinfo(name)
+                self.assertIn(os.path.basename(info.filename), ['1 Protocol document_1.png', '1 Application document_2.txt', '1  document_3.pdf'])
+                file = zf.read(name)
+                self.assertEqual(b'1234', file)

tests/study/test_study_service.py

@@ -129,7 +129,7 @@ class TestStudyService(BaseTest):
         self.assertEqual("UVA Compliance", documents["UVACompl_PRCAppr"]['category1'])
         self.assertEqual("PRC Approval", documents["UVACompl_PRCAppr"]['category2'])
         self.assertEqual("", documents["UVACompl_PRCAppr"]['category3'])
-        self.assertEqual("CRC", documents["UVACompl_PRCAppr"]['Who Uploads?'])
+        self.assertEqual("Study Team", documents["UVACompl_PRCAppr"]['who_uploads?'])
         self.assertEqual(0, documents["UVACompl_PRCAppr"]['count'])
         self.assertEqual(True, documents["UVACompl_PRCAppr"]['required'])
         self.assertEqual(6, documents["UVACompl_PRCAppr"]['id'])