cr-connect-workflow/tests/scripts/test_get_email_data.py

from tests.base_test import BaseTest
from crc import mail, session
from crc.models.workflow import WorkflowModel
from crc.services.email_service import EmailService
from crc.services.user_file_service import UserFileService


class TestGetEmailData(BaseTest):

    def test_email_data_validation(self):
        self.load_test_spec('empty_workflow', master_spec=True)
        self.create_reference_document()
        spec_model = self.load_test_spec('get_email_data')
        rv = self.app.get('/v1.0/workflow-specification/%s/validate' % spec_model.id, headers=self.logged_in_headers())
        self.assertEqual([], rv.json)

    def test_get_email_data_by_email_id(self):
        workflow = self.create_workflow('get_email_data')
        study = workflow.study
        with mail.record_messages() as outbox:
            # Send an email we can use for get_email_data
            email_model = EmailService.add_email(subject='My Email Subject',
                                                 sender='sender@example.com',
                                                 recipients=['joe@example.com'],
                                                 name='My Email Name',
                                                 content='asdf', content_html=None, study_id=study.id)

            workflow_api = self.get_workflow_api(workflow)
            task = workflow_api.next_task

            # Run workflow with get_email_data
            workflow_api = self.complete_form(workflow, task, {'email_id': email_model.id})

            # Make assertions
            task = workflow_api.next_task
            data = task.data
            self.assertIn('email_data', data)
            email_data = data['email_data']
            self.assertEqual(1, len(email_data))
            self.assertIn('doc_codes', email_data[0])
            self.assertEqual('My Email Subject', email_data[0]['subject'])
            self.assertEqual('sender@example.com', email_data[0]['sender'])
            self.assertEqual('[\'joe@example.com\']', email_data[0]['recipients'])
            self.assertEqual('My Email Name', email_data[0]['name'])
            # Make sure we remove content_html from email_data
            self.assertNotIn('content_html', email_data[0])

    def test_get_email_data_by_workflow_spec_id(self):
        workflow = self.create_workflow('get_email_data_by_workflow')
        study = workflow.study
        email_workflow = session.query(WorkflowModel).first()
        email_workflow_spec_id = email_workflow.workflow_spec_id

        with mail.record_messages() as outbox:
            # Send an email we can use for get_email_data
            email_model_one = EmailService.add_email(subject='My Email Subject',
                                                     sender='sender@example.com',
                                                     recipients=['joe@example.com'],
                                                     name='My Email Name',
                                                     content='asdf',
                                                     content_html=None,
                                                     study_id=study.id,
                                                     workflow_spec_id=email_workflow_spec_id)
            email_model_two = EmailService.add_email(subject='My Other Email Subject',
                                                     sender='sender2@example.com',
                                                     recipients=['joanne@example.com'],
                                                     name='My Email Name',
                                                     content='xyzpdq',
                                                     content_html=None,
                                                     study_id=study.id,
                                                     workflow_spec_id=email_workflow_spec_id)

            workflow_api = self.get_workflow_api(workflow)
            task = workflow_api.next_task

            # Run workflow with get_email_data
            workflow_api = self.complete_form(workflow, task, {'workflow_spec_id': email_workflow_spec_id})
            task = workflow_api.next_task
            data = task.data

            # Make assertions
            self.assertIn('email_data', data)
            email_data = data['email_data']
            self.assertEqual(2, len(email_data))
            self.assertIn('doc_codes', email_data[0])
            self.assertEqual('My Email Subject', email_data[0]['subject'])
            self.assertEqual('sender@example.com', email_data[0]['sender'])
            self.assertEqual('[\'joe@example.com\']', email_data[0]['recipients'])
            self.assertEqual('My Email Name', email_data[0]['name'])

            self.assertIn('doc_codes', email_data[1])
            self.assertEqual('My Other Email Subject', email_data[1]['subject'])
            self.assertEqual('sender2@example.com', email_data[1]['sender'])
            self.assertEqual('[\'joanne@example.com\']', email_data[1]['recipients'])
            self.assertEqual('My Email Name', email_data[1]['name'])

    def test_get_email_data_with_attachments(self):
        self.create_reference_document()
        irb_code_1 = 'Study_App_Doc'
        irb_code_2 = 'Study_Protocol_Document'

        workflow = self.create_workflow('get_email_data')
        workflow_api = self.get_workflow_api(workflow)
        first_task = workflow_api.next_task

        UserFileService.add_workflow_file(workflow_id=workflow.id,
                                          task_spec_name=first_task.name,
                                          name="something.png", content_type="text",
                                          binary_data=b'1234', irb_doc_code=irb_code_1)
        UserFileService.add_workflow_file(workflow_id=workflow.id,
                                          task_spec_name=first_task.name,
                                          name="another.png", content_type="text",
                                          binary_data=b'67890', irb_doc_code=irb_code_1)
        UserFileService.add_workflow_file(workflow_id=workflow.id,
                                          task_spec_name=first_task.name,
                                          name="anything.png", content_type="text",
                                          binary_data=b'5678', irb_doc_code=irb_code_2)
        attachment_files = [
            {'id': 1, 'name': 'something.png', 'type': 'image/png', 'data': b'1234', 'doc_code': 'Study_App_Doc'},
            {'id': 2, 'name': 'another.png', 'type': 'image/png', 'data': b'67890', 'doc_code': 'Study_App_Doc'},
            {'id': 3, 'name': 'anything.png', 'type': 'image/png', 'data': b'5678', 'doc_code': 'Study_Protocol_Document'}
        ]

        with mail.record_messages() as outbox:
            # Send an email we can use for get_email_data
            email_model_one = EmailService.add_email(subject='My Email Subject',
                                                     sender='sender@example.com',
                                                     recipients=['joe@example.com'],
                                                     name='My Email Name',
                                                     content='asdf',
                                                     content_html=None,
                                                     study_id=workflow.study_id,
                                                     workflow_spec_id=workflow.workflow_spec_id,
                                                     attachment_files=attachment_files)

            workflow_api = self.get_workflow_api(workflow)
            task = workflow_api.next_task

            # Run workflow with get_email_data
            workflow_api = self.complete_form(workflow, task, {'email_id': email_model_one.id})
            task = workflow_api.next_task
            data = task.data

            self.assertIn('email_data', data)
            email_data = data['email_data']
            self.assertEqual(1, len(email_data))

            email = email_data[0]
            self.assertIn('doc_codes', email)
            doc_codes = email['doc_codes']
            self.assertEqual(2, len(doc_codes))
            self.assertIn('Study_App_Doc', doc_codes)
            self.assertIn('Study_Protocol_Document', doc_codes)