delete a variety of pdf-related things

This commit is contained in:
stvnrlly
2023-02-09 13:18:28 -05:00
parent e35c68345b
commit 6f6353e0f2
15 changed files with 27 additions and 1085 deletions

View File

@@ -419,57 +419,6 @@ def populate_organisation_agreement_details_from_file(file_name):
db.session.commit() db.session.commit()
@notify_command(name='get-letter-details-from-zips-sent-file')
@click.argument('file_paths', required=True, nargs=-1)
@statsd(namespace="tasks")
def get_letter_details_from_zips_sent_file(file_paths):
"""Get notification details from letters listed in zips_sent file(s)
This takes one or more file paths for the zips_sent files in S3 as its parameters, for example:
get-letter-details-from-zips-sent-file '2019-04-01/zips_sent/filename_1' '2019-04-01/zips_sent/filename_2'
"""
rows_from_file = []
for path in file_paths:
file_contents = s3.get_s3_file(
bucket_name=current_app.config['LETTERS_PDF_BUCKET_NAME'],
file_location=path
)
rows_from_file.extend(json.loads(file_contents))
notification_references = tuple(row[18:34] for row in rows_from_file)
get_letters_data_from_references(notification_references)
@notify_command(name='get-notification-and-service-ids-for-letters-that-failed-to-print')
@click.option('-f', '--file_name', required=True,
help="""Full path of the file to upload, file should contain letter filenames, one per line""")
def get_notification_and_service_ids_for_letters_that_failed_to_print(file_name):
print("Getting service and notification ids for letter filenames list {}".format(file_name))
file = open(file_name)
references = tuple([row[7:23] for row in file])
get_letters_data_from_references(tuple(references))
file.close()
def get_letters_data_from_references(notification_references):
sql = """
SELECT id, service_id, template_id, reference, job_id, created_at
FROM notifications
WHERE reference IN :notification_references
ORDER BY service_id, job_id"""
result = db.session.execute(sql, {'notification_references': notification_references}).fetchall()
with open('zips_sent_details.csv', 'w') as csvfile:
csv_writer = csv.writer(csvfile)
csv_writer.writerow(['notification_id', 'service_id', 'template_id', 'reference', 'job_id', 'created_at'])
for row in result:
csv_writer.writerow(row)
@notify_command(name='associate-services-to-organisations') @notify_command(name='associate-services-to-organisations')
def associate_services_to_organisations(): def associate_services_to_organisations():
services = Service.get_history_model().query.filter_by( services = Service.get_history_model().query.filter_by(

View File

@@ -134,10 +134,6 @@ class Config(object):
MAX_FAILED_LOGIN_COUNT = 10 MAX_FAILED_LOGIN_COUNT = 10
API_RATE_LIMIT_ENABLED = True API_RATE_LIMIT_ENABLED = True
# be careful increasing this size without being sure that we won't see slowness in pysftp
MAX_LETTER_PDF_ZIP_FILESIZE = 40 * 1024 * 1024 # 40mb
MAX_LETTER_PDF_COUNT_PER_ZIP = 500
# Default data # Default data
CONFIG_FILES = path.dirname(__file__) + '/config_files/' CONFIG_FILES = path.dirname(__file__) + '/config_files/'

View File

@@ -10,7 +10,7 @@ register_errors(letter_job)
MAX_REFERENCES_PER_TASK = 5000 MAX_REFERENCES_PER_TASK = 5000
# TODO: return deactivation notice # TODO: return deprecation notice
# @letter_job.route('/letters/returned', methods=['POST']) # @letter_job.route('/letters/returned', methods=['POST'])
# def create_process_returned_letters_job(): # def create_process_returned_letters_job():
# references = validate(request.get_json(), letter_references)['references'] # references = validate(request.get_json(), letter_references)['references']

View File

@@ -1,221 +0,0 @@
import io
import json
import math
from datetime import datetime, timedelta
from enum import Enum
import boto3
from flask import current_app
from notifications_utils.letter_timings import LETTER_PROCESSING_DEADLINE
from notifications_utils.pdf import pdf_page_count
from notifications_utils.s3 import s3upload
from notifications_utils.timezones import convert_utc_to_local_timezone
from app.models import (
KEY_TYPE_TEST,
NOTIFICATION_VALIDATION_FAILED,
RESOLVE_POSTAGE_FOR_FILE_NAME,
SECOND_CLASS,
)
class ScanErrorType(Enum):
ERROR = 1
FAILURE = 2
LETTERS_PDF_FILE_LOCATION_STRUCTURE = \
'{folder}NOTIFY.{reference}.{duplex}.{letter_class}.{colour}.{date}.pdf'
PRECOMPILED_BUCKET_PREFIX = '{folder}NOTIFY.{reference}'
def get_folder_name(created_at):
print_datetime = convert_utc_to_local_timezone(created_at)
if print_datetime.time() > LETTER_PROCESSING_DEADLINE:
print_datetime += timedelta(days=1)
return '{}/'.format(print_datetime.date())
class LetterPDFNotFound(Exception):
pass
def find_letter_pdf_in_s3(notification):
bucket_name, prefix = get_bucket_name_and_prefix_for_notification(notification)
s3 = boto3.resource('s3')
bucket = s3.Bucket(bucket_name)
try:
item = next(x for x in bucket.objects.filter(Prefix=prefix))
except StopIteration:
raise LetterPDFNotFound(f'File not found in bucket {bucket_name} with prefix {prefix}', )
return item
def generate_letter_pdf_filename(reference, created_at, ignore_folder=False, postage=SECOND_CLASS):
upload_file_name = LETTERS_PDF_FILE_LOCATION_STRUCTURE.format(
folder='' if ignore_folder else get_folder_name(created_at),
reference=reference,
duplex="D",
letter_class=RESOLVE_POSTAGE_FOR_FILE_NAME[postage],
colour="C",
date=created_at.strftime('%Y%m%d%H%M%S')
).upper()
return upload_file_name
def get_bucket_name_and_prefix_for_notification(notification):
folder = ''
if notification.status == NOTIFICATION_VALIDATION_FAILED:
bucket_name = current_app.config['INVALID_PDF_BUCKET_NAME']
elif notification.key_type == KEY_TYPE_TEST:
bucket_name = current_app.config['TEST_LETTERS_BUCKET_NAME']
else:
bucket_name = current_app.config['LETTERS_PDF_BUCKET_NAME']
folder = get_folder_name(notification.created_at)
upload_file_name = PRECOMPILED_BUCKET_PREFIX.format(
folder=folder,
reference=notification.reference
).upper()
return bucket_name, upload_file_name
def get_reference_from_filename(filename):
# filename looks like '2018-01-13/NOTIFY.ABCDEF1234567890.D.2.C.20180113120000.PDF'
filename_parts = filename.split('.')
return filename_parts[1]
def upload_letter_pdf(notification, pdf_data, precompiled=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 = generate_letter_pdf_filename(
reference=notification.reference,
created_at=notification.created_at,
ignore_folder=precompiled or notification.key_type == KEY_TYPE_TEST,
postage=notification.postage
)
if precompiled:
bucket_name = current_app.config['LETTERS_SCAN_BUCKET_NAME']
elif notification.key_type == KEY_TYPE_TEST:
bucket_name = current_app.config['TEST_LETTERS_BUCKET_NAME']
else:
bucket_name = current_app.config['LETTERS_PDF_BUCKET_NAME']
s3upload(
filedata=pdf_data,
region=current_app.config['AWS_REGION'],
bucket_name=bucket_name,
file_location=upload_file_name
)
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_failed_pdf(source_filename, scan_error_type):
scan_bucket = current_app.config['LETTERS_SCAN_BUCKET_NAME']
target_filename = ('ERROR/' if scan_error_type == ScanErrorType.ERROR else 'FAILURE/') + source_filename
_move_s3_object(scan_bucket, source_filename, scan_bucket, target_filename)
def move_error_pdf_to_scan_bucket(source_filename):
scan_bucket = current_app.config['LETTERS_SCAN_BUCKET_NAME']
error_file = 'ERROR/' + source_filename
_move_s3_object(scan_bucket, error_file, scan_bucket, source_filename)
def move_scan_to_invalid_pdf_bucket(source_filename, message=None, invalid_pages=None, page_count=None):
metadata = {}
if message:
metadata["message"] = message
if invalid_pages:
metadata["invalid_pages"] = json.dumps(invalid_pages)
if page_count:
metadata["page_count"] = str(page_count)
_move_s3_object(
source_bucket=current_app.config['LETTERS_SCAN_BUCKET_NAME'],
source_filename=source_filename,
target_bucket=current_app.config['INVALID_PDF_BUCKET_NAME'],
target_filename=source_filename,
metadata=metadata
)
def move_uploaded_pdf_to_letters_bucket(source_filename, upload_filename):
_move_s3_object(
source_bucket=current_app.config['TRANSIENT_UPLOADED_LETTERS'],
source_filename=source_filename,
target_bucket=current_app.config['LETTERS_PDF_BUCKET_NAME'],
target_filename=upload_filename,
)
def get_file_names_from_error_bucket():
s3 = boto3.resource('s3')
scan_bucket = current_app.config['LETTERS_SCAN_BUCKET_NAME']
bucket = s3.Bucket(scan_bucket)
return bucket.objects.filter(Prefix="ERROR")
def get_letter_pdf_and_metadata(notification):
obj = find_letter_pdf_in_s3(notification).get()
return obj["Body"].read(), obj["Metadata"]
def _move_s3_object(source_bucket, source_filename, target_bucket, target_filename, metadata=None):
s3 = boto3.resource('s3')
copy_source = {'Bucket': source_bucket, 'Key': source_filename}
target_bucket = s3.Bucket(target_bucket)
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
put_args = {'ServerSideEncryption': 'AES256'}
if metadata:
put_args['Metadata'] = metadata
put_args["MetadataDirective"] = "REPLACE"
obj.copy(copy_source, ExtraArgs=put_args)
s3.Object(source_bucket, source_filename).delete()
current_app.logger.info("Moved letter PDF: {}/{} to {}/{}".format(
source_bucket, source_filename, target_bucket, target_filename))
def letter_print_day(created_at):
bst_print_datetime = convert_utc_to_local_timezone(created_at) + timedelta(hours=6, minutes=30)
bst_print_date = bst_print_datetime.date()
current_local_date = convert_utc_to_local_timezone(datetime.utcnow()).date()
if bst_print_date >= current_local_date:
return 'today'
else:
print_date = bst_print_datetime.strftime('%d %B').lstrip('0')
return 'on {}'.format(print_date)
def get_page_count(pdf):
return pdf_page_count(io.BytesIO(pdf))
def get_billable_units_for_letter_page_count(page_count):
if not page_count:
return 0
pages_per_sheet = 2
billable_units = math.ceil(page_count / pages_per_sheet)
return billable_units

View File

@@ -324,7 +324,6 @@ INTERNATIONAL_SMS_TYPE = 'international_sms'
INBOUND_SMS_TYPE = 'inbound_sms' INBOUND_SMS_TYPE = 'inbound_sms'
SCHEDULE_NOTIFICATIONS = 'schedule_notifications' SCHEDULE_NOTIFICATIONS = 'schedule_notifications'
EMAIL_AUTH = 'email_auth' EMAIL_AUTH = 'email_auth'
LETTERS_AS_PDF = 'letters_as_pdf'
PRECOMPILED_LETTER = 'precompiled_letter' PRECOMPILED_LETTER = 'precompiled_letter'
UPLOAD_DOCUMENT = 'upload_document' UPLOAD_DOCUMENT = 'upload_document'
EDIT_FOLDER_PERMISSIONS = 'edit_folder_permissions' EDIT_FOLDER_PERMISSIONS = 'edit_folder_permissions'
@@ -339,7 +338,6 @@ SERVICE_PERMISSION_TYPES = [
INBOUND_SMS_TYPE, INBOUND_SMS_TYPE,
SCHEDULE_NOTIFICATIONS, SCHEDULE_NOTIFICATIONS,
EMAIL_AUTH, EMAIL_AUTH,
LETTERS_AS_PDF,
UPLOAD_DOCUMENT, UPLOAD_DOCUMENT,
EDIT_FOLDER_PERMISSIONS, EDIT_FOLDER_PERMISSIONS,
UPLOAD_LETTERS, UPLOAD_LETTERS,

View File

@@ -2,10 +2,6 @@ import itertools
from datetime import datetime from datetime import datetime
from flask import Blueprint, current_app, jsonify, request from flask import Blueprint, current_app, jsonify, request
from notifications_utils.letter_timings import (
letter_can_be_cancelled,
too_late_to_cancel_letter,
)
from notifications_utils.timezones import convert_utc_to_local_timezone from notifications_utils.timezones import convert_utc_to_local_timezone
from sqlalchemy.exc import IntegrityError from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm.exc import NoResultFound from sqlalchemy.orm.exc import NoResultFound
@@ -97,10 +93,8 @@ from app.dao.services_dao import (
from app.dao.templates_dao import dao_get_template_by_id from app.dao.templates_dao import dao_get_template_by_id
from app.dao.users_dao import get_user_by_id from app.dao.users_dao import get_user_by_id
from app.errors import InvalidRequest, register_errors from app.errors import InvalidRequest, register_errors
from app.letters.utils import letter_print_day
from app.models import ( from app.models import (
KEY_TYPE_NORMAL, KEY_TYPE_NORMAL,
LETTER_TYPE,
NOTIFICATION_CANCELLED, NOTIFICATION_CANCELLED,
EmailBranding, EmailBranding,
LetterBranding, LetterBranding,
@@ -493,22 +487,8 @@ def cancel_notification_for_service(service_id, notification_id):
if not notification: if not notification:
raise InvalidRequest('Notification not found', status_code=404) raise InvalidRequest('Notification not found', status_code=404)
elif notification.notification_type != LETTER_TYPE: else:
raise InvalidRequest('Notification cannot be cancelled - only letters can be cancelled', status_code=400) raise InvalidRequest('Notification cannot be cancelled - only letters can be cancelled', status_code=400)
elif not letter_can_be_cancelled(notification.status, notification.created_at):
print_day = letter_print_day(notification.created_at)
if too_late_to_cancel_letter(notification.created_at):
message = "Its too late to cancel this letter. Printing started {} at 5.30pm".format(print_day)
elif notification.status == 'cancelled':
message = "This letter has already been cancelled."
else:
message = (
f"We could not cancel this letter. "
f"Letter status: {notification.status}, created_at: {notification.created_at}"
)
raise InvalidRequest(
message,
status_code=400)
updated_notification = notifications_dao.update_notification_status_by_id( updated_notification = notifications_dao.update_notification_status_by_id(
notification_id, notification_id,
@@ -745,7 +725,7 @@ def create_one_off_notification(service_id):
return jsonify(resp), 201 return jsonify(resp), 201
# TODO: return deactivation notice # TODO: return deprecation notice
# @service_blueprint.route('/<uuid:service_id>/send-pdf-letter', methods=['POST']) # @service_blueprint.route('/<uuid:service_id>/send-pdf-letter', methods=['POST'])
# def create_pdf_letter(service_id): # def create_pdf_letter(service_id):
# data = validate(request.get_json(), send_pdf_letter_request) # data = validate(request.get_json(), send_pdf_letter_request)

View File

@@ -49,11 +49,6 @@ class ValidationError(InvalidRequest):
self.message = message if message else self.message self.message = message if message else self.message
class PDFNotReadyError(BadRequestError):
def __init__(self):
super().__init__(message='PDF not available yet, try again later', status_code=400)
def register_errors(blueprint): def register_errors(blueprint):
@blueprint.errorhandler(InvalidEmailError) @blueprint.errorhandler(InvalidEmailError)
def invalid_format(error): def invalid_format(error):

View File

@@ -1,18 +1,8 @@
from io import BytesIO from flask import current_app, jsonify, request, url_for
from flask import current_app, jsonify, request, send_file, url_for
from app import api_user, authenticated_service from app import api_user, authenticated_service
from app.dao import notifications_dao from app.dao import notifications_dao
from app.letters.utils import get_letter_pdf_and_metadata
from app.models import (
LETTER_TYPE,
NOTIFICATION_PENDING_VIRUS_CHECK,
NOTIFICATION_TECHNICAL_FAILURE,
NOTIFICATION_VIRUS_SCAN_FAILED,
)
from app.schema_validation import validate from app.schema_validation import validate
from app.v2.errors import BadRequestError, PDFNotReadyError
from app.v2.notifications import v2_notification_blueprint from app.v2.notifications import v2_notification_blueprint
from app.v2.notifications.notification_schemas import ( from app.v2.notifications.notification_schemas import (
get_notifications_request, get_notifications_request,
@@ -30,32 +20,33 @@ def get_notification_by_id(notification_id):
return jsonify(notification.serialize()), 200 return jsonify(notification.serialize()), 200
@v2_notification_blueprint.route('/<notification_id>/pdf', methods=['GET']) # TODO: return deprecation notice
def get_pdf_for_notification(notification_id): # @v2_notification_blueprint.route('/<notification_id>/pdf', methods=['GET'])
_data = {"notification_id": notification_id} # def get_pdf_for_notification(notification_id):
validate(_data, notification_by_id) # _data = {"notification_id": notification_id}
notification = notifications_dao.get_notification_by_id( # validate(_data, notification_by_id)
notification_id, authenticated_service.id, _raise=True # notification = notifications_dao.get_notification_by_id(
) # notification_id, authenticated_service.id, _raise=True
# )
if notification.notification_type != LETTER_TYPE: # if notification.notification_type != LETTER_TYPE:
raise BadRequestError(message="Notification is not a letter") # raise BadRequestError(message="Notification is not a letter")
if notification.status == NOTIFICATION_VIRUS_SCAN_FAILED: # if notification.status == NOTIFICATION_VIRUS_SCAN_FAILED:
raise BadRequestError(message='File did not pass the virus scan') # raise BadRequestError(message='File did not pass the virus scan')
if notification.status == NOTIFICATION_TECHNICAL_FAILURE: # if notification.status == NOTIFICATION_TECHNICAL_FAILURE:
raise BadRequestError(message='PDF not available for letters in status {}'.format(notification.status)) # raise BadRequestError(message='PDF not available for letters in status {}'.format(notification.status))
if notification.status == NOTIFICATION_PENDING_VIRUS_CHECK: # if notification.status == NOTIFICATION_PENDING_VIRUS_CHECK:
raise PDFNotReadyError() # raise PDFNotReadyError()
try: # try:
pdf_data, metadata = get_letter_pdf_and_metadata(notification) # pdf_data, metadata = get_letter_pdf_and_metadata(notification)
except Exception: # except Exception:
raise PDFNotReadyError() # raise PDFNotReadyError()
return send_file(path_or_file=BytesIO(pdf_data), mimetype='application/pdf') # return send_file(path_or_file=BytesIO(pdf_data), mimetype='application/pdf')
@v2_notification_blueprint.route("", methods=['GET']) @v2_notification_blueprint.route("", methods=['GET'])

View File

@@ -1,4 +1,3 @@
import base64
import functools import functools
import uuid import uuid
from datetime import datetime from datetime import datetime
@@ -13,13 +12,10 @@ from app import (
authenticated_service, authenticated_service,
document_download_client, document_download_client,
encryption, encryption,
notify_celery,
) )
from app.celery.tasks import save_api_email, save_api_sms from app.celery.tasks import save_api_email, save_api_sms
from app.clients.document_download import DocumentDownloadError from app.clients.document_download import DocumentDownloadError
from app.config import QueueNames, TaskNames from app.config import QueueNames
from app.dao.dao_utils import transaction
from app.letters.utils import upload_letter_pdf
from app.models import ( from app.models import (
EMAIL_TYPE, EMAIL_TYPE,
KEY_TYPE_NORMAL, KEY_TYPE_NORMAL,
@@ -27,7 +23,6 @@ from app.models import (
KEY_TYPE_TEST, KEY_TYPE_TEST,
NOTIFICATION_CREATED, NOTIFICATION_CREATED,
NOTIFICATION_DELIVERED, NOTIFICATION_DELIVERED,
NOTIFICATION_PENDING_VIRUS_CHECK,
NOTIFICATION_SENDING, NOTIFICATION_SENDING,
PRIORITY, PRIORITY,
SMS_TYPE, SMS_TYPE,
@@ -321,6 +316,7 @@ def process_document_uploads(personalisation_data, service, simulated=False):
return personalisation_data, len(file_keys) return personalisation_data, len(file_keys)
# TODO: remove precompiled var
def process_letter_notification( def process_letter_notification(
*, letter_data, api_key, service, template, template_with_content, reply_to_text, precompiled=False *, letter_data, api_key, service, template, template_with_content, reply_to_text, precompiled=False
): ):
@@ -330,13 +326,6 @@ def process_letter_notification(
if not service.research_mode and service.restricted and api_key.key_type != KEY_TYPE_TEST: if not service.research_mode and 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) raise BadRequestError(message='Cannot send letters when service is in trial mode', status_code=403)
if precompiled:
return process_precompiled_letter_notifications(letter_data=letter_data,
api_key=api_key,
service=service,
template=template,
reply_to_text=reply_to_text)
postage = validate_address(service, letter_data['personalisation']) postage = validate_address(service, letter_data['personalisation'])
test_key = api_key.key_type == KEY_TYPE_TEST test_key = api_key.key_type == KEY_TYPE_TEST
@@ -375,40 +364,6 @@ def process_letter_notification(
return resp return resp
def process_precompiled_letter_notifications(*, letter_data, api_key, service, template, reply_to_text):
try:
status = NOTIFICATION_PENDING_VIRUS_CHECK
letter_content = base64.b64decode(letter_data['content'])
except ValueError:
raise BadRequestError(message='Cannot decode letter content (invalid base64 encoding)', status_code=400)
with transaction():
notification = create_letter_notification(letter_data=letter_data,
service=service,
template=template,
api_key=api_key,
status=status,
reply_to_text=reply_to_text)
filename = upload_letter_pdf(notification, letter_content, precompiled=True)
resp = {
'id': notification.id,
'reference': notification.client_reference,
'postage': notification.postage
}
# call task to add the filename to anti virus queue
if current_app.config['ANTIVIRUS_ENABLED']:
current_app.logger.info('Calling task scan-file for {}'.format(filename))
notify_celery.send_task(
name=TaskNames.SCAN_FILE,
kwargs={'filename': filename},
queue=QueueNames.ANTIVIRUS,
)
return resp
def get_reply_to_text(notification_type, form, template): def get_reply_to_text(notification_type, form, template):
reply_to = None reply_to = None
if notification_type == EMAIL_TYPE: if notification_type == EMAIL_TYPE:

View File

@@ -13,7 +13,6 @@ There are a bunch of queues:
- retry tasks - retry tasks
- notify internal tasks - notify internal tasks
- process ftp tasks - process ftp tasks
- create letters pdf tasks
- service callbacks - service callbacks
- service callbacks retry - service callbacks retry
- letter tasks - letter tasks
@@ -28,7 +27,6 @@ And these tasks:
- check if letters still in created - check if letters still in created
- check if letters still pending virus check - check if letters still pending virus check
- check job status - check job status
- collate letter pdfs to be sent
- create fake letter response file - create fake letter response file
- create nightly billing - create nightly billing
- create nightly billing for day - create nightly billing for day
@@ -44,7 +42,6 @@ And these tasks:
- delete verify codes - delete verify codes
- deliver email - deliver email
- deliver sms - deliver sms
- get pdf for templated letter
- process incomplete jobs - process incomplete jobs
- process job - process job
- process returned letters list - process returned letters list

View File

@@ -1,11 +1,7 @@
import uuid import uuid
from datetime import datetime, timedelta from datetime import datetime, timedelta
import boto3
import pytest
from flask import current_app
from freezegun import freeze_time from freezegun import freeze_time
from moto import mock_s3
from app.dao.notifications_dao import ( from app.dao.notifications_dao import (
insert_notification_history_delete_notifications, insert_notification_history_delete_notifications,
@@ -26,50 +22,6 @@ from tests.app.db import (
) )
@mock_s3
@freeze_time('2019-09-01 04:30')
@pytest.mark.skip(reason="Skipping letter-related functionality for now")
def test_move_notifications_deletes_letters_from_s3(sample_letter_template, mocker):
s3 = boto3.client('s3', region_name='eu-west-1')
bucket_name = current_app.config['LETTERS_PDF_BUCKET_NAME']
s3.create_bucket(
Bucket=bucket_name,
CreateBucketConfiguration={'LocationConstraint': 'eu-west-1'}
)
eight_days_ago = datetime.utcnow() - timedelta(days=8)
create_notification(template=sample_letter_template, status='delivered',
reference='LETTER_REF', created_at=eight_days_ago, sent_at=eight_days_ago)
filename = "{}/NOTIFY.LETTER_REF.D.2.C.{}.PDF".format(
str(eight_days_ago.date()),
eight_days_ago.strftime('%Y%m%d%H%M%S')
)
s3.put_object(Bucket=bucket_name, Key=filename, Body=b'foo')
move_notifications_to_notification_history('letter', sample_letter_template.service_id, datetime(2020, 1, 2))
with pytest.raises(s3.exceptions.NoSuchKey):
s3.get_object(Bucket=bucket_name, Key=filename)
@mock_s3
@freeze_time('2019-09-01 04:30')
@pytest.mark.skip(reason="Skipping letter-related functionality for now")
def test_move_notifications_copes_if_letter_not_in_s3(sample_letter_template, mocker):
s3 = boto3.client('s3', region_name='eu-west-1')
s3.create_bucket(
Bucket=current_app.config['LETTERS_PDF_BUCKET_NAME'],
CreateBucketConfiguration={'LocationConstraint': 'eu-west-1'}
)
eight_days_ago = datetime.utcnow() - timedelta(days=8)
create_notification(template=sample_letter_template, status='delivered', sent_at=eight_days_ago)
move_notifications_to_notification_history('letter', sample_letter_template.service_id, datetime(2020, 1, 2))
assert Notification.query.count() == 0
assert NotificationHistory.query.count() == 1
def test_move_notifications_does_nothing_if_notification_history_row_already_exists( def test_move_notifications_does_nothing_if_notification_history_row_already_exists(
sample_email_template, mocker sample_email_template, mocker
): ):
@@ -90,72 +42,6 @@ def test_move_notifications_does_nothing_if_notification_history_row_already_exi
assert history[0].status == 'delivered' assert history[0].status == 'delivered'
@pytest.mark.parametrize(
'notification_status', ['validation-failed', 'virus-scan-failed']
)
@pytest.mark.skip(reason="Skipping letter-related functionality for now")
def test_move_notifications_deletes_letters_not_sent_and_in_final_state_from_table_but_not_s3(
sample_service, mocker, notification_status
):
mock_s3_object = mocker.patch("app.dao.notifications_dao.find_letter_pdf_in_s3").return_value
letter_template = create_template(service=sample_service, template_type='letter')
create_notification(
template=letter_template,
status=notification_status,
reference='LETTER_REF',
created_at=datetime.utcnow() - timedelta(days=14)
)
assert Notification.query.count() == 1
assert NotificationHistory.query.count() == 0
move_notifications_to_notification_history('letter', sample_service.id, datetime.utcnow())
assert Notification.query.count() == 0
assert NotificationHistory.query.count() == 1
mock_s3_object.assert_not_called()
@mock_s3
@freeze_time('2020-12-24 04:30')
@pytest.mark.parametrize('notification_status', ['delivered', 'returned-letter', 'technical-failure'])
@pytest.mark.skip(reason="Skipping letter-related functionality for now")
def test_move_notifications_deletes_letters_sent_and_in_final_state_from_table_and_s3(
sample_service, mocker, notification_status
):
bucket_name = current_app.config['LETTERS_PDF_BUCKET_NAME']
s3 = boto3.client('s3', region_name='eu-west-1')
s3.create_bucket(
Bucket=bucket_name,
CreateBucketConfiguration={'LocationConstraint': 'eu-west-1'}
)
letter_template = create_template(service=sample_service, template_type='letter')
eight_days_ago = datetime.utcnow() - timedelta(days=8)
create_notification(
template=letter_template,
status=notification_status,
reference='LETTER_REF',
created_at=eight_days_ago,
sent_at=eight_days_ago
)
assert Notification.query.count() == 1
assert NotificationHistory.query.count() == 0
filename = "{}/NOTIFY.LETTER_REF.D.2.C.{}.PDF".format(
str(eight_days_ago.date()),
eight_days_ago.strftime('%Y%m%d%H%M%S')
)
s3.put_object(Bucket=bucket_name, Key=filename, Body=b'foo')
move_notifications_to_notification_history('letter', sample_service.id, datetime.utcnow())
assert Notification.query.count() == 0
assert NotificationHistory.query.count() == 1
with pytest.raises(s3.exceptions.NoSuchKey):
s3.get_object(Bucket=bucket_name, Key=filename)
def test_move_notifications_only_moves_notifications_older_than_provided_timestamp(sample_template): def test_move_notifications_only_moves_notifications_older_than_provided_timestamp(sample_template):
delete_time = datetime(2020, 6, 1, 12) delete_time = datetime(2020, 6, 1, 12)
one_second_before = delete_time - timedelta(seconds=1) one_second_before = delete_time - timedelta(seconds=1)

View File

@@ -1,382 +0,0 @@
from datetime import datetime
import boto3
import dateutil
import pytest
from flask import current_app
from freezegun import freeze_time
from moto import mock_s3
from app.letters.utils import (
LetterPDFNotFound,
ScanErrorType,
find_letter_pdf_in_s3,
generate_letter_pdf_filename,
get_billable_units_for_letter_page_count,
get_bucket_name_and_prefix_for_notification,
get_folder_name,
get_letter_pdf_and_metadata,
letter_print_day,
move_failed_pdf,
upload_letter_pdf,
)
from app.models import (
KEY_TYPE_NORMAL,
KEY_TYPE_TEST,
NOTIFICATION_VALIDATION_FAILED,
PRECOMPILED_TEMPLATE_NAME,
)
from tests.app.db import create_notification
FROZEN_DATE_TIME = "2018-03-14 17:00:00"
@pytest.skip(reason="Skipping letter-related functionality for now", allow_module_level=True)
@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.reference = 'foo'
with freeze_time(FROZEN_DATE_TIME):
sample_letter_notification.created_at = datetime.utcnow()
sample_letter_notification.updated_at = datetime.utcnow()
return sample_letter_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
@mock_s3
def test_find_letter_pdf_in_s3_returns_object(sample_notification):
bucket_name = current_app.config['LETTERS_PDF_BUCKET_NAME']
s3 = boto3.client('s3', region_name='eu-west-1')
s3.create_bucket(
Bucket=bucket_name,
CreateBucketConfiguration={'LocationConstraint': 'eu-west-1'}
)
_, prefix = get_bucket_name_and_prefix_for_notification(sample_notification)
s3.put_object(Bucket=bucket_name, Key=f'{prefix}-and-then-some', Body=b'f')
assert find_letter_pdf_in_s3(sample_notification).key == f'{prefix}-and-then-some'
@mock_s3
def test_find_letter_pdf_in_s3_raises_if_not_found(sample_notification):
bucket_name = current_app.config['LETTERS_PDF_BUCKET_NAME']
s3 = boto3.client('s3', region_name='eu-west-1')
s3.create_bucket(
Bucket=bucket_name,
CreateBucketConfiguration={'LocationConstraint': 'eu-west-1'}
)
with pytest.raises(LetterPDFNotFound):
find_letter_pdf_in_s3(sample_notification)
@pytest.mark.parametrize('created_at,folder', [
(datetime(2017, 1, 1, 17, 29), '2017-01-01'),
(datetime(2017, 1, 1, 17, 31), '2017-01-02'),
])
def test_get_bucket_name_and_prefix_for_notification_valid_notification(sample_notification, created_at, folder):
sample_notification.created_at = created_at
sample_notification.updated_at = created_at
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=folder,
reference=sample_notification.reference
).upper()
def test_get_bucket_name_and_prefix_for_notification_is_tomorrow_after_17_30(sample_notification):
sample_notification.created_at = datetime(2019, 8, 1, 17, 35)
sample_notification.sent_at = datetime(2019, 8, 2, 17, 45)
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='2019-08-02',
reference=sample_notification.reference
).upper()
def test_get_bucket_name_and_prefix_for_notification_is_today_before_17_30(sample_notification):
sample_notification.created_at = datetime(2019, 8, 1, 12, 00)
sample_notification.updated_at = datetime(2019, 8, 2, 12, 00)
sample_notification.sent_at = datetime(2019, 8, 3, 12, 00)
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='2019-08-01',
reference=sample_notification.reference
).upper()
@freeze_time(FROZEN_DATE_TIME)
def test_get_bucket_name_and_prefix_for_notification_precompiled_letter_using_test_key(
sample_precompiled_letter_notification_using_test_key
):
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()
@freeze_time(FROZEN_DATE_TIME)
def test_get_bucket_name_and_prefix_for_notification_templated_letter_using_test_key(sample_letter_notification):
sample_letter_notification.key_type = KEY_TYPE_TEST
bucket, bucket_prefix = get_bucket_name_and_prefix_for_notification(sample_letter_notification)
assert bucket == current_app.config['TEST_LETTERS_BUCKET_NAME']
assert bucket_prefix == 'NOTIFY.{}'.format(sample_letter_notification.reference).upper()
@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_name_and_prefix_for_notification(None)
@pytest.mark.parametrize('postage,expected_postage', [
('second', 2),
('first', 1),
])
def test_generate_letter_pdf_filename_returns_correct_postage_for_filename(
notify_api, postage, expected_postage):
created_at = datetime(2017, 12, 4, 17, 29)
filename = generate_letter_pdf_filename(reference='foo', created_at=created_at, postage=postage)
assert filename == '2017-12-04/NOTIFY.FOO.D.{}.C.20171204172900.PDF'.format(expected_postage)
def test_generate_letter_pdf_filename_returns_correct_filename_for_test_letters(
notify_api, mocker):
created_at = datetime(2017, 12, 4, 17, 29)
filename = generate_letter_pdf_filename(
reference='foo',
created_at=created_at,
ignore_folder=True
)
assert filename == 'NOTIFY.FOO.D.2.C.20171204172900.PDF'
def test_generate_letter_pdf_filename_returns_tomorrows_filename(notify_api, mocker):
created_at = datetime(2017, 12, 4, 17, 31)
filename = generate_letter_pdf_filename(reference='foo', created_at=created_at)
assert filename == '2017-12-05/NOTIFY.FOO.D.2.C.20171204173100.PDF'
@mock_s3
@pytest.mark.parametrize('bucket_config_name,filename_format', [
('TEST_LETTERS_BUCKET_NAME', 'NOTIFY.FOO.D.2.C.%Y%m%d%H%M%S.PDF'),
('LETTERS_PDF_BUCKET_NAME', '%Y-%m-%d/NOTIFY.FOO.D.2.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,
CreateBucketConfiguration={'LocationConstraint': 'eu-west-1'}
)
s3 = boto3.client('s3', region_name='eu-west-1')
s3.put_object(Bucket=bucket_name, Key=filename, Body=b'pdf_content')
file_data, metadata = get_letter_pdf_and_metadata(sample_precompiled_letter_notification_using_test_key)
assert file_data == 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 = generate_letter_pdf_filename(
reference=sample_letter_notification.reference,
created_at=sample_letter_notification.created_at,
ignore_folder=is_precompiled_letter
)
upload_letter_pdf(sample_letter_notification, b'\x00\x01', precompiled=is_precompiled_letter)
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']
)
@pytest.mark.parametrize('postage,expected_postage', [
('second', 2),
('first', 1)
])
def test_upload_letter_pdf_uses_postage_from_notification(
sample_letter_template, mocker, postage, expected_postage
):
letter_notification = create_notification(template=sample_letter_template, postage=postage)
mock_s3 = mocker.patch('app.letters.utils.s3upload')
filename = generate_letter_pdf_filename(
reference=letter_notification.reference,
created_at=letter_notification.created_at,
ignore_folder=False,
postage=letter_notification.postage
)
upload_letter_pdf(letter_notification, b'\x00\x01', precompiled=False)
mock_s3.assert_called_once_with(
bucket_name=current_app.config['LETTERS_PDF_BUCKET_NAME'],
file_location=filename,
filedata=b'\x00\x01',
region=current_app.config['AWS_REGION']
)
@mock_s3
@freeze_time(FROZEN_DATE_TIME)
def test_move_failed_pdf_error(notify_api):
filename = 'test.pdf'
bucket_name = current_app.config['LETTERS_SCAN_BUCKET_NAME']
conn = boto3.resource('s3', region_name='eu-west-1')
bucket = conn.create_bucket(
Bucket=bucket_name,
CreateBucketConfiguration={'LocationConstraint': 'eu-west-1'}
)
s3 = boto3.client('s3', region_name='eu-west-1')
s3.put_object(Bucket=bucket_name, Key=filename, Body=b'pdf_content')
move_failed_pdf(filename, ScanErrorType.ERROR)
assert 'ERROR/' + filename in [o.key for o in bucket.objects.all()]
assert filename not in [o.key for o in bucket.objects.all()]
@mock_s3
@freeze_time(FROZEN_DATE_TIME)
def test_move_failed_pdf_scan_failed(notify_api):
filename = 'test.pdf'
bucket_name = current_app.config['LETTERS_SCAN_BUCKET_NAME']
conn = boto3.resource('s3', region_name='eu-west-1')
bucket = conn.create_bucket(
Bucket=bucket_name,
CreateBucketConfiguration={'LocationConstraint': 'eu-west-1'}
)
s3 = boto3.client('s3', region_name='eu-west-1')
s3.put_object(Bucket=bucket_name, Key=filename, Body=b'pdf_content')
move_failed_pdf(filename, ScanErrorType.FAILURE)
assert 'FAILURE/' + filename in [o.key for o in bucket.objects.all()]
assert filename not in [o.key for o in bucket.objects.all()]
@pytest.mark.parametrize("timestamp, expected_folder_name",
[("2018-04-01 17:50:00", "2018-04-02/"),
("2018-07-02 16:29:00", "2018-07-02/"),
("2018-07-02 16:30:00", "2018-07-02/"),
("2018-07-02 16:31:00", "2018-07-03/"),
("2018-01-02 16:31:00", "2018-01-02/"),
("2018-01-02 17:31:00", "2018-01-03/"),
("2018-07-02 22:30:00", "2018-07-03/"),
("2018-07-02 23:30:00", "2018-07-03/"),
("2018-07-03 00:30:00", "2018-07-03/"),
("2018-01-02 22:30:00", "2018-01-03/"),
("2018-01-02 23:30:00", "2018-01-03/"),
("2018-01-03 00:30:00", "2018-01-03/"),
])
def test_get_folder_name_in_british_summer_time(notify_api, timestamp, expected_folder_name):
timestamp = dateutil.parser.parse(timestamp)
folder_name = get_folder_name(created_at=timestamp)
assert folder_name == expected_folder_name
@freeze_time('2017-07-07 20:00:00')
def test_letter_print_day_returns_today_if_letter_was_printed_after_1730_yesterday():
created_at = datetime(2017, 7, 6, 17, 30)
assert letter_print_day(created_at) == 'today'
@freeze_time('2017-07-07 16:30:00')
def test_letter_print_day_returns_today_if_letter_was_printed_today():
created_at = datetime(2017, 7, 7, 12, 0)
assert letter_print_day(created_at) == 'today'
@pytest.mark.parametrize('created_at, formatted_date', [
(datetime(2017, 7, 5, 16, 30), 'on 6 July'),
(datetime(2017, 7, 6, 16, 29), 'on 6 July'),
(datetime(2016, 8, 8, 10, 00), 'on 8 August'),
(datetime(2016, 12, 12, 17, 29), 'on 12 December'),
(datetime(2016, 12, 12, 17, 30), 'on 13 December'),
])
@freeze_time('2017-07-07 16:30:00')
def test_letter_print_day_returns_formatted_date_if_letter_printed_before_1730_yesterday(created_at, formatted_date):
assert letter_print_day(created_at) == formatted_date
@pytest.mark.parametrize('number_of_pages, expected_billable_units', [(2, 1), (3, 2), (10, 5)])
def test_get_billable_units_for_letter_page_count(number_of_pages, expected_billable_units):
result = get_billable_units_for_letter_page_count(number_of_pages)
assert result == expected_billable_units

View File

@@ -2484,32 +2484,6 @@ def test_send_one_off_notification(sample_service, admin_request, mocker):
assert response['id'] == str(noti.id) assert response['id'] == str(noti.id)
@pytest.mark.skip(reason="Skipping letter-related functionality for now")
def test_create_pdf_letter(mocker, sample_service_full_permissions, client, fake_uuid, notify_user):
mocker.patch('app.service.send_notification.utils_s3download')
mocker.patch('app.service.send_notification.get_page_count', return_value=1)
mocker.patch('app.service.send_notification.move_uploaded_pdf_to_letters_bucket')
user = sample_service_full_permissions.users[0]
data = json.dumps({
'filename': 'valid.pdf',
'created_by': str(user.id),
'file_id': fake_uuid,
'postage': 'second',
'recipient_address': 'Bugs%20Bunny%0A123%20Main%20Street%0ALooney%20Town'
})
response = client.post(
url_for('service.create_pdf_letter', service_id=sample_service_full_permissions.id),
data=data,
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()]
)
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 201
assert json_resp == {'id': fake_uuid}
def test_get_notification_for_service_includes_template_redacted(admin_request, sample_notification): def test_get_notification_for_service_includes_template_redacted(admin_request, sample_notification):
resp = admin_request.get( resp = admin_request.get(
'service.get_notification_for_service', 'service.get_notification_for_service',
@@ -3358,45 +3332,6 @@ def test_cancel_notification_for_service_raises_invalid_request_when_notificatio
assert response['result'] == 'error' assert response['result'] == 'error'
@pytest.mark.parametrize('notification_status', [
'cancelled',
'sending',
'sent',
'delivered',
'pending',
'failed',
'technical-failure',
'temporary-failure',
'permanent-failure',
'validation-failed',
'virus-scan-failed',
'returned-letter',
])
@freeze_time('2018-07-07 12:00:00')
def test_cancel_notification_for_service_raises_invalid_request_when_letter_is_in_wrong_state_to_be_cancelled(
admin_request,
sample_letter_notification,
notification_status,
):
sample_letter_notification.status = notification_status
sample_letter_notification.created_at = datetime.now()
response = admin_request.post(
'service.cancel_notification_for_service',
service_id=sample_letter_notification.service_id,
notification_id=sample_letter_notification.id,
_expected_status=400
)
if notification_status == 'cancelled':
assert response['message'] == 'This letter has already been cancelled.'
else:
assert response['message'] == (
f"We could not cancel this letter. "
f"Letter status: {notification_status}, created_at: 2018-07-07 12:00:00"
)
assert response['result'] == 'error'
@pytest.mark.skip(reason="Needs updating for TTS: Remove letters") @pytest.mark.skip(reason="Needs updating for TTS: Remove letters")
@pytest.mark.parametrize('notification_status', ['created', 'pending-virus-check']) @pytest.mark.parametrize('notification_status', ['created', 'pending-virus-check'])
@freeze_time('2018-07-07 16:00:00') @freeze_time('2018-07-07 16:00:00')
@@ -3416,23 +3351,6 @@ def test_cancel_notification_for_service_updates_letter_if_letter_is_in_cancella
assert response['status'] == 'cancelled' assert response['status'] == 'cancelled'
@freeze_time('2017-12-12 17:30:00')
def test_cancel_notification_for_service_raises_error_if_its_too_late_to_cancel(
admin_request,
sample_letter_notification,
):
sample_letter_notification.created_at = datetime(2017, 12, 11, 17, 0)
response = admin_request.post(
'service.cancel_notification_for_service',
service_id=sample_letter_notification.service_id,
notification_id=sample_letter_notification.id,
_expected_status=400
)
assert response['message'] == 'Its too late to cancel this letter. Printing started on 11 December at 5.30pm'
assert response['result'] == 'error'
@pytest.mark.skip(reason="Needs updating for TTS: Remove letters") @pytest.mark.skip(reason="Needs updating for TTS: Remove letters")
@pytest.mark.parametrize('created_at', [ @pytest.mark.parametrize('created_at', [
datetime(2018, 7, 6, 22, 30), # yesterday evening datetime(2018, 7, 6, 22, 30), # yesterday evening

View File

@@ -603,123 +603,3 @@ def test_get_all_notifications_renames_letter_statuses(
assert noti['status'] == 'accepted' assert noti['status'] == 'accepted'
else: else:
pytest.fail() pytest.fail()
@pytest.mark.parametrize('db_status,expected_status', [
('created', 'accepted'),
('sending', 'accepted'),
('delivered', 'received'),
('pending', 'pending'),
('technical-failure', 'technical-failure')
])
def test_get_notifications_renames_letter_statuses(client, sample_letter_template, db_status, expected_status):
letter_noti = create_notification(
sample_letter_template,
status=db_status,
personalisation={'address_line_1': 'Mr Foo', 'address_line_2': '1 Bar Street', 'postcode': 'N1'}
)
auth_header = create_service_authorization_header(service_id=letter_noti.service_id)
response = client.get(
path=url_for('v2_notifications.get_notification_by_id', notification_id=letter_noti.id),
headers=[('Content-Type', 'application/json'), auth_header]
)
json_response = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert json_response['status'] == expected_status
def test_get_pdf_for_notification_returns_pdf_content(
client,
sample_letter_notification,
mocker,
):
mock_get_letter_pdf = mocker.patch(
'app.v2.notifications.get_notifications.get_letter_pdf_and_metadata', return_value=(b'foo', {
"message": "",
"invalid_pages": "",
"page_count": "1"
})
)
sample_letter_notification.status = 'created'
auth_header = create_service_authorization_header(service_id=sample_letter_notification.service_id)
response = client.get(
path=url_for('v2_notifications.get_pdf_for_notification', notification_id=sample_letter_notification.id),
headers=[('Content-Type', 'application/json'), auth_header]
)
assert response.status_code == 200
assert response.get_data() == b'foo'
mock_get_letter_pdf.assert_called_once_with(sample_letter_notification)
def test_get_pdf_for_notification_returns_400_if_pdf_not_found(
client,
sample_letter_notification,
mocker,
):
# if no files are returned get_letter_pdf throws StopIteration as the iterator runs out
mock_get_letter_pdf = mocker.patch(
'app.v2.notifications.get_notifications.get_letter_pdf_and_metadata',
side_effect=(StopIteration, {})
)
sample_letter_notification.status = 'created'
auth_header = create_service_authorization_header(service_id=sample_letter_notification.service_id)
response = client.get(
path=url_for('v2_notifications.get_pdf_for_notification', notification_id=sample_letter_notification.id),
headers=[('Content-Type', 'application/json'), auth_header]
)
assert response.status_code == 400
assert response.json['errors'] == [{
'error': 'PDFNotReadyError',
'message': 'PDF not available yet, try again later'
}]
mock_get_letter_pdf.assert_called_once_with(sample_letter_notification)
@pytest.mark.parametrize('status, expected_message', [
('virus-scan-failed', 'File did not pass the virus scan'),
('technical-failure', 'PDF not available for letters in status technical-failure'),
])
def test_get_pdf_for_notification_only_returns_pdf_content_if_right_status(
client,
sample_letter_notification,
mocker,
status,
expected_message
):
mock_get_letter_pdf = mocker.patch(
'app.v2.notifications.get_notifications.get_letter_pdf_and_metadata', return_value=(b'foo', {
"message": "",
"invalid_pages": "",
"page_count": "1"
})
)
sample_letter_notification.status = status
auth_header = create_service_authorization_header(service_id=sample_letter_notification.service_id)
response = client.get(
path=url_for('v2_notifications.get_pdf_for_notification', notification_id=sample_letter_notification.id),
headers=[('Content-Type', 'application/json'), auth_header]
)
assert response.status_code == 400
assert response.json['errors'] == [{
'error': 'BadRequestError',
'message': expected_message
}]
assert mock_get_letter_pdf.called is False
def test_get_pdf_for_notification_fails_for_non_letters(client, sample_notification):
auth_header = create_service_authorization_header(service_id=sample_notification.service_id)
response = client.get(
path=url_for('v2_notifications.get_pdf_for_notification', notification_id=sample_notification.id),
headers=[('Content-Type', 'application/json'), auth_header]
)
assert response.status_code == 400
assert response.json['errors'] == [{'error': 'BadRequestError', 'message': 'Notification is not a letter'}]