notifications-api/tests/app/letters/test_letter_utils.py

import pytest
from datetime import datetime

import boto3
from flask import current_app
from freezegun import freeze_time
from moto import mock_s3

from app.letters.utils import (
    get_bucket_name_and_prefix_for_notification,
    get_letter_pdf_filename,
    get_letter_pdf,
    upload_letter_pdf,
    ScanErrorType, move_failed_pdf, get_folder_name
)
from app.models import KEY_TYPE_NORMAL, KEY_TYPE_TEST, PRECOMPILED_TEMPLATE_NAME, NOTIFICATION_VALIDATION_FAILED
from tests.app.db import create_notification

FROZEN_DATE_TIME = "2018-03-14 17:00:00"


@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()
    return sample_letter_notification


@pytest.fixture(name='sample_precompiled_letter_notification_using_test_key')
def _sample_precompiled_letter_notification_using_test_key(sample_precompiled_letter_notification):
    sample_precompiled_letter_notification.key_type = KEY_TYPE_TEST
    return sample_precompiled_letter_notification


def test_get_bucket_name_and_prefix_for_notification_valid_notification(sample_notification):

    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=sample_notification.created_at.date(),
        reference=sample_notification.reference
    ).upper()


@freeze_time(FROZEN_DATE_TIME)
def test_get_bucket_name_and_prefix_for_notification_precompiled_letter_using_test_key(
    sample_precompiled_letter_notification_using_test_key
):
    bucket, bucket_prefix = get_bucket_name_and_prefix_for_notification(
        sample_precompiled_letter_notification_using_test_key)

    assert bucket == current_app.config['TEST_LETTERS_BUCKET_NAME']
    assert bucket_prefix == 'NOTIFY.{}'.format(
        sample_precompiled_letter_notification_using_test_key.reference).upper()


@freeze_time(FROZEN_DATE_TIME)
def test_get_bucket_name_and_prefix_for_failed_validation(sample_precompiled_letter_notification):
    sample_precompiled_letter_notification.status = NOTIFICATION_VALIDATION_FAILED
    bucket, bucket_prefix = get_bucket_name_and_prefix_for_notification(sample_precompiled_letter_notification)

    assert bucket == current_app.config['INVALID_PDF_BUCKET_NAME']
    assert bucket_prefix == 'NOTIFY.{}'.format(
        sample_precompiled_letter_notification.reference).upper()


@freeze_time(FROZEN_DATE_TIME)
def test_get_bucket_name_and_prefix_for_test_noti_with_failed_validation(
    sample_precompiled_letter_notification_using_test_key
):
    sample_precompiled_letter_notification_using_test_key.status = NOTIFICATION_VALIDATION_FAILED
    bucket, bucket_prefix = get_bucket_name_and_prefix_for_notification(
        sample_precompiled_letter_notification_using_test_key
    )

    assert bucket == current_app.config['INVALID_PDF_BUCKET_NAME']
    assert bucket_prefix == 'NOTIFY.{}'.format(
        sample_precompiled_letter_notification_using_test_key.reference).upper()


def test_get_bucket_name_and_prefix_for_notification_invalid_notification():
    with pytest.raises(AttributeError):
        get_bucket_name_and_prefix_for_notification(None)


@pytest.mark.parametrize('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.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'


@freeze_time("2017-12-04 17:31:00")
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'


@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', region_name='eu-west-1')
    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.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.parametrize('postage,expected_postage', [
    ('second', 2),
    ('first', 1)
])
def test_upload_letter_pdf_uses_postage_from_notification(
    sample_letter_template, mocker, postage, expected_postage
):
    letter_notification = create_notification(template=sample_letter_template, postage=postage)
    mock_s3 = mocker.patch('app.letters.utils.s3upload')

    filename = 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']
    )


@mock_s3
@freeze_time(FROZEN_DATE_TIME)
def test_move_failed_pdf_error(notify_api):
    filename = 'test.pdf'
    bucket_name = current_app.config['LETTERS_SCAN_BUCKET_NAME']

    conn = boto3.resource('s3', region_name='eu-west-1')
    bucket = conn.create_bucket(Bucket=bucket_name)

    s3 = boto3.client('s3', region_name='eu-west-1')
    s3.put_object(Bucket=bucket_name, Key=filename, Body=b'pdf_content')

    move_failed_pdf(filename, ScanErrorType.ERROR)

    assert 'ERROR/' + filename in [o.key for o in bucket.objects.all()]
    assert filename not in [o.key for o in bucket.objects.all()]


@mock_s3
@freeze_time(FROZEN_DATE_TIME)
def test_move_failed_pdf_scan_failed(notify_api):
    filename = 'test.pdf'
    bucket_name = current_app.config['LETTERS_SCAN_BUCKET_NAME']

    conn = boto3.resource('s3', region_name='eu-west-1')
    bucket = conn.create_bucket(Bucket=bucket_name)

    s3 = boto3.client('s3', region_name='eu-west-1')
    s3.put_object(Bucket=bucket_name, Key=filename, Body=b'pdf_content')

    move_failed_pdf(filename, ScanErrorType.FAILURE)

    assert 'FAILURE/' + filename in [o.key for o in bucket.objects.all()]
    assert filename not in [o.key for o in bucket.objects.all()]


@pytest.mark.parametrize("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/"),
                          ])
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


def test_get_folder_name_returns_empty_string_for_test_letter():
    assert '' == get_folder_name(datetime.utcnow(), is_test_or_scan_letter=True)