mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-23 00:41:35 -05:00
8b6d28d3b0
application. If the Anti-virus app fails due to s3 errors or ClamAV so does not scan (even after retries) the file at all an error needs to be raised and the notification set to technical-failure. Files should be moved to a 'folder' a separate one for ERROR and FAILURE. * Added new letter task to process the error * Added a new method to letter utils.py to move a file into an error or failure folder based on the input * Added tests to test the task and the utils.py method
206 lines
7.5 KiB
Python
206 lines
7.5 KiB
Python
import pytest
|
|
from datetime import datetime
|
|
|
|
import boto3
|
|
from flask import current_app
|
|
from freezegun import freeze_time
|
|
from moto import mock_s3
|
|
|
|
from app.letters.utils import (
|
|
get_bucket_prefix_for_notification,
|
|
get_letter_pdf_filename,
|
|
get_letter_pdf,
|
|
upload_letter_pdf,
|
|
move_scanned_pdf_to_test_or_live_pdf_bucket,
|
|
ScanErrorType, move_failed_pdf)
|
|
from app.models import KEY_TYPE_NORMAL, KEY_TYPE_TEST, PRECOMPILED_TEMPLATE_NAME
|
|
from app.variables import Retention
|
|
|
|
FROZEN_DATE_TIME = "2018-03-14 17:00:00"
|
|
|
|
|
|
@pytest.fixture()
|
|
@freeze_time(FROZEN_DATE_TIME)
|
|
def sample_precompiled_letter_notification_using_test_key(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'
|
|
sample_letter_notification.created_at = datetime.utcnow()
|
|
return sample_letter_notification
|
|
|
|
|
|
def test_get_bucket_prefix_for_notification_valid_notification(sample_notification):
|
|
|
|
bucket_prefix = get_bucket_prefix_for_notification(sample_notification)
|
|
|
|
assert bucket_prefix == '{folder}/NOTIFY.{reference}'.format(
|
|
folder=sample_notification.created_at.date(),
|
|
reference=sample_notification.reference
|
|
).upper()
|
|
|
|
|
|
@freeze_time(FROZEN_DATE_TIME)
|
|
def test_get_bucket_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)
|
|
|
|
assert bucket_prefix == 'NOTIFY.{}'.format(
|
|
sample_precompiled_letter_notification_using_test_key.reference).upper()
|
|
|
|
|
|
def test_get_bucket_prefix_for_notification_invalid_notification():
|
|
with pytest.raises(AttributeError):
|
|
get_bucket_prefix_for_notification(None)
|
|
|
|
|
|
@pytest.mark.parametrize('crown_flag,expected_crown_text', [
|
|
(True, 'C'),
|
|
(False, 'N'),
|
|
])
|
|
@freeze_time("2017-12-04 17:29:00")
|
|
def test_get_letter_pdf_filename_returns_correct_filename(
|
|
notify_api, mocker, crown_flag, expected_crown_text):
|
|
filename = get_letter_pdf_filename(reference='foo', crown=crown_flag)
|
|
|
|
assert filename == '2017-12-04/NOTIFY.FOO.D.2.C.{}.20171204172900.PDF'.format(expected_crown_text)
|
|
|
|
|
|
@freeze_time("2017-12-04 17:29:00")
|
|
def test_get_letter_pdf_filename_returns_correct_filename_for_test_letters(
|
|
notify_api, mocker):
|
|
filename = get_letter_pdf_filename(reference='foo', crown='C', is_scan_letter=True)
|
|
|
|
assert filename == 'NOTIFY.FOO.D.2.C.C.20171204172900.PDF'
|
|
|
|
|
|
@freeze_time("2017-12-04 17:31:00")
|
|
def test_get_letter_pdf_filename_returns_tomorrows_filename(notify_api, mocker):
|
|
filename = get_letter_pdf_filename(reference='foo', crown=True)
|
|
|
|
assert filename == '2017-12-05/NOTIFY.FOO.D.2.C.C.20171204173100.PDF'
|
|
|
|
|
|
@mock_s3
|
|
@pytest.mark.parametrize('bucket_config_name,filename_format', [
|
|
('TEST_LETTERS_BUCKET_NAME', 'NOTIFY.FOO.D.2.C.C.%Y%m%d%H%M%S.PDF'),
|
|
('LETTERS_PDF_BUCKET_NAME', '%Y-%m-%d/NOTIFY.FOO.D.2.C.C.%Y%m%d%H%M%S.PDF')
|
|
])
|
|
@freeze_time(FROZEN_DATE_TIME)
|
|
def test_get_letter_pdf_gets_pdf_from_correct_bucket(
|
|
sample_precompiled_letter_notification_using_test_key,
|
|
bucket_config_name,
|
|
filename_format
|
|
):
|
|
if bucket_config_name == 'LETTERS_PDF_BUCKET_NAME':
|
|
sample_precompiled_letter_notification_using_test_key.key_type = KEY_TYPE_NORMAL
|
|
|
|
bucket_name = current_app.config[bucket_config_name]
|
|
filename = datetime.utcnow().strftime(filename_format)
|
|
conn = boto3.resource('s3', region_name='eu-west-1')
|
|
conn.create_bucket(Bucket=bucket_name)
|
|
s3 = boto3.client('s3', region_name='eu-west-1')
|
|
s3.put_object(Bucket=bucket_name, Key=filename, Body=b'pdf_content')
|
|
|
|
ret = get_letter_pdf(sample_precompiled_letter_notification_using_test_key)
|
|
|
|
assert ret == b'pdf_content'
|
|
|
|
|
|
@pytest.mark.parametrize('is_precompiled_letter,bucket_config_name', [
|
|
(False, 'LETTERS_PDF_BUCKET_NAME'),
|
|
(True, 'LETTERS_SCAN_BUCKET_NAME')
|
|
])
|
|
def test_upload_letter_pdf_to_correct_bucket(
|
|
sample_letter_notification, mocker, is_precompiled_letter, bucket_config_name
|
|
):
|
|
if is_precompiled_letter:
|
|
sample_letter_notification.template.hidden = True
|
|
sample_letter_notification.template.name = PRECOMPILED_TEMPLATE_NAME
|
|
|
|
mock_s3 = mocker.patch('app.letters.utils.s3upload')
|
|
|
|
filename = get_letter_pdf_filename(
|
|
reference=sample_letter_notification.reference,
|
|
crown=sample_letter_notification.service.crown,
|
|
is_scan_letter=is_precompiled_letter
|
|
)
|
|
|
|
upload_letter_pdf(sample_letter_notification, b'\x00\x01')
|
|
|
|
mock_s3.assert_called_once_with(
|
|
bucket_name=current_app.config[bucket_config_name],
|
|
file_location=filename,
|
|
filedata=b'\x00\x01',
|
|
region=current_app.config['AWS_REGION'],
|
|
tags={Retention.KEY: Retention.ONE_WEEK}
|
|
)
|
|
|
|
|
|
@mock_s3
|
|
@pytest.mark.parametrize('is_test_letter,bucket_config_name,folder_date_name', [
|
|
(False, 'LETTERS_PDF_BUCKET_NAME', '2018-03-14/'),
|
|
(True, 'TEST_LETTERS_BUCKET_NAME', '')
|
|
])
|
|
@freeze_time(FROZEN_DATE_TIME)
|
|
def test_move_scanned_letter_pdf_to_processing_bucket(
|
|
notify_api, is_test_letter, bucket_config_name, folder_date_name
|
|
):
|
|
filename = 'test.pdf'
|
|
source_bucket_name = current_app.config['LETTERS_SCAN_BUCKET_NAME']
|
|
target_bucket_name = current_app.config[bucket_config_name]
|
|
|
|
conn = boto3.resource('s3', region_name='eu-west-1')
|
|
source_bucket = conn.create_bucket(Bucket=source_bucket_name)
|
|
target_bucket = 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')
|
|
|
|
move_scanned_pdf_to_test_or_live_pdf_bucket(filename, is_test_letter=is_test_letter)
|
|
|
|
assert folder_date_name + filename in [o.key for o in target_bucket.objects.all()]
|
|
assert filename not in [o.key for o in source_bucket.objects.all()]
|
|
|
|
|
|
@mock_s3
|
|
@freeze_time(FROZEN_DATE_TIME)
|
|
def test_move_failed_pdf_error(notify_api):
|
|
filename = 'test.pdf'
|
|
source_bucket_name = current_app.config['LETTERS_SCAN_BUCKET_NAME']
|
|
target_bucket_name = current_app.config['LETTERS_SCAN_BUCKET_NAME']
|
|
|
|
conn = boto3.resource('s3', region_name='eu-west-1')
|
|
source_bucket = conn.create_bucket(Bucket=source_bucket_name)
|
|
target_bucket = 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')
|
|
|
|
move_failed_pdf(filename, ScanErrorType.ERROR)
|
|
|
|
assert 'ERROR/' + filename in [o.key for o in target_bucket.objects.all()]
|
|
assert filename not in [o.key for o in source_bucket.objects.all()]
|
|
|
|
|
|
@mock_s3
|
|
@freeze_time(FROZEN_DATE_TIME)
|
|
def test_move_failed_pdf_scan_failed(notify_api):
|
|
filename = 'test.pdf'
|
|
source_bucket_name = current_app.config['LETTERS_SCAN_BUCKET_NAME']
|
|
target_bucket_name = current_app.config['LETTERS_SCAN_BUCKET_NAME']
|
|
|
|
conn = boto3.resource('s3', region_name='eu-west-1')
|
|
source_bucket = conn.create_bucket(Bucket=source_bucket_name)
|
|
target_bucket = 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')
|
|
|
|
move_failed_pdf(filename, ScanErrorType.FAILURE)
|
|
|
|
assert 'FAILURE/' + filename in [o.key for o in target_bucket.objects.all()]
|
|
assert filename not in [o.key for o in source_bucket.objects.all()]
|