diff --git a/app/main/validators.py b/app/main/validators.py index 081556986..9d09fbe0c 100644 --- a/app/main/validators.py +++ b/app/main/validators.py @@ -12,7 +12,7 @@ from wtforms import ValidationError from app.main._commonly_used_passwords import commonly_used_passwords from app.models.spreadsheet import Spreadsheet -from app.utils import is_gov_user +from app.utils.user import is_gov_user class CommonlyUsedPassword: diff --git a/app/main/views/add_service.py b/app/main/views/add_service.py index 9802370c2..4b968c46d 100644 --- a/app/main/views/add_service.py +++ b/app/main/views/add_service.py @@ -6,7 +6,7 @@ from app import service_api_client from app.formatters import email_safe from app.main import main from app.main.forms import CreateNhsServiceForm, CreateServiceForm -from app.utils import user_is_gov_user, user_is_logged_in +from app.utils.user import user_is_gov_user, user_is_logged_in def _create_service(service_name, organisation_type, email_from, form): diff --git a/app/main/views/agreement.py b/app/main/views/agreement.py index 6fccac17b..930185924 100644 --- a/app/main/views/agreement.py +++ b/app/main/views/agreement.py @@ -8,7 +8,7 @@ from app.main import main from app.main.forms import AcceptAgreementForm from app.models.organisation import Organisation from app.s3_client.s3_mou_client import get_mou -from app.utils import user_has_permissions +from app.utils.user import user_has_permissions @main.route('/services//agreement') diff --git a/app/main/views/api_keys.py b/app/main/views/api_keys.py index 6df188b1e..840a753b4 100644 --- a/app/main/views/api_keys.py +++ b/app/main/views/api_keys.py @@ -23,7 +23,7 @@ from app.notify_client.api_key_api_client import ( KEY_TYPE_TEAM, KEY_TYPE_TEST, ) -from app.utils import user_has_permissions +from app.utils.user import user_has_permissions dummy_bearer_token = 'bearer_token_set' diff --git a/app/main/views/broadcast.py b/app/main/views/broadcast.py index cae93e45f..1ad578a58 100644 --- a/app/main/views/broadcast.py +++ b/app/main/views/broadcast.py @@ -19,7 +19,8 @@ from app.main.forms import ( SearchByNameForm, ) from app.models.broadcast_message import BroadcastMessage, BroadcastMessages -from app.utils import service_has_permission, user_has_permissions +from app.utils import service_has_permission +from app.utils.user import user_has_permissions def _get_back_link_from_view_broadcast_endpoint(): diff --git a/app/main/views/choose_account.py b/app/main/views/choose_account.py index 3626a66e6..1c26a0b8f 100644 --- a/app/main/views/choose_account.py +++ b/app/main/views/choose_account.py @@ -4,7 +4,8 @@ from flask_login import current_user from app import status_api_client from app.main import main from app.models.organisation import Organisations -from app.utils import PermanentRedirect, user_is_logged_in +from app.utils import PermanentRedirect +from app.utils.user import user_is_logged_in @main.route("/services") diff --git a/app/main/views/conversation.py b/app/main/views/conversation.py index a858688e7..6e7fd257f 100644 --- a/app/main/views/conversation.py +++ b/app/main/views/conversation.py @@ -8,7 +8,7 @@ from app import current_service, notification_api_client, service_api_client from app.main import main from app.main.forms import SearchByNameForm from app.models.template_list import TemplateList -from app.utils import user_has_permissions +from app.utils.user import user_has_permissions @main.route("/services//conversation/") diff --git a/app/main/views/dashboard.py b/app/main/views/dashboard.py index ab940050f..fb8b56ae8 100644 --- a/app/main/views/dashboard.py +++ b/app/main/views/dashboard.py @@ -30,13 +30,13 @@ from app.utils import ( DELIVERED_STATUSES, FAILURE_STATUSES, REQUESTED_STATUSES, - Spreadsheet, generate_next_dict, generate_previous_dict, get_current_financial_year, service_has_permission, - user_has_permissions, ) +from app.utils.csv import Spreadsheet +from app.utils.user import user_has_permissions @main.route("/services//dashboard") diff --git a/app/main/views/email_branding.py b/app/main/views/email_branding.py index c32227242..c3b5b10ec 100644 --- a/app/main/views/email_branding.py +++ b/app/main/views/email_branding.py @@ -11,7 +11,8 @@ from app.s3_client.s3_logo_client import ( persist_logo, upload_email_logo, ) -from app.utils import get_logo_cdn_domain, user_is_platform_admin +from app.utils import get_logo_cdn_domain +from app.utils.user import user_is_platform_admin @main.route("/email-branding", methods=['GET', 'POST']) diff --git a/app/main/views/find_services.py b/app/main/views/find_services.py index ff841e488..8156c7f3b 100644 --- a/app/main/views/find_services.py +++ b/app/main/views/find_services.py @@ -6,7 +6,7 @@ from flask import redirect, render_template, url_for from app import service_api_client from app.main import main from app.main.forms import SearchByNameForm -from app.utils import user_is_platform_admin +from app.utils.user import user_is_platform_admin @main.route("/find-services-by-name", methods=['GET', 'POST']) diff --git a/app/main/views/find_users.py b/app/main/views/find_users.py index 10850cf9f..00c8bdac2 100644 --- a/app/main/views/find_users.py +++ b/app/main/views/find_users.py @@ -7,7 +7,7 @@ from app.event_handlers import create_archive_user_event from app.main import main from app.main.forms import SearchUsersByEmailForm from app.models.user import User -from app.utils import user_is_platform_admin +from app.utils.user import user_is_platform_admin @main.route("/find-users-by-email", methods=['GET', 'POST']) diff --git a/app/main/views/history.py b/app/main/views/history.py index 1be4f1386..6c0603a7c 100644 --- a/app/main/views/history.py +++ b/app/main/views/history.py @@ -6,7 +6,7 @@ from flask import render_template, request from app import current_service, format_date_numeric from app.main import main from app.models.event import APIKeyEvent, APIKeyEvents, ServiceEvents -from app.utils import user_has_permissions +from app.utils.user import user_has_permissions @main.route("/services//history") diff --git a/app/main/views/inbound_number.py b/app/main/views/inbound_number.py index 3007db9e5..c56620b33 100644 --- a/app/main/views/inbound_number.py +++ b/app/main/views/inbound_number.py @@ -2,7 +2,7 @@ from flask import render_template from app import inbound_number_client from app.main import main -from app.utils import user_is_platform_admin +from app.utils.user import user_is_platform_admin @main.route('/inbound-sms-admin', methods=['GET', 'POST']) diff --git a/app/main/views/jobs.py b/app/main/views/jobs.py index d6009c8b8..66952030f 100644 --- a/app/main/views/jobs.py +++ b/app/main/views/jobs.py @@ -35,15 +35,17 @@ 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_letter_printing_statement, get_page_from_request, parse_filter_args, - printing_today_or_tomorrow, set_status_filters, - user_has_permissions, ) +from app.utils.csv import generate_notifications_csv +from app.utils.letters import ( + get_letter_printing_statement, + printing_today_or_tomorrow, +) +from app.utils.user import user_has_permissions @main.route("/services//jobs") diff --git a/app/main/views/letter_branding.py b/app/main/views/letter_branding.py index 76c694e30..571581ec9 100644 --- a/app/main/views/letter_branding.py +++ b/app/main/views/letter_branding.py @@ -25,7 +25,8 @@ from app.s3_client.s3_logo_client import ( persist_logo, upload_letter_temp_logo, ) -from app.utils import get_logo_cdn_domain, user_is_platform_admin +from app.utils import get_logo_cdn_domain +from app.utils.user import user_is_platform_admin @main.route("/letter-branding", methods=['GET']) diff --git a/app/main/views/manage_users.py b/app/main/views/manage_users.py index 8140288bc..3a75f1359 100644 --- a/app/main/views/manage_users.py +++ b/app/main/views/manage_users.py @@ -30,7 +30,7 @@ from app.main.forms import ( ) from app.models.roles_and_permissions import broadcast_permissions, permissions from app.models.user import InvitedUser, User -from app.utils import is_gov_user, user_has_permissions +from app.utils.user import is_gov_user, user_has_permissions @main.route("/services//users") diff --git a/app/main/views/notifications.py b/app/main/views/notifications.py index aac38a026..77113cb42 100644 --- a/app/main/views/notifications.py +++ b/app/main/views/notifications.py @@ -37,15 +37,17 @@ 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_letter_printing_statement, - get_letter_validation_error, - get_template, parse_filter_args, set_status_filters, - user_has_permissions, ) +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 @main.route("/services//notification/") diff --git a/app/main/views/organisations.py b/app/main/views/organisations.py index 0d8ae2a5c..12405cbd8 100644 --- a/app/main/views/organisations.py +++ b/app/main/views/organisations.py @@ -43,7 +43,7 @@ from app.main.views.dashboard import ( from app.main.views.service_settings import get_branding_as_value_and_label from app.models.organisation import Organisation, Organisations from app.models.user import InvitedOrgUser, User -from app.utils import user_has_permissions, user_is_platform_admin +from app.utils.user import user_has_permissions, user_is_platform_admin @main.route("/organisations", methods=['GET']) diff --git a/app/main/views/platform_admin.py b/app/main/views/platform_admin.py index 6f69de299..5ce4dd937 100644 --- a/app/main/views/platform_admin.py +++ b/app/main/views/platform_admin.py @@ -28,12 +28,12 @@ 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, - user_is_platform_admin, ) +from app.utils.csv import Spreadsheet +from app.utils.user import user_is_platform_admin COMPLAINT_THRESHOLD = 0.02 FAILURE_THRESHOLD = 3 diff --git a/app/main/views/providers.py b/app/main/views/providers.py index 0484f4463..977936cc9 100644 --- a/app/main/views/providers.py +++ b/app/main/views/providers.py @@ -8,7 +8,7 @@ from werkzeug.utils import redirect from app import format_date_numeric, provider_client from app.main import main from app.main.forms import ProviderForm, ProviderRatioForm -from app.utils import user_is_platform_admin +from app.utils.user import user_is_platform_admin PROVIDER_PRIORITY_MEANING_SWITCHOVER = datetime(2019, 11, 29, 11, 0).isoformat() diff --git a/app/main/views/returned_letters.py b/app/main/views/returned_letters.py index d4861733a..ba222a066 100644 --- a/app/main/views/returned_letters.py +++ b/app/main/views/returned_letters.py @@ -5,7 +5,7 @@ from flask import render_template from app import current_service, service_api_client from app.main import main from app.models.spreadsheet import Spreadsheet -from app.utils import user_has_permissions +from app.utils.user import user_has_permissions @main.route("/services//returned-letters") diff --git a/app/main/views/send.py b/app/main/views/send.py index 50ab717a7..6dd1c83e2 100644 --- a/app/main/views/send.py +++ b/app/main/views/send.py @@ -52,13 +52,12 @@ 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, - user_has_permissions, ) +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 = [ column.replace('_', ' ') diff --git a/app/main/views/service_settings.py b/app/main/views/service_settings.py index fd45e8ebc..d9251cfca 100644 --- a/app/main/views/service_settings.py +++ b/app/main/views/service_settings.py @@ -62,10 +62,8 @@ from app.main.forms import ( SetLetterBranding, SMSPrefixForm, ) -from app.utils import ( - DELIVERED_STATUSES, - FAILURE_STATUSES, - SENDING_STATUSES, +from app.utils import DELIVERED_STATUSES, FAILURE_STATUSES, SENDING_STATUSES +from app.utils.user import ( user_has_permissions, user_is_gov_user, user_is_platform_admin, diff --git a/app/main/views/templates.py b/app/main/views/templates.py index 2e69e29bb..a389af900 100644 --- a/app/main/views/templates.py +++ b/app/main/views/templates.py @@ -39,13 +39,9 @@ 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, - user_has_permissions, - user_is_platform_admin, -) +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 = { 'email': EmailTemplateForm, diff --git a/app/main/views/tour.py b/app/main/views/tour.py index d6223a0e8..0cf96b885 100644 --- a/app/main/views/tour.py +++ b/app/main/views/tour.py @@ -9,7 +9,8 @@ from app.main.views.send import ( get_placeholder_form_instance, get_recipient_and_placeholders_from_session, ) -from app.utils import get_template, user_has_permissions +from app.utils.templates import get_template +from app.utils.user import user_has_permissions @main.route("/services//tour/") diff --git a/app/main/views/uploads.py b/app/main/views/uploads.py index b2609bde6..628009e6c 100644 --- a/app/main/views/uploads.py +++ b/app/main/views/uploads.py @@ -46,18 +46,18 @@ 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, + 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, - get_page_from_request, - get_sample_template, - get_template, - unicode_truncate, - user_has_permissions, ) +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/main/views/user_profile.py b/app/main/views/user_profile.py index c96f306ef..03ace2424 100644 --- a/app/main/views/user_profile.py +++ b/app/main/views/user_profile.py @@ -27,7 +27,7 @@ from app.main.forms import ( TwoFactorForm, ) from app.models.user import User -from app.utils import ( +from app.utils.user import ( user_is_gov_user, user_is_logged_in, user_is_platform_admin, diff --git a/app/main/views/webauthn_credentials.py b/app/main/views/webauthn_credentials.py index 75ef6bb8d..d1b18a6d2 100644 --- a/app/main/views/webauthn_credentials.py +++ b/app/main/views/webauthn_credentials.py @@ -10,11 +10,8 @@ from app.main.views.two_factor import log_in_user from app.models.user import User from app.models.webauthn_credential import RegistrationError, WebAuthnCredential from app.notify_client.user_api_client import user_api_client -from app.utils import ( - is_less_than_days_ago, - redirect_to_sign_in, - user_is_platform_admin, -) +from app.utils import is_less_than_days_ago, redirect_to_sign_in +from app.utils.user import user_is_platform_admin @main.route('/webauthn/register') 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/models/job.py b/app/models/job.py index 9c3f916a5..89597a5e5 100644 --- a/app/models/job.py +++ b/app/models/job.py @@ -13,11 +13,8 @@ from app.models import JSONModel, ModelList, PaginatedModelList from app.notify_client.job_api_client import job_api_client from app.notify_client.notification_api_client import notification_api_client from app.notify_client.service_api_client import service_api_client -from app.utils import ( - get_letter_printing_statement, - is_less_than_days_ago, - set_status_filters, -) +from app.utils import is_less_than_days_ago, set_status_filters +from app.utils.letters import get_letter_printing_statement class Job(JSONModel): diff --git a/app/models/user.py b/app/models/user.py index 2f534fe1a..0d95e1624 100644 --- a/app/models/user.py +++ b/app/models/user.py @@ -15,7 +15,7 @@ from app.notify_client import InviteTokenError from app.notify_client.invite_api_client import invite_api_client from app.notify_client.org_invite_api_client import org_invite_api_client from app.notify_client.user_api_client import user_api_client -from app.utils import is_gov_user +from app.utils.user import is_gov_user def _get_service_id_from_view_args(): diff --git a/app/utils.py b/app/utils.py deleted file mode 100644 index 59f52d585..000000000 --- a/app/utils.py +++ /dev/null @@ -1,674 +0,0 @@ -import os -from datetime import datetime, timedelta -from functools import wraps -from itertools import chain -from urllib.parse import urlparse - -import pytz -from dateutil import parser -from flask import ( - abort, - current_app, - g, - make_response, - redirect, - request, - session, - url_for, -) -from flask_login import current_user, login_required -from notifications_utils.field import Field -from notifications_utils.formatters import unescaped_formatted_list -from notifications_utils.letter_timings import letter_can_be_cancelled -from notifications_utils.postal_address import PostalAddress -from notifications_utils.recipients import RecipientCSV -from notifications_utils.template import ( - BroadcastPreviewTemplate, - EmailPreviewTemplate, - LetterImageTemplate, - LetterPreviewTemplate, - SMSPreviewTemplate, -) -from notifications_utils.timezones import ( - convert_bst_to_utc, - convert_utc_to_bst, - 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 -from app.notify_client.organisations_api_client import organisations_client - -SENDING_STATUSES = ['created', 'pending', 'sending', 'pending-virus-check'] -DELIVERED_STATUSES = ['delivered', 'sent', 'returned-letter'] -FAILURE_STATUSES = ['failed', 'temporary-failure', 'permanent-failure', - 'technical-failure', 'virus-scan-failed', 'validation-failed'] -REQUESTED_STATUSES = SENDING_STATUSES + DELIVERED_STATUSES + FAILURE_STATUSES - -NOTIFICATION_TYPES = ["sms", "email", "letter", "broadcast"] - - -with open('{}/email_domains.txt'.format( - os.path.dirname(os.path.realpath(__file__)) -)) as email_domains: - GOVERNMENT_EMAIL_DOMAIN_NAMES = [line.strip() for line in email_domains] - - -user_is_logged_in = login_required - - -def user_has_permissions(*permissions, **permission_kwargs): - def wrap(func): - @wraps(func) - def wrap_func(*args, **kwargs): - if not current_user.is_authenticated: - return current_app.login_manager.unauthorized() - if not current_user.has_permissions(*permissions, **permission_kwargs): - abort(403) - return func(*args, **kwargs) - return wrap_func - return wrap - - -def service_has_permission(permission): - - from app import current_service - - def wrap(func): - @wraps(func) - def wrap_func(*args, **kwargs): - if not current_service or not current_service.has_permission(permission): - abort(403) - return func(*args, **kwargs) - return wrap_func - return wrap - - -def user_is_gov_user(f): - @wraps(f) - def wrapped(*args, **kwargs): - if not current_user.is_authenticated: - return current_app.login_manager.unauthorized() - if not current_user.is_gov_user: - abort(403) - return f(*args, **kwargs) - return wrapped - - -def user_is_platform_admin(f): - @wraps(f) - def wrapped(*args, **kwargs): - if not current_user.is_authenticated: - return current_app.login_manager.unauthorized() - if not current_user.platform_admin: - abort(403) - return f(*args, **kwargs) - return wrapped - - -def redirect_to_sign_in(f): - @wraps(f) - def wrapped(*args, **kwargs): - if 'user_details' not in session: - return redirect(url_for('main.sign_in')) - else: - return f(*args, **kwargs) - 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: - return int(request.args['page']) - except ValueError: - return None - else: - return 1 - - -def generate_previous_dict(view, service_id, page, url_args=None): - return generate_previous_next_dict(view, service_id, page - 1, 'Previous page', url_args or {}) - - -def generate_next_dict(view, service_id, page, url_args=None): - return generate_previous_next_dict(view, service_id, page + 1, 'Next page', url_args or {}) - - -def generate_previous_next_dict(view, service_id, page, title, url_args): - return { - 'url': url_for(view, service_id=service_id, page=page, **url_args), - 'title': title, - 'label': 'page {}'.format(page) - } - - -def get_help_argument(): - return request.args.get('help') if request.args.get('help') in ('1', '2', '3') else None - - -def email_address_ends_with(email_address, known_domains): - return any( - email_address.lower().endswith(( - "@{}".format(known), - ".{}".format(known), - )) - for known in known_domains - ) - - -def is_gov_user(email_address): - return email_address_ends_with( - email_address, GOVERNMENT_EMAIL_DOMAIN_NAMES - ) or email_address_ends_with( - email_address, organisations_client.get_domains() - ) - - -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() - ) - current_month = int(now.strftime('%-m')) - current_year = int(now.strftime('%Y')) - return current_year if current_month > 3 else current_year - 1 - - -def get_logo_cdn_domain(): - parsed_uri = urlparse(current_app.config['ADMIN_BASE_URL']) - - if parsed_uri.netloc.startswith('localhost'): - return 'static-logos.notify.tools' - - subdomain = parsed_uri.hostname.split('.')[0] - domain = parsed_uri.netloc[len(subdomain + '.'):] - - return "static-logos.{}".format(domain) - - -def parse_filter_args(filter_dict): - if not isinstance(filter_dict, MultiDict): - filter_dict = MultiDict(filter_dict) - - return MultiDict( - ( - key, - (','.join(filter_dict.getlist(key))).split(',') - ) - for key in filter_dict.keys() - if ''.join(filter_dict.getlist(key)) - ) - - -def set_status_filters(filter_args): - status_filters = filter_args.get('status', []) - return list(OrderedSet(chain( - (status_filters or REQUESTED_STATUSES), - DELIVERED_STATUSES if 'delivered' in status_filters else [], - SENDING_STATUSES if 'sending' in status_filters else [], - FAILURE_STATUSES if 'failed' in status_filters else [] - ))) - - -def unicode_truncate(s, length): - encoded = s.encode('utf-8')[:length] - return encoded.decode('utf-8', 'ignore') - - -def should_skip_template_page(template_type): - return ( - current_user.has_permissions('send_messages') - and not current_user.has_permissions('manage_templates', 'manage_api_keys') - and template_type != 'letter' - ) - - -def get_default_sms_sender(sms_senders): - return str(next(( - Field(x['sms_sender'], html='escape') - for x in sms_senders if x['is_default'] - ), "None")) - - -def printing_today_or_tomorrow(created_at): - print_cutoff = convert_bst_to_utc( - convert_utc_to_bst(datetime.utcnow()).replace(hour=17, minute=30) - ).replace(tzinfo=pytz.utc) - created_at = utc_string_to_aware_gmt_datetime(created_at) - - if created_at < print_cutoff: - return 'today' - else: - return 'tomorrow' - - -def get_letter_printing_statement(status, created_at, long_form=True): - created_at_dt = parser.parse(created_at).replace(tzinfo=None) - if letter_can_be_cancelled(status, created_at_dt): - decription = 'Printing starts' if long_form else 'Printing' - return f'{decription} {printing_today_or_tomorrow(created_at)} at 5:30pm' - else: - printed_datetime = utc_string_to_aware_gmt_datetime(created_at) + timedelta(hours=6, minutes=30) - if printed_datetime.date() == datetime.now().date(): - return 'Printed today at 5:30pm' - elif printed_datetime.date() == datetime.now().date() - timedelta(days=1): - return 'Printed yesterday at 5:30pm' - - printed_date = printed_datetime.strftime('%d %B').lstrip('0') - description = 'Printed on' if long_form else 'Printed' - - return f'{description} {printed_date} at 5:30pm' - - -LETTER_VALIDATION_MESSAGES = { - 'letter-not-a4-portrait-oriented': { - 'title': 'Your letter is not A4 portrait size', - 'detail': ( - 'You need to change the size or orientation of {invalid_pages}.
' - 'Files must meet our ' - '' - 'letter specification' - '.' - ), - 'summary': ( - 'Validation failed because {invalid_pages} {invalid_pages_are_or_is} not A4 portrait size.
' - 'Files must meet our ' - '' - 'letter specification' - '.' - ), - }, - 'content-outside-printable-area': { - 'title': 'Your content is outside the printable area', - 'detail': ( - 'You need to edit {invalid_pages}.
' - 'Files must meet our ' - '' - 'letter specification' - '.' - ), - 'summary': ( - 'Validation failed because content is outside the printable area on {invalid_pages}.
' - 'Files must meet our ' - '' - 'letter specification' - '.' - ), - }, - 'letter-too-long': { - 'title': 'Your letter is too long', - 'detail': ( - 'Letters must be 10 pages or less (5 double-sided sheets of paper).
' - 'Your letter is {page_count} pages long.' - ), - 'summary': ( - 'Validation failed because this letter is {page_count} pages long.
' - 'Letters must be 10 pages or less (5 double-sided sheets of paper).' - ), - }, - 'no-encoded-string': { - 'title': 'Sanitise failed - No encoded string' - }, - 'unable-to-read-the-file': { - 'title': 'There’s a problem with your file', - 'detail': ( - 'Notify cannot read this PDF.' - '
Save a new copy of your file and try again.' - ), - 'summary': ( - 'Validation failed because Notify cannot read this PDF.
' - 'Save a new copy of your file and try again.' - ), - }, - 'address-is-empty': { - 'title': 'The address block is empty', - 'detail': ( - 'You need to add a recipient address.
' - 'Files must meet our ' - '' - 'letter specification' - '.' - ), - 'summary': ( - 'Validation failed because the address block is empty.
' - 'Files must meet our ' - '' - 'letter specification' - '.' - ), - }, - 'not-a-real-uk-postcode': { - 'title': 'There’s a problem with the address for this letter', - 'detail': ( - 'The last line of the address must be a real UK postcode.' - ), - 'summary': ( - 'Validation failed because the last line of the address is not a real UK postcode.' - ), - }, - 'cant-send-international-letters': { - 'title': 'There’s a problem with the address for this letter', - 'detail': ( - 'You do not have permission to send letters to other countries.' - ), - 'summary': ( - 'Validation failed because your service cannot send letters to other countries.' - ), - }, - 'not-a-real-uk-postcode-or-country': { - 'title': 'There’s a problem with the address for this letter', - 'detail': ( - 'The last line of the address must be a UK postcode or ' - 'another country.' - ), - 'summary': ( - 'Validation failed because the last line of the address is ' - 'not a UK postcode or another country.' - ), - }, - 'not-enough-address-lines': { - 'title': 'There’s a problem with the address for this letter', - 'detail': ( - f'The address must be at least {PostalAddress.MIN_LINES} ' - f'lines long.' - ), - 'summary': ( - f'Validation failed because the address must be at least ' - f'{PostalAddress.MIN_LINES} lines long.' - ), - }, - 'too-many-address-lines': { - 'title': 'There’s a problem with the address for this letter', - 'detail': ( - f'The address must be no more than {PostalAddress.MAX_LINES} ' - f'lines long.' - ), - 'summary': ( - f'Validation failed because the address must be no more ' - f'than {PostalAddress.MAX_LINES} lines long.' - ), - }, - 'invalid-char-in-address': { - 'title': 'There’s a problem with the address for this letter', - 'detail': ( - "Address lines must not start with any of the following characters: @ ( ) = [ ] ” \\ / , < > ~" - ), - 'summary': ( - "Validation failed because address lines must not start with any of the " - "following characters: @ ( ) = [ ] ” \\ / , < > ~" - ), - }, - 'notify-tag-found-in-content': { - 'title': 'There’s a problem with your letter', - 'detail': ( - 'Your file includes a letter you’ve downloaded from Notify.
' - 'You need to edit {invalid_pages}.' - ), - 'summary': ( - 'Validation failed because your file includes a letter ' - 'you’ve downloaded from Notify on {invalid_pages}.' - ) - }, -} - - -def get_letter_validation_error(validation_message, invalid_pages=None, page_count=None): - if not invalid_pages: - invalid_pages = [] - if validation_message not in LETTER_VALIDATION_MESSAGES: - return {'title': 'Validation failed'} - - invalid_pages_are_or_is = 'is' if len(invalid_pages) == 1 else 'are' - - invalid_pages = unescaped_formatted_list( - invalid_pages, - before_each='', - after_each='', - prefix='page', - prefix_plural='pages' - ) - - return { - 'title': LETTER_VALIDATION_MESSAGES[validation_message]['title'], - 'detail': LETTER_VALIDATION_MESSAGES[validation_message]['detail'].format( - invalid_pages=invalid_pages, - invalid_pages_are_or_is=invalid_pages_are_or_is, - page_count=page_count, - letter_spec_guidance=url_for('.letter_specification') - ), - 'summary': LETTER_VALIDATION_MESSAGES[validation_message]['summary'].format( - invalid_pages=invalid_pages, - invalid_pages_are_or_is=invalid_pages_are_or_is, - page_count=page_count, - letter_spec_guidance=url_for('.letter_specification'), - ), - } - - -class PermanentRedirect(RequestRedirect): - """ - In Werkzeug 0.15.0 the status code for RequestRedirect changed from 301 to 308. - 308 status codes are not supported when Internet Explorer is used with Windows 7 - and Windows 8.1, so this class keeps the original status code of 301. - """ - code = 301 - - -def is_less_than_days_ago(date_from_db, number_of_days): - return ( - datetime.utcnow().astimezone(pytz.utc) - parser.parse(date_from_db) - ).days < number_of_days - - -def hide_from_search_engines(f): - @wraps(f) - def decorated_function(*args, **kwargs): - g.hide_from_search_engines = True - response = make_response(f(*args, **kwargs)) - response.headers['X-Robots-Tag'] = 'noindex' - return response - return decorated_function - - -# Function to merge two dict or lists with a JSON-like structure into one. -# JSON-like means they can contain all types JSON can: all the main primitives -# plus nested lists or dictionaries. -# Merge is additive. New values overwrite old and collections are added to. -def merge_jsonlike(source, destination): - def merge_items(source_item, destination_item): - if isinstance(source_item, dict) and isinstance(destination_item, dict): - merge_dicts(source_item, destination_item) - elif isinstance(source_item, list) and isinstance(destination_item, list): - merge_lists(source_item, destination_item) - else: # primitive value - return False - return True - - def merge_lists(source, destination): - last_src_idx = len(source) - 1 - for idx, item in enumerate(destination): - if idx <= last_src_idx: - # assign destination value if can't be merged into source - if merge_items(source[idx], destination[idx]) is False: - source[idx] = destination[idx] - else: - source.append(item) - - def merge_dicts(source, destination): - for key, value in destination.items(): - if key in source: - # assign destination value if can't be merged into source - if merge_items(source[key], value) is False: - source[key] = value - else: - source[key] = value - - merge_items(source, destination) diff --git a/app/utils/__init__.py b/app/utils/__init__.py new file mode 100644 index 000000000..3a2e27298 --- /dev/null +++ b/app/utils/__init__.py @@ -0,0 +1,211 @@ +from datetime import datetime +from functools import wraps +from itertools import chain +from urllib.parse import urlparse + +import pytz +from dateutil import parser +from flask import ( + abort, + current_app, + g, + make_response, + redirect, + request, + session, + url_for, +) +from flask_login import current_user +from notifications_utils.field import Field +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 + +SENDING_STATUSES = ['created', 'pending', 'sending', 'pending-virus-check'] +DELIVERED_STATUSES = ['delivered', 'sent', 'returned-letter'] +FAILURE_STATUSES = ['failed', 'temporary-failure', 'permanent-failure', + 'technical-failure', 'virus-scan-failed', 'validation-failed'] +REQUESTED_STATUSES = SENDING_STATUSES + DELIVERED_STATUSES + FAILURE_STATUSES + +NOTIFICATION_TYPES = ["sms", "email", "letter", "broadcast"] + + +def service_has_permission(permission): + + from app import current_service + + def wrap(func): + @wraps(func) + def wrap_func(*args, **kwargs): + if not current_service or not current_service.has_permission(permission): + abort(403) + return func(*args, **kwargs) + return wrap_func + return wrap + + +def redirect_to_sign_in(f): + @wraps(f) + def wrapped(*args, **kwargs): + if 'user_details' not in session: + return redirect(url_for('main.sign_in')) + else: + return f(*args, **kwargs) + return wrapped + + +def get_page_from_request(): + if 'page' in request.args: + try: + return int(request.args['page']) + except ValueError: + return None + else: + return 1 + + +def generate_previous_dict(view, service_id, page, url_args=None): + return generate_previous_next_dict(view, service_id, page - 1, 'Previous page', url_args or {}) + + +def generate_next_dict(view, service_id, page, url_args=None): + return generate_previous_next_dict(view, service_id, page + 1, 'Next page', url_args or {}) + + +def generate_previous_next_dict(view, service_id, page, title, url_args): + return { + 'url': url_for(view, service_id=service_id, page=page, **url_args), + 'title': title, + 'label': 'page {}'.format(page) + } + + +def get_help_argument(): + return request.args.get('help') if request.args.get('help') in ('1', '2', '3') else None + + +def get_current_financial_year(): + now = utc_string_to_aware_gmt_datetime( + datetime.utcnow() + ) + current_month = int(now.strftime('%-m')) + current_year = int(now.strftime('%Y')) + return current_year if current_month > 3 else current_year - 1 + + +def get_logo_cdn_domain(): + parsed_uri = urlparse(current_app.config['ADMIN_BASE_URL']) + + if parsed_uri.netloc.startswith('localhost'): + return 'static-logos.notify.tools' + + subdomain = parsed_uri.hostname.split('.')[0] + domain = parsed_uri.netloc[len(subdomain + '.'):] + + return "static-logos.{}".format(domain) + + +def parse_filter_args(filter_dict): + if not isinstance(filter_dict, MultiDict): + filter_dict = MultiDict(filter_dict) + + return MultiDict( + ( + key, + (','.join(filter_dict.getlist(key))).split(',') + ) + for key in filter_dict.keys() + if ''.join(filter_dict.getlist(key)) + ) + + +def set_status_filters(filter_args): + status_filters = filter_args.get('status', []) + return list(OrderedSet(chain( + (status_filters or REQUESTED_STATUSES), + DELIVERED_STATUSES if 'delivered' in status_filters else [], + SENDING_STATUSES if 'sending' in status_filters else [], + FAILURE_STATUSES if 'failed' in status_filters else [] + ))) + + +def unicode_truncate(s, length): + encoded = s.encode('utf-8')[:length] + return encoded.decode('utf-8', 'ignore') + + +def should_skip_template_page(template_type): + return ( + current_user.has_permissions('send_messages') + and not current_user.has_permissions('manage_templates', 'manage_api_keys') + and template_type != 'letter' + ) + + +def get_default_sms_sender(sms_senders): + return str(next(( + Field(x['sms_sender'], html='escape') + for x in sms_senders if x['is_default'] + ), "None")) + + +class PermanentRedirect(RequestRedirect): + """ + In Werkzeug 0.15.0 the status code for RequestRedirect changed from 301 to 308. + 308 status codes are not supported when Internet Explorer is used with Windows 7 + and Windows 8.1, so this class keeps the original status code of 301. + """ + code = 301 + + +def is_less_than_days_ago(date_from_db, number_of_days): + return ( + datetime.utcnow().astimezone(pytz.utc) - parser.parse(date_from_db) + ).days < number_of_days + + +def hide_from_search_engines(f): + @wraps(f) + def decorated_function(*args, **kwargs): + g.hide_from_search_engines = True + response = make_response(f(*args, **kwargs)) + response.headers['X-Robots-Tag'] = 'noindex' + return response + return decorated_function + + +# Function to merge two dict or lists with a JSON-like structure into one. +# JSON-like means they can contain all types JSON can: all the main primitives +# plus nested lists or dictionaries. +# Merge is additive. New values overwrite old and collections are added to. +def merge_jsonlike(source, destination): + def merge_items(source_item, destination_item): + if isinstance(source_item, dict) and isinstance(destination_item, dict): + merge_dicts(source_item, destination_item) + elif isinstance(source_item, list) and isinstance(destination_item, list): + merge_lists(source_item, destination_item) + else: # primitive value + return False + return True + + def merge_lists(source, destination): + last_src_idx = len(source) - 1 + for idx, item in enumerate(destination): + if idx <= last_src_idx: + # assign destination value if can't be merged into source + if merge_items(source[idx], destination[idx]) is False: + source[idx] = destination[idx] + else: + source.append(item) + + def merge_dicts(source, destination): + for key, value in destination.items(): + if key in source: + # assign destination value if can't be merged into source + if merge_items(source[key], value) is False: + source[key] = value + else: + source[key] = value + + merge_items(source, destination) 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/email_domains.txt b/app/utils/email_domains.txt similarity index 100% rename from app/email_domains.txt rename to app/utils/email_domains.txt diff --git a/app/utils/letters.py b/app/utils/letters.py new file mode 100644 index 000000000..81adb3222 --- /dev/null +++ b/app/utils/letters.py @@ -0,0 +1,228 @@ +from datetime import datetime, timedelta + +import pytz +from dateutil import parser +from flask import url_for +from notifications_utils.formatters import unescaped_formatted_list +from notifications_utils.letter_timings import letter_can_be_cancelled +from notifications_utils.postal_address import PostalAddress +from notifications_utils.timezones import ( + convert_bst_to_utc, + convert_utc_to_bst, + utc_string_to_aware_gmt_datetime, +) + + +def printing_today_or_tomorrow(created_at): + print_cutoff = convert_bst_to_utc( + convert_utc_to_bst(datetime.utcnow()).replace(hour=17, minute=30) + ).replace(tzinfo=pytz.utc) + created_at = utc_string_to_aware_gmt_datetime(created_at) + + if created_at < print_cutoff: + return 'today' + else: + return 'tomorrow' + + +def get_letter_printing_statement(status, created_at, long_form=True): + created_at_dt = parser.parse(created_at).replace(tzinfo=None) + if letter_can_be_cancelled(status, created_at_dt): + decription = 'Printing starts' if long_form else 'Printing' + return f'{decription} {printing_today_or_tomorrow(created_at)} at 5:30pm' + else: + printed_datetime = utc_string_to_aware_gmt_datetime(created_at) + timedelta(hours=6, minutes=30) + if printed_datetime.date() == datetime.now().date(): + return 'Printed today at 5:30pm' + elif printed_datetime.date() == datetime.now().date() - timedelta(days=1): + return 'Printed yesterday at 5:30pm' + + printed_date = printed_datetime.strftime('%d %B').lstrip('0') + description = 'Printed on' if long_form else 'Printed' + + return f'{description} {printed_date} at 5:30pm' + + +LETTER_VALIDATION_MESSAGES = { + 'letter-not-a4-portrait-oriented': { + 'title': 'Your letter is not A4 portrait size', + 'detail': ( + 'You need to change the size or orientation of {invalid_pages}.
' + 'Files must meet our ' + '' + 'letter specification' + '.' + ), + 'summary': ( + 'Validation failed because {invalid_pages} {invalid_pages_are_or_is} not A4 portrait size.
' + 'Files must meet our ' + '' + 'letter specification' + '.' + ), + }, + 'content-outside-printable-area': { + 'title': 'Your content is outside the printable area', + 'detail': ( + 'You need to edit {invalid_pages}.
' + 'Files must meet our ' + '' + 'letter specification' + '.' + ), + 'summary': ( + 'Validation failed because content is outside the printable area on {invalid_pages}.
' + 'Files must meet our ' + '' + 'letter specification' + '.' + ), + }, + 'letter-too-long': { + 'title': 'Your letter is too long', + 'detail': ( + 'Letters must be 10 pages or less (5 double-sided sheets of paper).
' + 'Your letter is {page_count} pages long.' + ), + 'summary': ( + 'Validation failed because this letter is {page_count} pages long.
' + 'Letters must be 10 pages or less (5 double-sided sheets of paper).' + ), + }, + 'no-encoded-string': { + 'title': 'Sanitise failed - No encoded string' + }, + 'unable-to-read-the-file': { + 'title': 'There’s a problem with your file', + 'detail': ( + 'Notify cannot read this PDF.' + '
Save a new copy of your file and try again.' + ), + 'summary': ( + 'Validation failed because Notify cannot read this PDF.
' + 'Save a new copy of your file and try again.' + ), + }, + 'address-is-empty': { + 'title': 'The address block is empty', + 'detail': ( + 'You need to add a recipient address.
' + 'Files must meet our ' + '' + 'letter specification' + '.' + ), + 'summary': ( + 'Validation failed because the address block is empty.
' + 'Files must meet our ' + '' + 'letter specification' + '.' + ), + }, + 'not-a-real-uk-postcode': { + 'title': 'There’s a problem with the address for this letter', + 'detail': ( + 'The last line of the address must be a real UK postcode.' + ), + 'summary': ( + 'Validation failed because the last line of the address is not a real UK postcode.' + ), + }, + 'cant-send-international-letters': { + 'title': 'There’s a problem with the address for this letter', + 'detail': ( + 'You do not have permission to send letters to other countries.' + ), + 'summary': ( + 'Validation failed because your service cannot send letters to other countries.' + ), + }, + 'not-a-real-uk-postcode-or-country': { + 'title': 'There’s a problem with the address for this letter', + 'detail': ( + 'The last line of the address must be a UK postcode or ' + 'another country.' + ), + 'summary': ( + 'Validation failed because the last line of the address is ' + 'not a UK postcode or another country.' + ), + }, + 'not-enough-address-lines': { + 'title': 'There’s a problem with the address for this letter', + 'detail': ( + f'The address must be at least {PostalAddress.MIN_LINES} ' + f'lines long.' + ), + 'summary': ( + f'Validation failed because the address must be at least ' + f'{PostalAddress.MIN_LINES} lines long.' + ), + }, + 'too-many-address-lines': { + 'title': 'There’s a problem with the address for this letter', + 'detail': ( + f'The address must be no more than {PostalAddress.MAX_LINES} ' + f'lines long.' + ), + 'summary': ( + f'Validation failed because the address must be no more ' + f'than {PostalAddress.MAX_LINES} lines long.' + ), + }, + 'invalid-char-in-address': { + 'title': 'There’s a problem with the address for this letter', + 'detail': ( + "Address lines must not start with any of the following characters: @ ( ) = [ ] ” \\ / , < > ~" + ), + 'summary': ( + "Validation failed because address lines must not start with any of the " + "following characters: @ ( ) = [ ] ” \\ / , < > ~" + ), + }, + 'notify-tag-found-in-content': { + 'title': 'There’s a problem with your letter', + 'detail': ( + 'Your file includes a letter you’ve downloaded from Notify.
' + 'You need to edit {invalid_pages}.' + ), + 'summary': ( + 'Validation failed because your file includes a letter ' + 'you’ve downloaded from Notify on {invalid_pages}.' + ) + }, +} + + +def get_letter_validation_error(validation_message, invalid_pages=None, page_count=None): + if not invalid_pages: + invalid_pages = [] + if validation_message not in LETTER_VALIDATION_MESSAGES: + return {'title': 'Validation failed'} + + invalid_pages_are_or_is = 'is' if len(invalid_pages) == 1 else 'are' + + invalid_pages = unescaped_formatted_list( + invalid_pages, + before_each='', + after_each='', + prefix='page', + prefix_plural='pages' + ) + + return { + 'title': LETTER_VALIDATION_MESSAGES[validation_message]['title'], + 'detail': LETTER_VALIDATION_MESSAGES[validation_message]['detail'].format( + invalid_pages=invalid_pages, + invalid_pages_are_or_is=invalid_pages_are_or_is, + page_count=page_count, + letter_spec_guidance=url_for('.letter_specification') + ), + 'summary': LETTER_VALIDATION_MESSAGES[validation_message]['summary'].format( + invalid_pages=invalid_pages, + invalid_pages_are_or_is=invalid_pages_are_or_is, + page_count=page_count, + letter_spec_guidance=url_for('.letter_specification'), + ), + } 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/app/utils/user.py b/app/utils/user.py new file mode 100644 index 000000000..577abb305 --- /dev/null +++ b/app/utils/user.py @@ -0,0 +1,68 @@ +import os +from functools import wraps + +from flask import abort, current_app +from flask_login import current_user, login_required + +from app.notify_client.organisations_api_client import organisations_client + +user_is_logged_in = login_required + + +with open('{}/email_domains.txt'.format( + os.path.dirname(os.path.realpath(__file__)) +)) as email_domains: + GOVERNMENT_EMAIL_DOMAIN_NAMES = [line.strip() for line in email_domains] + + +def user_has_permissions(*permissions, **permission_kwargs): + def wrap(func): + @wraps(func) + def wrap_func(*args, **kwargs): + if not current_user.is_authenticated: + return current_app.login_manager.unauthorized() + if not current_user.has_permissions(*permissions, **permission_kwargs): + abort(403) + return func(*args, **kwargs) + return wrap_func + return wrap + + +def user_is_gov_user(f): + @wraps(f) + def wrapped(*args, **kwargs): + if not current_user.is_authenticated: + return current_app.login_manager.unauthorized() + if not current_user.is_gov_user: + abort(403) + return f(*args, **kwargs) + return wrapped + + +def user_is_platform_admin(f): + @wraps(f) + def wrapped(*args, **kwargs): + if not current_user.is_authenticated: + return current_app.login_manager.unauthorized() + if not current_user.platform_admin: + abort(403) + return f(*args, **kwargs) + return wrapped + + +def is_gov_user(email_address): + return _email_address_ends_with( + email_address, GOVERNMENT_EMAIL_DOMAIN_NAMES + ) or _email_address_ends_with( + email_address, organisations_client.get_domains() + ) + + +def _email_address_ends_with(email_address, known_domains): + return any( + email_address.lower().endswith(( + "@{}".format(known), + ".{}".format(known), + )) + for known in known_domains + ) diff --git a/tests/app/main/test_errors_for_csv.py b/tests/app/main/test_errors_for_csv.py deleted file mode 100644 index d455c54b1..000000000 --- a/tests/app/main/test_errors_for_csv.py +++ /dev/null @@ -1,113 +0,0 @@ -from collections import namedtuple - -import pytest - -from app.utils import get_errors_for_csv - -MockRecipients = namedtuple( - 'RecipientCSV', - [ - 'rows_with_bad_recipients', - 'rows_with_missing_data', - 'rows_with_message_too_long', - 'rows_with_empty_message' - ] -) - - -@pytest.mark.parametrize( - "rows_with_bad_recipients, rows_with_missing_data, " - "rows_with_message_too_long, rows_with_empty_message, template_type, expected_errors", - [ - ( - [], [], [], [], - 'sms', - [] - ), - ( - {2}, [], [], [], - 'sms', - ['fix 1 phone number'] - ), - ( - {2, 4, 6}, [], [], [], - 'sms', - ['fix 3 phone numbers'] - ), - ( - {1}, [], [], [], - 'email', - ['fix 1 email address'] - ), - ( - {2, 4, 6}, [], [], [], - 'email', - ['fix 3 email addresses'] - ), - ( - {2}, [], [], [], - 'letter', - ['fix 1 address'] - ), - ( - {2, 4}, [], [], [], - 'letter', - ['fix 2 addresses'] - ), - ( - {2}, {3}, [], [], - 'sms', - [ - 'fix 1 phone number', - 'enter missing data in 1 row' - ] - ), - ( - {2, 4, 6, 8}, {3, 6, 9, 12}, [], [], - 'sms', - [ - 'fix 4 phone numbers', - 'enter missing data in 4 rows' - ] - ), - ( - {}, {}, {3}, [], - 'sms', - [ - 'shorten the message in 1 row' - ] - ), - ( - {}, {}, {3, 12}, [], - 'sms', - [ - 'shorten the messages in 2 rows' - ] - ), - ( - {}, {}, {}, {2}, - 'sms', - [ - 'check you have content for the empty message in 1 row' - ] - ), - ( - {}, {}, {}, {2, 4, 8}, - 'sms', - [ - 'check you have content for the empty messages in 3 rows' - ] - ), - ] -) -def test_get_errors_for_csv( - rows_with_bad_recipients, rows_with_missing_data, rows_with_message_too_long, rows_with_empty_message, - template_type, - expected_errors -): - assert get_errors_for_csv( - MockRecipients( - rows_with_bad_recipients, rows_with_missing_data, rows_with_message_too_long, rows_with_empty_message - ), - template_type - ) == expected_errors diff --git a/tests/app/main/test_permissions.py b/tests/app/main/test_permissions.py index 727dcdcd3..e7db91028 100644 --- a/tests/app/main/test_permissions.py +++ b/tests/app/main/test_permissions.py @@ -3,15 +3,12 @@ import inspect import re import pytest -from flask import current_app, request -from werkzeug.exceptions import Forbidden, Unauthorized +from flask import current_app -from app.main.views.index import index from app.models.roles_and_permissions import ( translate_permissions_from_admin_roles_to_db, translate_permissions_from_db_to_admin_roles, ) -from app.utils import user_has_permissions from tests import service_json from tests.conftest import ( ORGANISATION_ID, @@ -21,245 +18,6 @@ from tests.conftest import ( ) -def _test_permissions( - client, - usr, - permissions, - will_succeed, - kwargs=None, -): - request.view_args.update({'service_id': 'foo'}) - if usr: - client.login(usr) - - decorator = user_has_permissions(*permissions, **(kwargs or {})) - decorated_index = decorator(index) - - if will_succeed: - decorated_index() - else: - try: - if ( - decorated_index().location != '/sign-in?next=%2F' or - decorated_index().status_code != 302 - ): - pytest.fail("Failed to throw a forbidden or unauthorised exception") - except (Forbidden, Unauthorized): - pass - - -def test_user_has_permissions_on_endpoint_fail( - client, - mocker, - mock_get_service, -): - user = _user_with_permissions() - mocker.patch('app.user_api_client.get_user', return_value=user) - _test_permissions( - client, - user, - ['send_messages'], - will_succeed=False) - - -def test_user_has_permissions_success( - client, - mocker, -): - user = _user_with_permissions() - mocker.patch('app.user_api_client.get_user', return_value=user) - _test_permissions( - client, - user, - ['manage_service'], - will_succeed=True) - - -def test_user_has_permissions_or( - client, - mocker, -): - user = _user_with_permissions() - mocker.patch('app.user_api_client.get_user', return_value=user) - _test_permissions( - client, - user, - ['send_messages', 'manage_service'], - will_succeed=True) - - -def test_user_has_permissions_multiple( - client, - mocker, -): - user = _user_with_permissions() - mocker.patch('app.user_api_client.get_user', return_value=user) - _test_permissions( - client, - user, - ['manage_templates', 'manage_service'], - will_succeed=True) - - -def test_exact_permissions( - client, - mocker, -): - user = _user_with_permissions() - mocker.patch('app.user_api_client.get_user', return_value=user) - _test_permissions( - client, - user, - ['manage_service', 'manage_templates'], - will_succeed=True) - - -def test_platform_admin_user_can_access_page_that_has_no_permissions( - client, - platform_admin_user, - mocker, -): - mocker.patch('app.user_api_client.get_user', return_value=platform_admin_user) - _test_permissions( - client, - platform_admin_user, - [], - will_succeed=True) - - -def test_platform_admin_user_can_not_access_page( - client, - platform_admin_user, - mocker, - mock_get_service, -): - mocker.patch('app.user_api_client.get_user', return_value=platform_admin_user) - _test_permissions( - client, - platform_admin_user, - [], - will_succeed=False, - kwargs={'restrict_admin_usage': True}) - - -def test_no_user_returns_401_unauth( - client -): - from flask_login import current_user - assert not current_user.is_authenticated - _test_permissions( - client, - None, - [], - will_succeed=False) - - -def test_user_has_permissions_for_organisation( - client, - mocker, -): - user = _user_with_permissions() - user['organisations'] = ['org_1', 'org_2'] - mocker.patch('app.user_api_client.get_user', return_value=user) - client.login(user) - - request.view_args = {'org_id': 'org_2'} - - @user_has_permissions() - def index(): - pass - - index() - - -def test_platform_admin_can_see_orgs_they_dont_have( - client, - platform_admin_user, - mocker, -): - platform_admin_user['organisations'] = [] - mocker.patch('app.user_api_client.get_user', return_value=platform_admin_user) - client.login(platform_admin_user) - - request.view_args = {'org_id': 'org_2'} - - @user_has_permissions() - def index(): - pass - - index() - - -def test_cant_use_decorator_without_view_args( - client, - platform_admin_user, - mocker, -): - mocker.patch('app.user_api_client.get_user', return_value=platform_admin_user) - client.login(platform_admin_user) - - request.view_args = {} - - @user_has_permissions() - def index(): - pass - - with pytest.raises(NotImplementedError): - index() - - -def test_user_doesnt_have_permissions_for_organisation( - client, - mocker, -): - user = _user_with_permissions() - user['organisations'] = ['org_1', 'org_2'] - mocker.patch('app.user_api_client.get_user', return_value=user) - client.login(user) - - request.view_args = {'org_id': 'org_3'} - - @user_has_permissions() - def index(): - pass - - with pytest.raises(Forbidden): - index() - - -def test_user_with_no_permissions_to_service_goes_to_templates( - client, - mocker -): - user = _user_with_permissions() - mocker.patch('app.user_api_client.get_user', return_value=user) - client.login(user) - request.view_args = {'service_id': 'bar'} - - @user_has_permissions() - def index(): - pass - - index() - - -def _user_with_permissions(): - user_data = {'id': 999, - 'name': 'Test User', - 'password': 'somepassword', - 'email_address': 'test@user.gov.uk', - 'mobile_number': '+4412341234', - 'state': 'active', - 'failed_login_count': 0, - 'permissions': {'foo': ['manage_users', 'manage_templates', 'manage_settings']}, - 'platform_admin': False, - 'organisations': ['org_1', 'org_2'], - 'services': ['foo', 'bar'], - 'current_session_id': None, - } - return user_data - - def test_translate_permissions_from_db_to_admin_roles(): db_perms = ['send_texts', 'send_emails', 'send_letters', 'manage_templates', 'some_unknown_permission'] roles = translate_permissions_from_db_to_admin_roles(db_perms) diff --git a/tests/app/main/views/test_add_service.py b/tests/app/main/views/test_add_service.py index ced1d3270..82b943702 100644 --- a/tests/app/main/views/test_add_service.py +++ b/tests/app/main/views/test_add_service.py @@ -3,7 +3,7 @@ from flask import session, url_for from freezegun import freeze_time from notifications_python_client.errors import HTTPError -from app.utils import is_gov_user +from app.utils.user import is_gov_user from tests import organisation_json from tests.conftest import normalize_spaces diff --git a/tests/app/main/views/test_manage_users.py b/tests/app/main/views/test_manage_users.py index 44fbddb0f..513b26780 100644 --- a/tests/app/main/views/test_manage_users.py +++ b/tests/app/main/views/test_manage_users.py @@ -5,7 +5,7 @@ import pytest from flask import url_for import app -from app.utils import is_gov_user +from app.utils.user import is_gov_user from tests.conftest import ( ORGANISATION_ID, ORGANISATION_TWO_ID, diff --git a/tests/app/test_utils.py b/tests/app/test_utils.py index 52a7a4799..dd0b8da74 100644 --- a/tests/app/test_utils.py +++ b/tests/app/test_utils.py @@ -1,97 +1,16 @@ -from collections import OrderedDict -from csv import DictReader -from io import StringIO -from pathlib import Path - import pytest -from bs4 import BeautifulSoup -from flask import url_for 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_letter_printing_statement, - get_letter_validation_error, get_logo_cdn_domain, - get_sample_template, is_less_than_days_ago, merge_jsonlike, - printing_today_or_tomorrow, ) -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', [ @@ -129,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() @@ -354,246 +100,6 @@ def test_format_datetime_relative(time, human_readable_datetime): assert format_datetime_relative(time) == human_readable_datetime -@pytest.mark.parametrize('utc_datetime', [ - '2018-08-01T23:00:00+00:00', - '2018-08-01T16:29:00+00:00', - '2018-11-01T00:00:00+00:00', - '2018-11-01T10:00:00+00:00', - '2018-11-01T17:29:00+00:00', -]) -def test_printing_today_or_tomorrow_returns_today(utc_datetime): - with freeze_time(utc_datetime): - assert printing_today_or_tomorrow(utc_datetime) == 'today' - - -@pytest.mark.parametrize('utc_datetime', [ - '2018-08-01T22:59:00+00:00', - '2018-08-01T16:30:00+00:00', - '2018-11-01T17:30:00+00:00', - '2018-11-01T21:00:00+00:00', - '2018-11-01T23:59:00+00:00', -]) -def test_printing_today_or_tomorrow_returns_tomorrow(utc_datetime): - with freeze_time(utc_datetime): - assert printing_today_or_tomorrow(utc_datetime) == 'tomorrow' - - -@pytest.mark.parametrize('created_at, current_datetime', [ - ('2017-07-07T12:00:00+00:00', '2017-07-07 16:29:00'), # created today, summer - ('2017-07-06T23:30:00+00:00', '2017-07-07 16:29:00'), # created just after midnight, summer - ('2017-12-12T12:00:00+00:00', '2017-12-12 17:29:00'), # created today, winter - ('2017-12-12T21:30:00+00:00', '2017-12-13 17:29:00'), # created after 5:30 yesterday - ('2017-03-25T17:31:00+00:00', '2017-03-26 16:29:00'), # over clock change period on 2017-03-26 -]) -def test_get_letter_printing_statement_when_letter_prints_today(created_at, current_datetime): - with freeze_time(current_datetime): - statement = get_letter_printing_statement('created', created_at) - - assert statement == 'Printing starts today at 5:30pm' - - -@pytest.mark.parametrize('created_at, current_datetime', [ - ('2017-07-07T16:31:00+00:00', '2017-07-07 22:59:00'), # created today, summer - ('2017-12-12T17:31:00+00:00', '2017-12-12 23:59:00'), # created today, winter -]) -def test_get_letter_printing_statement_when_letter_prints_tomorrow(created_at, current_datetime): - with freeze_time(current_datetime): - statement = get_letter_printing_statement('created', created_at) - - assert statement == 'Printing starts tomorrow at 5:30pm' - - -@pytest.mark.parametrize('created_at, print_day', [ - ('2017-07-06T16:29:00+00:00', 'yesterday'), - ('2017-12-01T00:00:00+00:00', 'on 1 December'), - ('2017-03-26T12:00:00+00:00', 'on 26 March'), -]) -@freeze_time('2017-07-07 12:00:00') -def test_get_letter_printing_statement_for_letter_that_has_been_sent(created_at, print_day): - statement = get_letter_printing_statement('delivered', created_at) - - assert statement == 'Printed {} at 5:30pm'.format(print_day) - - -def test_get_letter_validation_error_for_unknown_error(): - assert get_letter_validation_error('Unknown error') == { - 'title': 'Validation failed' - } - - -@pytest.mark.parametrize('error_message, invalid_pages, expected_title, expected_content, expected_summary', [ - ( - 'letter-not-a4-portrait-oriented', - [2], - 'Your letter is not A4 portrait size', - ( - 'You need to change the size or orientation of page 2. ' - 'Files must meet our letter specification.' - ), - ( - 'Validation failed because page 2 is not A4 portrait size.' - 'Files must meet our letter specification.' - ), - ), - ( - 'letter-not-a4-portrait-oriented', - [2, 3, 4], - 'Your letter is not A4 portrait size', - ( - 'You need to change the size or orientation of pages 2, 3 and 4. ' - 'Files must meet our letter specification.' - ), - ( - 'Validation failed because pages 2, 3 and 4 are not A4 portrait size.' - 'Files must meet our letter specification.' - ), - ), - ( - 'content-outside-printable-area', - [2], - 'Your content is outside the printable area', - ( - 'You need to edit page 2.' - 'Files must meet our letter specification.' - ), - ( - 'Validation failed because content is outside the printable area ' - 'on page 2.' - 'Files must meet our letter specification.' - ), - ), - ( - 'letter-too-long', - None, - 'Your letter is too long', - ( - 'Letters must be 10 pages or less (5 double-sided sheets of paper). ' - 'Your letter is 13 pages long.' - ), - ( - 'Validation failed because this letter is 13 pages long.' - 'Letters must be 10 pages or less (5 double-sided sheets of paper).' - ), - ), - ( - 'unable-to-read-the-file', - None, - 'There’s a problem with your file', - ( - 'Notify cannot read this PDF.' - 'Save a new copy of your file and try again.' - ), - ( - 'Validation failed because Notify cannot read this PDF.' - 'Save a new copy of your file and try again.' - ), - ), - ( - 'address-is-empty', - None, - 'The address block is empty', - ( - 'You need to add a recipient address.' - 'Files must meet our letter specification.' - ), - ( - 'Validation failed because the address block is empty.' - 'Files must meet our letter specification.' - ), - ), - ( - 'not-a-real-uk-postcode', - None, - 'There’s a problem with the address for this letter', - ( - 'The last line of the address must be a real UK postcode.' - ), - ( - 'Validation failed because the last line of the address is not a real UK postcode.' - ), - ), - ( - 'cant-send-international-letters', - None, - 'There’s a problem with the address for this letter', - ( - 'You do not have permission to send letters to other countries.' - ), - ( - 'Validation failed because your service cannot send letters to other countries.' - ), - ), - ( - 'not-a-real-uk-postcode-or-country', - None, - 'There’s a problem with the address for this letter', - ( - 'The last line of the address must be a UK postcode or ' - 'another country.' - ), - ( - 'Validation failed because the last line of the address is ' - 'not a UK postcode or another country.' - ), - ), - ( - 'not-enough-address-lines', - None, - 'There’s a problem with the address for this letter', - ( - 'The address must be at least 3 lines long.' - ), - ( - 'Validation failed because the address must be at least 3 lines long.' - ), - ), - ( - 'too-many-address-lines', - None, - 'There’s a problem with the address for this letter', - ( - 'The address must be no more than 7 lines long.' - ), - ( - 'Validation failed because the address must be no more than 7 lines long.' - ), - ), - ( - 'invalid-char-in-address', - None, - 'There’s a problem with the address for this letter', - ( - 'Address lines must not start with any of the following characters: @ ( ) = [ ] ” \\ / , < > ~' - ), - ( - 'Validation failed because address lines must not start with any of the following ' - 'characters: @ ( ) = [ ] ” \\ / , < > ~' - ), - ), -]) -def test_get_letter_validation_error_for_known_errors( - client_request, - error_message, - invalid_pages, - expected_title, - expected_content, - expected_summary, -): - error = get_letter_validation_error(error_message, invalid_pages=invalid_pages, page_count=13) - detail = BeautifulSoup(error['detail'], 'html.parser') - summary = BeautifulSoup(error['summary'], 'html.parser') - - assert error['title'] == expected_title - - assert detail.text == expected_content - if detail.select_one('a'): - assert detail.select_one('a')['href'] == url_for('.letter_specification') - - assert summary.text == expected_summary - if summary.select_one('a'): - assert summary.select_one('a')['href'] == url_for('.letter_specification') - - @pytest.mark.parametrize("date_from_db, expected_result", [ ('2019-11-17T11:35:21.726132Z', True), ('2019-11-16T11:35:21.726132Z', False), @@ -604,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/__init__.py b/tests/app/utils/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/app/utils/test_csv.py b/tests/app/utils/test_csv.py new file mode 100644 index 000000000..68328e18a --- /dev/null +++ b/tests/app/utils/test_csv.py @@ -0,0 +1,361 @@ +from collections import OrderedDict, namedtuple +from csv import DictReader +from io import StringIO +from pathlib import Path + +import pytest + +from app.utils.csv import ( + Spreadsheet, + generate_notifications_csv, + get_errors_for_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 + + +MockRecipients = namedtuple( + 'RecipientCSV', + [ + 'rows_with_bad_recipients', + 'rows_with_missing_data', + 'rows_with_message_too_long', + 'rows_with_empty_message' + ] +) + + +@pytest.mark.parametrize( + "rows_with_bad_recipients, rows_with_missing_data, " + "rows_with_message_too_long, rows_with_empty_message, template_type, expected_errors", + [ + ( + [], [], [], [], + 'sms', + [] + ), + ( + {2}, [], [], [], + 'sms', + ['fix 1 phone number'] + ), + ( + {2, 4, 6}, [], [], [], + 'sms', + ['fix 3 phone numbers'] + ), + ( + {1}, [], [], [], + 'email', + ['fix 1 email address'] + ), + ( + {2, 4, 6}, [], [], [], + 'email', + ['fix 3 email addresses'] + ), + ( + {2}, [], [], [], + 'letter', + ['fix 1 address'] + ), + ( + {2, 4}, [], [], [], + 'letter', + ['fix 2 addresses'] + ), + ( + {2}, {3}, [], [], + 'sms', + [ + 'fix 1 phone number', + 'enter missing data in 1 row' + ] + ), + ( + {2, 4, 6, 8}, {3, 6, 9, 12}, [], [], + 'sms', + [ + 'fix 4 phone numbers', + 'enter missing data in 4 rows' + ] + ), + ( + {}, {}, {3}, [], + 'sms', + [ + 'shorten the message in 1 row' + ] + ), + ( + {}, {}, {3, 12}, [], + 'sms', + [ + 'shorten the messages in 2 rows' + ] + ), + ( + {}, {}, {}, {2}, + 'sms', + [ + 'check you have content for the empty message in 1 row' + ] + ), + ( + {}, {}, {}, {2, 4, 8}, + 'sms', + [ + 'check you have content for the empty messages in 3 rows' + ] + ), + ] +) +def test_get_errors_for_csv( + rows_with_bad_recipients, rows_with_missing_data, rows_with_message_too_long, rows_with_empty_message, + template_type, + expected_errors +): + assert get_errors_for_csv( + MockRecipients( + rows_with_bad_recipients, rows_with_missing_data, rows_with_message_too_long, rows_with_empty_message + ), + template_type + ) == expected_errors diff --git a/tests/app/utils/test_letters.py b/tests/app/utils/test_letters.py new file mode 100644 index 000000000..4f73b2fd3 --- /dev/null +++ b/tests/app/utils/test_letters.py @@ -0,0 +1,250 @@ +import pytest +from bs4 import BeautifulSoup +from flask import url_for +from freezegun import freeze_time + +from app.utils.letters import ( + get_letter_printing_statement, + get_letter_validation_error, + printing_today_or_tomorrow, +) + + +@pytest.mark.parametrize('utc_datetime', [ + '2018-08-01T23:00:00+00:00', + '2018-08-01T16:29:00+00:00', + '2018-11-01T00:00:00+00:00', + '2018-11-01T10:00:00+00:00', + '2018-11-01T17:29:00+00:00', +]) +def test_printing_today_or_tomorrow_returns_today(utc_datetime): + with freeze_time(utc_datetime): + assert printing_today_or_tomorrow(utc_datetime) == 'today' + + +@pytest.mark.parametrize('utc_datetime', [ + '2018-08-01T22:59:00+00:00', + '2018-08-01T16:30:00+00:00', + '2018-11-01T17:30:00+00:00', + '2018-11-01T21:00:00+00:00', + '2018-11-01T23:59:00+00:00', +]) +def test_printing_today_or_tomorrow_returns_tomorrow(utc_datetime): + with freeze_time(utc_datetime): + assert printing_today_or_tomorrow(utc_datetime) == 'tomorrow' + + +@pytest.mark.parametrize('created_at, current_datetime', [ + ('2017-07-07T12:00:00+00:00', '2017-07-07 16:29:00'), # created today, summer + ('2017-07-06T23:30:00+00:00', '2017-07-07 16:29:00'), # created just after midnight, summer + ('2017-12-12T12:00:00+00:00', '2017-12-12 17:29:00'), # created today, winter + ('2017-12-12T21:30:00+00:00', '2017-12-13 17:29:00'), # created after 5:30 yesterday + ('2017-03-25T17:31:00+00:00', '2017-03-26 16:29:00'), # over clock change period on 2017-03-26 +]) +def test_get_letter_printing_statement_when_letter_prints_today(created_at, current_datetime): + with freeze_time(current_datetime): + statement = get_letter_printing_statement('created', created_at) + + assert statement == 'Printing starts today at 5:30pm' + + +@pytest.mark.parametrize('created_at, current_datetime', [ + ('2017-07-07T16:31:00+00:00', '2017-07-07 22:59:00'), # created today, summer + ('2017-12-12T17:31:00+00:00', '2017-12-12 23:59:00'), # created today, winter +]) +def test_get_letter_printing_statement_when_letter_prints_tomorrow(created_at, current_datetime): + with freeze_time(current_datetime): + statement = get_letter_printing_statement('created', created_at) + + assert statement == 'Printing starts tomorrow at 5:30pm' + + +@pytest.mark.parametrize('created_at, print_day', [ + ('2017-07-06T16:29:00+00:00', 'yesterday'), + ('2017-12-01T00:00:00+00:00', 'on 1 December'), + ('2017-03-26T12:00:00+00:00', 'on 26 March'), +]) +@freeze_time('2017-07-07 12:00:00') +def test_get_letter_printing_statement_for_letter_that_has_been_sent(created_at, print_day): + statement = get_letter_printing_statement('delivered', created_at) + + assert statement == 'Printed {} at 5:30pm'.format(print_day) + + +def test_get_letter_validation_error_for_unknown_error(): + assert get_letter_validation_error('Unknown error') == { + 'title': 'Validation failed' + } + + +@pytest.mark.parametrize('error_message, invalid_pages, expected_title, expected_content, expected_summary', [ + ( + 'letter-not-a4-portrait-oriented', + [2], + 'Your letter is not A4 portrait size', + ( + 'You need to change the size or orientation of page 2. ' + 'Files must meet our letter specification.' + ), + ( + 'Validation failed because page 2 is not A4 portrait size.' + 'Files must meet our letter specification.' + ), + ), + ( + 'letter-not-a4-portrait-oriented', + [2, 3, 4], + 'Your letter is not A4 portrait size', + ( + 'You need to change the size or orientation of pages 2, 3 and 4. ' + 'Files must meet our letter specification.' + ), + ( + 'Validation failed because pages 2, 3 and 4 are not A4 portrait size.' + 'Files must meet our letter specification.' + ), + ), + ( + 'content-outside-printable-area', + [2], + 'Your content is outside the printable area', + ( + 'You need to edit page 2.' + 'Files must meet our letter specification.' + ), + ( + 'Validation failed because content is outside the printable area ' + 'on page 2.' + 'Files must meet our letter specification.' + ), + ), + ( + 'letter-too-long', + None, + 'Your letter is too long', + ( + 'Letters must be 10 pages or less (5 double-sided sheets of paper). ' + 'Your letter is 13 pages long.' + ), + ( + 'Validation failed because this letter is 13 pages long.' + 'Letters must be 10 pages or less (5 double-sided sheets of paper).' + ), + ), + ( + 'unable-to-read-the-file', + None, + 'There’s a problem with your file', + ( + 'Notify cannot read this PDF.' + 'Save a new copy of your file and try again.' + ), + ( + 'Validation failed because Notify cannot read this PDF.' + 'Save a new copy of your file and try again.' + ), + ), + ( + 'address-is-empty', + None, + 'The address block is empty', + ( + 'You need to add a recipient address.' + 'Files must meet our letter specification.' + ), + ( + 'Validation failed because the address block is empty.' + 'Files must meet our letter specification.' + ), + ), + ( + 'not-a-real-uk-postcode', + None, + 'There’s a problem with the address for this letter', + ( + 'The last line of the address must be a real UK postcode.' + ), + ( + 'Validation failed because the last line of the address is not a real UK postcode.' + ), + ), + ( + 'cant-send-international-letters', + None, + 'There’s a problem with the address for this letter', + ( + 'You do not have permission to send letters to other countries.' + ), + ( + 'Validation failed because your service cannot send letters to other countries.' + ), + ), + ( + 'not-a-real-uk-postcode-or-country', + None, + 'There’s a problem with the address for this letter', + ( + 'The last line of the address must be a UK postcode or ' + 'another country.' + ), + ( + 'Validation failed because the last line of the address is ' + 'not a UK postcode or another country.' + ), + ), + ( + 'not-enough-address-lines', + None, + 'There’s a problem with the address for this letter', + ( + 'The address must be at least 3 lines long.' + ), + ( + 'Validation failed because the address must be at least 3 lines long.' + ), + ), + ( + 'too-many-address-lines', + None, + 'There’s a problem with the address for this letter', + ( + 'The address must be no more than 7 lines long.' + ), + ( + 'Validation failed because the address must be no more than 7 lines long.' + ), + ), + ( + 'invalid-char-in-address', + None, + 'There’s a problem with the address for this letter', + ( + 'Address lines must not start with any of the following characters: @ ( ) = [ ] ” \\ / , < > ~' + ), + ( + 'Validation failed because address lines must not start with any of the following ' + 'characters: @ ( ) = [ ] ” \\ / , < > ~' + ), + ), +]) +def test_get_letter_validation_error_for_known_errors( + client_request, + error_message, + invalid_pages, + expected_title, + expected_content, + expected_summary, +): + error = get_letter_validation_error(error_message, invalid_pages=invalid_pages, page_count=13) + detail = BeautifulSoup(error['detail'], 'html.parser') + summary = BeautifulSoup(error['summary'], 'html.parser') + + assert error['title'] == expected_title + + assert detail.text == expected_content + if detail.select_one('a'): + assert detail.select_one('a')['href'] == url_for('.letter_specification') + + assert summary.text == expected_summary + if summary.select_one('a'): + assert summary.select_one('a')['href'] == url_for('.letter_specification') 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) diff --git a/tests/app/utils/test_user.py b/tests/app/utils/test_user.py new file mode 100644 index 000000000..f06a50642 --- /dev/null +++ b/tests/app/utils/test_user.py @@ -0,0 +1,245 @@ +import pytest +from flask import request +from werkzeug.exceptions import Forbidden, Unauthorized + +from app.main.views.index import index +from app.utils.user import user_has_permissions + + +def _test_permissions( + client, + usr, + permissions, + will_succeed, + kwargs=None, +): + request.view_args.update({'service_id': 'foo'}) + if usr: + client.login(usr) + + decorator = user_has_permissions(*permissions, **(kwargs or {})) + decorated_index = decorator(index) + + if will_succeed: + decorated_index() + else: + try: + if ( + decorated_index().location != '/sign-in?next=%2F' or + decorated_index().status_code != 302 + ): + pytest.fail("Failed to throw a forbidden or unauthorised exception") + except (Forbidden, Unauthorized): + pass + + +def test_user_has_permissions_on_endpoint_fail( + client, + mocker, + mock_get_service, +): + user = _user_with_permissions() + mocker.patch('app.user_api_client.get_user', return_value=user) + _test_permissions( + client, + user, + ['send_messages'], + will_succeed=False) + + +def test_user_has_permissions_success( + client, + mocker, +): + user = _user_with_permissions() + mocker.patch('app.user_api_client.get_user', return_value=user) + _test_permissions( + client, + user, + ['manage_service'], + will_succeed=True) + + +def test_user_has_permissions_or( + client, + mocker, +): + user = _user_with_permissions() + mocker.patch('app.user_api_client.get_user', return_value=user) + _test_permissions( + client, + user, + ['send_messages', 'manage_service'], + will_succeed=True) + + +def test_user_has_permissions_multiple( + client, + mocker, +): + user = _user_with_permissions() + mocker.patch('app.user_api_client.get_user', return_value=user) + _test_permissions( + client, + user, + ['manage_templates', 'manage_service'], + will_succeed=True) + + +def test_exact_permissions( + client, + mocker, +): + user = _user_with_permissions() + mocker.patch('app.user_api_client.get_user', return_value=user) + _test_permissions( + client, + user, + ['manage_service', 'manage_templates'], + will_succeed=True) + + +def test_platform_admin_user_can_access_page_that_has_no_permissions( + client, + platform_admin_user, + mocker, +): + mocker.patch('app.user_api_client.get_user', return_value=platform_admin_user) + _test_permissions( + client, + platform_admin_user, + [], + will_succeed=True) + + +def test_platform_admin_user_can_not_access_page( + client, + platform_admin_user, + mocker, + mock_get_service, +): + mocker.patch('app.user_api_client.get_user', return_value=platform_admin_user) + _test_permissions( + client, + platform_admin_user, + [], + will_succeed=False, + kwargs={'restrict_admin_usage': True}) + + +def test_no_user_returns_401_unauth( + client +): + from flask_login import current_user + assert not current_user.is_authenticated + _test_permissions( + client, + None, + [], + will_succeed=False) + + +def test_user_has_permissions_for_organisation( + client, + mocker, +): + user = _user_with_permissions() + user['organisations'] = ['org_1', 'org_2'] + mocker.patch('app.user_api_client.get_user', return_value=user) + client.login(user) + + request.view_args = {'org_id': 'org_2'} + + @user_has_permissions() + def index(): + pass + + index() + + +def test_platform_admin_can_see_orgs_they_dont_have( + client, + platform_admin_user, + mocker, +): + platform_admin_user['organisations'] = [] + mocker.patch('app.user_api_client.get_user', return_value=platform_admin_user) + client.login(platform_admin_user) + + request.view_args = {'org_id': 'org_2'} + + @user_has_permissions() + def index(): + pass + + index() + + +def test_cant_use_decorator_without_view_args( + client, + platform_admin_user, + mocker, +): + mocker.patch('app.user_api_client.get_user', return_value=platform_admin_user) + client.login(platform_admin_user) + + request.view_args = {} + + @user_has_permissions() + def index(): + pass + + with pytest.raises(NotImplementedError): + index() + + +def test_user_doesnt_have_permissions_for_organisation( + client, + mocker, +): + user = _user_with_permissions() + user['organisations'] = ['org_1', 'org_2'] + mocker.patch('app.user_api_client.get_user', return_value=user) + client.login(user) + + request.view_args = {'org_id': 'org_3'} + + @user_has_permissions() + def index(): + pass + + with pytest.raises(Forbidden): + index() + + +def test_user_with_no_permissions_to_service_goes_to_templates( + client, + mocker +): + user = _user_with_permissions() + mocker.patch('app.user_api_client.get_user', return_value=user) + client.login(user) + request.view_args = {'service_id': 'bar'} + + @user_has_permissions() + def index(): + pass + + index() + + +def _user_with_permissions(): + user_data = {'id': 999, + 'name': 'Test User', + 'password': 'somepassword', + 'email_address': 'test@user.gov.uk', + 'mobile_number': '+4412341234', + 'state': 'active', + 'failed_login_count': 0, + 'permissions': {'foo': ['manage_users', 'manage_templates', 'manage_settings']}, + 'platform_admin': False, + 'organisations': ['org_1', 'org_2'], + 'services': ['foo', 'bar'], + 'current_session_id': None, + } + return user_data