darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-02-04 02:11:11 -05:00
Code Issues Packages Projects Releases Wiki Activity

Make test for timeout with callbacks consistent

Browse Source 
This now matches the behaviour of the test above it: mocking out
the DAO function in order to focus on the specific behaviour of
the function under test.
This commit is contained in:
Ben Thorner
2021-11-25 18:14:02 +00:00
parent 5bf3fe6c0f
commit aea555fce2
1 changed files with 9 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
tests/app/celery/test_nightly_tasks.py
View File

@@ -179,18 +179,17 @@ def test_timeout_notifications_no_callbacks(mocker, sample_notification):
mock_update.assert_not_called() mock_update.assert_not_called()
def test_timeout_notifications_sends_status_update_to_service(client, sample_template, mocker):
callback_api = create_service_callback_api(service=sample_template.service) def test_timeout_notifications_with_callbacks(mocker, sample_notification):
mocked = mocker.patch('app.celery.service_callback_tasks.send_delivery_status_to_service.apply_async') mock_update = mocker.patch('app.celery.service_callback_tasks.send_delivery_status_to_service.apply_async')
notification = create_notification( mock_dao = mocker.patch('app.celery.nightly_tasks.dao_timeout_notifications')
template=sample_template, mock_dao.return_value = [sample_notification]
status='sending',
created_at=datetime.utcnow() - timedelta( callback_api = create_service_callback_api(service=sample_notification.service)
seconds=current_app.config.get('SENDING_NOTIFICATIONS_TIMEOUT_PERIOD') + 10))
timeout_notifications() timeout_notifications()
encrypted_data = create_delivery_status_callback_data(notification, callback_api) encrypted_data = create_delivery_status_callback_data(sample_notification, callback_api)
mocked.assert_called_once_with([str(notification.id), encrypted_data], queue=QueueNames.CALLBACKS) mock_update.assert_called_once_with([str(sample_notification.id), encrypted_data], queue=QueueNames.CALLBACKS)
def test_delete_inbound_sms_calls_child_task(notify_api, mocker): def test_delete_inbound_sms_calls_child_task(notify_api, mocker):