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

Merge pull request #1802 from alphagov/rc_add_virus_scan_error_task

Browse Source 
Added a new task to handle any error cases with the anti-virus
This commit is contained in:
Richard Chapman
2018-03-27 14:03:24 +01:00
committed by GitHub
parent c02b112e9b 023862dfdc
commit 20179af6e6
4 changed files with 109 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -19,8 +19,8 @@ from app.celery.letters_pdf_tasks import (
letter_in_created_state,
process_virus_scan_passed,
process_virus_scan_failed,
)
from app.letters.utils import get_letter_pdf_filename
process_virus_scan_error)
from app.letters.utils import get_letter_pdf_filename, ScanErrorType
from app.models import (
KEY_TYPE_NORMAL,
KEY_TYPE_TEST,
@@ -28,8 +28,8 @@ from app.models import (
NOTIFICATION_CREATED,
NOTIFICATION_DELIVERED,
NOTIFICATION_VIRUS_SCAN_FAILED,
NOTIFICATION_SENDING
)
NOTIFICATION_SENDING,
NOTIFICATION_TECHNICAL_FAILURE)
from tests.conftest import set_config_values
@@ -341,9 +341,20 @@ def test_process_letter_task_check_virus_scan_passed(
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_delete_pdf = mocker.patch('app.celery.letters_pdf_tasks.delete_pdf_from_letters_scan_bucket')
mock_move_failed_pdf = mocker.patch('app.celery.letters_pdf_tasks.move_failed_pdf')
process_virus_scan_failed(filename)
mock_delete_pdf.assert_called_once_with(filename)
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')
process_virus_scan_error(filename)
mock_move_failed_pdf.assert_called_once_with(filename, ScanErrorType.ERROR)
assert sample_letter_notification.status == NOTIFICATION_TECHNICAL_FAILURE

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

@@ -11,8 +11,8 @@ from app.letters.utils import (
get_letter_pdf_filename,
get_letter_pdf,
upload_letter_pdf,
move_scanned_pdf_to_test_or_live_pdf_bucket
)
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
@@ -163,3 +163,43 @@ def test_move_scanned_letter_pdf_to_processing_bucket(
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()]