mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-22 16:31:15 -05:00
f1abce22ae
be known. Added the notification id to the logging message so that the notification can be traced through the logging system by knowing the notification id, making it easier to debug. Also changed to raise an exception so that alerts are generated. This way we should get an email to say that there has been an error.
366 lines
15 KiB
Python
366 lines
15 KiB
Python
from flask import current_app
|
|
|
|
from unittest.mock import call
|
|
|
|
from freezegun import freeze_time
|
|
import pytest
|
|
import requests_mock
|
|
from botocore.exceptions import ClientError
|
|
from celery.exceptions import MaxRetriesExceededError
|
|
from requests import RequestException
|
|
from sqlalchemy.orm.exc import NoResultFound
|
|
|
|
from app.errors import VirusScanError
|
|
from app.variables import Retention
|
|
from app.celery.letters_pdf_tasks import (
|
|
create_letters_pdf,
|
|
get_letters_pdf,
|
|
collate_letter_pdfs_for_day,
|
|
group_letters,
|
|
letter_in_created_state,
|
|
process_virus_scan_passed,
|
|
process_virus_scan_failed,
|
|
process_virus_scan_error)
|
|
from app.letters.utils import get_letter_pdf_filename, ScanErrorType
|
|
from app.models import (
|
|
KEY_TYPE_NORMAL,
|
|
KEY_TYPE_TEST,
|
|
Notification,
|
|
NOTIFICATION_CREATED,
|
|
NOTIFICATION_DELIVERED,
|
|
NOTIFICATION_VIRUS_SCAN_FAILED,
|
|
NOTIFICATION_SENDING,
|
|
NOTIFICATION_TECHNICAL_FAILURE)
|
|
|
|
from tests.conftest import set_config_values
|
|
|
|
|
|
def test_should_have_decorated_tasks_functions():
|
|
assert create_letters_pdf.__wrapped__.__name__ == 'create_letters_pdf'
|
|
|
|
|
|
@pytest.mark.parametrize('personalisation', [{'name': 'test'}, None])
|
|
def test_get_letters_pdf_calls_notifications_template_preview_service_correctly(
|
|
notify_api, mocker, client, sample_letter_template, personalisation):
|
|
contact_block = 'Mr Foo,\n1 Test Street,\nLondon\nN1'
|
|
dvla_org_id = '002'
|
|
|
|
with set_config_values(notify_api, {
|
|
'TEMPLATE_PREVIEW_API_HOST': 'http://localhost/notifications-template-preview',
|
|
'TEMPLATE_PREVIEW_API_KEY': 'test-key'
|
|
}):
|
|
with requests_mock.Mocker() as request_mock:
|
|
mock_post = request_mock.post(
|
|
'http://localhost/notifications-template-preview/print.pdf', content=b'\x00\x01', status_code=200)
|
|
|
|
get_letters_pdf(
|
|
sample_letter_template, contact_block=contact_block, org_id=dvla_org_id, values=personalisation)
|
|
|
|
assert mock_post.last_request.json() == {
|
|
'values': personalisation,
|
|
'letter_contact_block': contact_block,
|
|
'dvla_org_id': dvla_org_id,
|
|
'template': {
|
|
'subject': sample_letter_template.subject,
|
|
'content': sample_letter_template.content
|
|
}
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize('page_count,expected_billable_units', [
|
|
('1', 1),
|
|
('2', 1),
|
|
('3', 2)
|
|
])
|
|
def test_get_letters_pdf_calculates_billing_units(
|
|
notify_api, mocker, client, sample_letter_template, page_count, expected_billable_units):
|
|
contact_block = 'Mr Foo,\n1 Test Street,\nLondon\nN1'
|
|
dvla_org_id = '002'
|
|
|
|
with set_config_values(notify_api, {
|
|
'TEMPLATE_PREVIEW_API_HOST': 'http://localhost/notifications-template-preview',
|
|
'TEMPLATE_PREVIEW_API_KEY': 'test-key'
|
|
}):
|
|
with requests_mock.Mocker() as request_mock:
|
|
request_mock.post(
|
|
'http://localhost/notifications-template-preview/print.pdf',
|
|
content=b'\x00\x01',
|
|
headers={'X-pdf-page-count': page_count},
|
|
status_code=200
|
|
)
|
|
|
|
_, billable_units = get_letters_pdf(
|
|
sample_letter_template, contact_block=contact_block, org_id=dvla_org_id, values=None)
|
|
|
|
assert billable_units == expected_billable_units
|
|
|
|
|
|
@freeze_time("2017-12-04 17:31:00")
|
|
def test_create_letters_pdf_calls_s3upload(mocker, sample_letter_notification):
|
|
mocker.patch('app.celery.letters_pdf_tasks.get_letters_pdf', return_value=(b'\x00\x01', '1'))
|
|
mock_s3 = mocker.patch('app.letters.utils.s3upload')
|
|
|
|
create_letters_pdf(sample_letter_notification.id)
|
|
|
|
filename = get_letter_pdf_filename(
|
|
reference=sample_letter_notification.reference,
|
|
crown=sample_letter_notification.service.crown
|
|
)
|
|
|
|
mock_s3.assert_called_with(
|
|
bucket_name=current_app.config['LETTERS_PDF_BUCKET_NAME'],
|
|
file_location=filename,
|
|
filedata=b'\x00\x01',
|
|
region=current_app.config['AWS_REGION'],
|
|
tags={Retention.KEY: Retention.ONE_WEEK}
|
|
)
|
|
|
|
|
|
def test_create_letters_pdf_sets_billable_units(mocker, sample_letter_notification):
|
|
mocker.patch('app.celery.letters_pdf_tasks.get_letters_pdf', return_value=(b'\x00\x01', 1))
|
|
mocker.patch('app.letters.utils.s3upload')
|
|
|
|
create_letters_pdf(sample_letter_notification.id)
|
|
noti = Notification.query.filter(Notification.reference == sample_letter_notification.reference).one()
|
|
assert noti.billable_units == 1
|
|
|
|
|
|
def test_create_letters_pdf_non_existent_notification(notify_api, mocker, fake_uuid):
|
|
with pytest.raises(expected_exception=NoResultFound):
|
|
create_letters_pdf(fake_uuid)
|
|
|
|
|
|
def test_create_letters_pdf_handles_request_errors(mocker, sample_letter_notification):
|
|
mock_get_letters_pdf = mocker.patch('app.celery.letters_pdf_tasks.get_letters_pdf', side_effect=RequestException)
|
|
mock_retry = mocker.patch('app.celery.letters_pdf_tasks.create_letters_pdf.retry')
|
|
|
|
create_letters_pdf(sample_letter_notification.id)
|
|
|
|
assert mock_get_letters_pdf.called
|
|
assert mock_retry.called
|
|
|
|
|
|
def test_create_letters_pdf_handles_s3_errors(mocker, sample_letter_notification):
|
|
mocker.patch('app.celery.letters_pdf_tasks.get_letters_pdf', return_value=(b'\x00\x01', 1))
|
|
error_response = {
|
|
'Error': {
|
|
'Code': 'InvalidParameterValue',
|
|
'Message': 'some error message from amazon',
|
|
'Type': 'Sender'
|
|
}
|
|
}
|
|
mock_s3 = mocker.patch('app.letters.utils.s3upload', side_effect=ClientError(error_response, 'operation_name'))
|
|
mock_retry = mocker.patch('app.celery.letters_pdf_tasks.create_letters_pdf.retry')
|
|
|
|
create_letters_pdf(sample_letter_notification.id)
|
|
|
|
assert mock_s3.called
|
|
assert mock_retry.called
|
|
|
|
|
|
def test_create_letters_pdf_sets_technical_failure_max_retries(mocker, sample_letter_notification):
|
|
mock_get_letters_pdf = mocker.patch('app.celery.letters_pdf_tasks.get_letters_pdf', side_effect=RequestException)
|
|
mock_retry = mocker.patch(
|
|
'app.celery.letters_pdf_tasks.create_letters_pdf.retry', side_effect=MaxRetriesExceededError)
|
|
mock_update_noti = mocker.patch('app.celery.letters_pdf_tasks.update_notification_status_by_id')
|
|
|
|
create_letters_pdf(sample_letter_notification.id)
|
|
|
|
assert mock_get_letters_pdf.called
|
|
assert mock_retry.called
|
|
mock_update_noti.assert_called_once_with(sample_letter_notification.id, 'technical-failure')
|
|
|
|
|
|
def test_collate_letter_pdfs_for_day(notify_api, mocker):
|
|
mock_s3 = mocker.patch('app.celery.tasks.s3.get_s3_bucket_objects')
|
|
mock_group_letters = mocker.patch('app.celery.letters_pdf_tasks.group_letters', return_value=[
|
|
[{'Key': 'A.PDF', 'Size': 1}, {'Key': 'B.pDf', 'Size': 2}],
|
|
[{'Key': 'C.pdf', 'Size': 3}]
|
|
])
|
|
mock_celery = mocker.patch('app.celery.letters_pdf_tasks.notify_celery.send_task')
|
|
|
|
collate_letter_pdfs_for_day('2017-01-02')
|
|
|
|
mock_s3.assert_called_once_with('test-letters-pdf', subfolder='2017-01-02')
|
|
mock_group_letters.assert_called_once_with(mock_s3.return_value)
|
|
assert mock_celery.call_args_list[0] == call(
|
|
name='zip-and-send-letter-pdfs',
|
|
kwargs={'filenames_to_zip': ['A.PDF', 'B.pDf']},
|
|
queue='process-ftp-tasks',
|
|
compression='zlib'
|
|
)
|
|
assert mock_celery.call_args_list[1] == call(
|
|
name='zip-and-send-letter-pdfs',
|
|
kwargs={'filenames_to_zip': ['C.pdf']},
|
|
queue='process-ftp-tasks',
|
|
compression='zlib'
|
|
)
|
|
|
|
|
|
def test_group_letters_splits_on_file_size(notify_api, mocker):
|
|
mocker.patch('app.celery.letters_pdf_tasks.letter_in_created_state', return_value=True)
|
|
letters = [
|
|
# ends under max but next one is too big
|
|
{'Key': 'A.pdf', 'Size': 1}, {'Key': 'B.pdf', 'Size': 2},
|
|
# ends on exactly max
|
|
{'Key': 'C.pdf', 'Size': 3}, {'Key': 'D.pdf', 'Size': 1}, {'Key': 'E.pdf', 'Size': 1},
|
|
# exactly max goes in next file
|
|
{'Key': 'F.pdf', 'Size': 5},
|
|
# if it's bigger than the max, still gets included
|
|
{'Key': 'G.pdf', 'Size': 6},
|
|
# whatever's left goes in last list
|
|
{'Key': 'H.pdf', 'Size': 1}, {'Key': 'I.pdf', 'Size': 1},
|
|
]
|
|
|
|
with set_config_values(notify_api, {'MAX_LETTER_PDF_ZIP_FILESIZE': 5}):
|
|
x = group_letters(letters)
|
|
|
|
assert next(x) == [{'Key': 'A.pdf', 'Size': 1}, {'Key': 'B.pdf', 'Size': 2}]
|
|
assert next(x) == [{'Key': 'C.pdf', 'Size': 3}, {'Key': 'D.pdf', 'Size': 1}, {'Key': 'E.pdf', 'Size': 1}]
|
|
assert next(x) == [{'Key': 'F.pdf', 'Size': 5}]
|
|
assert next(x) == [{'Key': 'G.pdf', 'Size': 6}]
|
|
assert next(x) == [{'Key': 'H.pdf', 'Size': 1}, {'Key': 'I.pdf', 'Size': 1}]
|
|
# make sure iterator is exhausted
|
|
assert next(x, None) is None
|
|
|
|
|
|
def test_group_letters_splits_on_file_count(notify_api, mocker):
|
|
mocker.patch('app.celery.letters_pdf_tasks.letter_in_created_state', return_value=True)
|
|
letters = [
|
|
{'Key': 'A.pdf', 'Size': 1},
|
|
{'Key': 'B.pdf', 'Size': 2},
|
|
{'Key': 'C.pdf', 'Size': 3},
|
|
{'Key': 'D.pdf', 'Size': 1},
|
|
{'Key': 'E.pdf', 'Size': 1},
|
|
{'Key': 'F.pdf', 'Size': 5},
|
|
{'Key': 'G.pdf', 'Size': 6},
|
|
{'Key': 'H.pdf', 'Size': 1},
|
|
{'Key': 'I.pdf', 'Size': 1},
|
|
]
|
|
|
|
with set_config_values(notify_api, {'MAX_LETTER_PDF_COUNT_PER_ZIP': 3}):
|
|
x = group_letters(letters)
|
|
|
|
assert next(x) == [{'Key': 'A.pdf', 'Size': 1}, {'Key': 'B.pdf', 'Size': 2}, {'Key': 'C.pdf', 'Size': 3}]
|
|
assert next(x) == [{'Key': 'D.pdf', 'Size': 1}, {'Key': 'E.pdf', 'Size': 1}, {'Key': 'F.pdf', 'Size': 5}]
|
|
assert next(x) == [{'Key': 'G.pdf', 'Size': 6}, {'Key': 'H.pdf', 'Size': 1}, {'Key': 'I.pdf', 'Size': 1}]
|
|
# make sure iterator is exhausted
|
|
assert next(x, None) is None
|
|
|
|
|
|
def test_group_letters_splits_on_file_size_and_file_count(notify_api, mocker):
|
|
mocker.patch('app.celery.letters_pdf_tasks.letter_in_created_state', return_value=True)
|
|
letters = [
|
|
# ends under max file size but next file is too big
|
|
{'Key': 'A.pdf', 'Size': 1},
|
|
{'Key': 'B.pdf', 'Size': 2},
|
|
# ends on exactly max number of files and file size
|
|
{'Key': 'C.pdf', 'Size': 3},
|
|
{'Key': 'D.pdf', 'Size': 1},
|
|
{'Key': 'E.pdf', 'Size': 1},
|
|
# exactly max file size goes in next file
|
|
{'Key': 'F.pdf', 'Size': 5},
|
|
# file size is within max but number of files reaches limit
|
|
{'Key': 'G.pdf', 'Size': 1},
|
|
{'Key': 'H.pdf', 'Size': 1},
|
|
{'Key': 'I.pdf', 'Size': 1},
|
|
# whatever's left goes in last list
|
|
{'Key': 'J.pdf', 'Size': 1},
|
|
]
|
|
|
|
with set_config_values(notify_api, {
|
|
'MAX_LETTER_PDF_ZIP_FILESIZE': 5,
|
|
'MAX_LETTER_PDF_COUNT_PER_ZIP': 3
|
|
}):
|
|
x = group_letters(letters)
|
|
|
|
assert next(x) == [{'Key': 'A.pdf', 'Size': 1}, {'Key': 'B.pdf', 'Size': 2}]
|
|
assert next(x) == [{'Key': 'C.pdf', 'Size': 3}, {'Key': 'D.pdf', 'Size': 1}, {'Key': 'E.pdf', 'Size': 1}]
|
|
assert next(x) == [{'Key': 'F.pdf', 'Size': 5}]
|
|
assert next(x) == [{'Key': 'G.pdf', 'Size': 1}, {'Key': 'H.pdf', 'Size': 1}, {'Key': 'I.pdf', 'Size': 1}]
|
|
assert next(x) == [{'Key': 'J.pdf', 'Size': 1}]
|
|
# make sure iterator is exhausted
|
|
assert next(x, None) is None
|
|
|
|
|
|
def test_group_letters_ignores_non_pdfs(notify_api, mocker):
|
|
mocker.patch('app.celery.letters_pdf_tasks.letter_in_created_state', return_value=True)
|
|
letters = [{'Key': 'A.zip'}]
|
|
assert list(group_letters(letters)) == []
|
|
|
|
|
|
def test_group_letters_ignores_notifications_already_sent(notify_api, mocker):
|
|
mock = mocker.patch('app.celery.letters_pdf_tasks.letter_in_created_state', return_value=False)
|
|
letters = [{'Key': 'A.pdf'}]
|
|
assert list(group_letters(letters)) == []
|
|
mock.assert_called_once_with('A.pdf')
|
|
|
|
|
|
def test_group_letters_with_no_letters(notify_api, mocker):
|
|
mocker.patch('app.celery.letters_pdf_tasks.letter_in_created_state', return_value=True)
|
|
assert list(group_letters([])) == []
|
|
|
|
|
|
def test_letter_in_created_state(sample_notification):
|
|
sample_notification.reference = 'ABCDEF1234567890'
|
|
filename = '2018-01-13/NOTIFY.ABCDEF1234567890.D.2.C.C.20180113120000.PDF'
|
|
|
|
assert letter_in_created_state(filename) is True
|
|
|
|
|
|
def test_letter_in_created_state_fails_if_notification_not_in_created(sample_notification):
|
|
sample_notification.reference = 'ABCDEF1234567890'
|
|
sample_notification.status = NOTIFICATION_SENDING
|
|
filename = '2018-01-13/NOTIFY.ABCDEF1234567890.D.2.C.C.20180113120000.PDF'
|
|
assert letter_in_created_state(filename) is False
|
|
|
|
|
|
def test_letter_in_created_state_fails_if_notification_doesnt_exist(sample_notification):
|
|
sample_notification.reference = 'QWERTY1234567890'
|
|
filename = '2018-01-13/NOTIFY.ABCDEF1234567890.D.2.C.C.20180113120000.PDF'
|
|
assert letter_in_created_state(filename) is False
|
|
|
|
|
|
@pytest.mark.parametrize('key_type,is_test_letter,noti_status', [
|
|
(KEY_TYPE_NORMAL, False, NOTIFICATION_CREATED),
|
|
(KEY_TYPE_TEST, True, NOTIFICATION_DELIVERED)
|
|
])
|
|
def test_process_letter_task_check_virus_scan_passed(
|
|
sample_letter_notification, mocker, key_type, is_test_letter, noti_status
|
|
):
|
|
filename = 'NOTIFY.{}'.format(sample_letter_notification.reference)
|
|
sample_letter_notification.status = 'pending-virus-check'
|
|
sample_letter_notification.key_type = key_type
|
|
mock_move_pdf = mocker.patch('app.celery.letters_pdf_tasks.move_scanned_pdf_to_test_or_live_pdf_bucket')
|
|
|
|
process_virus_scan_passed(filename)
|
|
|
|
mock_move_pdf.assert_called_once_with(filename, is_test_letter=is_test_letter)
|
|
assert sample_letter_notification.status == noti_status
|
|
|
|
|
|
def test_process_letter_task_check_virus_scan_failed(sample_letter_notification, mocker):
|
|
filename = 'NOTIFY.{}'.format(sample_letter_notification.reference)
|
|
sample_letter_notification.status = 'pending-virus-check'
|
|
mock_move_failed_pdf = mocker.patch('app.celery.letters_pdf_tasks.move_failed_pdf')
|
|
|
|
with pytest.raises(VirusScanError) as e:
|
|
process_virus_scan_failed(filename)
|
|
|
|
assert "Virus scan failed:" in str(e)
|
|
mock_move_failed_pdf.assert_called_once_with(filename, ScanErrorType.FAILURE)
|
|
assert sample_letter_notification.status == NOTIFICATION_VIRUS_SCAN_FAILED
|
|
|
|
|
|
def test_process_letter_task_check_virus_scan_error(sample_letter_notification, mocker):
|
|
filename = 'NOTIFY.{}'.format(sample_letter_notification.reference)
|
|
sample_letter_notification.status = 'pending-virus-check'
|
|
mock_move_failed_pdf = mocker.patch('app.celery.letters_pdf_tasks.move_failed_pdf')
|
|
|
|
with pytest.raises(VirusScanError) as e:
|
|
process_virus_scan_error(filename)
|
|
|
|
assert "Virus scan error:" in str(e)
|
|
mock_move_failed_pdf.assert_called_once_with(filename, ScanErrorType.ERROR)
|
|
assert sample_letter_notification.status == NOTIFICATION_TECHNICAL_FAILURE
|