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

Merge pull request #1783 from alphagov/ken-process-antivirus

Browse Source 
Send task to antivirus app and process antivirus callbacks
This commit is contained in:
kentsanggds
2018-03-21 16:39:55 +00:00
committed by GitHub
parent 40a3986720 d6f6669610
commit 5dc0248043
9 changed files with 336 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
app/celery/letters_pdf_tasks.py
View File

@@ -1,12 +1,14 @@
import math
from datetime import datetime
from botocore.exceptions import ClientError as BotoClientError
from flask import current_app
from notifications_utils.statsd_decorators import statsd
from requests import (
post as requests_post,
RequestException
)
from botocore.exceptions import ClientError as BotoClientError
from notifications_utils.statsd_decorators import statsd
from app import notify_celery
from app.aws import s3
@@ -16,9 +18,15 @@ from app.dao.notifications_dao import (
update_notification_status_by_id,
dao_update_notification,
dao_get_notifications_by_references,
dao_update_notifications_by_reference,
)
from app.letters.utils import upload_letter_pdf
from app.models import NOTIFICATION_CREATED
from app.letters.utils import (
delete_pdf_from_letters_scan_bucket,
get_reference_from_filename,
move_scanned_pdf_to_letters_pdf_bucket,
upload_letter_pdf
)
from app.models import NOTIFICATION_CREATED, NOTIFICATION_PERMANENT_FAILURE
@notify_celery.task(bind=True, name="create-letters-pdf", max_retries=15, default_retry_delay=300)
@@ -133,7 +141,7 @@ def group_letters(letter_pdfs):
def letter_in_created_state(filename):
# filename looks like '2018-01-13/NOTIFY.ABCDEF1234567890.D.2.C.C.20180113120000.PDF'
subfolder = filename.split('/')[0]
ref = filename.split('.')[1]
ref = get_reference_from_filename(filename)
notifications = dao_get_notifications_by_references([ref])
if notifications:
if notifications[0].status == NOTIFICATION_CREATED:
@@ -144,3 +152,42 @@ def letter_in_created_state(filename):
notifications[0].status
))
return False
@notify_celery.task(name='process-virus-scan-passed')
def process_virus_scan_passed(filename):
current_app.logger.info('Virus scan passed: {}'.format(filename))
move_scanned_pdf_to_letters_pdf_bucket(filename)
reference = get_reference_from_filename(filename)
updated_count = update_letter_pdf_status(reference, NOTIFICATION_CREATED)
if updated_count != 1:
raise Exception(
"There should only be one letter notification for each reference. Found {} notifications".format(
updated_count
)
)
@notify_celery.task(name='process-virus-scan-failed')
def process_virus_scan_failed(filename):
current_app.logger.error('Virus scan failed: {}'.format(filename))
delete_pdf_from_letters_scan_bucket(filename)
reference = get_reference_from_filename(filename)
updated_count = update_letter_pdf_status(reference, NOTIFICATION_PERMANENT_FAILURE)
if updated_count != 1:
raise Exception(
"There should only be one letter notification for each reference. Found {} notifications".format(
updated_count
)
)
def update_letter_pdf_status(reference, status):
return dao_update_notifications_by_reference(
references=[reference],
update_dict={
'status': status,
'updated_at': datetime.utcnow()
})

14
app/config.py
View File

