darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-07-31 19:58:46 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into update-page-count-after-antivirus-scan

Browse Source
This commit is contained in:
Rebecca Law
2018-10-18 16:01:59 +01:00
parent 4263117189 4dbedd02a3
commit 021a90f482
19 changed files with 338 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
tests/app/celery/test_letters_pdf_tasks.py
View File

@@ -33,11 +33,14 @@ from app.models import (
Notification,
NOTIFICATION_CREATED,
NOTIFICATION_DELIVERED,
NOTIFICATION_PENDING_VIRUS_CHECK,
NOTIFICATION_PERMANENT_FAILURE,
NOTIFICATION_SENDING,
NOTIFICATION_TECHNICAL_FAILURE,
NOTIFICATION_VALIDATION_FAILED,
NOTIFICATION_VIRUS_SCAN_FAILED,
)
from tests.app.db import create_notification
from tests.conftest import set_config_values
@@ -338,7 +341,7 @@ def test_letter_in_created_state_fails_if_notification_doesnt_exist(sample_notif
@freeze_time('2018-01-01 18:00')
@mock_s3
@pytest.mark.parametrize('key_type, noti_status,bucket_config_name,destination_folder', [
@pytest.mark.parametrize('key_type,noti_status,bucket_config_name,destination_folder', [
(KEY_TYPE_NORMAL, NOTIFICATION_CREATED, 'LETTERS_PDF_BUCKET_NAME', '2018-01-02/'),
(KEY_TYPE_TEST, NOTIFICATION_DELIVERED, 'TEST_LETTERS_BUCKET_NAME', '')
])
@@ -361,23 +364,61 @@ def test_process_letter_task_check_virus_scan_passed(
mocker.patch('app.celery.letters_pdf_tasks.pdf_page_count', return_value=1)
mock_s3upload = mocker.patch('app.celery.letters_pdf_tasks.s3upload')
mock_sanitise = mocker.patch('app.celery.letters_pdf_tasks._sanitise_precomiled_pdf')
mock_sanitise = mocker.patch('app.celery.letters_pdf_tasks._sanitise_precomiled_pdf', return_value="pdf_content")
process_virus_scan_passed(filename)
assert letter_notification.status == noti_status
assert letter_notification.billable_units == 1
mock_s3upload.assert_called_once_with(
filedata=b'pdf_content',
region='eu-west-1',
bucket_name=target_bucket_name,
file_location=destination_folder + filename
)
mock_sanitise.assert_called_once_with(
ANY,
letter_notification,
b'pdf_content'
)
mock_s3upload.assert_called_once_with(
bucket_name=target_bucket_name,
filedata=b'pdf_content',
file_location=destination_folder + filename,
region='eu-west-1',
)
@freeze_time('2018-01-01 18:00')
@mock_s3
@pytest.mark.parametrize('key_type,is_test_letter', [
(KEY_TYPE_NORMAL, False), (KEY_TYPE_TEST, True)
])
def test_process_letter_task_check_virus_scan_passed_when_sanitise_fails(
sample_letter_notification, mocker, key_type, is_test_letter
):
filename = 'NOTIFY.{}'.format(sample_letter_notification.reference)
source_bucket_name = current_app.config['LETTERS_SCAN_BUCKET_NAME']
target_bucket_name = current_app.config['INVALID_PDF_BUCKET_NAME']
conn = boto3.resource('s3', region_name='eu-west-1')
conn.create_bucket(Bucket=source_bucket_name)
conn.create_bucket(Bucket=target_bucket_name)
s3 = boto3.client('s3', region_name='eu-west-1')
s3.put_object(Bucket=source_bucket_name, Key=filename, Body=b'pdf_content')
sample_letter_notification.status = NOTIFICATION_PENDING_VIRUS_CHECK
sample_letter_notification.key_type = key_type
mock_move_s3 = mocker.patch('app.letters.utils._move_s3_object')
mock_sanitise = mocker.patch('app.celery.letters_pdf_tasks._sanitise_precomiled_pdf', return_value=None)
process_virus_scan_passed(filename)
assert sample_letter_notification.status == NOTIFICATION_VALIDATION_FAILED
mock_sanitise.assert_called_once_with(
ANY,
sample_letter_notification,
b'pdf_content'
)
mock_move_s3.assert_called_once_with(
source_bucket_name, filename,
target_bucket_name, filename
)
def test_get_page_count_set_notification_to_permanent_failure_when_not_pdf(
@@ -391,7 +432,7 @@ def test_get_page_count_set_notification_to_permanent_failure_when_not_pdf(
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'
sample_letter_notification.status = NOTIFICATION_PENDING_VIRUS_CHECK
mock_move_failed_pdf = mocker.patch('app.celery.letters_pdf_tasks.move_failed_pdf')
with pytest.raises(VirusScanError) as e:
@@ -404,7 +445,7 @@ def test_process_letter_task_check_virus_scan_failed(sample_letter_notification,
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'
sample_letter_notification.status = NOTIFICATION_PENDING_VIRUS_CHECK
mock_move_failed_pdf = mocker.patch('app.celery.letters_pdf_tasks.move_failed_pdf')
with pytest.raises(VirusScanError) as e:
@@ -436,6 +477,7 @@ def test_replay_letters_in_error_for_one_file(notify_api, mocker):
def test_sanitise_precompiled_pdf_returns_data_from_template_preview(rmock, sample_letter_notification):
sample_letter_notification.status = NOTIFICATION_PENDING_VIRUS_CHECK
rmock.post('http://localhost:9999/precompiled/sanitise', content=b'new_pdf', status_code=200)
mock_celery = Mock(**{'retry.side_effect': Retry})
@@ -446,6 +488,7 @@ def test_sanitise_precompiled_pdf_returns_data_from_template_preview(rmock, samp
def test_sanitise_precompiled_pdf_returns_none_on_validation_error(rmock, sample_letter_notification):
sample_letter_notification.status = NOTIFICATION_PENDING_VIRUS_CHECK
rmock.post('http://localhost:9999/precompiled/sanitise', content=b'new_pdf', status_code=400)
mock_celery = Mock(**{'retry.side_effect': Retry})
@@ -455,6 +498,7 @@ def test_sanitise_precompiled_pdf_returns_none_on_validation_error(rmock, sample
def test_sanitise_precompiled_pdf_retries_on_http_error(rmock, sample_letter_notification):
sample_letter_notification.status = NOTIFICATION_PENDING_VIRUS_CHECK
rmock.post('http://localhost:9999/precompiled/sanitise', content=b'new_pdf', status_code=500)
mock_celery = Mock(**{'retry.side_effect': Retry})
@@ -466,6 +510,7 @@ def test_sanitise_precompiled_pdf_sets_notification_to_technical_failure_after_t
rmock,
sample_letter_notification
):
sample_letter_notification.status = NOTIFICATION_PENDING_VIRUS_CHECK
rmock.post('http://localhost:9999/precompiled/sanitise', content=b'new_pdf', status_code=500)
mock_celery = Mock(**{'retry.side_effect': MaxRetriesExceededError})

28
tests/app/conftest.py
View File

@@ -189,24 +189,22 @@ def sample_service(
return service
@pytest.fixture(scope='function')
def sample_service_full_permissions(notify_db, notify_db_session):
return sample_service(
notify_db,
notify_db_session,
# ensure name doesn't clash with regular sample service
@pytest.fixture(scope='function', name='sample_service_full_permissions')
def _sample_service_full_permissions(notify_db_session):
service = create_service(
service_name="sample service full permissions",
permissions=set(SERVICE_PERMISSION_TYPES)
service_permissions=set(SERVICE_PERMISSION_TYPES)
)
@pytest.fixture(scope='function')
def sample_service_custom_letter_contact_block(notify_db, notify_db_session):
service = sample_service(notify_db, notify_db_session)
create_letter_contact(service, contact_block='((contact block))')
create_inbound_number('12345', service_id=service.id)
return service
@pytest.fixture(scope='function', name='sample_service_custom_letter_contact_block')
def _sample_service_custom_letter_contact_block(sample_service):
create_letter_contact(sample_service, contact_block='((contact block))')
return sample_service
@pytest.fixture(scope='function')
def sample_template(
notify_db,
@@ -226,7 +224,9 @@ def sample_template(
if user is None:
user = create_user()
if service is None:
service = sample_service(notify_db, notify_db_session, permissions=permissions)
service = Service.query.filter_by(name='Sample service').first()
if not service:
service = create_service(service_permissions=permissions)
if created_by is None:
created_by = create_user()

55
tests/app/letters/test_letter_utils.py
View File

@@ -7,33 +7,39 @@ from freezegun import freeze_time
from moto import mock_s3
from app.letters.utils import (
get_bucket_prefix_for_notification,
get_bucket_name_and_prefix_for_notification,
get_letter_pdf_filename,
get_letter_pdf,
upload_letter_pdf,
ScanErrorType, move_failed_pdf, get_folder_name
)
from app.models import KEY_TYPE_NORMAL, KEY_TYPE_TEST, PRECOMPILED_TEMPLATE_NAME
from app.models import KEY_TYPE_NORMAL, KEY_TYPE_TEST, PRECOMPILED_TEMPLATE_NAME, NOTIFICATION_VALIDATION_FAILED
from tests.app.db import create_notification
FROZEN_DATE_TIME = "2018-03-14 17:00:00"
@pytest.fixture()
def sample_precompiled_letter_notification_using_test_key(sample_letter_notification):
@pytest.fixture(name='sample_precompiled_letter_notification')
def _sample_precompiled_letter_notification(sample_letter_notification):
sample_letter_notification.template.hidden = True
sample_letter_notification.template.name = PRECOMPILED_TEMPLATE_NAME
sample_letter_notification.key_type = KEY_TYPE_TEST
sample_letter_notification.reference = 'foo'
with freeze_time(FROZEN_DATE_TIME):
sample_letter_notification.created_at = datetime.utcnow()
return sample_letter_notification
def test_get_bucket_prefix_for_notification_valid_notification(sample_notification):
@pytest.fixture(name='sample_precompiled_letter_notification_using_test_key')
def _sample_precompiled_letter_notification_using_test_key(sample_precompiled_letter_notification):
sample_precompiled_letter_notification.key_type = KEY_TYPE_TEST
return sample_precompiled_letter_notification
bucket_prefix = get_bucket_prefix_for_notification(sample_notification)
def test_get_bucket_name_and_prefix_for_notification_valid_notification(sample_notification):
bucket, bucket_prefix = get_bucket_name_and_prefix_for_notification(sample_notification)
assert bucket == current_app.config['LETTERS_PDF_BUCKET_NAME']
assert bucket_prefix == '{folder}/NOTIFY.{reference}'.format(
folder=sample_notification.created_at.date(),
reference=sample_notification.reference
@@ -41,19 +47,44 @@ def test_get_bucket_prefix_for_notification_valid_notification(sample_notificati
@freeze_time(FROZEN_DATE_TIME)
def test_get_bucket_prefix_for_notification_precompiled_letter_using_test_key(
def test_get_bucket_name_and_prefix_for_notification_precompiled_letter_using_test_key(
sample_precompiled_letter_notification_using_test_key
):
bucket_prefix = get_bucket_prefix_for_notification(
sample_precompiled_letter_notification_using_test_key, is_test_letter=True)
bucket, bucket_prefix = get_bucket_name_and_prefix_for_notification(
sample_precompiled_letter_notification_using_test_key)
assert bucket == current_app.config['TEST_LETTERS_BUCKET_NAME']
assert bucket_prefix == 'NOTIFY.{}'.format(
sample_precompiled_letter_notification_using_test_key.reference).upper()
def test_get_bucket_prefix_for_notification_invalid_notification():
@freeze_time(FROZEN_DATE_TIME)
def test_get_bucket_name_and_prefix_for_failed_validation(sample_precompiled_letter_notification):
sample_precompiled_letter_notification.status = NOTIFICATION_VALIDATION_FAILED
bucket, bucket_prefix = get_bucket_name_and_prefix_for_notification(sample_precompiled_letter_notification)
assert bucket == current_app.config['INVALID_PDF_BUCKET_NAME']
assert bucket_prefix == 'NOTIFY.{}'.format(
sample_precompiled_letter_notification.reference).upper()
@freeze_time(FROZEN_DATE_TIME)
def test_get_bucket_name_and_prefix_for_test_noti_with_failed_validation(
sample_precompiled_letter_notification_using_test_key
):
sample_precompiled_letter_notification_using_test_key.status = NOTIFICATION_VALIDATION_FAILED
bucket, bucket_prefix = get_bucket_name_and_prefix_for_notification(
sample_precompiled_letter_notification_using_test_key
)
assert bucket == current_app.config['INVALID_PDF_BUCKET_NAME']
assert bucket_prefix == 'NOTIFY.{}'.format(
sample_precompiled_letter_notification_using_test_key.reference).upper()
def test_get_bucket_name_and_prefix_for_notification_invalid_notification():
with pytest.raises(AttributeError):
get_bucket_prefix_for_notification(None)
get_bucket_name_and_prefix_for_notification(None)
@pytest.mark.parametrize('crown_flag,expected_crown_text', [

6
tests/app/user/test_rest_verify.py
View File

@@ -346,15 +346,15 @@ def test_reset_failed_login_count_returns_404_when_user_does_not_exist(client):
@pytest.mark.parametrize('data, expected_auth_url', (
(
{},
'http://localhost:6012/email-auth/.',
'http://localhost:6012/email-auth/%2E',
),
(
{'to': None},
'http://localhost:6012/email-auth/.',
'http://localhost:6012/email-auth/%2E',
),
(
{'to': None, 'email_auth_link_host': 'https://example.com'},
'https://example.com/email-auth/.',
'https://example.com/email-auth/%2E',
),
))
def test_send_user_email_code(

2
tests/app/v2/notifications/test_get_notifications.py
View File

@@ -437,7 +437,7 @@ def test_get_all_notifications_filter_by_status_invalid_status(client, sample_no
assert len(json_response['errors']) == 1
assert json_response['errors'][0]['message'] == "status elephant is not one of [cancelled, created, sending, " \
"sent, delivered, pending, failed, technical-failure, temporary-failure, permanent-failure, " \
"pending-virus-check, virus-scan-failed, returned-letter, accepted, received]"
"pending-virus-check, validation-failed, virus-scan-failed, returned-letter, accepted, received]"
def test_get_all_notifications_filter_by_multiple_statuses(client, sample_template):

4
tests/app/v2/notifications/test_notification_schemas.py
View File

@@ -44,7 +44,7 @@ def test_get_notifications_request_invalid_statuses(
partial_error_status = "is not one of " \
"[cancelled, created, sending, sent, delivered, pending, failed, " \
"technical-failure, temporary-failure, permanent-failure, pending-virus-check, " \
"virus-scan-failed, returned-letter, accepted, received]"
"validation-failed, virus-scan-failed, returned-letter, accepted, received]"
with pytest.raises(ValidationError) as e:
validate({'status': invalid_statuses + valid_statuses}, get_notifications_request)
@@ -92,7 +92,7 @@ def test_get_notifications_request_invalid_statuses_and_template_types():
for invalid_status in ["elephant", "giraffe"]:
assert "status {} is not one of [cancelled, created, sending, sent, delivered, " \
"pending, failed, technical-failure, temporary-failure, permanent-failure, " \
"pending-virus-check, virus-scan-failed, returned-letter, accepted, received]".format(
"pending-virus-check, validation-failed, virus-scan-failed, returned-letter, accepted, received]".format(
invalid_status
) in error_messages