From edbae894c5941d06c4bd77b8ed2734e77f270243 Mon Sep 17 00:00:00 2001 From: Jumana Bahrainwala Date: Thu, 30 May 2024 18:50:27 +0000 Subject: [PATCH 1/2] Remove letter code --- app/letters/rest.py | 8 +- app/letters/utils.py | 188 +--------- tests/app/letters/__init__.py | 0 tests/app/letters/test_letter_utils.py | 402 --------------------- tests/app/letters/test_returned_letters.py | 27 -- 5 files changed, 19 insertions(+), 606 deletions(-) delete mode 100644 tests/app/letters/__init__.py delete mode 100644 tests/app/letters/test_letter_utils.py delete mode 100644 tests/app/letters/test_returned_letters.py diff --git a/app/letters/rest.py b/app/letters/rest.py index 21c80f2432..87684e0b66 100644 --- a/app/letters/rest.py +++ b/app/letters/rest.py @@ -1,7 +1,5 @@ -from flask import Blueprint, jsonify, request +from flask import Blueprint -from app.letters.letter_schemas import letter_references -from app.schema_validation import validate from app.v2.errors import register_errors letter_job = Blueprint("letter-job", __name__) @@ -10,6 +8,4 @@ @letter_job.route("/letters/returned", methods=["POST"]) def create_process_returned_letters_job(): - references = validate(request.get_json(), letter_references) - - return jsonify(references=references["references"]), 200 + pass diff --git a/app/letters/utils.py b/app/letters/utils.py index 6369b22040..8d5bcab489 100644 --- a/app/letters/utils.py +++ b/app/letters/utils.py @@ -1,21 +1,6 @@ -import io -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, -) +from app.models import SECOND_CLASS class ScanErrorType(Enum): @@ -29,203 +14,64 @@ class ScanErrorType(Enum): def get_folder_name(_now, is_test_or_scan_letter=False): - if is_test_or_scan_letter: - folder_name = "" - else: - print_datetime = convert_utc_to_local_timezone(_now) - if print_datetime.time() > LETTER_PROCESSING_DEADLINE: - print_datetime += timedelta(days=1) - folder_name = "{}/".format(print_datetime.date()) - return folder_name + pass def get_letter_pdf_filename(reference, crown, is_scan_letter=False, postage=SECOND_CLASS): - now = datetime.utcnow() - - upload_file_name = LETTERS_PDF_FILE_LOCATION_STRUCTURE.format( - folder=get_folder_name(now, is_scan_letter), - reference=reference, - duplex="D", - letter_class=RESOLVE_POSTAGE_FOR_FILE_NAME[postage], - colour="C", - crown="C" if crown else "N", - date=now.strftime("%Y%m%d%H%M%S"), - ).upper() - - return upload_file_name + pass 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"] - if notification.sent_at: - folder = "{}/".format(notification.sent_at.date()) - elif notification.updated_at: - folder = get_folder_name(notification.updated_at, False) - else: - folder = get_folder_name(notification.created_at, False) - - upload_file_name = PRECOMPILED_BUCKET_PREFIX.format(folder=folder, reference=notification.reference).upper() - - return bucket_name, upload_file_name + pass 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] + pass 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 = get_letter_pdf_filename( - notification.reference, - notification.service.crown, - is_scan_letter=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 + pass 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) + pass def copy_redaction_failed_pdf(source_filename): - scan_bucket = current_app.config["LETTERS_SCAN_BUCKET_NAME"] - - target_filename = "REDACTION_FAILURE/" + source_filename - - _copy_s3_object(scan_bucket, source_filename, scan_bucket, target_filename) + pass 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) + pass def move_scan_to_invalid_pdf_bucket(source_filename): - scan_bucket = current_app.config["LETTERS_SCAN_BUCKET_NAME"] - invalid_pdf_bucket = current_app.config["INVALID_PDF_BUCKET_NAME"] - _move_s3_object(scan_bucket, source_filename, invalid_pdf_bucket, source_filename) + pass 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, - ) + pass 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") + pass def get_letter_pdf(notification): - bucket_name, prefix = get_bucket_name_and_prefix_for_notification(notification) - - s3 = boto3.resource("s3") - bucket = s3.Bucket(bucket_name) - item = next(x for x in bucket.objects.filter(Prefix=prefix)) - - obj = s3.Object(bucket_name=bucket_name, key=item.key) - return obj.get()["Body"].read() + pass def _move_s3_object(source_bucket, source_filename, target_bucket, target_filename): - 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 - obj.copy(copy_source, ExtraArgs={"ServerSideEncryption": "AES256"}) - - s3.Object(source_bucket, source_filename).delete() - - current_app.logger.info( - "Moved letter PDF: {}/{} to {}/{}".format(source_bucket, source_filename, target_bucket, target_filename) - ) + pass def _copy_s3_object(source_bucket, source_filename, target_bucket, target_filename): - 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 - obj.copy(copy_source, ExtraArgs={"ServerSideEncryption": "AES256"}) - - current_app.logger.info( - "Copied letter PDF: {}/{} to {}/{}".format(source_bucket, source_filename, target_bucket, target_filename) - ) + pass 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_bst_date = convert_utc_to_local_timezone(datetime.utcnow()).date() - - if bst_print_date >= current_bst_date: - return "today" - else: - print_date = bst_print_datetime.strftime("%d %B").lstrip("0") - return "on {}".format(print_date) + pass def get_page_count(pdf): - pages = pdf_page_count(io.BytesIO(pdf)) - pages_per_sheet = 2 - billable_units = math.ceil(pages / pages_per_sheet) - return billable_units + pass diff --git a/tests/app/letters/__init__.py b/tests/app/letters/__init__.py deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/tests/app/letters/test_letter_utils.py b/tests/app/letters/test_letter_utils.py deleted file mode 100644 index ad4c05e56f..0000000000 --- a/tests/app/letters/test_letter_utils.py +++ /dev/null @@ -1,402 +0,0 @@ -from datetime import datetime - -import boto3 -import pytest -from flask import current_app -from freezegun import freeze_time -from moto import mock_s3 - -from app.letters.utils import ( - ScanErrorType, - copy_redaction_failed_pdf, - get_bucket_name_and_prefix_for_notification, - get_folder_name, - get_letter_pdf, - get_letter_pdf_filename, - 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, save_notification - -FROZEN_DATE_TIME = "2018-03-14 17:00:00" - - -@pytest.mark.skip(reason="Letter tests") -@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.mark.skip(reason="Letter tests") -@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 - - -@pytest.mark.skip(reason="Letter tests") -@pytest.mark.parametrize( - "created_at,folder", - [ - (datetime(2017, 1, 1, 17, 29), "2017-01-01"), - (datetime(2017, 1, 1, 17, 31), "2017-01-02"), - ], -) -@pytest.mark.skip(reason="Letter feature") -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() - - -@pytest.mark.skip(reason="Letter tests") -def test_get_bucket_name_and_prefix_for_notification_get_from_sent_at_date( - 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() - ) - - -@pytest.mark.skip(reason="Letter tests") -def test_get_bucket_name_and_prefix_for_notification_from_created_at_date( - 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-03", reference=sample_notification.reference).upper() - ) - - -@pytest.mark.skip(reason="Letter tests") -@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() - - -@pytest.mark.skip(reason="Letter tests") -@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() - - -@pytest.mark.skip(reason="Letter tests") -@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() - - -@pytest.mark.skip(reason="Letter tests") -@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() - - -@pytest.mark.skip(reason="Letter tests") -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.skip(reason="Letter tests") -@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) - - -@pytest.mark.skip(reason="Letter tests") -@pytest.mark.parametrize( - "postage,expected_postage", - [ - ("second", 2), - ("first", 1), - ], -) -@freeze_time("2017-12-04 17:29:00") -def test_get_letter_pdf_filename_returns_correct_postage_for_filename(notify_api, postage, expected_postage): - filename = get_letter_pdf_filename(reference="foo", crown=True, postage=postage) - - assert filename == "2017-12-04/NOTIFY.FOO.D.{}.C.C.20171204172900.PDF".format(expected_postage) - - -@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" - - -@pytest.mark.skip(reason="Letter tests") -@freeze_time("2017-12-04 17:31:00") -@pytest.mark.skip(reason="Letter feature") -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" - - -@pytest.mark.skip(reason="Letter tests") -@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") - 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.skip(reason="Letter tests") -@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", 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.skip(reason="Letter tests") -@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 = save_notification(create_notification(template=sample_letter_template, postage=postage)) - mock_s3 = mocker.patch("app.letters.utils.s3upload") - - filename = get_letter_pdf_filename( - reference=letter_notification.reference, - crown=letter_notification.service.crown, - is_scan_letter=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"], - ) - - -@pytest.mark.skip(reason="Letter tests") -@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") - bucket = 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") - - 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()] - - -@pytest.mark.skip(reason="Letter tests") -@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") - bucket = 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") - - 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.skip(reason="Letter tests") -@mock_s3 -@freeze_time(FROZEN_DATE_TIME) -def test_copy_redaction_failed_pdf(notify_api): - filename = "test.pdf" - bucket_name = current_app.config["LETTERS_SCAN_BUCKET_NAME"] - - conn = boto3.resource("s3") - bucket = 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") - - copy_redaction_failed_pdf(filename) - - assert "REDACTION_FAILURE/" + filename in [o.key for o in bucket.objects.all()] - assert filename in [o.key for o in bucket.objects.all()] - - -@pytest.mark.skip(reason="Letter tests") -@pytest.mark.parametrize( - "freeze_date, 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/"), - ], -) -@pytest.mark.skip(reason="Letter feature") -def test_get_folder_name_in_british_summer_time(notify_api, freeze_date, expected_folder_name): - with freeze_time(freeze_date): - now = datetime.utcnow() - folder_name = get_folder_name(_now=now, is_test_or_scan_letter=False) - assert folder_name == expected_folder_name - - -@pytest.mark.skip(reason="Letter tests") -def test_get_folder_name_returns_empty_string_for_test_letter(): - assert "" == get_folder_name(datetime.utcnow(), is_test_or_scan_letter=True) - - -@pytest.mark.skip(reason="Letter tests") -@freeze_time("2017-07-07 20:00:00") -@pytest.mark.skip(reason="Letter feature") -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" - - -@pytest.mark.skip(reason="Letter tests") -@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.skip(reason="Letter tests") -@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") -@pytest.mark.skip(reason="Letter feature") -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 diff --git a/tests/app/letters/test_returned_letters.py b/tests/app/letters/test_returned_letters.py deleted file mode 100644 index 7cdf223641..0000000000 --- a/tests/app/letters/test_returned_letters.py +++ /dev/null @@ -1,27 +0,0 @@ -import pytest - - -@pytest.mark.skip(reason="Deprecated: LETTER CODE") -@pytest.mark.parametrize( - "status, references", - [ - (200, ["1234567890ABCDEF", "1234567890ABCDEG"]), - (400, ["1234567890ABCDEFG", "1234567890ABCDEG"]), - (400, ["1234567890ABCDE", "1234567890ABCDEG"]), - (400, ["1234567890ABCDE\u26d4", "1234567890ABCDEG"]), - (400, ["NOTIFY0001234567890ABCDEF", "1234567890ABCDEG"]), - ], -) -def test_process_returned_letters(status, references, admin_request, mocker): - mock_celery = mocker.patch("app.letters.rest.process_returned_letters_list.apply_async") - - response = admin_request.post( - "letter-job.create_process_returned_letters_job", - _data={"references": references}, - _expected_status=status, - ) - - if status != 200: - assert "{} does not match".format(references[0]) in response["errors"][0]["message"] - else: - mock_celery.assert_called_once_with([references], queue="database-tasks") From 95fa47f3afd6321d1e7abd7fe46ea7a78131adad Mon Sep 17 00:00:00 2001 From: Jumana Bahrainwala Date: Thu, 30 May 2024 20:13:56 +0000 Subject: [PATCH 2/2] remove more letter code --- app/celery/letters_pdf_tasks.py | 361 +------ app/dao/notifications_dao.py | 40 - app/service/rest.py | 56 +- tests/app/celery/test_letters_pdf_tasks.py | 923 ------------------ ...t_notification_dao_delete_notifications.py | 28 +- tests/app/service/test_rest.py | 359 ------- .../test_send_pdf_letter_notification.py | 111 --- 7 files changed, 19 insertions(+), 1859 deletions(-) delete mode 100644 tests/app/celery/test_letters_pdf_tasks.py delete mode 100644 tests/app/service/test_send_pdf_letter_notification.py diff --git a/app/celery/letters_pdf_tasks.py b/app/celery/letters_pdf_tasks.py index 059f0fc940..e40a10b366 100644 --- a/app/celery/letters_pdf_tasks.py +++ b/app/celery/letters_pdf_tasks.py @@ -1,392 +1,51 @@ -import base64 -import math -from base64 import urlsafe_b64encode -from datetime import datetime -from hashlib import sha512 -from json import JSONDecodeError -from uuid import UUID - -from botocore.exceptions import ClientError as BotoClientError -from flask import current_app -from notifications_utils.s3 import s3upload from notifications_utils.statsd_decorators import statsd -from PyPDF2.utils import PdfReadError -from requests import RequestException -from requests import post as requests_post from app import notify_celery -from app.aws import s3 -from app.config import QueueNames, TaskNames from app.cronitor import cronitor -from app.dao.notifications_dao import ( - dao_get_notification_by_reference, - dao_get_notifications_by_references, - dao_update_notification, - dao_update_notifications_by_reference, - get_notification_by_id, - update_notification_status_by_id, -) -from app.errors import VirusScanError -from app.letters.utils import ( - ScanErrorType, - copy_redaction_failed_pdf, - get_file_names_from_error_bucket, - get_folder_name, - get_page_count, - get_reference_from_filename, - move_error_pdf_to_scan_bucket, - move_failed_pdf, - move_scan_to_invalid_pdf_bucket, - upload_letter_pdf, -) -from app.models import ( - KEY_TYPE_TEST, - NOTIFICATION_CREATED, - NOTIFICATION_DELIVERED, - NOTIFICATION_TECHNICAL_FAILURE, - NOTIFICATION_VALIDATION_FAILED, - NOTIFICATION_VIRUS_SCAN_FAILED, -) -from celery.exceptions import MaxRetriesExceededError @notify_celery.task(bind=True, name="create-letters-pdf", max_retries=15, default_retry_delay=300) @statsd(namespace="tasks") def create_letters_pdf(self, notification_id): - try: - notification = get_notification_by_id(notification_id, _raise=True) - pdf_data, billable_units = get_letters_pdf( - notification.template, - contact_block=notification.reply_to_text, - filename=notification.service.letter_branding and notification.service.letter_branding.filename, - values=notification.personalisation, - ) - - upload_letter_pdf(notification, pdf_data) - - if notification.key_type != KEY_TYPE_TEST: - notification.billable_units = billable_units - dao_update_notification(notification) - - current_app.logger.info( - "Letter notification reference {reference}: billable units set to {billable_units}".format( - reference=str(notification.reference), billable_units=billable_units - ) - ) - - except (RequestException, BotoClientError): - try: - current_app.logger.exception("Letters PDF notification creation for id: {} failed".format(notification_id)) - self.retry(queue=QueueNames.RETRY) - except MaxRetriesExceededError: - current_app.logger.error( - "RETRY FAILED: task create_letters_pdf failed for notification {}".format(notification_id), - ) - update_notification_status_by_id(notification_id, "technical-failure") + pass def get_letters_pdf(template, contact_block, filename, values): - template_for_letter_print = { - "subject": template.subject, - "content": template.content, - } - - data = { - "letter_contact_block": contact_block, - "template": template_for_letter_print, - "values": values, - "filename": filename, - } - resp = requests_post( - "{}/print.pdf".format(current_app.config["TEMPLATE_PREVIEW_API_HOST"]), - json=data, - headers={"Authorization": "Token {}".format(current_app.config["TEMPLATE_PREVIEW_API_KEY"])}, - ) - resp.raise_for_status() - - pages_per_sheet = 2 - billable_units = math.ceil(int(resp.headers.get("X-pdf-page-count", 0)) / pages_per_sheet) - - return resp.content, billable_units + pass @notify_celery.task(name="collate-letter-pdfs-for-day") @cronitor("collate-letter-pdfs-for-day") def collate_letter_pdfs_for_day(date=None): - if not date: - # Using the truncated date is ok because UTC to BST does not make a difference to the date, - # since it is triggered mid afternoon. - date = datetime.utcnow().strftime("%Y-%m-%d") - - letter_pdfs = sorted( - s3.get_s3_bucket_objects(current_app.config["LETTERS_PDF_BUCKET_NAME"], subfolder=date), - key=lambda letter: letter["Key"], - ) - for i, letters in enumerate(group_letters(letter_pdfs)): - filenames = [letter["Key"] for letter in letters] - - hash = urlsafe_b64encode(sha512("".join(filenames).encode()).digest())[:20].decode() - # eg NOTIFY.2018-12-31.001.Wjrui5nAvObjPd-3GEL-.ZIP - dvla_filename = "NOTIFY.{date}.{num:03}.{hash}.ZIP".format(date=date, num=i + 1, hash=hash) - - current_app.logger.info( - "Calling task zip-and-send-letter-pdfs for {} pdfs to upload {} with total size {:,} bytes".format( - len(filenames), dvla_filename, sum(letter["Size"] for letter in letters) - ) - ) - notify_celery.send_task( - name=TaskNames.ZIP_AND_SEND_LETTER_PDFS, - kwargs={"filenames_to_zip": filenames, "upload_filename": dvla_filename}, - queue=QueueNames.PROCESS_FTP, - compression="zlib", - ) + pass def group_letters(letter_pdfs): - """ - Group letters in chunks of MAX_LETTER_PDF_ZIP_FILESIZE. Will add files to lists, never going over that size. - If a single file is (somehow) larger than MAX_LETTER_PDF_ZIP_FILESIZE that'll be in a list on it's own. - If there are no files, will just exit (rather than yielding an empty list). - """ - running_filesize = 0 - list_of_files = [] - for letter in letter_pdfs: - if letter["Key"].lower().endswith(".pdf") and letter_in_created_state(letter["Key"]): - if ( - running_filesize + letter["Size"] > current_app.config["MAX_LETTER_PDF_ZIP_FILESIZE"] - or len(list_of_files) >= current_app.config["MAX_LETTER_PDF_COUNT_PER_ZIP"] - ): - yield list_of_files - running_filesize = 0 - list_of_files = [] - - running_filesize += letter["Size"] - list_of_files.append(letter) - - if list_of_files: - yield list_of_files + pass 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 = get_reference_from_filename(filename) - notifications = dao_get_notifications_by_references([ref]) - if notifications: - if notifications[0].status == NOTIFICATION_CREATED: - return True - current_app.logger.info( - "Collating letters for {} but notification with reference {} already in {}".format( - subfolder, ref, notifications[0].status - ) - ) - return False + pass @notify_celery.task(bind=True, name="process-virus-scan-passed", max_retries=15, default_retry_delay=300) def process_virus_scan_passed(self, filename): - reference = get_reference_from_filename(filename) - notification = dao_get_notification_by_reference(reference) - current_app.logger.info("notification id {} Virus scan passed: {}".format(notification.id, filename)) - - is_test_key = notification.key_type == KEY_TYPE_TEST - - scan_pdf_object = s3.get_s3_object(current_app.config["LETTERS_SCAN_BUCKET_NAME"], filename) - old_pdf = scan_pdf_object.get()["Body"].read() - - try: - billable_units = get_page_count(old_pdf) - except PdfReadError: - current_app.logger.exception(msg="Invalid PDF received for notification_id: {}".format(notification.id)) - _move_invalid_letter_and_update_status(notification, filename, scan_pdf_object) - return - - sanitise_response = _sanitise_precompiled_pdf(self, notification, old_pdf) - if not sanitise_response: - new_pdf = None - else: - sanitise_response = sanitise_response.json() - try: - new_pdf = base64.b64decode(sanitise_response["file"].encode()) - except JSONDecodeError: - new_pdf = sanitise_response.content - - redaction_failed_message = sanitise_response.get("redaction_failed_message") - if redaction_failed_message and not is_test_key: - current_app.logger.info("{} for notification id {} ({})".format(redaction_failed_message, notification.id, filename)) - copy_redaction_failed_pdf(filename) - - # TODO: Remove this once CYSP update their template to not cross over the margins - if notification.service_id == UUID("fe44178f-3b45-4625-9f85-2264a36dd9ec"): # CYSP - # Check your state pension submit letters with good addresses and notify tags, so just use their supplied pdf - new_pdf = old_pdf - - if not new_pdf: - current_app.logger.info("Invalid precompiled pdf received {} ({})".format(notification.id, filename)) - _move_invalid_letter_and_update_status(notification, filename, scan_pdf_object) - return - else: - current_app.logger.info("Validation was successful for precompiled pdf {} ({})".format(notification.id, filename)) - - current_app.logger.info("notification id {} ({}) sanitised and ready to send".format(notification.id, filename)) - - try: - _upload_pdf_to_test_or_live_pdf_bucket(new_pdf, filename, is_test_letter=is_test_key) - - update_letter_pdf_status( - reference=reference, - status=NOTIFICATION_DELIVERED if is_test_key else NOTIFICATION_CREATED, - billable_units=billable_units, - ) - scan_pdf_object.delete() - except BotoClientError: - current_app.logger.exception("Error uploading letter to live pdf bucket for notification: {}".format(notification.id)) - update_notification_status_by_id(notification.id, NOTIFICATION_TECHNICAL_FAILURE) - - -def _move_invalid_letter_and_update_status(notification, filename, scan_pdf_object): - try: - move_scan_to_invalid_pdf_bucket(filename) - scan_pdf_object.delete() - - update_letter_pdf_status( - reference=notification.reference, - status=NOTIFICATION_VALIDATION_FAILED, - billable_units=0, - ) - except BotoClientError: - current_app.logger.exception("Error when moving letter with id {} to invalid PDF bucket".format(notification.id)) - update_notification_status_by_id(notification.id, NOTIFICATION_TECHNICAL_FAILURE) - - -def _upload_pdf_to_test_or_live_pdf_bucket(pdf_data, filename, is_test_letter): - target_bucket_config = "TEST_LETTERS_BUCKET_NAME" if is_test_letter else "LETTERS_PDF_BUCKET_NAME" - target_bucket_name = current_app.config[target_bucket_config] - target_filename = get_folder_name(datetime.utcnow(), is_test_letter) + filename - - s3upload( - filedata=pdf_data, - region=current_app.config["AWS_REGION"], - bucket_name=target_bucket_name, - file_location=target_filename, - ) - - -def _sanitise_precompiled_pdf(self, notification, precompiled_pdf): - try: - response = requests_post( - "{}/precompiled/sanitise".format(current_app.config["TEMPLATE_PREVIEW_API_HOST"]), - data=precompiled_pdf, - headers={ - "Authorization": "Token {}".format(current_app.config["TEMPLATE_PREVIEW_API_KEY"]), - "Service-ID": str(notification.service_id), - "Notification-ID": str(notification.id), - }, - ) - response.raise_for_status() - return response - except RequestException as ex: - if ex.response is not None and ex.response.status_code == 400: - message = "sanitise_precompiled_pdf validation error for notification: {}. ".format(notification.id) - if "message" in response.json(): - message += response.json()["message"] - - current_app.logger.info(message) - return None - - try: - current_app.logger.exception("sanitise_precompiled_pdf failed for notification: {}".format(notification.id)) - self.retry(queue=QueueNames.RETRY) - except MaxRetriesExceededError: - current_app.logger.error( - "RETRY FAILED: sanitise_precompiled_pdf failed for notification {}".format(notification.id), - ) - - notification.status = NOTIFICATION_TECHNICAL_FAILURE - dao_update_notification(notification) - raise + pass @notify_celery.task(name="process-virus-scan-failed") def process_virus_scan_failed(filename): - move_failed_pdf(filename, ScanErrorType.FAILURE) - reference = get_reference_from_filename(filename) - notification = dao_get_notification_by_reference(reference) - updated_count = update_letter_pdf_status(reference, NOTIFICATION_VIRUS_SCAN_FAILED, billable_units=0) - - if updated_count != 1: - raise Exception( - "There should only be one letter notification for each reference. Found {} notifications".format(updated_count) - ) - - error = VirusScanError("notification id {} Virus scan failed: {}".format(notification.id, filename)) - current_app.logger.exception(error) - raise error + pass @notify_celery.task(name="process-virus-scan-error") def process_virus_scan_error(filename): - move_failed_pdf(filename, ScanErrorType.ERROR) - reference = get_reference_from_filename(filename) - notification = dao_get_notification_by_reference(reference) - updated_count = update_letter_pdf_status(reference, NOTIFICATION_TECHNICAL_FAILURE, billable_units=0) - - if updated_count != 1: - raise Exception( - "There should only be one letter notification for each reference. Found {} notifications".format(updated_count) - ) - error = VirusScanError("notification id {} Virus scan error: {}".format(notification.id, filename)) - current_app.logger.exception(error) - raise error + pass def update_letter_pdf_status(reference, status, billable_units): - return dao_update_notifications_by_reference( - references=[reference], - update_dict={ - "status": status, - "billable_units": billable_units, - "updated_at": datetime.utcnow(), - }, - )[0] + pass def replay_letters_in_error(filename=None): - # This method can be used to replay letters that end up in the ERROR directory. - # We had an incident where clamAV was not processing the virus scan. - if filename: - move_error_pdf_to_scan_bucket(filename) - # call task to add the filename to anti virus queue - current_app.logger.info("Calling scan_file for: {}".format(filename)) - - if current_app.config["ANTIVIRUS_ENABLED"]: - notify_celery.send_task( - name=TaskNames.SCAN_FILE, - kwargs={"filename": filename}, - queue=QueueNames.ANTIVIRUS, - ) - else: - # stub out antivirus in dev - process_virus_scan_passed.apply_async( - kwargs={"filename": filename}, - queue=QueueNames.LETTERS, - ) - else: - error_files = get_file_names_from_error_bucket() - for item in error_files: - moved_file_name = item.key.split("/")[1] - current_app.logger.info("Calling scan_file for: {}".format(moved_file_name)) - move_error_pdf_to_scan_bucket(moved_file_name) - # call task to add the filename to anti virus queue - if current_app.config["ANTIVIRUS_ENABLED"]: - notify_celery.send_task( - name=TaskNames.SCAN_FILE, - kwargs={"filename": moved_file_name}, - queue=QueueNames.ANTIVIRUS, - ) - else: - # stub out antivirus in dev - process_virus_scan_passed.apply_async( - kwargs={"filename": moved_file_name}, - queue=QueueNames.LETTERS, - ) + pass diff --git a/app/dao/notifications_dao.py b/app/dao/notifications_dao.py index 9e20bca476..e88df7cbee 100644 --- a/app/dao/notifications_dao.py +++ b/app/dao/notifications_dao.py @@ -2,7 +2,6 @@ import string from datetime import datetime, timedelta -from boto.exception import BotoClientError from flask import current_app from itsdangerous import BadSignature from notifications_utils.international_billing_rates import INTERNATIONAL_BILLING_RATES @@ -25,11 +24,9 @@ from werkzeug.datastructures import MultiDict from app import create_uuid, db, signer_personalisation -from app.aws.s3 import get_s3_bucket_objects, remove_s3_object from app.dao.dao_utils import transactional from app.dao.date_util import utc_midnight_n_days_ago from app.errors import InvalidRequest -from app.letters.utils import LETTERS_PDF_FILE_LOCATION_STRUCTURE from app.models import ( EMAIL_TYPE, KEY_TYPE_TEST, @@ -391,9 +388,6 @@ def delete_notifications_older_than_retention_by_type(notification_type, qry_lim convert_utc_to_local_timezone(datetime.utcnow()).date() ) - timedelta(days=f.days_of_retention) - if notification_type == LETTER_TYPE: - _delete_letters_from_s3(notification_type, f.service_id, days_of_retention, qry_limit) - insert_update_notification_history(notification_type, days_of_retention, f.service_id) current_app.logger.info("Deleting {} notifications for service id: {}".format(notification_type, f.service_id)) @@ -409,8 +403,6 @@ def delete_notifications_older_than_retention_by_type(notification_type, qry_lim for row in service_ids_to_purge: service_id = row._mapping["id"] - if notification_type == LETTER_TYPE: - _delete_letters_from_s3(notification_type, service_id, seven_days_ago, qry_limit) insert_update_notification_history(notification_type, seven_days_ago, service_id) deleted += _delete_notifications(notification_type, seven_days_ago, service_id, qry_limit) @@ -486,38 +478,6 @@ def insert_update_notification_history(notification_type, date_to_delete_from, s db.session.commit() -def _delete_letters_from_s3(notification_type, service_id, date_to_delete_from, query_limit): - letters_to_delete_from_s3 = ( - db.session.query(Notification) - .filter( - Notification.notification_type == notification_type, - Notification.created_at < date_to_delete_from, - Notification.service_id == service_id, - ) - .limit(query_limit) - .all() - ) - for letter in letters_to_delete_from_s3: - bucket_name = current_app.config["LETTERS_PDF_BUCKET_NAME"] - if letter.sent_at: - sent_at = str(letter.sent_at.date()) - prefix = LETTERS_PDF_FILE_LOCATION_STRUCTURE.format( - folder=sent_at + "/", - reference=letter.reference, - duplex="D", - letter_class="2", - colour="C", - crown="C" if letter.service.crown else "N", - date="", - ).upper()[:-5] - s3_objects = get_s3_bucket_objects(bucket_name=bucket_name, subfolder=prefix) - for s3_object in s3_objects: - try: - remove_s3_object(bucket_name, s3_object["Key"]) - except BotoClientError: - current_app.logger.exception("Could not delete S3 object with filename: {}".format(s3_object["Key"])) - - @statsd(namespace="dao") @transactional def dao_delete_notifications_by_id(notification_id): diff --git a/app/service/rest.py b/app/service/rest.py index 8ecf13f47d..8ba79097e3 100644 --- a/app/service/rest.py +++ b/app/service/rest.py @@ -11,7 +11,6 @@ over_email_daily_limit_cache_key, over_sms_daily_limit_cache_key, ) -from notifications_utils.letter_timings import letter_can_be_cancelled from notifications_utils.timezones import convert_utc_to_local_timezone from sqlalchemy import func from sqlalchemy.exc import IntegrityError @@ -53,13 +52,6 @@ dao_get_reply_to_by_service_id, update_reply_to_email_address, ) -from app.dao.service_letter_contact_dao import ( - add_letter_contact_for_service, - archive_letter_contact, - dao_get_letter_contact_by_id, - dao_get_letter_contacts_by_service_id, - update_letter_contact, -) from app.dao.service_safelist_dao import ( dao_add_and_commit_safelisted_contacts, dao_fetch_service_safelist, @@ -93,7 +85,6 @@ from app.dao.templates_dao import dao_get_template_by_id from app.dao.users_dao import get_user_by_id from app.errors import InvalidRequest, register_errors -from app.letters.utils import letter_print_day from app.models import ( KEY_TYPE_NORMAL, LETTER_TYPE, @@ -117,10 +108,7 @@ service_schema, ) from app.service import statistics -from app.service.send_notification import ( - send_one_off_notification, - send_pdf_letter_notification, -) +from app.service.send_notification import send_one_off_notification from app.service.sender import send_notification_to_service_users from app.service.service_data_retention_schema import ( add_service_data_retention_request, @@ -128,7 +116,6 @@ ) from app.service.service_senders_schema import ( add_service_email_reply_to_request, - add_service_letter_contact_block_request, add_service_sms_sender_request, ) from app.service.utils import ( @@ -575,13 +562,6 @@ def cancel_notification_for_service(service_id, notification_id): "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) - - raise InvalidRequest( - "It’s too late to cancel this letter. Printing started {} at 5.30pm".format(print_day), - status_code=400, - ) updated_notification = notifications_dao.update_notification_status_by_id( notification_id, @@ -793,8 +773,7 @@ def create_one_off_notification(service_id): @service_blueprint.route("//send-pdf-letter", methods=["POST"]) def create_pdf_letter(service_id): - resp = send_pdf_letter_notification(service_id, request.get_json()) - return jsonify(resp), 201 + pass @service_blueprint.route("//email-reply-to", methods=["GET"]) @@ -872,41 +851,22 @@ def delete_service_reply_to_email_address(service_id, reply_to_email_id): @service_blueprint.route("//letter-contact", methods=["GET"]) def get_letter_contacts(service_id): - result = dao_get_letter_contacts_by_service_id(service_id) - return jsonify([i.serialize() for i in result]), 200 + pass @service_blueprint.route("//letter-contact/", methods=["GET"]) def get_letter_contact_by_id(service_id, letter_contact_id): - result = dao_get_letter_contact_by_id(service_id=service_id, letter_contact_id=letter_contact_id) - return jsonify(result.serialize()), 200 + pass @service_blueprint.route("//letter-contact", methods=["POST"]) def add_service_letter_contact(service_id): - # validate the service exists, throws ResultNotFound exception. - dao_fetch_service_by_id(service_id) - form = validate(request.get_json(), add_service_letter_contact_block_request) - new_letter_contact = add_letter_contact_for_service( - service_id=service_id, - contact_block=form["contact_block"], - is_default=form.get("is_default", True), - ) - return jsonify(data=new_letter_contact.serialize()), 201 + pass @service_blueprint.route("//letter-contact/", methods=["POST"]) def update_service_letter_contact(service_id, letter_contact_id): - # validate the service exists, throws ResultNotFound exception. - dao_fetch_service_by_id(service_id) - form = validate(request.get_json(), add_service_letter_contact_block_request) - new_reply_to = update_letter_contact( - service_id=service_id, - letter_contact_id=letter_contact_id, - contact_block=form["contact_block"], - is_default=form.get("is_default", True), - ) - return jsonify(data=new_reply_to.serialize()), 200 + pass @service_blueprint.route( @@ -914,9 +874,7 @@ def update_service_letter_contact(service_id, letter_contact_id): methods=["POST"], ) def delete_service_letter_contact(service_id, letter_contact_id): - archived_letter_contact = archive_letter_contact(service_id, letter_contact_id) - - return jsonify(data=archived_letter_contact.serialize()), 200 + pass @service_blueprint.route("//sms-sender", methods=["POST"]) diff --git a/tests/app/celery/test_letters_pdf_tasks.py b/tests/app/celery/test_letters_pdf_tasks.py deleted file mode 100644 index 34eb4589c8..0000000000 --- a/tests/app/celery/test_letters_pdf_tasks.py +++ /dev/null @@ -1,923 +0,0 @@ -import base64 -from unittest.mock import ANY, Mock, call - -import boto3 -import pytest -import requests_mock -from botocore.exceptions import ClientError -from flask import current_app -from freezegun import freeze_time -from moto import mock_s3 -from PyPDF2.utils import PdfReadError -from requests import RequestException -from sqlalchemy.orm.exc import NoResultFound - -from app.celery.letters_pdf_tasks import ( - _move_invalid_letter_and_update_status, - _sanitise_precompiled_pdf, - collate_letter_pdfs_for_day, - create_letters_pdf, - get_letters_pdf, - group_letters, - letter_in_created_state, - process_virus_scan_error, - process_virus_scan_failed, - process_virus_scan_passed, - replay_letters_in_error, -) -from app.errors import VirusScanError -from app.letters.utils import ScanErrorType -from app.models import ( - KEY_TYPE_NORMAL, - KEY_TYPE_TEST, - NOTIFICATION_CREATED, - NOTIFICATION_DELIVERED, - NOTIFICATION_PENDING_VIRUS_CHECK, - NOTIFICATION_SENDING, - NOTIFICATION_TECHNICAL_FAILURE, - NOTIFICATION_VALIDATION_FAILED, - NOTIFICATION_VIRUS_SCAN_FAILED, - Notification, -) -from celery.exceptions import MaxRetriesExceededError, Retry -from tests.app.db import create_letter_branding, create_notification, save_notification -from tests.conftest import set_config_values - - -@pytest.mark.skip(reason="Letter tests") -def test_should_have_decorated_tasks_functions(): - assert create_letters_pdf.__wrapped__.__name__ == "create_letters_pdf" - assert collate_letter_pdfs_for_day.__wrapped__.__name__ == "collate_letter_pdfs_for_day" - assert process_virus_scan_passed.__wrapped__.__name__ == "process_virus_scan_passed" - assert process_virus_scan_failed.__wrapped__.__name__ == "process_virus_scan_failed" - assert process_virus_scan_error.__wrapped__.__name__ == "process_virus_scan_error" - - -@pytest.mark.skip(reason="Letter tests") -@pytest.mark.parametrize("personalisation", [{"name": "test"}, None]) -def test_get_letters_pdf_calls_notifications_template_preview_service_correctly( - notify_api, mocker, client, sample_letter_template, personalisation -): - contact_block = "Mr Foo,\n1 Test Street,\nLondon\nN1" - filename = "opg" - - with set_config_values( - notify_api, - { - "TEMPLATE_PREVIEW_API_HOST": "http://localhost/notifications-template-preview", - "TEMPLATE_PREVIEW_API_KEY": "test-key", - }, - ): - with requests_mock.Mocker() as request_mock: - mock_post = request_mock.post( - "http://localhost/notifications-template-preview/print.pdf", - content=b"\x00\x01", - status_code=200, - ) - - get_letters_pdf( - sample_letter_template, - contact_block=contact_block, - filename=filename, - values=personalisation, - ) - - assert mock_post.last_request.json() == { - "values": personalisation, - "letter_contact_block": contact_block, - "filename": filename, - "template": { - "subject": sample_letter_template.subject, - "content": sample_letter_template.content, - }, - } - - -@pytest.mark.skip(reason="Letter tests") -@pytest.mark.parametrize("page_count,expected_billable_units", [("1", 1), ("2", 1), ("3", 2)]) -def test_get_letters_pdf_calculates_billing_units( - notify_api, - mocker, - client, - sample_letter_template, - page_count, - expected_billable_units, -): - contact_block = "Mr Foo,\n1 Test Street,\nLondon\nN1" - filename = "opg" - - with set_config_values( - notify_api, - { - "TEMPLATE_PREVIEW_API_HOST": "http://localhost/notifications-template-preview", - "TEMPLATE_PREVIEW_API_KEY": "test-key", - }, - ): - with requests_mock.Mocker() as request_mock: - request_mock.post( - "http://localhost/notifications-template-preview/print.pdf", - content=b"\x00\x01", - headers={"X-pdf-page-count": page_count}, - status_code=200, - ) - - _, billable_units = get_letters_pdf( - sample_letter_template, - contact_block=contact_block, - filename=filename, - values=None, - ) - - assert billable_units == expected_billable_units - - -@pytest.mark.skip(reason="Letter tests") -@freeze_time("2017-12-04 17:31:00") -def test_create_letters_pdf_calls_s3upload(mocker, sample_letter_notification): - mocker.patch("app.celery.letters_pdf_tasks.get_letters_pdf", return_value=(b"\x00\x01", "1")) - mock_s3 = mocker.patch("app.letters.utils.s3upload") - - create_letters_pdf(sample_letter_notification.id) - - mock_s3.assert_called_with( - bucket_name=current_app.config["LETTERS_PDF_BUCKET_NAME"], - file_location="2017-12-04/NOTIFY.FOO.D.2.C.C.20171204173100.PDF", - filedata=b"\x00\x01", - region=current_app.config["AWS_REGION"], - ) - - -@pytest.mark.skip(reason="Letter tests") -@freeze_time("2017-12-04 17:31:00") -def test_create_letters_pdf_calls_s3upload_for_test_letters(mocker, sample_letter_notification): - mocker.patch("app.celery.letters_pdf_tasks.get_letters_pdf", return_value=(b"\x00\x01", "1")) - mock_s3 = mocker.patch("app.letters.utils.s3upload") - sample_letter_notification.key_type = "test" - - create_letters_pdf(sample_letter_notification.id) - - mock_s3.assert_called_with( - bucket_name=current_app.config["TEST_LETTERS_BUCKET_NAME"], - file_location="NOTIFY.FOO.D.2.C.C.20171204173100.PDF", - filedata=b"\x00\x01", - region=current_app.config["AWS_REGION"], - ) - - -@pytest.mark.skip(reason="Letter tests") -def test_create_letters_pdf_sets_billable_units(mocker, sample_letter_notification): - mocker.patch("app.celery.letters_pdf_tasks.get_letters_pdf", return_value=(b"\x00\x01", 1)) - mocker.patch("app.letters.utils.s3upload") - - create_letters_pdf(sample_letter_notification.id) - noti = Notification.query.filter(Notification.reference == sample_letter_notification.reference).one() - assert noti.billable_units == 1 - - -@pytest.mark.skip(reason="Letter tests") -def test_create_letters_pdf_non_existent_notification(notify_api, mocker, fake_uuid): - with pytest.raises(expected_exception=NoResultFound): - create_letters_pdf(fake_uuid) - - -@pytest.mark.skip(reason="Letter tests") -def test_create_letters_pdf_handles_request_errors(mocker, sample_letter_notification): - mock_get_letters_pdf = mocker.patch("app.celery.letters_pdf_tasks.get_letters_pdf", side_effect=RequestException) - mock_retry = mocker.patch("app.celery.letters_pdf_tasks.create_letters_pdf.retry") - - create_letters_pdf(sample_letter_notification.id) - - assert mock_get_letters_pdf.called - assert mock_retry.called - - -@pytest.mark.skip(reason="Letter tests") -def test_create_letters_pdf_handles_s3_errors(mocker, sample_letter_notification): - mocker.patch("app.celery.letters_pdf_tasks.get_letters_pdf", return_value=(b"\x00\x01", 1)) - error_response = { - "Error": { - "Code": "InvalidParameterValue", - "Message": "some error message from amazon", - "Type": "Sender", - } - } - mock_s3 = mocker.patch( - "app.letters.utils.s3upload", - side_effect=ClientError(error_response, "operation_name"), - ) - mock_retry = mocker.patch("app.celery.letters_pdf_tasks.create_letters_pdf.retry") - - create_letters_pdf(sample_letter_notification.id) - - assert mock_s3.called - assert mock_retry.called - - -@pytest.mark.skip(reason="Letter tests") -def test_create_letters_pdf_sets_technical_failure_max_retries(mocker, sample_letter_notification): - mock_get_letters_pdf = mocker.patch("app.celery.letters_pdf_tasks.get_letters_pdf", side_effect=RequestException) - mock_retry = mocker.patch( - "app.celery.letters_pdf_tasks.create_letters_pdf.retry", - side_effect=MaxRetriesExceededError, - ) - mock_update_noti = mocker.patch("app.celery.letters_pdf_tasks.update_notification_status_by_id") - - create_letters_pdf(sample_letter_notification.id) - - assert mock_get_letters_pdf.called - assert mock_retry.called - mock_update_noti.assert_called_once_with(sample_letter_notification.id, "technical-failure") - - -@pytest.mark.skip(reason="Letter tests") -def test_create_letters_gets_the_right_logo_when_service_has_no_logo(notify_api, mocker, sample_letter_notification): - mock_get_letters_pdf = mocker.patch("app.celery.letters_pdf_tasks.get_letters_pdf", return_value=(b"\x00\x01", 1)) - mocker.patch("app.letters.utils.s3upload") - mocker.patch("app.celery.letters_pdf_tasks.update_notification_status_by_id") - - create_letters_pdf(sample_letter_notification.id) - mock_get_letters_pdf.assert_called_once_with( - sample_letter_notification.template, - contact_block=sample_letter_notification.reply_to_text, - filename=None, - values=sample_letter_notification.personalisation, - ) - - -# We only need this while we are migrating to the new letter_branding model -@pytest.mark.skip(reason="Letter tests") -def test_create_letters_gets_the_right_logo_when_service_has_letter_branding_logo(notify_api, mocker, sample_letter_notification): - letter_branding = create_letter_branding(name="test brand", filename="test-brand") - sample_letter_notification.service.letter_branding = letter_branding - mock_get_letters_pdf = mocker.patch("app.celery.letters_pdf_tasks.get_letters_pdf", return_value=(b"\x00\x01", 1)) - mocker.patch("app.letters.utils.s3upload") - mocker.patch("app.celery.letters_pdf_tasks.update_notification_status_by_id") - - create_letters_pdf(sample_letter_notification.id) - mock_get_letters_pdf.assert_called_once_with( - sample_letter_notification.template, - contact_block=sample_letter_notification.reply_to_text, - filename=sample_letter_notification.service.letter_branding.filename, - values=sample_letter_notification.personalisation, - ) - - -@pytest.mark.skip(reason="Letter tests") -def test_collate_letter_pdfs_for_day(notify_api, mocker): - mock_s3 = mocker.patch( - "app.celery.tasks.s3.get_s3_bucket_objects", - return_value=[ - {"Key": "B.pDf", "Size": 2}, - {"Key": "A.PDF", "Size": 1}, - {"Key": "C.pdf", "Size": 3}, - ], - ) - mock_group_letters = mocker.patch( - "app.celery.letters_pdf_tasks.group_letters", - return_value=[ - [{"Key": "A.PDF", "Size": 1}, {"Key": "B.pDf", "Size": 2}], - [{"Key": "C.pdf", "Size": 3}], - ], - ) - mock_celery = mocker.patch("app.celery.letters_pdf_tasks.notify_celery.send_task") - - collate_letter_pdfs_for_day("2017-01-02") - - mock_s3.assert_called_once_with("test-letters-pdf", subfolder="2017-01-02") - mock_group_letters.assert_called_once_with(sorted(mock_s3.return_value, key=lambda x: x["Key"])) - assert mock_celery.call_args_list[0] == call( - name="zip-and-send-letter-pdfs", - kwargs={ - "filenames_to_zip": ["A.PDF", "B.pDf"], - "upload_filename": "NOTIFY.2017-01-02.001.oqdjIM2-NAUU9Sm5Slmi.ZIP", - }, - queue="process-ftp-tasks", - compression="zlib", - ) - assert mock_celery.call_args_list[1] == call( - name="zip-and-send-letter-pdfs", - kwargs={ - "filenames_to_zip": ["C.pdf"], - "upload_filename": "NOTIFY.2017-01-02.002.tdr7hcdPieiqjkVoS4kU.ZIP", - }, - queue="process-ftp-tasks", - compression="zlib", - ) - - -@pytest.mark.skip(reason="Letter tests") -@freeze_time("2018-09-12 17:50:00") -def test_collate_letter_pdfs_for_day_works_without_date_param(notify_api, mocker): - mock_s3 = mocker.patch("app.celery.tasks.s3.get_s3_bucket_objects") - collate_letter_pdfs_for_day() - expected_date = "2018-09-12" - mock_s3.assert_called_once_with("test-letters-pdf", subfolder=expected_date) - - -@pytest.mark.skip(reason="Letter tests") -def test_group_letters_splits_on_file_size(notify_api, mocker): - mocker.patch("app.celery.letters_pdf_tasks.letter_in_created_state", return_value=True) - letters = [ - # ends under max but next one is too big - {"Key": "A.pdf", "Size": 1}, - {"Key": "B.pdf", "Size": 2}, - # ends on exactly max - {"Key": "C.pdf", "Size": 3}, - {"Key": "D.pdf", "Size": 1}, - {"Key": "E.pdf", "Size": 1}, - # exactly max goes in next file - {"Key": "F.pdf", "Size": 5}, - # if it's bigger than the max, still gets included - {"Key": "G.pdf", "Size": 6}, - # whatever's left goes in last list - {"Key": "H.pdf", "Size": 1}, - {"Key": "I.pdf", "Size": 1}, - ] - - with set_config_values(notify_api, {"MAX_LETTER_PDF_ZIP_FILESIZE": 5}): - x = group_letters(letters) - - assert next(x) == [{"Key": "A.pdf", "Size": 1}, {"Key": "B.pdf", "Size": 2}] - assert next(x) == [ - {"Key": "C.pdf", "Size": 3}, - {"Key": "D.pdf", "Size": 1}, - {"Key": "E.pdf", "Size": 1}, - ] - assert next(x) == [{"Key": "F.pdf", "Size": 5}] - assert next(x) == [{"Key": "G.pdf", "Size": 6}] - assert next(x) == [{"Key": "H.pdf", "Size": 1}, {"Key": "I.pdf", "Size": 1}] - # make sure iterator is exhausted - assert next(x, None) is None - - -@pytest.mark.skip(reason="Letter tests") -def test_group_letters_splits_on_file_count(notify_api, mocker): - mocker.patch("app.celery.letters_pdf_tasks.letter_in_created_state", return_value=True) - letters = [ - {"Key": "A.pdf", "Size": 1}, - {"Key": "B.pdf", "Size": 2}, - {"Key": "C.pdf", "Size": 3}, - {"Key": "D.pdf", "Size": 1}, - {"Key": "E.pdf", "Size": 1}, - {"Key": "F.pdf", "Size": 5}, - {"Key": "G.pdf", "Size": 6}, - {"Key": "H.pdf", "Size": 1}, - {"Key": "I.pdf", "Size": 1}, - ] - - with set_config_values(notify_api, {"MAX_LETTER_PDF_COUNT_PER_ZIP": 3}): - x = group_letters(letters) - - assert next(x) == [ - {"Key": "A.pdf", "Size": 1}, - {"Key": "B.pdf", "Size": 2}, - {"Key": "C.pdf", "Size": 3}, - ] - assert next(x) == [ - {"Key": "D.pdf", "Size": 1}, - {"Key": "E.pdf", "Size": 1}, - {"Key": "F.pdf", "Size": 5}, - ] - assert next(x) == [ - {"Key": "G.pdf", "Size": 6}, - {"Key": "H.pdf", "Size": 1}, - {"Key": "I.pdf", "Size": 1}, - ] - # make sure iterator is exhausted - assert next(x, None) is None - - -@pytest.mark.skip(reason="Letter tests") -def test_group_letters_splits_on_file_size_and_file_count(notify_api, mocker): - mocker.patch("app.celery.letters_pdf_tasks.letter_in_created_state", return_value=True) - letters = [ - # ends under max file size but next file is too big - {"Key": "A.pdf", "Size": 1}, - {"Key": "B.pdf", "Size": 2}, - # ends on exactly max number of files and file size - {"Key": "C.pdf", "Size": 3}, - {"Key": "D.pdf", "Size": 1}, - {"Key": "E.pdf", "Size": 1}, - # exactly max file size goes in next file - {"Key": "F.pdf", "Size": 5}, - # file size is within max but number of files reaches limit - {"Key": "G.pdf", "Size": 1}, - {"Key": "H.pdf", "Size": 1}, - {"Key": "I.pdf", "Size": 1}, - # whatever's left goes in last list - {"Key": "J.pdf", "Size": 1}, - ] - - with set_config_values( - notify_api, - {"MAX_LETTER_PDF_ZIP_FILESIZE": 5, "MAX_LETTER_PDF_COUNT_PER_ZIP": 3}, - ): - x = group_letters(letters) - - assert next(x) == [{"Key": "A.pdf", "Size": 1}, {"Key": "B.pdf", "Size": 2}] - assert next(x) == [ - {"Key": "C.pdf", "Size": 3}, - {"Key": "D.pdf", "Size": 1}, - {"Key": "E.pdf", "Size": 1}, - ] - assert next(x) == [{"Key": "F.pdf", "Size": 5}] - assert next(x) == [ - {"Key": "G.pdf", "Size": 1}, - {"Key": "H.pdf", "Size": 1}, - {"Key": "I.pdf", "Size": 1}, - ] - assert next(x) == [{"Key": "J.pdf", "Size": 1}] - # make sure iterator is exhausted - assert next(x, None) is None - - -@pytest.mark.skip(reason="Letter tests") -def test_group_letters_ignores_non_pdfs(notify_api, mocker): - mocker.patch("app.celery.letters_pdf_tasks.letter_in_created_state", return_value=True) - letters = [{"Key": "A.zip"}] - assert list(group_letters(letters)) == [] - - -@pytest.mark.skip(reason="Letter tests") -def test_group_letters_ignores_notifications_already_sent(notify_api, mocker): - mock = mocker.patch("app.celery.letters_pdf_tasks.letter_in_created_state", return_value=False) - letters = [{"Key": "A.pdf"}] - assert list(group_letters(letters)) == [] - mock.assert_called_once_with("A.pdf") - - -@pytest.mark.skip(reason="Letter tests") -def test_group_letters_with_no_letters(notify_api, mocker): - mocker.patch("app.celery.letters_pdf_tasks.letter_in_created_state", return_value=True) - assert list(group_letters([])) == [] - - -@pytest.mark.skip(reason="Letter tests") -def test_letter_in_created_state(sample_notification): - sample_notification.reference = "ABCDEF1234567890" - filename = "2018-01-13/NOTIFY.ABCDEF1234567890.D.2.C.C.20180113120000.PDF" - - assert letter_in_created_state(filename) is True - - -@pytest.mark.skip(reason="Letter tests") -def test_letter_in_created_state_fails_if_notification_not_in_created( - sample_notification, -): - sample_notification.reference = "ABCDEF1234567890" - sample_notification.status = NOTIFICATION_SENDING - filename = "2018-01-13/NOTIFY.ABCDEF1234567890.D.2.C.C.20180113120000.PDF" - assert letter_in_created_state(filename) is False - - -@pytest.mark.skip(reason="Letter tests") -def test_letter_in_created_state_fails_if_notification_doesnt_exist( - sample_notification, -): - sample_notification.reference = "QWERTY1234567890" - filename = "2018-01-13/NOTIFY.ABCDEF1234567890.D.2.C.C.20180113120000.PDF" - assert letter_in_created_state(filename) is False - - -@pytest.mark.skip(reason="Letter tests") -@freeze_time("2018-01-01 18:00") -@mock_s3 -@pytest.mark.parametrize( - "key_type,noti_status,bucket_config_name,destination_folder", - [ - ( - KEY_TYPE_NORMAL, - NOTIFICATION_CREATED, - "LETTERS_PDF_BUCKET_NAME", - "2018-01-01/", - ), - (KEY_TYPE_TEST, NOTIFICATION_DELIVERED, "TEST_LETTERS_BUCKET_NAME", ""), - ], -) -def test_process_letter_task_check_virus_scan_passed( - sample_letter_template, - mocker, - key_type, - noti_status, - bucket_config_name, - destination_folder, -): - letter_notification = save_notification( - create_notification( - template=sample_letter_template, - billable_units=0, - status="pending-virus-check", - key_type=key_type, - reference="{} letter".format(key_type), - ) - ) - filename = "NOTIFY.{}".format(letter_notification.reference) - source_bucket_name = current_app.config["LETTERS_SCAN_BUCKET_NAME"] - target_bucket_name = current_app.config[bucket_config_name] - - conn = boto3.resource("s3") - conn.create_bucket(Bucket=source_bucket_name) - conn.create_bucket(Bucket=target_bucket_name) - - s3 = boto3.client("s3", region_name="ca-central-1") - s3.put_object(Bucket=source_bucket_name, Key=filename, Body=b"old_pdf") - - mock_get_page_count = mocker.patch("app.celery.letters_pdf_tasks.get_page_count", return_value=1) - mock_s3upload = mocker.patch("app.celery.letters_pdf_tasks.s3upload") - endpoint = "http://localhost:9999/precompiled/sanitise" - with requests_mock.mock() as rmock: - rmock.request( - "POST", - endpoint, - json={ - "file": base64.b64encode(b"new_pdf").decode("utf-8"), - "validation_passed": True, - "errors": { - "content_outside_of_printable_area": [], - "document_not_a4_size_portrait_orientation": [], - }, - }, - status_code=200, - ) - process_virus_scan_passed(filename) - - assert letter_notification.status == noti_status - assert letter_notification.billable_units == 1 - assert rmock.called - assert rmock.request_history[0].url == endpoint - - mock_s3upload.assert_called_once_with( - bucket_name=target_bucket_name, - filedata=b"new_pdf", - file_location=destination_folder + filename, - region="ca-central-1", - ) - mock_get_page_count.assert_called_once_with(b"old_pdf") - - -@pytest.mark.skip(reason="Letter tests") -@freeze_time("2018-01-01 18:00") -@mock_s3 -@pytest.mark.parametrize("key_type", [KEY_TYPE_NORMAL, KEY_TYPE_TEST]) -def test_process_letter_task_check_virus_scan_passed_when_sanitise_fails(sample_letter_notification, mocker, key_type): - filename = "NOTIFY.{}".format(sample_letter_notification.reference) - source_bucket_name = current_app.config["LETTERS_SCAN_BUCKET_NAME"] - target_bucket_name = current_app.config["INVALID_PDF_BUCKET_NAME"] - - conn = boto3.resource("s3") - conn.create_bucket(Bucket=source_bucket_name) - conn.create_bucket(Bucket=target_bucket_name) - - s3 = boto3.client("s3", region_name="ca-central-1") - s3.put_object(Bucket=source_bucket_name, Key=filename, Body=b"pdf_content") - - sample_letter_notification.status = NOTIFICATION_PENDING_VIRUS_CHECK - sample_letter_notification.key_type = key_type - mock_move_s3 = mocker.patch("app.letters.utils._move_s3_object") - mock_sanitise = mocker.patch("app.celery.letters_pdf_tasks._sanitise_precompiled_pdf", return_value=None) - mock_get_page_count = mocker.patch("app.celery.letters_pdf_tasks.get_page_count", return_value=2) - - process_virus_scan_passed(filename) - - assert sample_letter_notification.status == NOTIFICATION_VALIDATION_FAILED - assert sample_letter_notification.billable_units == 0 - mock_sanitise.assert_called_once_with(ANY, sample_letter_notification, b"pdf_content") - mock_move_s3.assert_called_once_with(source_bucket_name, filename, target_bucket_name, filename) - - mock_get_page_count.assert_called_once_with(b"pdf_content") - - -@pytest.mark.skip(reason="Letter tests") -@freeze_time("2018-01-01 18:00") -@mock_s3 -@pytest.mark.parametrize( - "key_type,notification_status,bucket_config_name", - [ - (KEY_TYPE_NORMAL, NOTIFICATION_CREATED, "LETTERS_PDF_BUCKET_NAME"), - (KEY_TYPE_TEST, NOTIFICATION_DELIVERED, "TEST_LETTERS_BUCKET_NAME"), - ], -) -def test_process_letter_task_check_virus_scan_passed_when_redaction_fails( - sample_letter_notification, - mocker, - key_type, - notification_status, - bucket_config_name, -): - filename = "NOTIFY.{}".format(sample_letter_notification.reference) - bucket_name = current_app.config["LETTERS_SCAN_BUCKET_NAME"] - target_bucket_name = current_app.config[bucket_config_name] - - conn = boto3.resource("s3") - conn.create_bucket(Bucket=bucket_name) - conn.create_bucket(Bucket=target_bucket_name) - - s3 = boto3.client("s3", region_name="eu-west-1") - s3.put_object(Bucket=bucket_name, Key=filename, Body=b"pdf_content") - - sample_letter_notification.status = NOTIFICATION_PENDING_VIRUS_CHECK - sample_letter_notification.key_type = key_type - mock_copy_s3 = mocker.patch("app.letters.utils._copy_s3_object") - mocker.patch("app.celery.letters_pdf_tasks.get_page_count", return_value=2) - - endpoint = "http://localhost:9999/precompiled/sanitise" - with requests_mock.mock() as rmock: - rmock.request( - "POST", - endpoint, - json={ - "file": base64.b64encode(b"new_pdf").decode("utf-8"), - "validation_passed": True, - "redaction_failed_message": "No matches for address block during redaction procedure", - "errors": { - "content_outside_of_printable_area": [], - "document_not_a4_size_portrait_orientation": [], - }, - }, - status_code=200, - ) - process_virus_scan_passed(filename) - - assert sample_letter_notification.billable_units == 2 - assert sample_letter_notification.status == notification_status - if key_type == KEY_TYPE_NORMAL: - mock_copy_s3.assert_called_once_with(bucket_name, filename, bucket_name, "REDACTION_FAILURE/" + filename) - else: - mock_copy_s3.assert_not_called() - - -@pytest.mark.skip(reason="Letter tests") -@freeze_time("2018-01-01 18:00") -@mock_s3 -@pytest.mark.parametrize("key_type", [KEY_TYPE_NORMAL, KEY_TYPE_TEST]) -def test_process_letter_task_check_virus_scan_passed_when_file_cannot_be_opened(sample_letter_notification, mocker, key_type): - filename = "NOTIFY.{}".format(sample_letter_notification.reference) - source_bucket_name = current_app.config["LETTERS_SCAN_BUCKET_NAME"] - target_bucket_name = current_app.config["INVALID_PDF_BUCKET_NAME"] - - conn = boto3.resource("s3") - conn.create_bucket(Bucket=source_bucket_name) - conn.create_bucket(Bucket=target_bucket_name) - - s3 = boto3.client("s3", region_name="ca-central-1") - s3.put_object(Bucket=source_bucket_name, Key=filename, Body=b"pdf_content") - - sample_letter_notification.status = NOTIFICATION_PENDING_VIRUS_CHECK - sample_letter_notification.key_type = key_type - mock_move_s3 = mocker.patch("app.letters.utils._move_s3_object") - - mock_get_page_count = mocker.patch("app.celery.letters_pdf_tasks.get_page_count", side_effect=PdfReadError) - mock_sanitise = mocker.patch("app.celery.letters_pdf_tasks._sanitise_precompiled_pdf") - - process_virus_scan_passed(filename) - - mock_sanitise.assert_not_called() - mock_get_page_count.assert_called_once_with(b"pdf_content") - mock_move_s3.assert_called_once_with(source_bucket_name, filename, target_bucket_name, filename) - assert sample_letter_notification.status == NOTIFICATION_VALIDATION_FAILED - assert sample_letter_notification.billable_units == 0 - - -@pytest.mark.skip(reason="Letter tests") -@mock_s3 -def test_process_virus_scan_passed_logs_error_and_sets_tech_failure_if_s3_error_uploading_to_live_bucket( - mocker, - sample_letter_notification, -): - mock_logger = mocker.patch("app.celery.tasks.current_app.logger.exception") - - sample_letter_notification.status = NOTIFICATION_PENDING_VIRUS_CHECK - filename = "NOTIFY.{}".format(sample_letter_notification.reference) - - source_bucket_name = current_app.config["LETTERS_SCAN_BUCKET_NAME"] - conn = boto3.resource("s3") - conn.create_bucket(Bucket=source_bucket_name) - - s3 = boto3.client("s3", region_name="ca-central-1") - s3.put_object(Bucket=source_bucket_name, Key=filename, Body=b"pdf_content") - - mocker.patch("app.celery.letters_pdf_tasks.get_page_count", return_value=1) - - error_response = { - "Error": { - "Code": "InvalidParameterValue", - "Message": "some error message from amazon", - "Type": "Sender", - } - } - mocker.patch( - "app.celery.letters_pdf_tasks._upload_pdf_to_test_or_live_pdf_bucket", - side_effect=ClientError(error_response, "operation_name"), - ) - - endpoint = "http://localhost:9999/precompiled/sanitise" - with requests_mock.mock() as rmock: - rmock.request( - "POST", - endpoint, - json={ - "file": base64.b64encode(b"new_pdf").decode("utf-8"), - "validation_passed": True, - "errors": { - "content_outside_of_printable_area": [], - "document_not_a4_size_portrait_orientation": [], - }, - }, - status_code=200, - ) - process_virus_scan_passed(filename) - - assert sample_letter_notification.status == NOTIFICATION_TECHNICAL_FAILURE - mock_logger.assert_called_once_with( - "Error uploading letter to live pdf bucket for notification: {}".format(sample_letter_notification.id) - ) - - -@pytest.mark.skip(reason="Letter tests") -def test_move_invalid_letter_and_update_status_logs_error_and_sets_tech_failure_state_if_s3_error( - mocker, - sample_letter_notification, -): - error_response = { - "Error": { - "Code": "InvalidParameterValue", - "Message": "some error message from amazon", - "Type": "Sender", - } - } - mocker.patch( - "app.celery.letters_pdf_tasks.move_scan_to_invalid_pdf_bucket", - side_effect=ClientError(error_response, "operation_name"), - ) - mock_logger = mocker.patch("app.celery.tasks.current_app.logger.exception") - - _move_invalid_letter_and_update_status(sample_letter_notification, "filename", mocker.Mock()) - - assert sample_letter_notification.status == NOTIFICATION_TECHNICAL_FAILURE - mock_logger.assert_called_once_with( - "Error when moving letter with id {} to invalid PDF bucket".format(sample_letter_notification.id) - ) - - -@pytest.mark.skip(reason="Letter tests") -def test_process_letter_task_check_virus_scan_failed(sample_letter_notification, mocker): - filename = "NOTIFY.{}".format(sample_letter_notification.reference) - sample_letter_notification.status = NOTIFICATION_PENDING_VIRUS_CHECK - mock_move_failed_pdf = mocker.patch("app.celery.letters_pdf_tasks.move_failed_pdf") - - with pytest.raises(VirusScanError) as e: - process_virus_scan_failed(filename) - - assert "Virus scan failed:" in str(e) - mock_move_failed_pdf.assert_called_once_with(filename, ScanErrorType.FAILURE) - assert sample_letter_notification.status == NOTIFICATION_VIRUS_SCAN_FAILED - - -@pytest.mark.skip(reason="Letter tests") -def test_process_letter_task_check_virus_scan_error(sample_letter_notification, mocker): - filename = "NOTIFY.{}".format(sample_letter_notification.reference) - sample_letter_notification.status = NOTIFICATION_PENDING_VIRUS_CHECK - mock_move_failed_pdf = mocker.patch("app.celery.letters_pdf_tasks.move_failed_pdf") - - with pytest.raises(VirusScanError) as e: - process_virus_scan_error(filename) - - assert "Virus scan error:" in str(e.value) - mock_move_failed_pdf.assert_called_once_with(filename, ScanErrorType.ERROR) - assert sample_letter_notification.status == NOTIFICATION_TECHNICAL_FAILURE - - -@pytest.mark.skip(reason="Letter tests") -def test_replay_letters_in_error_for_all_letters_in_error_bucket(notify_api, mocker): - mockObject = boto3.resource("s3").Object("ERROR", "ERROR/file_name") - mocker.patch( - "app.celery.letters_pdf_tasks.get_file_names_from_error_bucket", - return_value=[mockObject], - ) - mock_move = mocker.patch("app.celery.letters_pdf_tasks.move_error_pdf_to_scan_bucket") - mock_celery = mocker.patch("app.celery.letters_pdf_tasks.notify_celery.send_task") - replay_letters_in_error() - mock_move.assert_called_once_with("file_name") - mock_celery.assert_called_once_with(name="scan-file", kwargs={"filename": "file_name"}, queue="antivirus-tasks") - - -@pytest.mark.skip(reason="Letter tests") -def test_replay_letters_in_error_for_one_file(notify_api, mocker): - mockObject = boto3.resource("s3").Object("ERROR", "ERROR/file_name") - mocker.patch( - "app.celery.letters_pdf_tasks.get_file_names_from_error_bucket", - return_value=[mockObject], - ) - mock_move = mocker.patch("app.celery.letters_pdf_tasks.move_error_pdf_to_scan_bucket") - mock_celery = mocker.patch("app.celery.letters_pdf_tasks.notify_celery.send_task") - replay_letters_in_error("file_name") - mock_move.assert_called_once_with("file_name") - mock_celery.assert_called_once_with(name="scan-file", kwargs={"filename": "file_name"}, queue="antivirus-tasks") - - -@pytest.mark.skip(reason="Letter tests") -def test_sanitise_precompiled_pdf_returns_data_from_template_preview(rmock, sample_letter_notification): - sample_letter_notification.status = NOTIFICATION_PENDING_VIRUS_CHECK - endpoint = "http://localhost:9999/precompiled/sanitise" - with requests_mock.mock() as rmock: - rmock.request( - "POST", - endpoint, - json={ - "file": base64.b64encode(b"new_pdf").decode("utf-8"), - "validation_passed": True, - "errors": { - "content_outside_of_printable_area": [], - "document_not_a4_size_portrait_orientation": [], - }, - }, - status_code=200, - ) - mock_celery = Mock(**{"retry.side_effect": Retry}) - response = _sanitise_precompiled_pdf(mock_celery, sample_letter_notification, b"old_pdf") - assert rmock.called - assert rmock.request_history[0].url == endpoint - - assert base64.b64decode(response.json()["file"].encode()) == b"new_pdf" - assert rmock.last_request.text == "old_pdf" - - -@pytest.mark.skip(reason="Letter tests") -def test_sanitise_precompiled_pdf_returns_none_on_validation_error(rmock, sample_letter_notification): - sample_letter_notification.status = NOTIFICATION_PENDING_VIRUS_CHECK - - endpoint = "http://localhost:9999/precompiled/sanitise" - with requests_mock.mock() as rmock: - rmock.request( - "POST", - endpoint, - json={ - "file": base64.b64encode(b"nyan").decode("utf-8"), - "validation_passed": False, - "errors": { - "content_outside_of_printable_area": [1], - "document_not_a4_size_portrait_orientation": [], - }, - }, - status_code=400, - ) - mock_celery = Mock(**{"retry.side_effect": Retry}) - response = _sanitise_precompiled_pdf(mock_celery, sample_letter_notification, b"old_pdf") - assert rmock.called - assert rmock.request_history[0].url == endpoint - - assert response is None - - -@pytest.mark.skip(reason="Letter tests") -def test_sanitise_precompiled_pdf_passes_the_service_id_and_notification_id_to_template_preview( - mocker, - sample_letter_notification, -): - tp_mock = mocker.patch("app.celery.letters_pdf_tasks.requests_post") - sample_letter_notification.status = NOTIFICATION_PENDING_VIRUS_CHECK - mock_celery = Mock(**{"retry.side_effect": Retry}) - _sanitise_precompiled_pdf(mock_celery, sample_letter_notification, b"old_pdf") - - service_id = str(sample_letter_notification.service_id) - notification_id = str(sample_letter_notification.id) - - tp_mock.assert_called_once_with( - "http://localhost:9999/precompiled/sanitise", - data=b"old_pdf", - headers={ - "Authorization": "Token my-secret-key", - "Service-ID": service_id, - "Notification-ID": notification_id, - }, - ) - - -@pytest.mark.skip(reason="Letter tests") -def test_sanitise_precompiled_pdf_retries_on_http_error(rmock, sample_letter_notification): - sample_letter_notification.status = NOTIFICATION_PENDING_VIRUS_CHECK - rmock.post( - "http://localhost:9999/precompiled/sanitise", - content=b"new_pdf", - status_code=500, - ) - mock_celery = Mock(**{"retry.side_effect": Retry}) - - with pytest.raises(Retry): - _sanitise_precompiled_pdf(mock_celery, sample_letter_notification, b"old_pdf") - - -@pytest.mark.skip(reason="Letter tests") -def test_sanitise_precompiled_pdf_sets_notification_to_technical_failure_after_too_many_errors(rmock, sample_letter_notification): - sample_letter_notification.status = NOTIFICATION_PENDING_VIRUS_CHECK - rmock.post( - "http://localhost:9999/precompiled/sanitise", - content=b"new_pdf", - status_code=500, - ) - mock_celery = Mock(**{"retry.side_effect": MaxRetriesExceededError}) - - with pytest.raises(MaxRetriesExceededError): - _sanitise_precompiled_pdf(mock_celery, sample_letter_notification, b"old_pdf") - - assert sample_letter_notification.status == NOTIFICATION_TECHNICAL_FAILURE diff --git a/tests/app/dao/notification_dao/test_notification_dao_delete_notifications.py b/tests/app/dao/notification_dao/test_notification_dao_delete_notifications.py index 581da5add7..2601522739 100644 --- a/tests/app/dao/notification_dao/test_notification_dao_delete_notifications.py +++ b/tests/app/dao/notification_dao/test_notification_dao_delete_notifications.py @@ -1,7 +1,6 @@ from datetime import date, datetime, timedelta import pytest -from flask import current_app from freezegun import freeze_time from app.dao.notifications_dao import ( @@ -99,7 +98,7 @@ def _create_templates(sample_service): @pytest.mark.parametrize("month, delete_run_time", [(4, "2016-04-10 23:40"), (1, "2016-01-11 00:40")]) @pytest.mark.parametrize( "notification_type, expected_sms_count, expected_email_count, expected_letter_count", - [("sms", 7, 10, 10), ("email", 10, 7, 10), ("letter", 10, 10, 7)], + [("sms", 7, 10, 10), ("email", 10, 7, 10)], ) def test_should_delete_notifications_by_type_after_seven_days( sample_service, @@ -111,7 +110,6 @@ def test_should_delete_notifications_by_type_after_seven_days( expected_email_count, expected_letter_count, ): - mocker.patch("app.dao.notifications_dao.get_s3_bucket_objects") email_template, letter_template, sms_template = _create_templates(sample_service) # create one notification a day between 1st and 10th from 11:00 to 19:00 of each type for i in range(1, 11): @@ -157,7 +155,6 @@ def test_should_delete_notifications_by_type_after_seven_days( @freeze_time("2016-01-10 12:00:00.000000") def test_should_not_delete_notification_history(sample_service, mocker): - mocker.patch("app.dao.notifications_dao.get_s3_bucket_objects") with freeze_time("2016-01-01 12:00"): email_template, letter_template, sms_template = _create_templates(sample_service) save_notification(create_notification(template=email_template, status="permanent-failure")) @@ -169,22 +166,13 @@ def test_should_not_delete_notification_history(sample_service, mocker): assert NotificationHistory.query.count() == 1 -@pytest.mark.parametrize("notification_type", ["sms", "email", "letter"]) +@pytest.mark.parametrize("notification_type", ["sms", "email"]) def test_delete_notifications_for_days_of_retention(sample_service, notification_type, mocker): - mock_get_s3 = mocker.patch("app.dao.notifications_dao.get_s3_bucket_objects") create_test_data(notification_type, sample_service) assert Notification.query.count() == 9 delete_notifications_older_than_retention_by_type(notification_type) assert Notification.query.count() == 7 assert Notification.query.filter_by(notification_type=notification_type).count() == 1 - if notification_type == "letter": - mock_get_s3.assert_called_with( - bucket_name=current_app.config["LETTERS_PDF_BUCKET_NAME"], - subfolder="{}/NOTIFY.LETTER_REF.D.2.C.C".format(str(datetime.utcnow().date())), - ) - assert mock_get_s3.call_count == 2 - else: - mock_get_s3.assert_not_called() def test_delete_notifications_inserts_notification_history(sample_service): @@ -197,7 +185,6 @@ def test_delete_notifications_inserts_notification_history(sample_service): def test_delete_notifications_updates_notification_history(sample_email_template, mocker): - mocker.patch("app.dao.notifications_dao.get_s3_bucket_objects") notification = save_notification( create_notification(template=sample_email_template, created_at=datetime.utcnow() - timedelta(days=8)) ) @@ -232,7 +219,6 @@ def test_delete_notifications_keep_data_for_days_of_retention_is_longer(sample_s def test_delete_notifications_with_test_keys(sample_template, mocker): - mocker.patch("app.dao.notifications_dao.get_s3_bucket_objects") save_notification( create_notification( template=sample_template, @@ -279,18 +265,8 @@ def test_delete_notifications_delete_notification_type_for_default_time_if_no_da assert Notification.query.filter_by(notification_type="email").count() == 1 -def test_delete_notifications_does_try_to_delete_from_s3_when_letter_has_not_been_sent(sample_service, mocker): - mock_get_s3 = mocker.patch("app.dao.notifications_dao.get_s3_bucket_objects") - letter_template = create_template(service=sample_service, template_type="letter") - - save_notification(create_notification(template=letter_template, status="sending", reference="LETTER_REF")) - delete_notifications_older_than_retention_by_type("email", qry_limit=1) - mock_get_s3.assert_not_called() - - @freeze_time("2016-01-10 12:00:00.000000") def test_should_not_delete_notification_if_history_does_not_exist(sample_service, mocker): - mocker.patch("app.dao.notifications_dao.get_s3_bucket_objects") mocker.patch("app.dao.notifications_dao.insert_update_notification_history") with freeze_time("2016-01-01 12:00"): email_template, letter_template, sms_template = _create_templates(sample_service) diff --git a/tests/app/service/test_rest.py b/tests/app/service/test_rest.py index 342943b433..4c2069a6e2 100644 --- a/tests/app/service/test_rest.py +++ b/tests/app/service/test_rest.py @@ -38,7 +38,6 @@ Notification, Service, ServiceEmailReplyTo, - ServiceLetterContact, ServicePermission, ServiceSmsSender, User, @@ -57,7 +56,6 @@ create_ft_notification_status, create_inbound_number, create_letter_branding, - create_letter_contact, create_notification, create_organisation, create_reply_to_email, @@ -662,47 +660,6 @@ def test_cant_update_service_org_type_to_random_value(client, sample_service): assert resp.status_code == 500 -def test_update_service_letter_branding(client, notify_db, sample_service): - letter_branding = create_letter_branding(name="test brand", filename="test-brand") - data = {"letter_branding": str(letter_branding.id)} - - auth_header = create_authorization_header() - - resp = client.post( - "/service/{}".format(sample_service.id), - data=json.dumps(data), - headers=[("Content-Type", "application/json"), auth_header], - ) - result = resp.json - assert resp.status_code == 200 - assert result["data"]["letter_branding"] == str(letter_branding.id) - - -def test_update_service_remove_letter_branding(client, notify_db, sample_service): - letter_branding = create_letter_branding(name="test brand", filename="test-brand") - sample_service - data = {"letter_branding": str(letter_branding.id)} - - auth_header = create_authorization_header() - - client.post( - "/service/{}".format(sample_service.id), - data=json.dumps(data), - headers=[("Content-Type", "application/json"), auth_header], - ) - - data = {"letter_branding": None} - resp = client.post( - "/service/{}".format(sample_service.id), - data=json.dumps(data), - headers=[("Content-Type", "application/json"), auth_header], - ) - - result = resp.json - assert resp.status_code == 200 - assert result["data"]["letter_branding"] is None - - def test_update_service_remove_email_branding(admin_request, notify_db, sample_service): brand = EmailBranding(colour="#000000", logo="justice-league.png", name="Justice League") sample_service.email_branding = brand @@ -2429,25 +2386,6 @@ def test_send_one_off_notification(sample_service, admin_request, mocker): assert response["id"] == str(noti.id) -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}) - - response = client.post( - url_for("service.create_pdf_letter", service_id=sample_service_full_permissions.id), - data=data, - headers=[("Content-Type", "application/json"), create_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): resp = admin_request.get( "service.get_notification_for_service", @@ -2963,218 +2901,6 @@ def test_get_email_reply_to_address(client, notify_db, notify_db_session): assert json.loads(response.get_data(as_text=True)) == reply_to.serialize() -def test_get_letter_contacts_when_there_are_no_letter_contacts(client, sample_service): - response = client.get( - "/service/{}/letter-contact".format(sample_service.id), - headers=[create_authorization_header()], - ) - - assert json.loads(response.get_data(as_text=True)) == [] - assert response.status_code == 200 - - -def test_get_letter_contacts_with_one_letter_contact(client, notify_db, notify_db_session): - service = create_service() - create_letter_contact(service, "Aberdeen, AB23 1XH") - - response = client.get( - "/service/{}/letter-contact".format(service.id), - headers=[create_authorization_header()], - ) - json_response = json.loads(response.get_data(as_text=True)) - - assert len(json_response) == 1 - assert json_response[0]["contact_block"] == "Aberdeen, AB23 1XH" - assert json_response[0]["is_default"] - assert json_response[0]["created_at"] - assert not json_response[0]["updated_at"] - assert response.status_code == 200 - - -def test_get_letter_contacts_with_multiple_letter_contacts(client, notify_db, notify_db_session): - service = create_service() - letter_contact_a = create_letter_contact(service, "Aberdeen, AB23 1XH") - letter_contact_b = create_letter_contact(service, "London, E1 8QS", False) - - response = client.get( - "/service/{}/letter-contact".format(service.id), - headers=[create_authorization_header()], - ) - json_response = json.loads(response.get_data(as_text=True)) - - assert len(json_response) == 2 - assert response.status_code == 200 - - assert json_response[0]["id"] == str(letter_contact_a.id) - assert json_response[0]["service_id"] == str(letter_contact_a.service_id) - assert json_response[0]["contact_block"] == "Aberdeen, AB23 1XH" - assert json_response[0]["is_default"] - assert json_response[0]["created_at"] - assert not json_response[0]["updated_at"] - - assert json_response[1]["id"] == str(letter_contact_b.id) - assert json_response[1]["service_id"] == str(letter_contact_b.service_id) - assert json_response[1]["contact_block"] == "London, E1 8QS" - assert not json_response[1]["is_default"] - assert json_response[1]["created_at"] - assert not json_response[1]["updated_at"] - - -def test_get_letter_contact_by_id(client, notify_db, notify_db_session): - service = create_service() - letter_contact = create_letter_contact(service, "London, E1 8QS") - - response = client.get( - "/service/{}/letter-contact/{}".format(service.id, letter_contact.id), - headers=[("Content-Type", "application/json"), create_authorization_header()], - ) - - assert response.status_code == 200 - assert json.loads(response.get_data(as_text=True)) == letter_contact.serialize() - - -def test_get_letter_contact_return_404_when_invalid_contact_id(client, notify_db, notify_db_session): - service = create_service() - - response = client.get( - "/service/{}/letter-contact/{}".format(service.id, "93d59f88-4aa1-453c-9900-f61e2fc8a2de"), - headers=[("Content-Type", "application/json"), create_authorization_header()], - ) - - assert response.status_code == 404 - - -def test_add_service_contact_block(client, sample_service): - data = json.dumps({"contact_block": "London, E1 8QS", "is_default": True}) - response = client.post( - "/service/{}/letter-contact".format(sample_service.id), - data=data, - headers=[("Content-Type", "application/json"), create_authorization_header()], - ) - - assert response.status_code == 201 - json_resp = json.loads(response.get_data(as_text=True)) - results = ServiceLetterContact.query.all() - assert len(results) == 1 - assert json_resp["data"] == results[0].serialize() - - -def test_add_service_letter_contact_can_add_multiple_addresses(client, sample_service): - first = json.dumps({"contact_block": "London, E1 8QS", "is_default": True}) - client.post( - "/service/{}/letter-contact".format(sample_service.id), - data=first, - headers=[("Content-Type", "application/json"), create_authorization_header()], - ) - - second = json.dumps({"contact_block": "Aberdeen, AB23 1XH", "is_default": True}) - response = client.post( - "/service/{}/letter-contact".format(sample_service.id), - data=second, - headers=[("Content-Type", "application/json"), create_authorization_header()], - ) - assert response.status_code == 201 - json_resp = json.loads(response.get_data(as_text=True)) - results = ServiceLetterContact.query.all() - assert len(results) == 2 - default = [x for x in results if x.is_default] - assert json_resp["data"] == default[0].serialize() - first_letter_contact_not_default = [x for x in results if not x.is_default] - assert first_letter_contact_not_default[0].contact_block == "London, E1 8QS" - - -def test_add_service_letter_contact_block_fine_if_no_default(client, sample_service): - data = json.dumps({"contact_block": "London, E1 8QS", "is_default": False}) - response = client.post( - "/service/{}/letter-contact".format(sample_service.id), - data=data, - headers=[("Content-Type", "application/json"), create_authorization_header()], - ) - assert response.status_code == 201 - - -def test_add_service_letter_contact_block_404s_when_invalid_service_id(client, notify_db, notify_db_session): - response = client.post( - "/service/{}/letter-contact".format(uuid.uuid4()), - data={}, - headers=[("Content-Type", "application/json"), create_authorization_header()], - ) - - assert response.status_code == 404 - result = json.loads(response.get_data(as_text=True)) - assert result["result"] == "error" - assert result["message"] == "No result found" - - -def test_update_service_letter_contact(client, sample_service): - original_letter_contact = create_letter_contact(service=sample_service, contact_block="Aberdeen, AB23 1XH") - data = json.dumps({"contact_block": "London, E1 8QS", "is_default": True}) - response = client.post( - "/service/{}/letter-contact/{}".format(sample_service.id, original_letter_contact.id), - data=data, - headers=[("Content-Type", "application/json"), create_authorization_header()], - ) - - assert response.status_code == 200 - json_resp = json.loads(response.get_data(as_text=True)) - results = ServiceLetterContact.query.all() - assert len(results) == 1 - assert json_resp["data"] == results[0].serialize() - - -def test_update_service_letter_contact_returns_200_when_no_default(client, sample_service): - original_reply_to = create_letter_contact(service=sample_service, contact_block="Aberdeen, AB23 1XH") - data = json.dumps({"contact_block": "London, E1 8QS", "is_default": False}) - response = client.post( - "/service/{}/letter-contact/{}".format(sample_service.id, original_reply_to.id), - data=data, - headers=[("Content-Type", "application/json"), create_authorization_header()], - ) - assert response.status_code == 200 - - -def test_update_service_letter_contact_returns_404_when_invalid_service_id(client, notify_db, notify_db_session): - response = client.post( - "/service/{}/letter-contact/{}".format(uuid.uuid4(), uuid.uuid4()), - data={}, - headers=[("Content-Type", "application/json"), create_authorization_header()], - ) - - assert response.status_code == 404 - result = json.loads(response.get_data(as_text=True)) - assert result["result"] == "error" - assert result["message"] == "No result found" - - -def test_delete_service_letter_contact_can_archive_letter_contact(admin_request, notify_db_session): - service = create_service() - create_letter_contact(service=service, contact_block="Edinburgh, ED1 1AA") - letter_contact = create_letter_contact(service=service, contact_block="Swansea, SN1 3CC", is_default=False) - - admin_request.post( - "service.delete_service_letter_contact", - service_id=service.id, - letter_contact_id=letter_contact.id, - ) - - assert letter_contact.archived is True - - -def test_delete_service_letter_contact_returns_200_if_archiving_template_default(admin_request, notify_db_session): - service = create_service() - create_letter_contact(service=service, contact_block="Edinburgh, ED1 1AA") - letter_contact = create_letter_contact(service=service, contact_block="Swansea, SN1 3CC", is_default=False) - create_template(service=service, template_type="letter", reply_to=letter_contact.id) - - response = admin_request.post( - "service.delete_service_letter_contact", - service_id=service.id, - letter_contact_id=letter_contact.id, - _expected_status=200, - ) - assert response["data"]["archived"] is True - - def test_add_service_sms_sender_can_add_multiple_senders(client, notify_db_session): service = create_service() data = { @@ -3480,91 +3206,6 @@ def test_cancel_notification_for_service_raises_invalid_request_when_notificatio 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 - - 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"] == "It’s too late to cancel this letter. Printing started today at 5.30pm" - assert response["result"] == "error" - - -@pytest.mark.parametrize("notification_status", ["created", "pending-virus-check"]) -@freeze_time("2018-07-07 16:00:00") -def test_cancel_notification_for_service_updates_letter_if_letter_is_in_cancellable_state( - 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, - ) - 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"] == "It’s too late to cancel this letter. Printing started on 11 December at 5.30pm" - assert response["result"] == "error" - - -@freeze_time("2018-7-7 16:00:00") -def test_cancel_notification_for_service_updates_letter_if_still_time_to_cancel( - admin_request, - sample_letter_notification, -): - sample_letter_notification.created_at = datetime(2018, 7, 7, 10, 0) - - response = admin_request.post( - "service.cancel_notification_for_service", - service_id=sample_letter_notification.service_id, - notification_id=sample_letter_notification.id, - ) - assert response["status"] == "cancelled" - - def test_get_monthly_notification_data_by_service(mocker, admin_request): dao_mock = mocker.patch( "app.service.rest.fact_notification_status_dao.fetch_monthly_notification_statuses_per_service", diff --git a/tests/app/service/test_send_pdf_letter_notification.py b/tests/app/service/test_send_pdf_letter_notification.py deleted file mode 100644 index b236c83cab..0000000000 --- a/tests/app/service/test_send_pdf_letter_notification.py +++ /dev/null @@ -1,111 +0,0 @@ -import uuid - -import pytest -from freezegun import freeze_time -from notifications_utils.s3 import S3ObjectNotFound - -from app.dao.notifications_dao import get_notification_by_id -from app.models import EMAIL_TYPE, LETTER_TYPE, UPLOAD_LETTERS -from app.service.send_notification import send_pdf_letter_notification -from app.v2.errors import BadRequestError, TooManyRequestsError -from tests.app.db import create_service - - -@pytest.mark.parametrize( - "permissions", - [ - [EMAIL_TYPE], - [LETTER_TYPE], - [UPLOAD_LETTERS], - ], -) -def test_send_pdf_letter_notification_raises_error_if_service_does_not_have_permission( - notify_db_session, - fake_uuid, - permissions, -): - service = create_service(service_permissions=permissions) - post_data = {"filename": "valid.pdf", "created_by": fake_uuid, "file_id": fake_uuid} - - with pytest.raises(BadRequestError): - send_pdf_letter_notification(service.id, post_data) - - -def test_send_pdf_letter_notification_raises_error_if_service_is_over_daily_message_limit( - mocker, - sample_service_full_permissions, - fake_uuid, -): - mocker.patch( - "app.service.send_notification.check_service_over_daily_message_limit", - side_effect=TooManyRequestsError(10), - ) - post_data = {"filename": "valid.pdf", "created_by": fake_uuid, "file_id": fake_uuid} - - with pytest.raises(TooManyRequestsError): - send_pdf_letter_notification(sample_service_full_permissions.id, post_data) - - -def test_send_pdf_letter_notification_validates_created_by(sample_service_full_permissions, fake_uuid, sample_user): - post_data = { - "filename": "valid.pdf", - "created_by": sample_user.id, - "file_id": fake_uuid, - } - - with pytest.raises(BadRequestError): - send_pdf_letter_notification(sample_service_full_permissions.id, post_data) - - -def test_send_pdf_letter_notification_raises_error_when_pdf_is_not_in_transient_letter_bucket( - mocker, - sample_service_full_permissions, - fake_uuid, - notify_user, -): - user = sample_service_full_permissions.users[0] - post_data = {"filename": "valid.pdf", "created_by": user.id, "file_id": fake_uuid} - mocker.patch( - "app.service.send_notification.utils_s3download", - side_effect=S3ObjectNotFound({}, ""), - ) - - with pytest.raises(S3ObjectNotFound): - send_pdf_letter_notification(sample_service_full_permissions.id, post_data) - - -@freeze_time("2019-08-02 11:00:00") -def test_send_pdf_letter_notification_creates_notification_and_moves_letter( - mocker, - sample_service_full_permissions, - notify_user, -): - user = sample_service_full_permissions.users[0] - filename = "valid.pdf" - file_id = uuid.uuid4() - post_data = {"filename": filename, "created_by": user.id, "file_id": file_id} - - mocker.patch("app.service.send_notification.utils_s3download") - mocker.patch("app.service.send_notification.get_page_count", return_value=1) - s3_mock = mocker.patch("app.service.send_notification.move_uploaded_pdf_to_letters_bucket") - - result = send_pdf_letter_notification(sample_service_full_permissions.id, post_data) - - notification = get_notification_by_id(file_id) - - assert notification.id == file_id - assert notification.api_key_id is None - assert notification.client_reference == filename - assert notification.created_by_id == user.id - assert notification.postage == "second" - assert notification.notification_type == LETTER_TYPE - assert notification.billable_units == 1 - assert notification.to == filename - assert notification.service_id == sample_service_full_permissions.id - - assert result == {"id": str(notification.id)} - - s3_mock.assert_called_once_with( - "service-{}/{}.pdf".format(sample_service_full_permissions.id, file_id), - "2019-08-02/NOTIFY.{}.D.2.C.C.20190802110000.PDF".format(notification.reference), - )