@@ -32,6 +32,7 @@ class QueueNames(object):
CREATE_LETTERS_PDF = 'create-letters-pdf-tasks'
CALLBACKS = 'service-callbacks'
LETTERS = 'letter-tasks'
ANTIVIRUS = 'antivirus-tasks'
@staticmethod
def all_queues():
@@ -56,6 +57,7 @@ class TaskNames(object):
DVLA_JOBS = 'send-jobs-to-dvla'
PROCESS_INCOMPLETE_JOBS = 'process-incomplete-jobs'
ZIP_AND_SEND_LETTER_PDFS = 'zip-and-send-letter-pdfs'
SCAN_FILE = 'scan-file'
class Config(object):
@@ -328,6 +330,8 @@ class Development(Config):
LETTERS_PDF_BUCKET_NAME = 'development-letters-pdf'
TEST_LETTERS_BUCKET_NAME = 'development-test-letters'
DVLA_RESPONSE_BUCKET_NAME = 'notify.tools-ftp'
LETTERS_PDF_BUCKET_NAME = 'development-letters-pdf'
LETTERS_SCAN_BUCKET_NAME = 'development-letters-scan'
ADMIN_CLIENT_SECRET = 'dev-notify-secret-key'
SECRET_KEY = 'dev-notify-secret-key'
@@ -365,6 +369,8 @@ class Test(Development):
LETTERS_PDF_BUCKET_NAME = 'test-letters-pdf'
TEST_LETTERS_BUCKET_NAME = 'test-test-letters'
DVLA_RESPONSE_BUCKET_NAME = 'test.notify.com-ftp'
LETTERS_PDF_BUCKET_NAME = 'test-letters-pdf'
LETTERS_SCAN_BUCKET_NAME = 'test-letters-scan'
# this is overriden in jenkins and on cloudfoundry
SQLALCHEMY_DATABASE_URI = os.getenv('SQLALCHEMY_DATABASE_URI', 'postgresql://localhost/test_notification_api')
@@ -393,6 +399,8 @@ class Preview(Config):
LETTERS_PDF_BUCKET_NAME = 'preview-letters-pdf'
TEST_LETTERS_BUCKET_NAME = 'preview-test-letters'
DVLA_RESPONSE_BUCKET_NAME = 'notify.works-ftp'
LETTERS_PDF_BUCKET_NAME = 'preview-letters-pdf'
LETTERS_SCAN_BUCKET_NAME = 'preview-letters-scan'
FROM_NUMBER = 'preview'
API_RATE_LIMIT_ENABLED = True
CHECK_PROXY_HEADER = True
@@ -405,6 +413,8 @@ class Staging(Config):
LETTERS_PDF_BUCKET_NAME = 'staging-letters-pdf'
TEST_LETTERS_BUCKET_NAME = 'staging-test-letters'
DVLA_RESPONSE_BUCKET_NAME = 'staging-notify.works-ftp'
LETTERS_PDF_BUCKET_NAME = 'staging-letters-pdf'
LETTERS_SCAN_BUCKET_NAME = 'staging-letters-scan'
STATSD_ENABLED = True
FROM_NUMBER = 'stage'
API_RATE_LIMIT_ENABLED = True
@@ -419,6 +429,8 @@ class Live(Config):
LETTERS_PDF_BUCKET_NAME = 'production-letters-pdf'
TEST_LETTERS_BUCKET_NAME = 'production-test-letters'
DVLA_RESPONSE_BUCKET_NAME = 'notifications.service.gov.uk-ftp'
LETTERS_PDF_BUCKET_NAME = 'production-letters-pdf'
LETTERS_SCAN_BUCKET_NAME = 'production-letters-scan'
STATSD_ENABLED = True
FROM_NUMBER = 'GOVUK'
FUNCTIONAL_TEST_PROVIDER_SERVICE_ID = '6c1d81bb-dae2-4ee9-80b0-89a4aae9f649'
@@ -440,6 +452,8 @@ class Sandbox(CloudFoundryConfig):
LETTERS_PDF_BUCKET_NAME = 'cf-sandbox-letters-pdf'
TEST_LETTERS_BUCKET_NAME = 'cf-sandbox-test-letters'
DVLA_RESPONSE_BUCKET_NAME = 'notify.works-ftp'
LETTERS_PDF_BUCKET_NAME = 'cf-sandbox-letters-pdf'
LETTERS_SCAN_BUCKET_NAME = 'cf-sandbox-letters-scan'
FROM_NUMBER = 'sandbox'
REDIS_ENABLED = False

