From 0326005aebf2fbef80e365b11d44c18d937a3a1f Mon Sep 17 00:00:00 2001 From: Ben Thorner Date: Wed, 9 Jun 2021 15:15:35 +0100 Subject: [PATCH] Extract template / csv utility code into modules This follows a similar approach to the previous commits, noting that one module depends on the other, so we have to extract both together. --- app/main/views/dashboard.py | 2 +- app/main/views/jobs.py | 2 +- app/main/views/notifications.py | 4 +- app/main/views/platform_admin.py | 2 +- app/main/views/send.py | 5 +- app/main/views/templates.py | 7 +- app/main/views/tour.py | 2 +- app/main/views/uploads.py | 6 +- app/models/contact_list.py | 2 +- app/utils/__init__.py | 177 ------------------ app/utils/csv.py | 109 +++++++++++ app/utils/templates.py | 70 +++++++ tests/app/main/test_errors_for_csv.py | 2 +- tests/app/test_utils.py | 255 -------------------------- tests/app/utils/test_csv.py | 248 +++++++++++++++++++++++++ tests/app/utils/test_templates.py | 10 + 16 files changed, 451 insertions(+), 452 deletions(-) create mode 100644 app/utils/csv.py create mode 100644 app/utils/templates.py create mode 100644 tests/app/utils/test_csv.py create mode 100644 tests/app/utils/test_templates.py diff --git a/app/main/views/dashboard.py b/app/main/views/dashboard.py index 33101ec5a..fb8b56ae8 100644 --- a/app/main/views/dashboard.py +++ b/app/main/views/dashboard.py @@ -30,12 +30,12 @@ from app.utils import ( DELIVERED_STATUSES, FAILURE_STATUSES, REQUESTED_STATUSES, - Spreadsheet, generate_next_dict, generate_previous_dict, get_current_financial_year, service_has_permission, ) +from app.utils.csv import Spreadsheet from app.utils.user import user_has_permissions diff --git a/app/main/views/jobs.py b/app/main/views/jobs.py index 9110d7186..66952030f 100644 --- a/app/main/views/jobs.py +++ b/app/main/views/jobs.py @@ -35,12 +35,12 @@ from app.main.forms import SearchNotificationsForm from app.models.job import Job from app.utils import ( generate_next_dict, - generate_notifications_csv, generate_previous_dict, get_page_from_request, parse_filter_args, set_status_filters, ) +from app.utils.csv import generate_notifications_csv from app.utils.letters import ( get_letter_printing_statement, printing_today_or_tomorrow, diff --git a/app/main/views/notifications.py b/app/main/views/notifications.py index 4bc29fd47..77113cb42 100644 --- a/app/main/views/notifications.py +++ b/app/main/views/notifications.py @@ -37,16 +37,16 @@ from app.template_previews import get_page_count_for_letter from app.utils import ( DELIVERED_STATUSES, FAILURE_STATUSES, - generate_notifications_csv, get_help_argument, - get_template, parse_filter_args, set_status_filters, ) +from app.utils.csv import generate_notifications_csv from app.utils.letters import ( get_letter_printing_statement, get_letter_validation_error, ) +from app.utils.templates import get_template from app.utils.user import user_has_permissions diff --git a/app/main/views/platform_admin.py b/app/main/views/platform_admin.py index 544da3a58..0c6907e27 100644 --- a/app/main/views/platform_admin.py +++ b/app/main/views/platform_admin.py @@ -28,11 +28,11 @@ from app.statistics_utils import ( get_formatted_percentage_two_dp, ) from app.utils import ( - Spreadsheet, generate_next_dict, generate_previous_dict, get_page_from_request, ) +from app.utils.csv import Spreadsheet from app.utils.user import user_is_platform_admin COMPLAINT_THRESHOLD = 0.02 diff --git a/app/main/views/send.py b/app/main/views/send.py index ddd8ce73a..6dd1c83e2 100644 --- a/app/main/views/send.py +++ b/app/main/views/send.py @@ -52,12 +52,11 @@ from app.s3_client.s3_csv_client import ( from app.template_previews import TemplatePreview, get_page_count_for_letter from app.utils import ( PermanentRedirect, - Spreadsheet, - get_errors_for_csv, - get_template, should_skip_template_page, unicode_truncate, ) +from app.utils.csv import Spreadsheet, get_errors_for_csv +from app.utils.templates import get_template from app.utils.user import user_has_permissions letter_address_columns = [ diff --git a/app/main/views/templates.py b/app/main/views/templates.py index bb01de36b..a389af900 100644 --- a/app/main/views/templates.py +++ b/app/main/views/templates.py @@ -39,11 +39,8 @@ from app.main.views.send import get_sender_details from app.models.service import Service from app.models.template_list import TemplateList, TemplateLists from app.template_previews import TemplatePreview, get_page_count_for_letter -from app.utils import ( - NOTIFICATION_TYPES, - get_template, - should_skip_template_page, -) +from app.utils import NOTIFICATION_TYPES, should_skip_template_page +from app.utils.templates import get_template from app.utils.user import user_has_permissions, user_is_platform_admin form_objects = { diff --git a/app/main/views/tour.py b/app/main/views/tour.py index 9317367e3..0cf96b885 100644 --- a/app/main/views/tour.py +++ b/app/main/views/tour.py @@ -9,7 +9,7 @@ from app.main.views.send import ( get_placeholder_form_instance, get_recipient_and_placeholders_from_session, ) -from app.utils import get_template +from app.utils.templates import get_template from app.utils.user import user_has_permissions diff --git a/app/main/views/uploads.py b/app/main/views/uploads.py index b59b0be9a..628009e6c 100644 --- a/app/main/views/uploads.py +++ b/app/main/views/uploads.py @@ -46,19 +46,17 @@ from app.s3_client.s3_letter_upload_client import ( ) from app.template_previews import TemplatePreview, sanitise_letter from app.utils import ( - Spreadsheet, generate_next_dict, generate_previous_dict, - get_errors_for_csv, get_page_from_request, - get_sample_template, - get_template, unicode_truncate, ) +from app.utils.csv import Spreadsheet, get_errors_for_csv from app.utils.letters import ( get_letter_printing_statement, get_letter_validation_error, ) +from app.utils.templates import get_sample_template, get_template from app.utils.user import user_has_permissions MAX_FILE_UPLOAD_SIZE = 2 * 1024 * 1024 # 2MB diff --git a/app/models/contact_list.py b/app/models/contact_list.py index bc18841dc..16ece39e4 100644 --- a/app/models/contact_list.py +++ b/app/models/contact_list.py @@ -16,7 +16,7 @@ from app.s3_client.s3_csv_client import ( s3upload, set_metadata_on_csv_upload, ) -from app.utils import get_sample_template +from app.utils.templates import get_sample_template class ContactList(JSONModel): diff --git a/app/utils/__init__.py b/app/utils/__init__.py index 20e50aa8d..3a2e27298 100644 --- a/app/utils/__init__.py +++ b/app/utils/__init__.py @@ -17,21 +17,11 @@ from flask import ( ) from flask_login import current_user from notifications_utils.field import Field -from notifications_utils.recipients import RecipientCSV -from notifications_utils.template import ( - BroadcastPreviewTemplate, - EmailPreviewTemplate, - LetterImageTemplate, - LetterPreviewTemplate, - SMSPreviewTemplate, -) from notifications_utils.timezones import utc_string_to_aware_gmt_datetime from orderedset._orderedset import OrderedSet from werkzeug.datastructures import MultiDict from werkzeug.routing import RequestRedirect -from app.models.spreadsheet import Spreadsheet - SENDING_STATUSES = ['created', 'pending', 'sending', 'pending-virus-check'] DELIVERED_STATUSES = ['delivered', 'sent', 'returned-letter'] FAILURE_STATUSES = ['failed', 'temporary-failure', 'permanent-failure', @@ -65,122 +55,6 @@ def redirect_to_sign_in(f): return wrapped -def get_errors_for_csv(recipients, template_type): - - errors = [] - - if any(recipients.rows_with_bad_recipients): - number_of_bad_recipients = len(list(recipients.rows_with_bad_recipients)) - if 'sms' == template_type: - if 1 == number_of_bad_recipients: - errors.append("fix 1 phone number") - else: - errors.append("fix {} phone numbers".format(number_of_bad_recipients)) - elif 'email' == template_type: - if 1 == number_of_bad_recipients: - errors.append("fix 1 email address") - else: - errors.append("fix {} email addresses".format(number_of_bad_recipients)) - elif 'letter' == template_type: - if 1 == number_of_bad_recipients: - errors.append("fix 1 address") - else: - errors.append("fix {} addresses".format(number_of_bad_recipients)) - - if any(recipients.rows_with_missing_data): - number_of_rows_with_missing_data = len(list(recipients.rows_with_missing_data)) - if 1 == number_of_rows_with_missing_data: - errors.append("enter missing data in 1 row") - else: - errors.append("enter missing data in {} rows".format(number_of_rows_with_missing_data)) - - if any(recipients.rows_with_message_too_long): - number_of_rows_with_message_too_long = len(list(recipients.rows_with_message_too_long)) - if 1 == number_of_rows_with_message_too_long: - errors.append("shorten the message in 1 row") - else: - errors.append("shorten the messages in {} rows".format(number_of_rows_with_message_too_long)) - - if any(recipients.rows_with_empty_message): - number_of_rows_with_empty_message = len(list(recipients.rows_with_empty_message)) - if 1 == number_of_rows_with_empty_message: - errors.append("check you have content for the empty message in 1 row") - else: - errors.append("check you have content for the empty messages in {} rows".format( - number_of_rows_with_empty_message - )) - - return errors - - -def get_sample_template(template_type): - if template_type == 'email': - return EmailPreviewTemplate({'content': 'any', 'subject': '', 'template_type': 'email'}) - if template_type == 'sms': - return SMSPreviewTemplate({'content': 'any', 'template_type': 'sms'}) - if template_type == 'letter': - return LetterImageTemplate( - {'content': 'any', 'subject': '', 'template_type': 'letter'}, postage='second', image_url='x', page_count=1 - ) - - -def generate_notifications_csv(**kwargs): - from app import notification_api_client - from app.s3_client.s3_csv_client import s3download - if 'page' not in kwargs: - kwargs['page'] = 1 - - if kwargs.get('job_id'): - original_file_contents = s3download(kwargs['service_id'], kwargs['job_id']) - original_upload = RecipientCSV( - original_file_contents, - template=get_sample_template(kwargs['template_type']), - ) - original_column_headers = original_upload.column_headers - fieldnames = ['Row number'] + original_column_headers + ['Template', 'Type', 'Job', 'Status', 'Time'] - else: - fieldnames = ['Recipient', 'Reference', 'Template', 'Type', 'Sent by', 'Sent by email', 'Job', 'Status', 'Time'] - - yield ','.join(fieldnames) + '\n' - - while kwargs['page']: - notifications_resp = notification_api_client.get_notifications_for_service(**kwargs) - for notification in notifications_resp['notifications']: - if kwargs.get('job_id'): - values = [ - notification['row_number'], - ] + [ - original_upload[notification['row_number'] - 1].get(header).data - for header in original_column_headers - ] + [ - notification['template_name'], - notification['template_type'], - notification['job_name'], - notification['status'], - notification['created_at'], - ] - else: - values = [ - # the recipient for precompiled letters is the full address block - notification['recipient'].splitlines()[0].lstrip().rstrip(' ,'), - notification['client_reference'], - notification['template_name'], - notification['template_type'], - notification['created_by_name'] or '', - notification['created_by_email_address'] or '', - notification['job_name'] or '', - notification['status'], - notification['created_at'] - ] - yield Spreadsheet.from_rows([map(str, values)]).as_csv_data - - if notifications_resp['links'].get('next'): - kwargs['page'] += 1 - else: - return - raise Exception("Should never reach here") - - def get_page_from_request(): if 'page' in request.args: try: @@ -211,57 +85,6 @@ def get_help_argument(): return request.args.get('help') if request.args.get('help') in ('1', '2', '3') else None -def get_template( - template, - service, - show_recipient=False, - letter_preview_url=None, - page_count=1, - redact_missing_personalisation=False, - email_reply_to=None, - sms_sender=None, -): - if 'email' == template['template_type']: - return EmailPreviewTemplate( - template, - from_name=service.name, - from_address='{}@notifications.service.gov.uk'.format(service.email_from), - show_recipient=show_recipient, - redact_missing_personalisation=redact_missing_personalisation, - reply_to=email_reply_to, - ) - if 'sms' == template['template_type']: - return SMSPreviewTemplate( - template, - prefix=service.name, - show_prefix=service.prefix_sms, - sender=sms_sender, - show_sender=bool(sms_sender), - show_recipient=show_recipient, - redact_missing_personalisation=redact_missing_personalisation, - ) - if 'letter' == template['template_type']: - if letter_preview_url: - return LetterImageTemplate( - template, - image_url=letter_preview_url, - page_count=int(page_count), - contact_block=template['reply_to_text'], - postage=template['postage'], - ) - else: - return LetterPreviewTemplate( - template, - contact_block=template['reply_to_text'], - admin_base_url=current_app.config['ADMIN_BASE_URL'], - redact_missing_personalisation=redact_missing_personalisation, - ) - if 'broadcast' == template['template_type']: - return BroadcastPreviewTemplate( - template, - ) - - def get_current_financial_year(): now = utc_string_to_aware_gmt_datetime( datetime.utcnow() diff --git a/app/utils/csv.py b/app/utils/csv.py new file mode 100644 index 000000000..ba0e2c57b --- /dev/null +++ b/app/utils/csv.py @@ -0,0 +1,109 @@ +from notifications_utils.recipients import RecipientCSV + +from app.models.spreadsheet import Spreadsheet +from app.utils.templates import get_sample_template + + +def get_errors_for_csv(recipients, template_type): + + errors = [] + + if any(recipients.rows_with_bad_recipients): + number_of_bad_recipients = len(list(recipients.rows_with_bad_recipients)) + if 'sms' == template_type: + if 1 == number_of_bad_recipients: + errors.append("fix 1 phone number") + else: + errors.append("fix {} phone numbers".format(number_of_bad_recipients)) + elif 'email' == template_type: + if 1 == number_of_bad_recipients: + errors.append("fix 1 email address") + else: + errors.append("fix {} email addresses".format(number_of_bad_recipients)) + elif 'letter' == template_type: + if 1 == number_of_bad_recipients: + errors.append("fix 1 address") + else: + errors.append("fix {} addresses".format(number_of_bad_recipients)) + + if any(recipients.rows_with_missing_data): + number_of_rows_with_missing_data = len(list(recipients.rows_with_missing_data)) + if 1 == number_of_rows_with_missing_data: + errors.append("enter missing data in 1 row") + else: + errors.append("enter missing data in {} rows".format(number_of_rows_with_missing_data)) + + if any(recipients.rows_with_message_too_long): + number_of_rows_with_message_too_long = len(list(recipients.rows_with_message_too_long)) + if 1 == number_of_rows_with_message_too_long: + errors.append("shorten the message in 1 row") + else: + errors.append("shorten the messages in {} rows".format(number_of_rows_with_message_too_long)) + + if any(recipients.rows_with_empty_message): + number_of_rows_with_empty_message = len(list(recipients.rows_with_empty_message)) + if 1 == number_of_rows_with_empty_message: + errors.append("check you have content for the empty message in 1 row") + else: + errors.append("check you have content for the empty messages in {} rows".format( + number_of_rows_with_empty_message + )) + + return errors + + +def generate_notifications_csv(**kwargs): + from app import notification_api_client + from app.s3_client.s3_csv_client import s3download + if 'page' not in kwargs: + kwargs['page'] = 1 + + if kwargs.get('job_id'): + original_file_contents = s3download(kwargs['service_id'], kwargs['job_id']) + original_upload = RecipientCSV( + original_file_contents, + template=get_sample_template(kwargs['template_type']), + ) + original_column_headers = original_upload.column_headers + fieldnames = ['Row number'] + original_column_headers + ['Template', 'Type', 'Job', 'Status', 'Time'] + else: + fieldnames = ['Recipient', 'Reference', 'Template', 'Type', 'Sent by', 'Sent by email', 'Job', 'Status', 'Time'] + + yield ','.join(fieldnames) + '\n' + + while kwargs['page']: + notifications_resp = notification_api_client.get_notifications_for_service(**kwargs) + for notification in notifications_resp['notifications']: + if kwargs.get('job_id'): + values = [ + notification['row_number'], + ] + [ + original_upload[notification['row_number'] - 1].get(header).data + for header in original_column_headers + ] + [ + notification['template_name'], + notification['template_type'], + notification['job_name'], + notification['status'], + notification['created_at'], + ] + else: + values = [ + # the recipient for precompiled letters is the full address block + notification['recipient'].splitlines()[0].lstrip().rstrip(' ,'), + notification['client_reference'], + notification['template_name'], + notification['template_type'], + notification['created_by_name'] or '', + notification['created_by_email_address'] or '', + notification['job_name'] or '', + notification['status'], + notification['created_at'] + ] + yield Spreadsheet.from_rows([map(str, values)]).as_csv_data + + if notifications_resp['links'].get('next'): + kwargs['page'] += 1 + else: + return + raise Exception("Should never reach here") diff --git a/app/utils/templates.py b/app/utils/templates.py new file mode 100644 index 000000000..607a04fb9 --- /dev/null +++ b/app/utils/templates.py @@ -0,0 +1,70 @@ +from flask import current_app +from notifications_utils.template import ( + BroadcastPreviewTemplate, + EmailPreviewTemplate, + LetterImageTemplate, + LetterPreviewTemplate, + SMSPreviewTemplate, +) + + +def get_sample_template(template_type): + if template_type == 'email': + return EmailPreviewTemplate({'content': 'any', 'subject': '', 'template_type': 'email'}) + if template_type == 'sms': + return SMSPreviewTemplate({'content': 'any', 'template_type': 'sms'}) + if template_type == 'letter': + return LetterImageTemplate( + {'content': 'any', 'subject': '', 'template_type': 'letter'}, postage='second', image_url='x', page_count=1 + ) + + +def get_template( + template, + service, + show_recipient=False, + letter_preview_url=None, + page_count=1, + redact_missing_personalisation=False, + email_reply_to=None, + sms_sender=None, +): + if 'email' == template['template_type']: + return EmailPreviewTemplate( + template, + from_name=service.name, + from_address='{}@notifications.service.gov.uk'.format(service.email_from), + show_recipient=show_recipient, + redact_missing_personalisation=redact_missing_personalisation, + reply_to=email_reply_to, + ) + if 'sms' == template['template_type']: + return SMSPreviewTemplate( + template, + prefix=service.name, + show_prefix=service.prefix_sms, + sender=sms_sender, + show_sender=bool(sms_sender), + show_recipient=show_recipient, + redact_missing_personalisation=redact_missing_personalisation, + ) + if 'letter' == template['template_type']: + if letter_preview_url: + return LetterImageTemplate( + template, + image_url=letter_preview_url, + page_count=int(page_count), + contact_block=template['reply_to_text'], + postage=template['postage'], + ) + else: + return LetterPreviewTemplate( + template, + contact_block=template['reply_to_text'], + admin_base_url=current_app.config['ADMIN_BASE_URL'], + redact_missing_personalisation=redact_missing_personalisation, + ) + if 'broadcast' == template['template_type']: + return BroadcastPreviewTemplate( + template, + ) diff --git a/tests/app/main/test_errors_for_csv.py b/tests/app/main/test_errors_for_csv.py index d455c54b1..9fc7b94fb 100644 --- a/tests/app/main/test_errors_for_csv.py +++ b/tests/app/main/test_errors_for_csv.py @@ -2,7 +2,7 @@ from collections import namedtuple import pytest -from app.utils import get_errors_for_csv +from app.utils.csv import get_errors_for_csv MockRecipients = namedtuple( 'RecipientCSV', diff --git a/tests/app/test_utils.py b/tests/app/test_utils.py index 320c8f144..dd0b8da74 100644 --- a/tests/app/test_utils.py +++ b/tests/app/test_utils.py @@ -1,92 +1,16 @@ -from collections import OrderedDict -from csv import DictReader -from io import StringIO -from pathlib import Path - import pytest from freezegun import freeze_time -from notifications_utils.template import Template from app import format_datetime_relative from app.formatters import email_safe, round_to_significant_figures from app.utils import ( - Spreadsheet, generate_next_dict, - generate_notifications_csv, generate_previous_dict, get_current_financial_year, get_logo_cdn_domain, - get_sample_template, is_less_than_days_ago, merge_jsonlike, ) -from tests.conftest import fake_uuid - - -def _get_notifications_csv( - row_number=1, - recipient='foo@bar.com', - template_name='foo', - template_type='sms', - job_name='bar.csv', - status='Delivered', - created_at='1943-04-19 12:00:00', - rows=1, - with_links=False, - job_id=fake_uuid, - created_by_name=None, - created_by_email_address=None, -): - - def _get( - service_id, - page=1, - job_id=None, - template_type=template_type, - ): - links = {} - if with_links: - links = { - 'prev': '/service/{}/notifications?page=0'.format(service_id), - 'next': '/service/{}/notifications?page=1'.format(service_id), - 'last': '/service/{}/notifications?page=2'.format(service_id) - } - - data = { - 'notifications': [{ - "row_number": row_number + i, - "to": recipient, - "recipient": recipient, - "client_reference": 'ref 1234', - "template_name": template_name, - "template_type": template_type, - "template": {"name": template_name, "template_type": template_type}, - "job_name": job_name, - "status": status, - "created_at": created_at, - "updated_at": None, - "created_by_name": created_by_name, - "created_by_email_address": created_by_email_address, - } for i in range(rows)], - 'total': rows, - 'page_size': 50, - 'links': links - } - - return data - - return _get - - -@pytest.fixture(scope='function') -def _get_notifications_csv_mock( - mocker, - api_user_active, -): - return mocker.patch( - 'app.notification_api_client.get_notifications_for_service', - side_effect=_get_notifications_csv() - ) @pytest.mark.parametrize('service_name, safe_email', [ @@ -124,179 +48,6 @@ def test_generate_previous_next_dict_adds_other_url_args(client): assert 'notifications/blah' in ret['url'] -def test_can_create_spreadsheet_from_large_excel_file(): - with open(str(Path.cwd() / 'tests' / 'spreadsheet_files' / 'excel 2007.xlsx'), 'rb') as xl: - ret = Spreadsheet.from_file(xl, filename='xl.xlsx') - assert ret.as_csv_data - - -def test_can_create_spreadsheet_from_dict(): - assert Spreadsheet.from_dict(OrderedDict( - foo='bar', - name='Jane', - )).as_csv_data == ( - "foo,name\r\n" - "bar,Jane\r\n" - ) - - -def test_can_create_spreadsheet_from_dict_with_filename(): - assert Spreadsheet.from_dict({}, filename='empty.csv').as_dict['file_name'] == "empty.csv" - - -@pytest.mark.parametrize('args, kwargs', ( - ( - ('hello', ['hello']), - {}, - ), - ( - (), - {'csv_data': 'hello', 'rows': ['hello']} - ), -)) -def test_spreadsheet_checks_for_bad_arguments(args, kwargs): - with pytest.raises(TypeError) as exception: - Spreadsheet(*args, **kwargs) - assert str(exception.value) == 'Spreadsheet must be created from either rows or CSV data' - - -@pytest.mark.parametrize('created_by_name, expected_content', [ - ( - None, [ - 'Recipient,Reference,Template,Type,Sent by,Sent by email,Job,Status,Time\n', - 'foo@bar.com,ref 1234,foo,sms,,sender@email.gov.uk,,Delivered,1943-04-19 12:00:00\r\n', - ] - ), - ( - 'Anne Example', [ - 'Recipient,Reference,Template,Type,Sent by,Sent by email,Job,Status,Time\n', - 'foo@bar.com,ref 1234,foo,sms,Anne Example,sender@email.gov.uk,,Delivered,1943-04-19 12:00:00\r\n', - ] - ), -]) -def test_generate_notifications_csv_without_job( - notify_admin, - mocker, - created_by_name, - expected_content, -): - mocker.patch( - 'app.notification_api_client.get_notifications_for_service', - side_effect=_get_notifications_csv( - created_by_name=created_by_name, - created_by_email_address="sender@email.gov.uk", - job_id=None, - job_name=None - ) - ) - assert list(generate_notifications_csv(service_id=fake_uuid)) == expected_content - - -@pytest.mark.parametrize('original_file_contents, expected_column_headers, expected_1st_row', [ - ( - """ - phone_number - 07700900123 - """, - ['Row number', 'phone_number', 'Template', 'Type', 'Job', 'Status', 'Time'], - ['1', '07700900123', 'foo', 'sms', 'bar.csv', 'Delivered', '1943-04-19 12:00:00'], - ), - ( - """ - phone_number, a, b, c - 07700900123, 🐜,🐝,🦀 - """, - ['Row number', 'phone_number', 'a', 'b', 'c', 'Template', 'Type', 'Job', 'Status', 'Time'], - ['1', '07700900123', '🐜', '🐝', '🦀', 'foo', 'sms', 'bar.csv', 'Delivered', '1943-04-19 12:00:00'], - ), - ( - """ - "phone_number", "a", "b", "c" - "07700900123","🐜,🐜","🐝,🐝","🦀" - """, - ['Row number', 'phone_number', 'a', 'b', 'c', 'Template', 'Type', 'Job', 'Status', 'Time'], - ['1', '07700900123', '🐜,🐜', '🐝,🐝', '🦀', 'foo', 'sms', 'bar.csv', 'Delivered', '1943-04-19 12:00:00'], - ), -]) -def test_generate_notifications_csv_returns_correct_csv_file( - notify_admin, - mocker, - _get_notifications_csv_mock, - original_file_contents, - expected_column_headers, - expected_1st_row, -): - mocker.patch( - 'app.s3_client.s3_csv_client.s3download', - return_value=original_file_contents, - ) - csv_content = generate_notifications_csv(service_id='1234', job_id=fake_uuid, template_type='sms') - csv_file = DictReader(StringIO('\n'.join(csv_content))) - assert csv_file.fieldnames == expected_column_headers - assert next(csv_file) == dict(zip(expected_column_headers, expected_1st_row)) - - -def test_generate_notifications_csv_only_calls_once_if_no_next_link( - notify_admin, - _get_notifications_csv_mock, -): - list(generate_notifications_csv(service_id='1234')) - - assert _get_notifications_csv_mock.call_count == 1 - - -@pytest.mark.parametrize("job_id", ["some", None]) -def test_generate_notifications_csv_calls_twice_if_next_link( - notify_admin, - mocker, - job_id, -): - - mocker.patch( - 'app.s3_client.s3_csv_client.s3download', - return_value=""" - phone_number - 07700900000 - 07700900001 - 07700900002 - 07700900003 - 07700900004 - 07700900005 - 07700900006 - 07700900007 - 07700900008 - 07700900009 - """ - ) - - service_id = '1234' - response_with_links = _get_notifications_csv(rows=7, with_links=True) - response_with_no_links = _get_notifications_csv(rows=3, row_number=8, with_links=False) - - mock_get_notifications = mocker.patch( - 'app.notification_api_client.get_notifications_for_service', - side_effect=[ - response_with_links(service_id), - response_with_no_links(service_id), - ] - ) - - csv_content = generate_notifications_csv( - service_id=service_id, - job_id=job_id or fake_uuid, - template_type='sms', - ) - csv = list(DictReader(StringIO('\n'.join(csv_content)))) - - assert len(csv) == 10 - assert csv[0]['phone_number'] == '07700900000' - assert csv[9]['phone_number'] == '07700900009' - assert mock_get_notifications.call_count == 2 - # mock_calls[0][2] is the kwargs from first call - assert mock_get_notifications.mock_calls[0][2]['page'] == 1 - assert mock_get_notifications.mock_calls[1][2]['page'] == 2 - - def test_get_cdn_domain_on_localhost(client, mocker): mocker.patch.dict('app.current_app.config', values={'ADMIN_BASE_URL': 'http://localhost:6012'}) domain = get_logo_cdn_domain() @@ -359,12 +110,6 @@ def test_is_less_than_days_ago(date_from_db, expected_result): assert is_less_than_days_ago(date_from_db, 90) == expected_result -@pytest.mark.parametrize("template_type", ["sms", "letter", "email"]) -def test_get_sample_template_returns_template(template_type): - template = get_sample_template(template_type) - assert isinstance(template, Template) - - @pytest.mark.parametrize("source_object, destination_object, expected_result", [ # simple dicts: ({"a": "b"}, {"c": "d"}, {"a": "b", "c": "d"}), diff --git a/tests/app/utils/test_csv.py b/tests/app/utils/test_csv.py new file mode 100644 index 000000000..cbae49dc3 --- /dev/null +++ b/tests/app/utils/test_csv.py @@ -0,0 +1,248 @@ +from collections import OrderedDict +from csv import DictReader +from io import StringIO +from pathlib import Path + +import pytest + +from app.utils.csv import Spreadsheet, generate_notifications_csv +from tests.conftest import fake_uuid + + +def _get_notifications_csv( + row_number=1, + recipient='foo@bar.com', + template_name='foo', + template_type='sms', + job_name='bar.csv', + status='Delivered', + created_at='1943-04-19 12:00:00', + rows=1, + with_links=False, + job_id=fake_uuid, + created_by_name=None, + created_by_email_address=None, +): + + def _get( + service_id, + page=1, + job_id=None, + template_type=template_type, + ): + links = {} + if with_links: + links = { + 'prev': '/service/{}/notifications?page=0'.format(service_id), + 'next': '/service/{}/notifications?page=1'.format(service_id), + 'last': '/service/{}/notifications?page=2'.format(service_id) + } + + data = { + 'notifications': [{ + "row_number": row_number + i, + "to": recipient, + "recipient": recipient, + "client_reference": 'ref 1234', + "template_name": template_name, + "template_type": template_type, + "template": {"name": template_name, "template_type": template_type}, + "job_name": job_name, + "status": status, + "created_at": created_at, + "updated_at": None, + "created_by_name": created_by_name, + "created_by_email_address": created_by_email_address, + } for i in range(rows)], + 'total': rows, + 'page_size': 50, + 'links': links + } + + return data + + return _get + + +@pytest.fixture(scope='function') +def _get_notifications_csv_mock( + mocker, + api_user_active, +): + return mocker.patch( + 'app.notification_api_client.get_notifications_for_service', + side_effect=_get_notifications_csv() + ) + + +def test_can_create_spreadsheet_from_large_excel_file(): + with open(str(Path.cwd() / 'tests' / 'spreadsheet_files' / 'excel 2007.xlsx'), 'rb') as xl: + ret = Spreadsheet.from_file(xl, filename='xl.xlsx') + assert ret.as_csv_data + + +def test_can_create_spreadsheet_from_dict(): + assert Spreadsheet.from_dict(OrderedDict( + foo='bar', + name='Jane', + )).as_csv_data == ( + "foo,name\r\n" + "bar,Jane\r\n" + ) + + +def test_can_create_spreadsheet_from_dict_with_filename(): + assert Spreadsheet.from_dict({}, filename='empty.csv').as_dict['file_name'] == "empty.csv" + + +@pytest.mark.parametrize('args, kwargs', ( + ( + ('hello', ['hello']), + {}, + ), + ( + (), + {'csv_data': 'hello', 'rows': ['hello']} + ), +)) +def test_spreadsheet_checks_for_bad_arguments(args, kwargs): + with pytest.raises(TypeError) as exception: + Spreadsheet(*args, **kwargs) + assert str(exception.value) == 'Spreadsheet must be created from either rows or CSV data' + + +@pytest.mark.parametrize('created_by_name, expected_content', [ + ( + None, [ + 'Recipient,Reference,Template,Type,Sent by,Sent by email,Job,Status,Time\n', + 'foo@bar.com,ref 1234,foo,sms,,sender@email.gov.uk,,Delivered,1943-04-19 12:00:00\r\n', + ] + ), + ( + 'Anne Example', [ + 'Recipient,Reference,Template,Type,Sent by,Sent by email,Job,Status,Time\n', + 'foo@bar.com,ref 1234,foo,sms,Anne Example,sender@email.gov.uk,,Delivered,1943-04-19 12:00:00\r\n', + ] + ), +]) +def test_generate_notifications_csv_without_job( + notify_admin, + mocker, + created_by_name, + expected_content, +): + mocker.patch( + 'app.notification_api_client.get_notifications_for_service', + side_effect=_get_notifications_csv( + created_by_name=created_by_name, + created_by_email_address="sender@email.gov.uk", + job_id=None, + job_name=None + ) + ) + assert list(generate_notifications_csv(service_id=fake_uuid)) == expected_content + + +@pytest.mark.parametrize('original_file_contents, expected_column_headers, expected_1st_row', [ + ( + """ + phone_number + 07700900123 + """, + ['Row number', 'phone_number', 'Template', 'Type', 'Job', 'Status', 'Time'], + ['1', '07700900123', 'foo', 'sms', 'bar.csv', 'Delivered', '1943-04-19 12:00:00'], + ), + ( + """ + phone_number, a, b, c + 07700900123, 🐜,🐝,🦀 + """, + ['Row number', 'phone_number', 'a', 'b', 'c', 'Template', 'Type', 'Job', 'Status', 'Time'], + ['1', '07700900123', '🐜', '🐝', '🦀', 'foo', 'sms', 'bar.csv', 'Delivered', '1943-04-19 12:00:00'], + ), + ( + """ + "phone_number", "a", "b", "c" + "07700900123","🐜,🐜","🐝,🐝","🦀" + """, + ['Row number', 'phone_number', 'a', 'b', 'c', 'Template', 'Type', 'Job', 'Status', 'Time'], + ['1', '07700900123', '🐜,🐜', '🐝,🐝', '🦀', 'foo', 'sms', 'bar.csv', 'Delivered', '1943-04-19 12:00:00'], + ), +]) +def test_generate_notifications_csv_returns_correct_csv_file( + notify_admin, + mocker, + _get_notifications_csv_mock, + original_file_contents, + expected_column_headers, + expected_1st_row, +): + mocker.patch( + 'app.s3_client.s3_csv_client.s3download', + return_value=original_file_contents, + ) + csv_content = generate_notifications_csv(service_id='1234', job_id=fake_uuid, template_type='sms') + csv_file = DictReader(StringIO('\n'.join(csv_content))) + assert csv_file.fieldnames == expected_column_headers + assert next(csv_file) == dict(zip(expected_column_headers, expected_1st_row)) + + +def test_generate_notifications_csv_only_calls_once_if_no_next_link( + notify_admin, + _get_notifications_csv_mock, +): + list(generate_notifications_csv(service_id='1234')) + + assert _get_notifications_csv_mock.call_count == 1 + + +@pytest.mark.parametrize("job_id", ["some", None]) +def test_generate_notifications_csv_calls_twice_if_next_link( + notify_admin, + mocker, + job_id, +): + + mocker.patch( + 'app.s3_client.s3_csv_client.s3download', + return_value=""" + phone_number + 07700900000 + 07700900001 + 07700900002 + 07700900003 + 07700900004 + 07700900005 + 07700900006 + 07700900007 + 07700900008 + 07700900009 + """ + ) + + service_id = '1234' + response_with_links = _get_notifications_csv(rows=7, with_links=True) + response_with_no_links = _get_notifications_csv(rows=3, row_number=8, with_links=False) + + mock_get_notifications = mocker.patch( + 'app.notification_api_client.get_notifications_for_service', + side_effect=[ + response_with_links(service_id), + response_with_no_links(service_id), + ] + ) + + csv_content = generate_notifications_csv( + service_id=service_id, + job_id=job_id or fake_uuid, + template_type='sms', + ) + csv = list(DictReader(StringIO('\n'.join(csv_content)))) + + assert len(csv) == 10 + assert csv[0]['phone_number'] == '07700900000' + assert csv[9]['phone_number'] == '07700900009' + assert mock_get_notifications.call_count == 2 + # mock_calls[0][2] is the kwargs from first call + assert mock_get_notifications.mock_calls[0][2]['page'] == 1 + assert mock_get_notifications.mock_calls[1][2]['page'] == 2 diff --git a/tests/app/utils/test_templates.py b/tests/app/utils/test_templates.py new file mode 100644 index 000000000..6b4ddac3c --- /dev/null +++ b/tests/app/utils/test_templates.py @@ -0,0 +1,10 @@ +import pytest +from notifications_utils.template import Template + +from app.utils.templates import get_sample_template + + +@pytest.mark.parametrize("template_type", ["sms", "letter", "email"]) +def test_get_sample_template_returns_template(template_type): + template = get_sample_template(template_type) + assert isinstance(template, Template)