darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2025-12-22 16:31:15 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add optional queue param to send_notification_to_queue

Browse Source 
We want to use this function for sending internal notifications to a different
queue.
This commit is contained in:
Jenny Duckett
2016-12-09 12:10:42 +00:00
parent 32a21bfd3a
commit d5d079a150
2 changed files with 30 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
tests/app/notifications/test_process_notification.py
View File

@@ -135,24 +135,28 @@ def test_persist_notification_with_optionals(sample_job, sample_api_key, mocker)
assert persisted_notification.reference is None
@pytest.mark.parametrize('research_mode, queue, notification_type, key_type',
[(True, 'research-mode', 'sms', 'normal'),
(True, 'research-mode', 'email', 'normal'),
(True, 'research-mode', 'email', 'team'),
(False, 'send-sms', 'sms', 'normal'),
(False, 'send-email', 'email', 'normal'),
(False, 'send-sms', 'sms', 'team'),
(False, 'research-mode', 'sms', 'test')])
@pytest.mark.parametrize('research_mode, requested_queue, expected_queue, notification_type, key_type',
[(True, None, 'research-mode', 'sms', 'normal'),
(True, None, 'research-mode', 'email', 'normal'),
(True, None, 'research-mode', 'email', 'team'),
(False, None, 'send-sms', 'sms', 'normal'),
(False, None, 'send-email', 'email', 'normal'),
(False, None, 'send-sms', 'sms', 'team'),
(False, None, 'research-mode', 'sms', 'test'),
(True, 'notify', 'research-mode', 'email', 'normal'),
(False, 'notify', 'notify', 'sms', 'normal'),
(False, 'notify', 'notify', 'email', 'normal'),
(False, 'notify', 'research-mode', 'sms', 'test')])
def test_send_notification_to_queue(notify_db, notify_db_session,
research_mode, notification_type,
queue, key_type, mocker):
research_mode, requested_queue, expected_queue,
notification_type, key_type, mocker):
mocked = mocker.patch('app.celery.provider_tasks.deliver_{}.apply_async'.format(notification_type))
template = sample_template(notify_db, notify_db_session) if notification_type == 'sms' \
else sample_email_template(notify_db, notify_db_session)
notification = sample_notification(notify_db, notify_db_session, template=template, key_type=key_type)
send_notification_to_queue(notification=notification, research_mode=research_mode)
send_notification_to_queue(notification=notification, research_mode=research_mode, queue=requested_queue)
mocked.assert_called_once_with([str(notification.id)], queue=queue)
mocked.assert_called_once_with([str(notification.id)], queue=expected_queue)
def test_send_notification_to_queue_throws_exception_deletes_notification(sample_notification, mocker):