54
app/letters/utils.py
View File

@@ -15,8 +15,8 @@ LETTERS_PDF_FILE_LOCATION_STRUCTURE = \
PRECOMPILED_BUCKET_PREFIX = '{folder}NOTIFY.{reference}'
def get_folder_name(_now, is_test_letter):
if is_test_letter:
def get_folder_name(_now, is_test_or_scan_letter=False):
if is_test_or_scan_letter:
folder_name = ''
else:
print_datetime = _now
@@ -26,11 +26,11 @@ def get_folder_name(_now, is_test_letter):
return folder_name
def get_letter_pdf_filename(reference, crown, is_test_letter=False):
def get_letter_pdf_filename(reference, crown, is_test_or_scan_letter=False):
now = datetime.utcnow()
upload_file_name = LETTERS_PDF_FILE_LOCATION_STRUCTURE.format(
folder=get_folder_name(now, is_test_letter),
folder=get_folder_name(now, is_test_or_scan_letter),
reference=reference,
duplex="D",
letter_class="2",
@@ -52,17 +52,28 @@ def get_bucket_prefix_for_notification(notification, is_test_letter=False):
return upload_file_name
def get_reference_from_filename(filename):
# filename looks like '2018-01-13/NOTIFY.ABCDEF1234567890.D.2.C.C.20180113120000.PDF'
filename_parts = filename.split('.')
return filename_parts[1]
def upload_letter_pdf(notification, pdf_data, is_test_letter=False):
current_app.logger.info("PDF Letter {} reference {} created at {}, {} bytes".format(
notification.id, notification.reference, notification.created_at, len(pdf_data)))
upload_file_name = get_letter_pdf_filename(
notification.reference, notification.service.crown, is_test_letter)
notification.reference,
notification.service.crown,
is_test_or_scan_letter=is_test_letter or notification.template.is_precompiled_letter)
if is_test_letter:
bucket_name = current_app.config['TEST_LETTERS_BUCKET_NAME']
else:
bucket_name = current_app.config['LETTERS_PDF_BUCKET_NAME']
if notification.template.is_precompiled_letter:
bucket_name = current_app.config['LETTERS_SCAN_BUCKET_NAME']
else:
bucket_name = current_app.config['LETTERS_PDF_BUCKET_NAME']
s3upload(
filedata=pdf_data,
@@ -74,6 +85,37 @@ def upload_letter_pdf(notification, pdf_data, is_test_letter=False):
current_app.logger.info("Uploaded letters PDF {} to {} for notification id {}".format(
upload_file_name, bucket_name, notification.id))
return upload_file_name
def move_scanned_pdf_to_letters_pdf_bucket(filename):
source_bucket_name = current_app.config['LETTERS_SCAN_BUCKET_NAME']
target_bucket_name = current_app.config['LETTERS_PDF_BUCKET_NAME']
s3 = boto3.resource('s3')
copy_source = {'Bucket': source_bucket_name, 'Key': filename}
target_filename = get_folder_name(datetime.utcnow()) + filename
target_bucket = s3.Bucket(target_bucket_name)
obj = target_bucket.Object(target_filename)
# Tags are copied across but the expiration time is reset in the destination bucket
# e.g. if a file has 5 days left to expire on a ONE_WEEK retention in the source bucket,
# in the destination bucket the expiration time will be reset to 7 days left to expire
obj.copy(copy_source, ExtraArgs={'ServerSideEncryption': 'AES256'})
s3.Object(source_bucket_name, filename).delete()
current_app.logger.info("Moved letter PDF: {}/{} to {}/{}".format(
source_bucket_name, filename, target_bucket_name, target_filename))
def delete_pdf_from_letters_scan_bucket(filename):
bucket_name = current_app.config['LETTERS_SCAN_BUCKET_NAME']
s3 = boto3.resource('s3')
s3.Object(bucket_name, filename).delete()
current_app.logger.info("Deleted letter PDF: {}/{}".format(bucket_name, filename))
def get_letter_pdf(notification):

32
app/v2/notifications/post_notifications.py
View File

@@ -4,12 +4,11 @@ import io
import math
from flask import request, jsonify, current_app, abort
from notifications_utils.pdf import pdf_page_count, PdfReadError
from notifications_utils.recipients import try_validate_and_format_phone_number
from app import api_user, authenticated_service
from app.config import QueueNames
from app import api_user, authenticated_service, notify_celery
from app.config import QueueNames, TaskNames
from app.dao.notifications_dao import dao_update_notification, update_notification_status_by_reference
from app.dao.templates_dao import dao_create_template
from app.dao.users_dao import get_user_by_id
@@ -25,7 +24,8 @@ from app.models import (
KEY_TYPE_TEAM,
NOTIFICATION_CREATED,
NOTIFICATION_SENDING,
NOTIFICATION_DELIVERED
NOTIFICATION_DELIVERED,
NOTIFICATION_PENDING_VIRUS_CHECK,
)
from app.celery.letters_pdf_tasks import create_letters_pdf
from app.celery.research_mode_tasks import create_fake_letter_response_file
@@ -220,8 +220,15 @@ def process_letter_notification(*, letter_data, api_key, template, reply_to_text
if not api_key.service.research_mode and api_key.service.restricted and api_key.key_type != KEY_TYPE_TEST:
raise BadRequestError(message='Cannot send letters when service is in trial mode', status_code=403)
should_send = not (api_key.service.research_mode or api_key.key_type == KEY_TYPE_TEST)
# if we don't want to actually send the letter, then start it off in SENDING so we don't pick it up
status = NOTIFICATION_CREATED if should_send else NOTIFICATION_SENDING
if precompiled:
try:
if should_send:
status = NOTIFICATION_PENDING_VIRUS_CHECK
letter_content = base64.b64decode(letter_data['content'])
pages = pdf_page_count(io.BytesIO(letter_content))
except ValueError:
@@ -230,10 +237,6 @@ def process_letter_notification(*, letter_data, api_key, template, reply_to_text
current_app.logger.exception(msg='Invalid PDF received')
raise BadRequestError(message='Letter content is not a valid PDF', status_code=400)
should_send = not (api_key.service.research_mode or api_key.key_type == KEY_TYPE_TEST)
# if we don't want to actually send the letter, then start it off in SENDING so we don't pick it up
status = NOTIFICATION_CREATED if should_send else NOTIFICATION_SENDING
notification = create_letter_notification(letter_data=letter_data,
template=template,
api_key=api_key,
@@ -242,10 +245,21 @@ def process_letter_notification(*, letter_data, api_key, template, reply_to_text
if should_send:
if precompiled:
upload_letter_pdf(notification, letter_content)
filename = upload_letter_pdf(notification, letter_content)
pages_per_sheet = 2
notification.billable_units = math.ceil(pages / pages_per_sheet)
dao_update_notification(notification)
current_app.logger.info(
'Calling task scan-file for {}'.format(filename)
)
# call task to add the filename to anti virus queue
notify_celery.send_task(
name=TaskNames.SCAN_FILE,
kwargs={'filename': filename},
queue=QueueNames.ANTIVIRUS,
)
else:
create_letters_pdf.apply_async(
[str(notification.id)],

2
requirements_for_test.txt
View File

@@ -1,6 +1,6 @@
-r requirements.txt
flake8==3.5.0
moto==1.1.25
moto==1.2
pytest==3.4.2
pytest-env==0.6.2
pytest-mock==1.7.1

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

@@ -16,10 +16,17 @@ from app.celery.letters_pdf_tasks import (
get_letters_pdf,
collate_letter_pdfs_for_day,
group_letters,
letter_in_created_state
letter_in_created_state,
process_virus_scan_passed,
process_virus_scan_failed,
)
from app.letters.utils import get_letter_pdf_filename
from app.models import Notification, NOTIFICATION_SENDING
from app.models import (
Notification,
NOTIFICATION_CREATED,
NOTIFICATION_PERMANENT_FAILURE,
NOTIFICATION_SENDING
)
from tests.conftest import set_config_values
@@ -308,3 +315,25 @@ def test_letter_in_created_state_fails_if_notification_doesnt_exist(sample_notif
sample_notification.reference = 'QWERTY1234567890'
filename = '2018-01-13/NOTIFY.ABCDEF1234567890.D.2.C.C.20180113120000.PDF'
assert letter_in_created_state(filename) is False
def test_process_letter_task_check_virus_scan_passed(sample_letter_notification, mocker):
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')
process_virus_scan_passed(filename)
mock_move_pdf.assert_called_once_with(filename)
assert sample_letter_notification.status == NOTIFICATION_CREATED
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')
process_virus_scan_failed(filename)
mock_delete_pdf.assert_called_once_with(filename)
assert sample_letter_notification.status == NOTIFICATION_PERMANENT_FAILURE

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

@@ -6,8 +6,15 @@ 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
from app.letters.utils import (
get_bucket_prefix_for_notification,
get_letter_pdf_filename,
get_letter_pdf,
upload_letter_pdf,
move_scanned_pdf_to_letters_pdf_bucket
)
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"
@@ -64,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_letter=True)
filename = get_letter_pdf_filename(reference='foo', crown='C', is_test_or_scan_letter=True)
assert filename == 'NOTIFY.FOO.D.2.C.C.20171204172900.PDF'
@@ -100,3 +107,53 @@ def test_get_letter_pdf_gets_pdf_from_correct_bucket(
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_test_or_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
@freeze_time(FROZEN_DATE_TIME)
def test_move_scanned_letter_pdf_to_processing_bucket(notify_api):
filename = 'test.pdf'
source_bucket_name = current_app.config['LETTERS_SCAN_BUCKET_NAME']
target_bucket_name = current_app.config['LETTERS_PDF_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_scanned_pdf_to_letters_pdf_bucket(filename)
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()]

113
tests/app/v2/notifications/test_post_letter_notifications.py
View File

@@ -6,15 +6,20 @@ from flask import url_for
import pytest
from app.config import QueueNames
from app.models import EMAIL_TYPE
from app.models import Job
from app.models import KEY_TYPE_NORMAL
from app.models import KEY_TYPE_TEAM
from app.models import KEY_TYPE_TEST
from app.models import LETTER_TYPE
from app.models import Notification
from app.models import NOTIFICATION_SENDING, NOTIFICATION_DELIVERED
from app.models import SMS_TYPE
from app.models import (
Job,
Notification,
EMAIL_TYPE,
KEY_TYPE_NORMAL,
KEY_TYPE_TEAM,
KEY_TYPE_TEST,
LETTER_TYPE,
NOTIFICATION_CREATED,
NOTIFICATION_SENDING,
NOTIFICATION_DELIVERED,
NOTIFICATION_PENDING_VIRUS_CHECK,
SMS_TYPE,
)
from app.schema_validation import validate
from app.v2.errors import RateLimitError
from app.v2.notifications.notification_schemas import post_letter_response
@@ -70,6 +75,7 @@ def test_post_letter_notification_returns_201(client, sample_letter_template, mo
assert validate(resp_json, post_letter_response) == resp_json
assert Job.query.count() == 0
notification = Notification.query.one()
assert notification.status == NOTIFICATION_CREATED
notification_id = notification.id
assert resp_json['id'] == str(notification_id)
assert resp_json['reference'] == reference
@@ -429,3 +435,92 @@ def test_post_letter_notification_persists_notification_reply_to_text(
notifications = Notification.query.all()
assert len(notifications) == 1
assert notifications[0].reply_to_text == service_address
def test_post_precompiled_letter_requires_permission(client, sample_service, notify_user, mocker):
mocker.patch('app.v2.notifications.post_notifications.upload_letter_pdf')
data = {
"reference": "letter-reference",
"content": "bGV0dGVyLWNvbnRlbnQ="
}
auth_header = create_authorization_header(service_id=sample_service.id)
response = client.post(
path="v2/notifications/letter",
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 400, response.get_data(as_text=True)
resp_json = json.loads(response.get_data(as_text=True))
assert resp_json['errors'][0]['message'] == 'Cannot send precompiled_letters'
def test_post_precompiled_letter_with_invalid_base64(client, notify_user, mocker):
sample_service = create_service(service_permissions=['letter', 'precompiled_letter'])
mocker.patch('app.v2.notifications.post_notifications.upload_letter_pdf')
data = {
"reference": "letter-reference",
"content": "hi"
}
auth_header = create_authorization_header(service_id=sample_service.id)
response = client.post(
path="v2/notifications/letter",
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 400, response.get_data(as_text=True)
resp_json = json.loads(response.get_data(as_text=True))
assert resp_json['errors'][0]['message'] == 'Cannot decode letter content (invalid base64 encoding)'
assert not Notification.query.first()
def test_post_precompiled_letter_notification_returns_201(client, notify_user, mocker):
sample_service = create_service(service_permissions=['letter', 'precompiled_letter'])
s3mock = mocker.patch('app.v2.notifications.post_notifications.upload_letter_pdf')
mocker.patch('app.v2.notifications.post_notifications.pdf_page_count', return_value=5)
mocker.patch("app.letters.rest.notify_celery.send_task")
data = {
"reference": "letter-reference",
"content": "bGV0dGVyLWNvbnRlbnQ="
}
auth_header = create_authorization_header(service_id=sample_service.id)
response = client.post(
path="v2/notifications/letter",
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 201, response.get_data(as_text=True)
s3mock.assert_called_once_with(ANY, b'letter-content')
notification = Notification.query.first()
assert notification.billable_units == 3
assert notification.status == NOTIFICATION_PENDING_VIRUS_CHECK
resp_json = json.loads(response.get_data(as_text=True))
assert resp_json == {'id': str(notification.id), 'reference': 'letter-reference'}
def test_post_precompiled_letter_notification_returns_400_with_invalid_pdf(client, notify_user, mocker):
sample_service = create_service(service_permissions=['letter', 'precompiled_letter'])
s3mock = mocker.patch('app.v2.notifications.post_notifications.upload_letter_pdf')
data = {
"reference": "letter-reference",
"content": "bGV0dGVyLWNvbnRlbnQ="
}
auth_header = create_authorization_header(service_id=sample_service.id)
response = client.post(
path="v2/notifications/letter",
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
resp_json = json.loads(response.get_data(as_text=True))
assert response.status_code == 400, response.get_data(as_text=True)
assert resp_json['errors'][0]['message'] == 'Letter content is not a valid PDF'
assert s3mock.called is False
assert Notification.query.count() == 0

97
tests/app/v2/notifications/test_post_notifications.py
View File

@@ -1,5 +1,4 @@
import uuid
from unittest.mock import ANY
import pytest
from freezegun import freeze_time
@@ -7,8 +6,9 @@ from freezegun import freeze_time
from app.dao.service_sms_sender_dao import dao_update_service_sms_sender
from app.models import (
ScheduledNotification,
SCHEDULE_NOTIFICATIONS,
EMAIL_TYPE,
NOTIFICATION_CREATED,
SCHEDULE_NOTIFICATIONS,
SMS_TYPE
)
from flask import json, current_app
@@ -55,6 +55,7 @@ def test_post_sms_notification_returns_201(client, sample_template_with_placehol
assert validate(resp_json, post_sms_response) == resp_json
notifications = Notification.query.all()
assert len(notifications) == 1
assert notifications[0].status == NOTIFICATION_CREATED
notification_id = notifications[0].id
assert resp_json['id'] == str(notification_id)
assert resp_json['reference'] == reference
@@ -306,6 +307,7 @@ def test_post_email_notification_returns_201(client, sample_email_template_with_
resp_json = json.loads(response.get_data(as_text=True))
assert validate(resp_json, post_email_response) == resp_json
notification = Notification.query.one()
assert notification.status == NOTIFICATION_CREATED
assert resp_json['id'] == str(notification.id)
assert resp_json['reference'] == reference
assert notification.reference is None
@@ -695,94 +697,3 @@ def test_post_email_notification_with_invalid_reply_to_id_returns_400(client, sa
assert 'email_reply_to_id {} does not exist in database for service id {}'. \
format(fake_uuid, sample_email_template.service_id) in resp_json['errors'][0]['message']
assert 'BadRequestError' in resp_json['errors'][0]['error']
def test_post_precompiled_letter_requires_permission(client, sample_service, notify_user, mocker):
mocker.patch('app.v2.notifications.post_notifications.upload_letter_pdf')
data = {
"reference": "letter-reference",
"content": "bGV0dGVyLWNvbnRlbnQ="
}
auth_header = create_authorization_header(service_id=sample_service.id)
response = client.post(
path="v2/notifications/letter",
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 400, response.get_data(as_text=True)
resp_json = json.loads(response.get_data(as_text=True))
assert resp_json['errors'][0]['message'] == 'Cannot send precompiled_letters'
def test_post_precompiled_letter_with_invalid_base64(client, notify_user, mocker):
sample_service = create_service(service_permissions=['letter', 'precompiled_letter'])
mocker.patch('app.v2.notifications.post_notifications.upload_letter_pdf')
data = {
"reference": "letter-reference",
"content": "hi"
}
auth_header = create_authorization_header(service_id=sample_service.id)
response = client.post(
path="v2/notifications/letter",
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 400, response.get_data(as_text=True)
resp_json = json.loads(response.get_data(as_text=True))
assert resp_json['errors'][0]['message'] == 'Cannot decode letter content (invalid base64 encoding)'
assert not Notification.query.first()
def test_post_precompiled_letter_notification_returns_201(client, notify_user, mocker):
sample_service = create_service(service_permissions=['letter', 'precompiled_letter'])
s3mock = mocker.patch('app.v2.notifications.post_notifications.upload_letter_pdf')
mocker.patch('app.v2.notifications.post_notifications.pdf_page_count', return_value=5)
data = {
"reference": "letter-reference",
"content": "bGV0dGVyLWNvbnRlbnQ="
}
auth_header = create_authorization_header(service_id=sample_service.id)
response = client.post(
path="v2/notifications/letter",
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 201, response.get_data(as_text=True)
s3mock.assert_called_once_with(ANY, b'letter-content')
notification = Notification.query.first()
assert notification.billable_units == 3
resp_json = json.loads(response.get_data(as_text=True))
assert resp_json == {
'id': str(notification.id),
'reference': 'letter-reference'
}
def test_post_precompiled_letter_notification_returns_400_with_invalid_pdf(client, notify_user, mocker):
sample_service = create_service(service_permissions=['letter', 'precompiled_letter'])
s3mock = mocker.patch('app.v2.notifications.post_notifications.upload_letter_pdf')
data = {
"reference": "letter-reference",
"content": "bGV0dGVyLWNvbnRlbnQ="
}
auth_header = create_authorization_header(service_id=sample_service.id)
response = client.post(
path="v2/notifications/letter",
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
resp_json = json.loads(response.get_data(as_text=True))
assert response.status_code == 400, response.get_data(as_text=True)
assert resp_json['errors'][0]['message'] == 'Letter content is not a valid PDF'
assert s3mock.called is False
assert Notification.query.count() == 0