mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-24 01:11:38 -05:00
Add antivirus check on precompiled letters sent with test key
- precompiled PDFs sent by test key uploaded to scan bucket - set status to VIRUS-SCAN-FAILED for pdfs failing virus scan rather than PERMANENT-FAILURE - Make call to AV app for precompiled letters sent via a test key, and set notification status to PENDING-VIRUS-SCAN
This commit is contained in:
@@ -22,9 +22,12 @@ from app.celery.letters_pdf_tasks import (
|
||||
)
|
||||
from app.letters.utils import get_letter_pdf_filename
|
||||
from app.models import (
|
||||
KEY_TYPE_NORMAL,
|
||||
KEY_TYPE_TEST,
|
||||
Notification,
|
||||
NOTIFICATION_CREATED,
|
||||
NOTIFICATION_PERMANENT_FAILURE,
|
||||
NOTIFICATION_DELIVERED,
|
||||
NOTIFICATION_VIRUS_SCAN_FAILED,
|
||||
NOTIFICATION_SENDING
|
||||
)
|
||||
|
||||
@@ -317,15 +320,22 @@ def test_letter_in_created_state_fails_if_notification_doesnt_exist(sample_notif
|
||||
assert letter_in_created_state(filename) is False
|
||||
|
||||
|
||||
def test_process_letter_task_check_virus_scan_passed(sample_letter_notification, mocker):
|
||||
@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'
|
||||
mock_move_pdf = mocker.patch('app.celery.letters_pdf_tasks.move_scanned_pdf_to_letters_pdf_bucket')
|
||||
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)
|
||||
assert sample_letter_notification.status == NOTIFICATION_CREATED
|
||||
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):
|
||||
@@ -336,4 +346,4 @@ def test_process_letter_task_check_virus_scan_failed(sample_letter_notification,
|
||||
process_virus_scan_failed(filename)
|
||||
|
||||
mock_delete_pdf.assert_called_once_with(filename)
|
||||
assert sample_letter_notification.status == NOTIFICATION_PERMANENT_FAILURE
|
||||
assert sample_letter_notification.status == NOTIFICATION_VIRUS_SCAN_FAILED
|
||||
|
||||
@@ -11,7 +11,7 @@ from app.letters.utils import (
|
||||
get_letter_pdf_filename,
|
||||
get_letter_pdf,
|
||||
upload_letter_pdf,
|
||||
move_scanned_pdf_to_letters_pdf_bucket
|
||||
move_scanned_pdf_to_test_or_live_pdf_bucket
|
||||
)
|
||||
from app.models import KEY_TYPE_NORMAL, KEY_TYPE_TEST, PRECOMPILED_TEMPLATE_NAME
|
||||
from app.variables import Retention
|
||||
@@ -71,7 +71,7 @@ def test_get_letter_pdf_filename_returns_correct_filename(
|
||||
@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_test_or_scan_letter=True)
|
||||
filename = get_letter_pdf_filename(reference='foo', crown='C', is_scan_letter=True)
|
||||
|
||||
assert filename == 'NOTIFY.FOO.D.2.C.C.20171204172900.PDF'
|
||||
|
||||
@@ -125,7 +125,7 @@ def test_upload_letter_pdf_to_correct_bucket(
|
||||
filename = get_letter_pdf_filename(
|
||||
reference=sample_letter_notification.reference,
|
||||
crown=sample_letter_notification.service.crown,
|
||||
is_test_or_scan_letter=is_precompiled_letter
|
||||
is_scan_letter=is_precompiled_letter
|
||||
)
|
||||
|
||||
upload_letter_pdf(sample_letter_notification, b'\x00\x01')
|
||||
@@ -140,11 +140,17 @@ def test_upload_letter_pdf_to_correct_bucket(
|
||||
|
||||
|
||||
@mock_s3
|
||||
@pytest.mark.parametrize('is_test_letter,bucket_config_name', [
|
||||
(False, 'LETTERS_PDF_BUCKET_NAME'),
|
||||
(True, 'TEST_LETTERS_BUCKET_NAME')
|
||||
])
|
||||
@freeze_time(FROZEN_DATE_TIME)
|
||||
def test_move_scanned_letter_pdf_to_processing_bucket(notify_api):
|
||||
def test_move_scanned_letter_pdf_to_processing_bucket(
|
||||
notify_api, is_test_letter, bucket_config_name
|
||||
):
|
||||
filename = 'test.pdf'
|
||||
source_bucket_name = current_app.config['LETTERS_SCAN_BUCKET_NAME']
|
||||
target_bucket_name = current_app.config['LETTERS_PDF_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)
|
||||
@@ -153,7 +159,7 @@ def test_move_scanned_letter_pdf_to_processing_bucket(notify_api):
|
||||
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_letters_pdf_bucket(filename)
|
||||
move_scanned_pdf_to_test_or_live_pdf_bucket(filename, is_test_letter=is_test_letter)
|
||||
|
||||
assert '2018-03-14/' + filename in [o.key for o in target_bucket.objects.all()]
|
||||
assert filename not in [o.key for o in source_bucket.objects.all()]
|
||||
|
||||
@@ -5,7 +5,7 @@ from flask import json
|
||||
from flask import url_for
|
||||
import pytest
|
||||
|
||||
from app.config import QueueNames
|
||||
from app.config import TaskNames, QueueNames
|
||||
from app.models import (
|
||||
Job,
|
||||
Notification,
|
||||
@@ -399,8 +399,9 @@ def test_post_letter_notification_is_delivered_and_has_pdf_uploaded_to_test_lett
|
||||
mocker
|
||||
):
|
||||
sample_letter_service = create_service(service_permissions=['letter', 'precompiled_letter'])
|
||||
s3mock = mocker.patch('app.v2.notifications.post_notifications.upload_letter_pdf')
|
||||
s3mock = mocker.patch('app.v2.notifications.post_notifications.upload_letter_pdf', return_value='test.pdf')
|
||||
mocker.patch('app.v2.notifications.post_notifications.pdf_page_count', return_value=1)
|
||||
mock_celery = mocker.patch("app.letters.rest.notify_celery.send_task")
|
||||
data = {
|
||||
"reference": "letter-reference",
|
||||
"content": "bGV0dGVyLWNvbnRlbnQ="
|
||||
@@ -413,8 +414,13 @@ def test_post_letter_notification_is_delivered_and_has_pdf_uploaded_to_test_lett
|
||||
precompiled=True)
|
||||
|
||||
notification = Notification.query.one()
|
||||
assert notification.status == NOTIFICATION_DELIVERED
|
||||
s3mock.assert_called_once_with(ANY, b'letter-content', is_test_letter=True)
|
||||
assert notification.status == NOTIFICATION_PENDING_VIRUS_CHECK
|
||||
s3mock.assert_called_once_with(ANY, b'letter-content')
|
||||
mock_celery.assert_called_once_with(
|
||||
name=TaskNames.SCAN_FILE,
|
||||
kwargs={'filename': 'test.pdf'},
|
||||
queue=QueueNames.ANTIVIRUS
|
||||
)
|
||||
|
||||
|
||||
def test_post_letter_notification_persists_notification_reply_to_text(
|
||||
|
||||
Reference in New Issue
Block a user